rt/aio_misc.c

   1 /* Handle general operations.
   2    Copyright (C) 1997 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 <pthread.h>
  24 #include <stdlib.h>
  25 #include <unistd.h>
  26 #include <sys/stat.h>
  27
  28 #include "aio_misc.h"
  29
  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 __aio_free_req (struct requestlist *elem)
 148 {
 149   elem->running = no;
 150   elem->next_prio = freelist;
 151   freelist = elem;
 152 }
 153
 154
 155 struct requestlist *
 156 __aio_find_req (aiocb_union *elem)
 157 {
 158   struct requestlist *runp = requests;
 159   int fildes = elem->aiocb.aio_fildes;
 160
 161   while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
 162     runp = runp->next_fd;
 163
 164   if (runp != NULL)
 165     if (runp->aiocbp->aiocb.aio_fildes != fildes)
 166       runp = NULL;
 167     else
 168       while (runp != NULL && runp->aiocbp != elem)
 169         runp = runp->next_prio;
 170
 171   return runp;
 172 }
 173
 174
 175 struct requestlist *
 176 __aio_find_req_fd (int fildes)
 177 {
 178   struct requestlist *runp = requests;
 179
 180   while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
 181     runp = runp->next_fd;
 182
 183   return (runp != NULL && runp->aiocbp->aiocb.aio_fildes == fildes
 184           ? runp : NULL);
 185 }
 186
 187
 188 /* The thread handler.  */
 189 static void *handle_fildes_io (void *arg);
 190
 191
 192 /* User optimization.  */
 193 void
 194 __aio_init (const struct aioinit *init)
 195 {
 196   /* Get the mutex.  */
 197   pthread_mutex_lock (&__aio_requests_mutex);
 198
 199   /* Only allow writing new values if the table is not yet allocated.  */
 200   if (pool == NULL)
 201     {
 202       optim.aio_threads = init->aio_threads < 1 ? 1 : init->aio_threads;
 203       optim.aio_num = (init->aio_num < ENTRIES_PER_ROW
 204                        ? ENTRIES_PER_ROW
 205                        : init->aio_num & ~ENTRIES_PER_ROW);
 206     }
 207
 208   /* Release the mutex.  */
 209   pthread_mutex_unlock (&__aio_requests_mutex);
 210 }
 211 weak_alias (__aio_init, aio_init)
 212
 213
 214 /* The main function of the async I/O handling.  It enqueues requests
 215    and if necessary starts and handles threads.  */
 216 struct requestlist *
 217 __aio_enqueue_request (aiocb_union *aiocbp, int operation)
 218 {
 219   int result = 0;
 220   int policy, prio;
 221   struct sched_param param;
 222   struct requestlist *last, *runp, *newp;
 223   int running = no;
 224
 225   if (aiocbp->aiocb.aio_reqprio < 0
 226       || aiocbp->aiocb.aio_reqprio > AIO_PRIO_DELTA_MAX)
 227     {
 228       /* Invalid priority value.  */
 229       __set_errno (EINVAL);
 230       aiocbp->aiocb.__error_code = EINVAL;
 231       aiocbp->aiocb.__return_value = -1;
 232       return NULL;
 233     }
 234
 235   /* Compute priority for this request.  */
 236   pthread_getschedparam (pthread_self (), &policy, &param);
 237   prio = param.sched_priority - aiocbp->aiocb.aio_reqprio;
 238
 239   /* Get the mutex.  */
 240   pthread_mutex_lock (&__aio_requests_mutex);
 241
 242   last = NULL;
 243   runp = requests;
 244   /* First look whether the current file descriptor is currently
 245      worked with.  */
 246   while (runp != NULL
 247          && runp->aiocbp->aiocb.aio_fildes < aiocbp->aiocb.aio_fildes)
 248     {
 249       last = runp;
 250       runp = runp->next_fd;
 251     }
 252
 253   /* Get a new element for the waiting list.  */
 254   newp = get_elem ();
 255   if (newp == NULL)
 256     {
 257       __set_errno (EAGAIN);
 258       pthread_mutex_unlock (&__aio_requests_mutex);
 259       return NULL;
 260     }
 261   newp->aiocbp = aiocbp;
 262   newp->waiting = NULL;
 263
 264   aiocbp->aiocb.__abs_prio = prio;
 265   aiocbp->aiocb.__policy = policy;
 266   aiocbp->aiocb.aio_lio_opcode = operation;
 267   aiocbp->aiocb.__error_code = EINPROGRESS;
 268   aiocbp->aiocb.__return_value = 0;
 269
 270   if (runp != NULL
 271       && runp->aiocbp->aiocb.aio_fildes == aiocbp->aiocb.aio_fildes)
 272     {
 273       /* The current file descriptor is worked on.  It makes no sense
 274          to start another thread since this new thread would fight
 275          with the running thread for the resources.  But we also cannot
 276          say that the thread processing this desriptor shall imeediately
 277          after finishing the current job process this request if there
 278          are other threads in the running queue which have a higher
 279          priority.  */
 280
 281       /* Simply enqueue it after the running one according to the
 282          priority.  */
 283       while (runp->next_prio != NULL
 284              && runp->next_prio->aiocbp->aiocb.__abs_prio >= prio)
 285         runp = runp->next_prio;
 286
 287       newp->next_prio = runp->next_prio;
 288       runp->next_prio = newp;
 289
 290       running = queued;
 291     }
 292   else
 293     {
 294       /* Enqueue this request for a new descriptor.  */
 295       if (last == NULL)
 296         {
 297           newp->last_fd = NULL;
 298           newp->next_fd = requests;
 299           if (requests != NULL)
 300             requests->last_fd = newp;
 301           requests = newp;
 302         }
 303       else
 304         {
 305           newp->next_fd = last->next_fd;
 306           newp->last_fd = last;
 307           last->next_fd = newp;
 308           if (newp->next_fd != NULL)
 309             newp->next_fd->last_fd = newp;
 310         }
 311
 312       newp->next_prio = NULL;
 313     }
 314
 315   if (running == no)
 316     {
 317       /* We try to create a new thread for this file descriptor.  The
 318          function which gets called will handle all available requests
 319          for this descriptor and when all are processed it will
 320          terminate.
 321
 322          If no new thread can be created or if the specified limit of
 323          threads for AIO is reached we queue the request.  */
 324
 325       /* See if we can create a thread.  */
 326       if (nthreads < optim.aio_threads)
 327         {
 328           pthread_t thid;
 329           pthread_attr_t attr;
 330
 331           /* Make sure the thread is created detached.  */
 332           pthread_attr_init (&attr);
 333           pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
 334
 335           /* Now try to start a thread.  */
 336           if (pthread_create (&thid, &attr, handle_fildes_io, newp) == 0)
 337             {
 338               /* We managed to enqueue the request.  All errors which can
 339                  happen now can be recognized by calls to `aio_return' and
 340                  `aio_error'.  */
 341               running = allocated;
 342               ++nthreads;
 343             }
 344           else if (nthreads == 0)
 345             /* We cannot create a thread in the moment and there is
 346                also no thread running.  This is a problem.  `errno' is
 347                set to EAGAIN if this is only a temporary problem.  */
 348             result = -1;
 349         }
 350     }
 351
 352   /* Enqueue the request in the run queue if it is not yet running.  */
 353   if (running < yes && result == 0)
 354     {
 355       if (runlist == NULL || runlist->aiocbp->aiocb.__abs_prio < prio)
 356         {
 357           newp->next_run = runlist;
 358           runlist = newp;
 359         }
 360       else
 361         {
 362           runp = runlist;
 363
 364           while (runp->next_run != NULL
 365                  && runp->next_run->aiocbp->aiocb.__abs_prio >= prio)
 366             runp = runp->next_run;
 367
 368           newp->next_run = runp->next_run;
 369           runp->next_run = newp;
 370         }
 371     }
 372
 373   if (result == 0)
 374     newp->running = running;
 375   else
 376     {
 377       /* Something went wrong.  */
 378       __aio_free_req (newp);
 379       newp = NULL;
 380     }
 381
 382   /* Release the mutex.  */
 383   pthread_mutex_unlock (&__aio_requests_mutex);
 384
 385   return newp;
 386 }
 387
 388
 389 static void *
 390 handle_fildes_io (void *arg)
 391 {
 392   pthread_t self = pthread_self ();
 393   struct sched_param param;
 394   struct requestlist *runp = (struct requestlist *) arg;
 395   aiocb_union *aiocbp = runp->aiocbp;
 396   int policy;
 397   int fildes = runp->aiocbp->aiocb.aio_fildes;
 398
 399   pthread_getschedparam (self, &policy, &param);
 400
 401   do
 402     {
 403       /* Change the priority to the requested value (if necessary).  */
 404       if (aiocbp->aiocb.__abs_prio != param.sched_priority
 405           || aiocbp->aiocb.__policy != policy)
 406         {
 407           param.sched_priority = aiocbp->aiocb.__abs_prio;
 408           policy = aiocbp->aiocb.__policy;
 409           pthread_setschedparam (self, policy, &param);
 410         }
 411
 412       /* Process request pointed to by RUNP.  We must not be disturbed
 413          by signals.  */
 414       if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_READ)
 415         {
 416           if (aiocbp->aiocb.aio_lio_opcode & 128)
 417             aiocbp->aiocb.__return_value =
 418               TEMP_FAILURE_RETRY (__pread64 (fildes,
 419                                              (void *) aiocbp->aiocb64.aio_buf,
 420                                              aiocbp->aiocb64.aio_nbytes,
 421                                              aiocbp->aiocb64.aio_offset));
 422           else
 423             aiocbp->aiocb.__return_value =
 424               TEMP_FAILURE_RETRY (__pread (fildes,
 425                                            (void *) aiocbp->aiocb.aio_buf,
 426                                            aiocbp->aiocb.aio_nbytes,
 427                                            aiocbp->aiocb.aio_offset));
 428         }
 429       else if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_WRITE)
 430         {
 431           if (aiocbp->aiocb.aio_lio_opcode & 128)
 432             aiocbp->aiocb.__return_value =
 433               TEMP_FAILURE_RETRY (__pwrite64 (fildes,
 434                                               (const void *) aiocbp->aiocb64.aio_buf,
 435                                               aiocbp->aiocb64.aio_nbytes,
 436                                               aiocbp->aiocb64.aio_offset));
 437           else
 438             aiocbp->aiocb.__return_value =
 439               TEMP_FAILURE_RETRY (__pwrite (fildes,
 440                                             (const void *) aiocbp->aiocb.aio_buf,
 441                                             aiocbp->aiocb.aio_nbytes,
 442                                             aiocbp->aiocb.aio_offset));
 443         }
 444       else if (aiocbp->aiocb.aio_lio_opcode == LIO_DSYNC)
 445         aiocbp->aiocb.__return_value = TEMP_FAILURE_RETRY (fdatasync (fildes));
 446       else if (aiocbp->aiocb.aio_lio_opcode == LIO_SYNC)
 447         aiocbp->aiocb.__return_value = TEMP_FAILURE_RETRY (fsync (fildes));
 448       else
 449         {
 450           /* This is an invalid opcode.  */
 451           aiocbp->aiocb.__return_value = -1;
 452           __set_errno (EINVAL);
 453         }
 454
 455       /* Get the mutex.  */
 456       pthread_mutex_lock (&__aio_requests_mutex);
 457
 458       if (aiocbp->aiocb.__return_value == -1)
 459         aiocbp->aiocb.__error_code = errno;
 460       else
 461         aiocbp->aiocb.__error_code = 0;
 462
 463       /* Send the signal to notify about finished processing of the
 464          request.  */
 465       __aio_notify (runp);
 466
 467       /* Now dequeue the current request.  */
 468       if (runp->next_prio == NULL)
 469         {
 470           /* No outstanding request for this descriptor.  Process the
 471              runlist if necessary.  */
 472           if (runp->next_fd != NULL)
 473             runp->next_fd->last_fd = runp->last_fd;
 474           if (runp->last_fd != NULL)
 475             runp->last_fd->next_fd = runp->next_fd;
 476         }
 477       else
 478         {
 479           runp->next_prio->last_fd = runp->last_fd;
 480           runp->next_prio->next_fd = runp->next_fd;
 481           runp->next_prio->running = yes;
 482           if (runp->next_fd != NULL)
 483             runp->next_fd->last_fd = runp->next_prio;
 484           if (runp->last_fd != NULL)
 485             runp->last_fd->next_fd = runp->next_prio;
 486         }
 487
 488       /* Free the old element.  */
 489       __aio_free_req (runp);
 490
 491       runp = freelist;
 492       if (runp != NULL)
 493         {
 494           /* We must not run requests which are not marked `running'.  */
 495           if (runp->running == yes)
 496             {
 497               freelist = runp->next_run;
 498               runp->running = allocated;
 499             }
 500           else
 501             {
 502               struct requestlist *old;
 503
 504               do
 505                 {
 506                   old = runp;
 507                   runp = runp->next_run;
 508                 }
 509               while (runp != NULL && runp->running != yes);
 510
 511               if (runp != NULL)
 512                 old->next_run = runp->next_run;
 513             }
 514         }
 515
 516       /* If no request to work on we will stop the thread.  */
 517       if (runp == NULL)
 518         --nthreads;
 519
 520       /* Release the mutex.  */
 521       pthread_mutex_unlock (&__aio_requests_mutex);
 522     }
 523   while (runp != NULL);
 524
 525   pthread_exit (NULL);
 526 }
 527
 528
 529 /* Free allocated resources.  */
 530 static void
 531 __attribute__ ((unused))
 532 free_res (void)
 533 {
 534   size_t row;
 535
 536   /* The first block of rows as specified in OPTIM is allocated in
 537      one chunk.  */
 538   free (pool[0]);
 539
 540   for (row = optim.aio_num / ENTRIES_PER_ROW; row < pool_tab_size; ++row)
 541     free (pool[row]);
 542
 543   free (pool);
 544 }
 545
 546 text_set_element (__libc_subfreeres, free_res);