Pretty print.
[kopensolaris-gnu/glibc.git] / rt / aio_misc.c
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.
5
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.
10
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.
15
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.  */
20
21 #include <aio.h>
22 #include <errno.h>
23 #include <limits.h>
24 #include <pthread.h>
25 #include <stdlib.h>
26 #include <unistd.h>
27 #include <sys/stat.h>
28
29 #include "aio_misc.h"
30
31 /* Pool of request list entries.  */
32 static struct requestlist **pool;
33
34 /* Number of total and allocated pool entries.  */
35 static size_t pool_tab_size;
36 static size_t pool_size;
37
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
42
43 /* The row table is incremented in units of this.  */
44 #define ROW_STEP        8
45
46 /* List of available entries.  */
47 static struct requestlist *freelist;
48
49 /* List of request waiting to be processed.  */
50 static struct requestlist *runlist;
51
52 /* Structure list of all currently processed requests.  */
53 static struct requestlist *requests;
54
55 /* Number of threads currently running.  */
56 static int nthreads;
57
58
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 =
62 {
63   20,   /* int aio_threads;     Maximal number of threads.  */
64   256,  /* int aio_num;         Number of expected simultanious requests. */
65   0,
66   0,
67   0,
68   0,
69   { 0, }
70 };
71
72
73 /* Since the list is global we need a mutex protecting it.  */
74 pthread_mutex_t __aio_requests_mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
75
76
77 /* Functions to handle request list pool.  */
78 static struct requestlist *
79 get_elem (void)
80 {
81   struct requestlist *result;
82
83   if (freelist == NULL)
84     {
85       struct requestlist *new_row;
86       size_t new_size;
87
88       /* Compute new size.  */
89       new_size = pool_size ? pool_size + ENTRIES_PER_ROW : optim.aio_num;
90
91       if ((new_size / ENTRIES_PER_ROW) >= pool_tab_size)
92         {
93           size_t new_tab_size = new_size / ENTRIES_PER_ROW;
94           struct requestlist **new_tab;
95
96           new_tab = (struct requestlist **)
97             realloc (pool, (new_tab_size * sizeof (struct requestlist *)));
98
99           if (new_tab == NULL)
100             return NULL;
101
102           pool_tab_size = new_tab_size;
103           pool = new_tab;
104         }
105
106       if (pool_size == 0)
107         {
108           size_t cnt;
109
110           new_row = (struct requestlist *)
111             calloc (new_size, sizeof (struct requestlist));
112
113           if (new_row == NULL)
114             return NULL;
115
116           for (cnt = 0; cnt < new_size / ENTRIES_PER_ROW; ++cnt)
117             pool[cnt] = &new_row[cnt * ENTRIES_PER_ROW];
118         }
119       else
120         {
121           /* Allocat one new row.  */
122           new_row = (struct requestlist *)
123             calloc (ENTRIES_PER_ROW, sizeof (struct requestlist));
124           if (new_row == NULL)
125             return NULL;
126
127           pool[new_size / ENTRIES_PER_ROW] = new_row;
128         }
129
130       /* Put all the new entries in the freelist.  */
131       do
132         {
133           new_row->next_prio = freelist;
134           freelist = new_row++;
135         }
136       while (++pool_size < new_size);
137     }
138
139   result = freelist;
140   freelist = freelist->next_prio;
141
142   return result;
143 }
144
145
146 void
147 internal_function
148 __aio_free_request (struct requestlist *elem)
149 {
150   elem->running = no;
151   elem->next_prio = freelist;
152   freelist = elem;
153 }
154
155
156 struct requestlist *
157 internal_function
158 __aio_find_req (aiocb_union *elem)
159 {
160   struct requestlist *runp = requests;
161   int fildes = elem->aiocb.aio_fildes;
162
163   while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
164     runp = runp->next_fd;
165
166   if (runp != NULL)
167     {
168       if (runp->aiocbp->aiocb.aio_fildes != fildes)
169         runp = NULL;
170       else
171         while (runp != NULL && runp->aiocbp != elem)
172           runp = runp->next_prio;
173     }
174
175   return runp;
176 }
177
178
179 struct requestlist *
180 internal_function
181 __aio_find_req_fd (int fildes)
182 {
183   struct requestlist *runp = requests;
184
185   while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
186     runp = runp->next_fd;
187
188   return (runp != NULL && runp->aiocbp->aiocb.aio_fildes == fildes
189           ? runp : NULL);
190 }
191
192
193 /* The thread handler.  */
194 static void *handle_fildes_io (void *arg);
195
196
197 /* User optimization.  */
198 void
199 __aio_init (const struct aioinit *init)
200 {
201   /* Get the mutex.  */
202   pthread_mutex_lock (&__aio_requests_mutex);
203
204   /* Only allow writing new values if the table is not yet allocated.  */
205   if (pool == NULL)
206     {
207       optim.aio_threads = init->aio_threads < 1 ? 1 : init->aio_threads;
208       optim.aio_num = (init->aio_num < ENTRIES_PER_ROW
209                        ? ENTRIES_PER_ROW
210                        : init->aio_num & ~ENTRIES_PER_ROW);
211     }
212
213   /* Release the mutex.  */
214   pthread_mutex_unlock (&__aio_requests_mutex);
215 }
216 weak_alias (__aio_init, aio_init)
217
218
219 /* The main function of the async I/O handling.  It enqueues requests
220    and if necessary starts and handles threads.  */
221 struct requestlist *
222 internal_function
223 __aio_enqueue_request (aiocb_union *aiocbp, int operation)
224 {
225   int result = 0;
226   int policy, prio;
227   struct sched_param param;
228   struct requestlist *last, *runp, *newp;
229   int running = no;
230
231   if (aiocbp->aiocb.aio_reqprio < 0
232       || aiocbp->aiocb.aio_reqprio > AIO_PRIO_DELTA_MAX)
233     {
234       /* Invalid priority value.  */
235       __set_errno (EINVAL);
236       aiocbp->aiocb.__error_code = EINVAL;
237       aiocbp->aiocb.__return_value = -1;
238       return NULL;
239     }
240
241   /* Compute priority for this request.  */
242   pthread_getschedparam (pthread_self (), &policy, &param);
243   prio = param.sched_priority - aiocbp->aiocb.aio_reqprio;
244
245   /* Get the mutex.  */
246   pthread_mutex_lock (&__aio_requests_mutex);
247
248   last = NULL;
249   runp = requests;
250   /* First look whether the current file descriptor is currently
251      worked with.  */
252   while (runp != NULL
253          && runp->aiocbp->aiocb.aio_fildes < aiocbp->aiocb.aio_fildes)
254     {
255       last = runp;
256       runp = runp->next_fd;
257     }
258
259   /* Get a new element for the waiting list.  */
260   newp = get_elem ();
261   if (newp == NULL)
262     {
263       __set_errno (EAGAIN);
264       pthread_mutex_unlock (&__aio_requests_mutex);
265       return NULL;
266     }
267   newp->aiocbp = aiocbp;
268   newp->waiting = NULL;
269
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;
275
276   if (runp != NULL
277       && runp->aiocbp->aiocb.aio_fildes == aiocbp->aiocb.aio_fildes)
278     {
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
285          priority.  */
286
287       /* Simply enqueue it after the running one according to the
288          priority.  */
289       while (runp->next_prio != NULL
290              && runp->next_prio->aiocbp->aiocb.__abs_prio >= prio)
291         runp = runp->next_prio;
292
293       newp->next_prio = runp->next_prio;
294       runp->next_prio = newp;
295
296       running = queued;
297     }
298   else
299     {
300       /* Enqueue this request for a new descriptor.  */
301       if (last == NULL)
302         {
303           newp->last_fd = NULL;
304           newp->next_fd = requests;
305           if (requests != NULL)
306             requests->last_fd = newp;
307           requests = newp;
308         }
309       else
310         {
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;
316         }
317
318       newp->next_prio = NULL;
319     }
320
321   if (running == no)
322     {
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
326          terminate.
327
328          If no new thread can be created or if the specified limit of
329          threads for AIO is reached we queue the request.  */
330
331       /* See if we can create a thread.  */
332       if (nthreads < optim.aio_threads)
333         {
334           pthread_t thid;
335           pthread_attr_t attr;
336
337           /* Make sure the thread is created detached.  */
338           pthread_attr_init (&attr);
339           pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
340
341           /* Now try to start a thread.  */
342           if (pthread_create (&thid, &attr, handle_fildes_io, newp) == 0)
343             {
344               /* We managed to enqueue the request.  All errors which can
345                  happen now can be recognized by calls to `aio_return' and
346                  `aio_error'.  */
347               running = allocated;
348               ++nthreads;
349             }
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.  */
354             result = -1;
355         }
356     }
357
358   /* Enqueue the request in the run queue if it is not yet running.  */
359   if (running < yes && result == 0)
360     {
361       if (runlist == NULL || runlist->aiocbp->aiocb.__abs_prio < prio)
362         {
363           newp->next_run = runlist;
364           runlist = newp;
365         }
366       else
367         {
368           runp = runlist;
369
370           while (runp->next_run != NULL
371                  && runp->next_run->aiocbp->aiocb.__abs_prio >= prio)
372             runp = runp->next_run;
373
374           newp->next_run = runp->next_run;
375           runp->next_run = newp;
376         }
377     }
378
379   if (result == 0)
380     newp->running = running;
381   else
382     {
383       /* Something went wrong.  */
384       __aio_free_request (newp);
385       newp = NULL;
386     }
387
388   /* Release the mutex.  */
389   pthread_mutex_unlock (&__aio_requests_mutex);
390
391   return newp;
392 }
393
394
395 static void *
396 handle_fildes_io (void *arg)
397 {
398   pthread_t self = pthread_self ();
399   struct sched_param param;
400   struct requestlist *runp = (struct requestlist *) arg;
401   aiocb_union *aiocbp;
402   int policy;
403   int fildes;
404
405   pthread_getschedparam (self, &policy, &param);
406
407   do
408     {
409       /* Update our variables.  */
410       aiocbp = runp->aiocbp;
411       fildes = aiocbp->aiocb.aio_fildes;
412
413       /* Change the priority to the requested value (if necessary).  */
414       if (aiocbp->aiocb.__abs_prio != param.sched_priority
415           || aiocbp->aiocb.__policy != policy)
416         {
417           param.sched_priority = aiocbp->aiocb.__abs_prio;
418           policy = aiocbp->aiocb.__policy;
419           pthread_setschedparam (self, policy, &param);
420         }
421
422       /* Process request pointed to by RUNP.  We must not be disturbed
423          by signals.  */
424       if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_READ)
425         {
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));
432           else
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));
438         }
439       else if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_WRITE)
440         {
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));
447           else
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));
453         }
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));
458       else
459         {
460           /* This is an invalid opcode.  */
461           aiocbp->aiocb.__return_value = -1;
462           __set_errno (EINVAL);
463         }
464
465       /* Get the mutex.  */
466       pthread_mutex_lock (&__aio_requests_mutex);
467
468       if (aiocbp->aiocb.__return_value == -1)
469         aiocbp->aiocb.__error_code = errno;
470       else
471         aiocbp->aiocb.__error_code = 0;
472
473       /* Send the signal to notify about finished processing of the
474          request.  */
475       __aio_notify (runp);
476
477       /* Now dequeue the current request.  */
478       if (runp->next_prio == NULL)
479         {
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;
486           else
487             requests = runp->next_fd;
488         }
489       else
490         {
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;
498           else
499             requests = runp->next_prio;
500         }
501
502       /* Free the old element.  */
503       __aio_free_request (runp);
504
505       runp = runlist;
506       if (runp != NULL)
507         {
508           /* We must not run requests which are not marked `running'.  */
509           if (runp->running == yes)
510             runlist = runp->next_run;
511           else
512             {
513               struct requestlist *old;
514
515               do
516                 {
517                   old = runp;
518                   runp = runp->next_run;
519                 }
520               while (runp != NULL && runp->running != yes);
521
522               if (runp != NULL)
523                 old->next_run = runp->next_run;
524             }
525         }
526
527       /* If no request to work on we will stop the thread.  */
528       if (runp == NULL)
529         --nthreads;
530       else
531         runp->running = allocated;
532
533       /* Release the mutex.  */
534       pthread_mutex_unlock (&__aio_requests_mutex);
535     }
536   while (runp != NULL);
537
538   pthread_exit (NULL);
539 }
540
541
542 /* Free allocated resources.  */
543 static void
544 __attribute__ ((unused))
545 free_res (void)
546 {
547   size_t row;
548
549   /* The first block of rows as specified in OPTIM is allocated in
550      one chunk.  */
551   free (pool[0]);
552
553   for (row = optim.aio_num / ENTRIES_PER_ROW; row < pool_tab_size; ++row)
554     free (pool[row]);
555
556   free (pool);
557 }
558
559 text_set_element (__libc_subfreeres, free_res);