1 /* Handle general operations.
2 Copyright (C) 1997, 1998 Free Software Foundation, Inc.
3 This file is part of the GNU C Library.
4 Contributed by Ulrich Drepper <drepper@cygnus.com>, 1997.
6 The GNU C Library is free software; you can redistribute it and/or
7 modify it under the terms of the GNU Library General Public License as
8 published by the Free Software Foundation; either version 2 of the
9 License, or (at your option) any later version.
11 The GNU C Library is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 Library General Public License for more details.
16 You should have received a copy of the GNU Library General Public
17 License along with the GNU C Library; see the file COPYING.LIB. If not,
18 write to the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
19 Boston, MA 02111-1307, USA. */
31 /* Pool of request list entries. */
32 static struct requestlist **pool;
34 /* Number of total and allocated pool entries. */
35 static size_t pool_tab_size;
36 static size_t pool_size;
38 /* We implement a two dimensional array but allocate each row separately.
39 The macro below determines how many entries should be used per row.
40 It should better be a power of two. */
41 #define ENTRIES_PER_ROW 16
43 /* The row table is incremented in units of this. */
46 /* List of available entries. */
47 static struct requestlist *freelist;
49 /* List of request waiting to be processed. */
50 static struct requestlist *runlist;
52 /* Structure list of all currently processed requests. */
53 static struct requestlist *requests;
55 /* Number of threads currently running. */
59 /* These are the values used to optimize the use of AIO. The user can
60 overwrite them by using the `aio_init' function. */
61 static struct aioinit optim =
63 20, /* int aio_threads; Maximal number of threads. */
64 256, /* int aio_num; Number of expected simultanious requests. */
73 /* Since the list is global we need a mutex protecting it. */
74 pthread_mutex_t __aio_requests_mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
77 /* Functions to handle request list pool. */
78 static struct requestlist *
81 struct requestlist *result;
85 struct requestlist *new_row;
88 /* Compute new size. */
89 new_size = pool_size ? pool_size + ENTRIES_PER_ROW : optim.aio_num;
91 if ((new_size / ENTRIES_PER_ROW) >= pool_tab_size)
93 size_t new_tab_size = new_size / ENTRIES_PER_ROW;
94 struct requestlist **new_tab;
96 new_tab = (struct requestlist **)
97 realloc (pool, (new_tab_size * sizeof (struct requestlist *)));
102 pool_tab_size = new_tab_size;
110 new_row = (struct requestlist *)
111 calloc (new_size, sizeof (struct requestlist));
116 for (cnt = 0; cnt < new_size / ENTRIES_PER_ROW; ++cnt)
117 pool[cnt] = &new_row[cnt * ENTRIES_PER_ROW];
121 /* Allocat one new row. */
122 new_row = (struct requestlist *)
123 calloc (ENTRIES_PER_ROW, sizeof (struct requestlist));
127 pool[new_size / ENTRIES_PER_ROW] = new_row;
130 /* Put all the new entries in the freelist. */
133 new_row->next_prio = freelist;
134 freelist = new_row++;
136 while (++pool_size < new_size);
140 freelist = freelist->next_prio;
148 __aio_free_request (struct requestlist *elem)
151 elem->next_prio = freelist;
158 __aio_find_req (aiocb_union *elem)
160 struct requestlist *runp = requests;
161 int fildes = elem->aiocb.aio_fildes;
163 while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
164 runp = runp->next_fd;
168 if (runp->aiocbp->aiocb.aio_fildes != fildes)
171 while (runp != NULL && runp->aiocbp != elem)
172 runp = runp->next_prio;
181 __aio_find_req_fd (int fildes)
183 struct requestlist *runp = requests;
185 while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
186 runp = runp->next_fd;
188 return (runp != NULL && runp->aiocbp->aiocb.aio_fildes == fildes
193 /* The thread handler. */
194 static void *handle_fildes_io (void *arg);
197 /* User optimization. */
199 __aio_init (const struct aioinit *init)
202 pthread_mutex_lock (&__aio_requests_mutex);
204 /* Only allow writing new values if the table is not yet allocated. */
207 optim.aio_threads = init->aio_threads < 1 ? 1 : init->aio_threads;
208 optim.aio_num = (init->aio_num < ENTRIES_PER_ROW
210 : init->aio_num & ~ENTRIES_PER_ROW);
213 /* Release the mutex. */
214 pthread_mutex_unlock (&__aio_requests_mutex);
216 weak_alias (__aio_init, aio_init)
219 /* The main function of the async I/O handling. It enqueues requests
220 and if necessary starts and handles threads. */
223 __aio_enqueue_request (aiocb_union *aiocbp, int operation)
227 struct sched_param param;
228 struct requestlist *last, *runp, *newp;
231 if (aiocbp->aiocb.aio_reqprio < 0
232 || aiocbp->aiocb.aio_reqprio > AIO_PRIO_DELTA_MAX)
234 /* Invalid priority value. */
235 __set_errno (EINVAL);
236 aiocbp->aiocb.__error_code = EINVAL;
237 aiocbp->aiocb.__return_value = -1;
241 /* Compute priority for this request. */
242 pthread_getschedparam (pthread_self (), &policy, ¶m);
243 prio = param.sched_priority - aiocbp->aiocb.aio_reqprio;
246 pthread_mutex_lock (&__aio_requests_mutex);
250 /* First look whether the current file descriptor is currently
253 && runp->aiocbp->aiocb.aio_fildes < aiocbp->aiocb.aio_fildes)
256 runp = runp->next_fd;
259 /* Get a new element for the waiting list. */
263 __set_errno (EAGAIN);
264 pthread_mutex_unlock (&__aio_requests_mutex);
267 newp->aiocbp = aiocbp;
268 newp->waiting = NULL;
270 aiocbp->aiocb.__abs_prio = prio;
271 aiocbp->aiocb.__policy = policy;
272 aiocbp->aiocb.aio_lio_opcode = operation;
273 aiocbp->aiocb.__error_code = EINPROGRESS;
274 aiocbp->aiocb.__return_value = 0;
277 && runp->aiocbp->aiocb.aio_fildes == aiocbp->aiocb.aio_fildes)
279 /* The current file descriptor is worked on. It makes no sense
280 to start another thread since this new thread would fight
281 with the running thread for the resources. But we also cannot
282 say that the thread processing this desriptor shall immediately
283 after finishing the current job process this request if there
284 are other threads in the running queue which have a higher
287 /* Simply enqueue it after the running one according to the
289 while (runp->next_prio != NULL
290 && runp->next_prio->aiocbp->aiocb.__abs_prio >= prio)
291 runp = runp->next_prio;
293 newp->next_prio = runp->next_prio;
294 runp->next_prio = newp;
300 /* Enqueue this request for a new descriptor. */
303 newp->last_fd = NULL;
304 newp->next_fd = requests;
305 if (requests != NULL)
306 requests->last_fd = newp;
311 newp->next_fd = last->next_fd;
312 newp->last_fd = last;
313 last->next_fd = newp;
314 if (newp->next_fd != NULL)
315 newp->next_fd->last_fd = newp;
318 newp->next_prio = NULL;
323 /* We try to create a new thread for this file descriptor. The
324 function which gets called will handle all available requests
325 for this descriptor and when all are processed it will
328 If no new thread can be created or if the specified limit of
329 threads for AIO is reached we queue the request. */
331 /* See if we can create a thread. */
332 if (nthreads < optim.aio_threads)
337 /* Make sure the thread is created detached. */
338 pthread_attr_init (&attr);
339 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
341 /* Now try to start a thread. */
342 if (pthread_create (&thid, &attr, handle_fildes_io, newp) == 0)
344 /* We managed to enqueue the request. All errors which can
345 happen now can be recognized by calls to `aio_return' and
350 else if (nthreads == 0)
351 /* We cannot create a thread in the moment and there is
352 also no thread running. This is a problem. `errno' is
353 set to EAGAIN if this is only a temporary problem. */
358 /* Enqueue the request in the run queue if it is not yet running. */
359 if (running < yes && result == 0)
361 if (runlist == NULL || runlist->aiocbp->aiocb.__abs_prio < prio)
363 newp->next_run = runlist;
370 while (runp->next_run != NULL
371 && runp->next_run->aiocbp->aiocb.__abs_prio >= prio)
372 runp = runp->next_run;
374 newp->next_run = runp->next_run;
375 runp->next_run = newp;
380 newp->running = running;
383 /* Something went wrong. */
384 __aio_free_request (newp);
388 /* Release the mutex. */
389 pthread_mutex_unlock (&__aio_requests_mutex);
396 handle_fildes_io (void *arg)
398 pthread_t self = pthread_self ();
399 struct sched_param param;
400 struct requestlist *runp = (struct requestlist *) arg;
405 pthread_getschedparam (self, &policy, ¶m);
409 /* Update our variables. */
410 aiocbp = runp->aiocbp;
411 fildes = aiocbp->aiocb.aio_fildes;
413 /* Change the priority to the requested value (if necessary). */
414 if (aiocbp->aiocb.__abs_prio != param.sched_priority
415 || aiocbp->aiocb.__policy != policy)
417 param.sched_priority = aiocbp->aiocb.__abs_prio;
418 policy = aiocbp->aiocb.__policy;
419 pthread_setschedparam (self, policy, ¶m);
422 /* Process request pointed to by RUNP. We must not be disturbed
424 if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_READ)
426 if (aiocbp->aiocb.aio_lio_opcode & 128)
427 aiocbp->aiocb.__return_value =
428 TEMP_FAILURE_RETRY (__pread64 (fildes,
429 (void *) aiocbp->aiocb64.aio_buf,
430 aiocbp->aiocb64.aio_nbytes,
431 aiocbp->aiocb64.aio_offset));
433 aiocbp->aiocb.__return_value =
434 TEMP_FAILURE_RETRY (pread (fildes,
435 (void *) aiocbp->aiocb.aio_buf,
436 aiocbp->aiocb.aio_nbytes,
437 aiocbp->aiocb.aio_offset));
439 else if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_WRITE)
441 if (aiocbp->aiocb.aio_lio_opcode & 128)
442 aiocbp->aiocb.__return_value =
443 TEMP_FAILURE_RETRY (__pwrite64 (fildes,
444 (const void *) aiocbp->aiocb64.aio_buf,
445 aiocbp->aiocb64.aio_nbytes,
446 aiocbp->aiocb64.aio_offset));
448 aiocbp->aiocb.__return_value =
449 TEMP_FAILURE_RETRY (pwrite (fildes,
450 (const void *) aiocbp->aiocb.aio_buf,
451 aiocbp->aiocb.aio_nbytes,
452 aiocbp->aiocb.aio_offset));
454 else if (aiocbp->aiocb.aio_lio_opcode == LIO_DSYNC)
455 aiocbp->aiocb.__return_value = TEMP_FAILURE_RETRY (fdatasync (fildes));
456 else if (aiocbp->aiocb.aio_lio_opcode == LIO_SYNC)
457 aiocbp->aiocb.__return_value = TEMP_FAILURE_RETRY (fsync (fildes));
460 /* This is an invalid opcode. */
461 aiocbp->aiocb.__return_value = -1;
462 __set_errno (EINVAL);
466 pthread_mutex_lock (&__aio_requests_mutex);
468 if (aiocbp->aiocb.__return_value == -1)
469 aiocbp->aiocb.__error_code = errno;
471 aiocbp->aiocb.__error_code = 0;
473 /* Send the signal to notify about finished processing of the
477 /* Now dequeue the current request. */
478 if (runp->next_prio == NULL)
480 /* No outstanding request for this descriptor. Remove this
481 descriptor from the list. */
482 if (runp->next_fd != NULL)
483 runp->next_fd->last_fd = runp->last_fd;
484 if (runp->last_fd != NULL)
485 runp->last_fd->next_fd = runp->next_fd;
487 requests = runp->next_fd;
491 runp->next_prio->last_fd = runp->last_fd;
492 runp->next_prio->next_fd = runp->next_fd;
493 runp->next_prio->running = yes;
494 if (runp->next_fd != NULL)
495 runp->next_fd->last_fd = runp->next_prio;
496 if (runp->last_fd != NULL)
497 runp->last_fd->next_fd = runp->next_prio;
499 requests = runp->next_prio;
502 /* Free the old element. */
503 __aio_free_request (runp);
508 /* We must not run requests which are not marked `running'. */
509 if (runp->running == yes)
510 runlist = runp->next_run;
513 struct requestlist *old;
518 runp = runp->next_run;
520 while (runp != NULL && runp->running != yes);
523 old->next_run = runp->next_run;
527 /* If no request to work on we will stop the thread. */
531 runp->running = allocated;
533 /* Release the mutex. */
534 pthread_mutex_unlock (&__aio_requests_mutex);
536 while (runp != NULL);
542 /* Free allocated resources. */
544 __attribute__ ((unused))
549 /* The first block of rows as specified in OPTIM is allocated in
553 for (row = optim.aio_num / ENTRIES_PER_ROW; row < pool_tab_size; ++row)
559 text_set_element (__libc_subfreeres, free_res);