1 /* Handle general operations.
2 Copyright (C) 1997-2001, 2003, 2004, 2006, 2007, 2009, 2011
3 Free Software Foundation, Inc.
4 This file is part of the GNU C Library.
5 Contributed by Ulrich Drepper <drepper@cygnus.com>, 1997.
7 The GNU C Library is free software; you can redistribute it and/or
8 modify it under the terms of the GNU Lesser General Public
9 License as published by the Free Software Foundation; either
10 version 2.1 of the License, or (at your option) any later version.
12 The GNU C Library is distributed in the hope that it will be useful,
13 but WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 Lesser General Public License for more details.
17 You should have received a copy of the GNU Lesser General Public
18 License along with the GNU C Library; if not, write to the Free
19 Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
29 #include <sys/param.h>
34 #ifndef aio_create_helper_thread
35 # define aio_create_helper_thread __aio_create_helper_thread
38 __aio_create_helper_thread (pthread_t
*threadp
, void *(*tf
) (void *), void *arg
)
42 /* Make sure the thread is created detached. */
43 pthread_attr_init (&attr
);
44 pthread_attr_setdetachstate (&attr
, PTHREAD_CREATE_DETACHED
);
46 int ret
= pthread_create (threadp
, &attr
, tf
, arg
);
48 (void) pthread_attr_destroy (&attr
);
53 static void add_request_to_runlist (struct requestlist
*newrequest
);
55 /* Pool of request list entries. */
56 static struct requestlist
**pool
;
58 /* Number of total and allocated pool entries. */
59 static size_t pool_max_size
;
60 static size_t pool_size
;
62 /* We implement a two dimensional array but allocate each row separately.
63 The macro below determines how many entries should be used per row.
64 It should better be a power of two. */
65 #define ENTRIES_PER_ROW 32
67 /* How many rows we allocate at once. */
70 /* List of available entries. */
71 static struct requestlist
*freelist
;
73 /* List of request waiting to be processed. */
74 static struct requestlist
*runlist
;
76 /* Structure list of all currently processed requests. */
77 static struct requestlist
*requests
;
79 /* Number of threads currently running. */
82 /* Number of threads waiting for work to arrive. */
83 static int idle_thread_count
;
86 /* These are the values used to optimize the use of AIO. The user can
87 overwrite them by using the `aio_init' function. */
88 static struct aioinit optim
=
90 20, /* int aio_threads; Maximal number of threads. */
91 64, /* int aio_num; Number of expected simultaneous requests. */
101 /* Since the list is global we need a mutex protecting it. */
102 pthread_mutex_t __aio_requests_mutex
= PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP
;
104 /* When you add a request to the list and there are idle threads present,
105 you signal this condition variable. When a thread finishes work, it waits
106 on this condition variable for a time before it actually exits. */
107 pthread_cond_t __aio_new_request_notification
= PTHREAD_COND_INITIALIZER
;
110 /* Functions to handle request list pool. */
111 static struct requestlist
*
114 struct requestlist
*result
;
116 if (freelist
== NULL
)
118 struct requestlist
*new_row
;
121 assert (sizeof (struct aiocb
) == sizeof (struct aiocb64
));
123 if (pool_size
+ 1 >= pool_max_size
)
125 size_t new_max_size
= pool_max_size
+ ROWS_STEP
;
126 struct requestlist
**new_tab
;
128 new_tab
= (struct requestlist
**)
129 realloc (pool
, new_max_size
* sizeof (struct requestlist
*));
134 pool_max_size
= new_max_size
;
138 /* Allocate the new row. */
139 cnt
= pool_size
== 0 ? optim
.aio_num
: ENTRIES_PER_ROW
;
140 new_row
= (struct requestlist
*) calloc (cnt
,
141 sizeof (struct requestlist
));
145 pool
[pool_size
++] = new_row
;
147 /* Put all the new entries in the freelist. */
150 new_row
->next_prio
= freelist
;
151 freelist
= new_row
++;
157 freelist
= freelist
->next_prio
;
165 __aio_free_request (struct requestlist
*elem
)
168 elem
->next_prio
= freelist
;
175 __aio_find_req (aiocb_union
*elem
)
177 struct requestlist
*runp
= requests
;
178 int fildes
= elem
->aiocb
.aio_fildes
;
180 while (runp
!= NULL
&& runp
->aiocbp
->aiocb
.aio_fildes
< fildes
)
181 runp
= runp
->next_fd
;
185 if (runp
->aiocbp
->aiocb
.aio_fildes
!= fildes
)
188 while (runp
!= NULL
&& runp
->aiocbp
!= elem
)
189 runp
= runp
->next_prio
;
198 __aio_find_req_fd (int fildes
)
200 struct requestlist
*runp
= requests
;
202 while (runp
!= NULL
&& runp
->aiocbp
->aiocb
.aio_fildes
< fildes
)
203 runp
= runp
->next_fd
;
205 return (runp
!= NULL
&& runp
->aiocbp
->aiocb
.aio_fildes
== fildes
212 __aio_remove_request (struct requestlist
*last
, struct requestlist
*req
,
215 assert (req
->running
== yes
|| req
->running
== queued
216 || req
->running
== done
);
219 last
->next_prio
= all
? NULL
: req
->next_prio
;
222 if (all
|| req
->next_prio
== NULL
)
224 if (req
->last_fd
!= NULL
)
225 req
->last_fd
->next_fd
= req
->next_fd
;
227 requests
= req
->next_fd
;
228 if (req
->next_fd
!= NULL
)
229 req
->next_fd
->last_fd
= req
->last_fd
;
233 if (req
->last_fd
!= NULL
)
234 req
->last_fd
->next_fd
= req
->next_prio
;
236 requests
= req
->next_prio
;
238 if (req
->next_fd
!= NULL
)
239 req
->next_fd
->last_fd
= req
->next_prio
;
241 req
->next_prio
->last_fd
= req
->last_fd
;
242 req
->next_prio
->next_fd
= req
->next_fd
;
244 /* Mark this entry as runnable. */
245 req
->next_prio
->running
= yes
;
248 if (req
->running
== yes
)
250 struct requestlist
*runp
= runlist
;
258 runlist
= runp
->next_run
;
260 last
->next_run
= runp
->next_run
;
264 runp
= runp
->next_run
;
271 /* The thread handler. */
272 static void *handle_fildes_io (void *arg
);
275 /* User optimization. */
277 __aio_init (const struct aioinit
*init
)
280 pthread_mutex_lock (&__aio_requests_mutex
);
282 /* Only allow writing new values if the table is not yet allocated. */
285 optim
.aio_threads
= init
->aio_threads
< 1 ? 1 : init
->aio_threads
;
286 assert (powerof2 (ENTRIES_PER_ROW
));
287 optim
.aio_num
= (init
->aio_num
< ENTRIES_PER_ROW
289 : init
->aio_num
& ~(ENTRIES_PER_ROW
- 1));
292 if (init
->aio_idle_time
!= 0)
293 optim
.aio_idle_time
= init
->aio_idle_time
;
295 /* Release the mutex. */
296 pthread_mutex_unlock (&__aio_requests_mutex
);
298 weak_alias (__aio_init
, aio_init
)
301 /* The main function of the async I/O handling. It enqueues requests
302 and if necessary starts and handles threads. */
305 __aio_enqueue_request (aiocb_union
*aiocbp
, int operation
)
309 struct sched_param param
;
310 struct requestlist
*last
, *runp
, *newp
;
313 if (operation
== LIO_SYNC
|| operation
== LIO_DSYNC
)
314 aiocbp
->aiocb
.aio_reqprio
= 0;
315 else if (aiocbp
->aiocb
.aio_reqprio
< 0
316 || aiocbp
->aiocb
.aio_reqprio
> AIO_PRIO_DELTA_MAX
)
318 /* Invalid priority value. */
319 __set_errno (EINVAL
);
320 aiocbp
->aiocb
.__error_code
= EINVAL
;
321 aiocbp
->aiocb
.__return_value
= -1;
325 /* Compute priority for this request. */
326 pthread_getschedparam (pthread_self (), &policy
, ¶m
);
327 prio
= param
.sched_priority
- aiocbp
->aiocb
.aio_reqprio
;
330 pthread_mutex_lock (&__aio_requests_mutex
);
334 /* First look whether the current file descriptor is currently
337 && runp
->aiocbp
->aiocb
.aio_fildes
< aiocbp
->aiocb
.aio_fildes
)
340 runp
= runp
->next_fd
;
343 /* Get a new element for the waiting list. */
347 pthread_mutex_unlock (&__aio_requests_mutex
);
348 __set_errno (EAGAIN
);
351 newp
->aiocbp
= aiocbp
;
352 #ifdef BROKEN_THREAD_SIGNALS
353 newp
->caller_pid
= (aiocbp
->aiocb
.aio_sigevent
.sigev_notify
== SIGEV_SIGNAL
356 newp
->waiting
= NULL
;
358 aiocbp
->aiocb
.__abs_prio
= prio
;
359 aiocbp
->aiocb
.__policy
= policy
;
360 aiocbp
->aiocb
.aio_lio_opcode
= operation
;
361 aiocbp
->aiocb
.__error_code
= EINPROGRESS
;
362 aiocbp
->aiocb
.__return_value
= 0;
365 && runp
->aiocbp
->aiocb
.aio_fildes
== aiocbp
->aiocb
.aio_fildes
)
367 /* The current file descriptor is worked on. It makes no sense
368 to start another thread since this new thread would fight
369 with the running thread for the resources. But we also cannot
370 say that the thread processing this desriptor shall immediately
371 after finishing the current job process this request if there
372 are other threads in the running queue which have a higher
375 /* Simply enqueue it after the running one according to the
378 while (runp
->next_prio
!= NULL
379 && runp
->next_prio
->aiocbp
->aiocb
.__abs_prio
>= prio
)
382 runp
= runp
->next_prio
;
385 newp
->next_prio
= runp
->next_prio
;
386 runp
->next_prio
= newp
;
393 /* Enqueue this request for a new descriptor. */
396 newp
->last_fd
= NULL
;
397 newp
->next_fd
= requests
;
398 if (requests
!= NULL
)
399 requests
->last_fd
= newp
;
404 newp
->next_fd
= last
->next_fd
;
405 newp
->last_fd
= last
;
406 last
->next_fd
= newp
;
407 if (newp
->next_fd
!= NULL
)
408 newp
->next_fd
->last_fd
= newp
;
411 newp
->next_prio
= NULL
;
417 /* We try to create a new thread for this file descriptor. The
418 function which gets called will handle all available requests
419 for this descriptor and when all are processed it will
422 If no new thread can be created or if the specified limit of
423 threads for AIO is reached we queue the request. */
425 /* See if we need to and are able to create a thread. */
426 if (nthreads
< optim
.aio_threads
&& idle_thread_count
== 0)
430 running
= newp
->running
= allocated
;
432 /* Now try to start a thread. */
433 result
= aio_create_helper_thread (&thid
, handle_fildes_io
, newp
);
435 /* We managed to enqueue the request. All errors which can
436 happen now can be recognized by calls to `aio_return' and
441 /* Reset the running flag. The new request is not running. */
442 running
= newp
->running
= yes
;
446 /* We cannot create a thread in the moment and there is
447 also no thread running. This is a problem. `errno' is
448 set to EAGAIN if this is only a temporary problem. */
449 __aio_remove_request (last
, newp
, 0);
457 /* Enqueue the request in the run queue if it is not yet running. */
458 if (running
== yes
&& result
== 0)
460 add_request_to_runlist (newp
);
462 /* If there is a thread waiting for work, then let it know that we
463 have just given it something to do. */
464 if (idle_thread_count
> 0)
465 pthread_cond_signal (&__aio_new_request_notification
);
469 newp
->running
= running
;
472 /* Something went wrong. */
473 __aio_free_request (newp
);
474 aiocbp
->aiocb
.__error_code
= result
;
475 __set_errno (result
);
479 /* Release the mutex. */
480 pthread_mutex_unlock (&__aio_requests_mutex
);
487 handle_fildes_io (void *arg
)
489 pthread_t self
= pthread_self ();
490 struct sched_param param
;
491 struct requestlist
*runp
= (struct requestlist
*) arg
;
496 pthread_getschedparam (self
, &policy
, ¶m
);
500 /* If runp is NULL, then we were created to service the work queue
501 in general, not to handle any particular request. In that case we
502 skip the "do work" stuff on the first pass, and go directly to the
503 "get work off the work queue" part of this loop, which is near the
506 pthread_mutex_lock (&__aio_requests_mutex
);
509 /* Hopefully this request is marked as running. */
510 assert (runp
->running
== allocated
);
512 /* Update our variables. */
513 aiocbp
= runp
->aiocbp
;
514 fildes
= aiocbp
->aiocb
.aio_fildes
;
516 /* Change the priority to the requested value (if necessary). */
517 if (aiocbp
->aiocb
.__abs_prio
!= param
.sched_priority
518 || aiocbp
->aiocb
.__policy
!= policy
)
520 param
.sched_priority
= aiocbp
->aiocb
.__abs_prio
;
521 policy
= aiocbp
->aiocb
.__policy
;
522 pthread_setschedparam (self
, policy
, ¶m
);
525 /* Process request pointed to by RUNP. We must not be disturbed
527 if ((aiocbp
->aiocb
.aio_lio_opcode
& 127) == LIO_READ
)
529 if (sizeof (off_t
) != sizeof (off64_t
)
530 && aiocbp
->aiocb
.aio_lio_opcode
& 128)
531 aiocbp
->aiocb
.__return_value
=
532 TEMP_FAILURE_RETRY (__pread64 (fildes
, (void *)
533 aiocbp
->aiocb64
.aio_buf
,
534 aiocbp
->aiocb64
.aio_nbytes
,
535 aiocbp
->aiocb64
.aio_offset
));
537 aiocbp
->aiocb
.__return_value
=
538 TEMP_FAILURE_RETRY (pread (fildes
,
539 (void *) aiocbp
->aiocb
.aio_buf
,
540 aiocbp
->aiocb
.aio_nbytes
,
541 aiocbp
->aiocb
.aio_offset
));
543 if (aiocbp
->aiocb
.__return_value
== -1 && errno
== ESPIPE
)
544 /* The Linux kernel is different from others. It returns
545 ESPIPE if using pread on a socket. Other platforms
546 simply ignore the offset parameter and behave like
548 aiocbp
->aiocb
.__return_value
=
549 TEMP_FAILURE_RETRY (read (fildes
,
550 (void *) aiocbp
->aiocb64
.aio_buf
,
551 aiocbp
->aiocb64
.aio_nbytes
));
553 else if ((aiocbp
->aiocb
.aio_lio_opcode
& 127) == LIO_WRITE
)
555 if (sizeof (off_t
) != sizeof (off64_t
)
556 && aiocbp
->aiocb
.aio_lio_opcode
& 128)
557 aiocbp
->aiocb
.__return_value
=
558 TEMP_FAILURE_RETRY (__pwrite64 (fildes
, (const void *)
559 aiocbp
->aiocb64
.aio_buf
,
560 aiocbp
->aiocb64
.aio_nbytes
,
561 aiocbp
->aiocb64
.aio_offset
));
563 aiocbp
->aiocb
.__return_value
=
564 TEMP_FAILURE_RETRY (__libc_pwrite (fildes
, (const void *)
565 aiocbp
->aiocb
.aio_buf
,
566 aiocbp
->aiocb
.aio_nbytes
,
567 aiocbp
->aiocb
.aio_offset
));
569 if (aiocbp
->aiocb
.__return_value
== -1 && errno
== ESPIPE
)
570 /* The Linux kernel is different from others. It returns
571 ESPIPE if using pwrite on a socket. Other platforms
572 simply ignore the offset parameter and behave like
574 aiocbp
->aiocb
.__return_value
=
575 TEMP_FAILURE_RETRY (write (fildes
,
576 (void *) aiocbp
->aiocb64
.aio_buf
,
577 aiocbp
->aiocb64
.aio_nbytes
));
579 else if (aiocbp
->aiocb
.aio_lio_opcode
== LIO_DSYNC
)
580 aiocbp
->aiocb
.__return_value
=
581 TEMP_FAILURE_RETRY (fdatasync (fildes
));
582 else if (aiocbp
->aiocb
.aio_lio_opcode
== LIO_SYNC
)
583 aiocbp
->aiocb
.__return_value
=
584 TEMP_FAILURE_RETRY (fsync (fildes
));
587 /* This is an invalid opcode. */
588 aiocbp
->aiocb
.__return_value
= -1;
589 __set_errno (EINVAL
);
593 pthread_mutex_lock (&__aio_requests_mutex
);
595 /* In theory we would need here a write memory barrier since the
596 callers test using aio_error() whether the request finished
597 and once this value != EINPROGRESS the field __return_value
598 must be committed to memory.
600 But since the pthread_mutex_lock call involves write memory
601 barriers as well it is not necessary. */
603 if (aiocbp
->aiocb
.__return_value
== -1)
604 aiocbp
->aiocb
.__error_code
= errno
;
606 aiocbp
->aiocb
.__error_code
= 0;
608 /* Send the signal to notify about finished processing of the
612 /* For debugging purposes we reset the running flag of the
614 assert (runp
->running
== allocated
);
615 runp
->running
= done
;
617 /* Now dequeue the current request. */
618 __aio_remove_request (NULL
, runp
, 0);
619 if (runp
->next_prio
!= NULL
)
620 add_request_to_runlist (runp
->next_prio
);
622 /* Free the old element. */
623 __aio_free_request (runp
);
628 /* If the runlist is empty, then we sleep for a while, waiting for
629 something to arrive in it. */
630 if (runp
== NULL
&& optim
.aio_idle_time
>= 0)
633 struct timespec wakeup_time
;
636 gettimeofday (&now
, NULL
);
637 wakeup_time
.tv_sec
= now
.tv_sec
+ optim
.aio_idle_time
;
638 wakeup_time
.tv_nsec
= now
.tv_usec
* 1000;
639 if (wakeup_time
.tv_nsec
> 1000000000)
641 wakeup_time
.tv_nsec
-= 1000000000;
642 ++wakeup_time
.tv_sec
;
644 pthread_cond_timedwait (&__aio_new_request_notification
,
645 &__aio_requests_mutex
,
655 assert (runp
->running
== yes
);
656 runp
->running
= allocated
;
657 runlist
= runp
->next_run
;
659 /* If we have a request to process, and there's still another in
660 the run list, then we need to either wake up or create a new
661 thread to service the request that is still in the run list. */
664 /* There are at least two items in the work queue to work on.
665 If there are other idle threads, then we should wake them
666 up for these other work elements; otherwise, we should try
667 to create a new thread. */
668 if (idle_thread_count
> 0)
669 pthread_cond_signal (&__aio_new_request_notification
);
670 else if (nthreads
< optim
.aio_threads
)
675 /* Make sure the thread is created detached. */
676 pthread_attr_init (&attr
);
677 pthread_attr_setdetachstate (&attr
, PTHREAD_CREATE_DETACHED
);
679 /* Now try to start a thread. If we fail, no big deal,
680 because we know that there is at least one thread (us)
681 that is working on AIO operations. */
682 if (pthread_create (&thid
, &attr
, handle_fildes_io
, NULL
)
689 /* Release the mutex. */
690 pthread_mutex_unlock (&__aio_requests_mutex
);
692 while (runp
!= NULL
);
698 /* Free allocated resources. */
699 libc_freeres_fn (free_res
)
703 for (row
= 0; row
< pool_max_size
; ++row
)
710 /* Add newrequest to the runlist. The __abs_prio flag of newrequest must
711 be correctly set to do this. Also, you had better set newrequest's
712 "running" flag to "yes" before you release your lock or you'll throw an
715 add_request_to_runlist (struct requestlist
*newrequest
)
717 int prio
= newrequest
->aiocbp
->aiocb
.__abs_prio
;
718 struct requestlist
*runp
;
720 if (runlist
== NULL
|| runlist
->aiocbp
->aiocb
.__abs_prio
< prio
)
722 newrequest
->next_run
= runlist
;
723 runlist
= newrequest
;
729 while (runp
->next_run
!= NULL
730 && runp
->next_run
->aiocbp
->aiocb
.__abs_prio
>= prio
)
731 runp
= runp
->next_run
;
733 newrequest
->next_run
= runp
->next_run
;
734 runp
->next_run
= newrequest
;