rt/aio_misc.c

   1 /* Handle general operations.
   2    Copyright (C) 1997, 1998, 1999, 2000 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 <assert.h>
  23 #include <errno.h>
  24 #include <limits.h>
  25 #include <pthread.h>
  26 #include <stdlib.h>
  27 #include <unistd.h>
  28 #include <sys/stat.h>
  29 #include <sys/time.h>
  30
  31 #include "aio_misc.h"
  32
  33 static void add_request_to_runlist (struct requestlist *newrequest);
  34
  35 /* Pool of request list entries.  */
  36 static struct requestlist **pool;
  37
  38 /* Number of total and allocated pool entries.  */
  39 static size_t pool_tab_size;
  40 static size_t pool_size;
  41
  42 /* We implement a two dimensional array but allocate each row separately.
  43    The macro below determines how many entries should be used per row.
  44    It should better be a power of two.  */
  45 #define ENTRIES_PER_ROW 16
  46
  47 /* The row table is incremented in units of this.  */
  48 #define ROW_STEP        8
  49
  50 /* List of available entries.  */
  51 static struct requestlist *freelist;
  52
  53 /* List of request waiting to be processed.  */
  54 static struct requestlist *runlist;
  55
  56 /* Structure list of all currently processed requests.  */
  57 static struct requestlist *requests;
  58
  59 /* Number of threads currently running.  */
  60 static int nthreads;
  61
  62 /* Number of threads waiting for work to arrive. */
  63 static int idle_thread_count;
  64
  65
  66 /* These are the values used to optimize the use of AIO.  The user can
  67    overwrite them by using the `aio_init' function.  */
  68 static struct aioinit optim =
  69 {
  70   20,   /* int aio_threads;     Maximal number of threads.  */
  71   256,  /* int aio_num;         Number of expected simultanious requests. */
  72   0,
  73   0,
  74   0,
  75   0,
  76   1,
  77   0
  78 };
  79
  80
  81 /* Since the list is global we need a mutex protecting it.  */
  82 pthread_mutex_t __aio_requests_mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
  83
  84 /* When you add a request to the list and there are idle threads present,
  85    you signal this condition variable. When a thread finishes work, it waits
  86    on this condition variable for a time before it actually exits. */
  87 pthread_cond_t __aio_new_request_notification = PTHREAD_COND_INITIALIZER;
  88
  89
  90 /* Functions to handle request list pool.  */
  91 static struct requestlist *
  92 get_elem (void)
  93 {
  94   struct requestlist *result;
  95
  96   if (freelist == NULL)
  97     {
  98       struct requestlist *new_row;
  99       size_t new_size;
 100
 101       assert (sizeof (struct aiocb) == sizeof (struct aiocb64));
 102
 103       /* Compute new size.  */
 104       new_size = pool_size ? pool_size + ENTRIES_PER_ROW : optim.aio_num;
 105
 106       if ((new_size / ENTRIES_PER_ROW) >= pool_tab_size)
 107         {
 108           size_t new_tab_size = new_size / ENTRIES_PER_ROW;
 109           struct requestlist **new_tab;
 110
 111           new_tab = (struct requestlist **)
 112             realloc (pool, (new_tab_size * sizeof (struct requestlist *)));
 113
 114           if (new_tab == NULL)
 115             return NULL;
 116
 117           pool_tab_size = new_tab_size;
 118           pool = new_tab;
 119         }
 120
 121       if (pool_size == 0)
 122         {
 123           size_t cnt;
 124
 125           new_row = (struct requestlist *)
 126             calloc (new_size, sizeof (struct requestlist));
 127
 128           if (new_row == NULL)
 129             return NULL;
 130
 131           for (cnt = 0; cnt < new_size / ENTRIES_PER_ROW; ++cnt)
 132             pool[cnt] = &new_row[cnt * ENTRIES_PER_ROW];
 133         }
 134       else
 135         {
 136           /* Allocat one new row.  */
 137           new_row = (struct requestlist *)
 138             calloc (ENTRIES_PER_ROW, sizeof (struct requestlist));
 139           if (new_row == NULL)
 140             return NULL;
 141
 142           pool[new_size / ENTRIES_PER_ROW - 1] = new_row;
 143         }
 144
 145       /* Put all the new entries in the freelist.  */
 146       do
 147         {
 148           new_row->next_prio = freelist;
 149           freelist = new_row++;
 150         }
 151       while (++pool_size < new_size);
 152     }
 153
 154   result = freelist;
 155   freelist = freelist->next_prio;
 156
 157   return result;
 158 }
 159
 160
 161 void
 162 internal_function
 163 __aio_free_request (struct requestlist *elem)
 164 {
 165   elem->running = no;
 166   elem->next_prio = freelist;
 167   freelist = elem;
 168 }
 169
 170
 171 struct requestlist *
 172 internal_function
 173 __aio_find_req (aiocb_union *elem)
 174 {
 175   struct requestlist *runp = requests;
 176   int fildes = elem->aiocb.aio_fildes;
 177
 178   while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
 179     runp = runp->next_fd;
 180
 181   if (runp != NULL)
 182     {
 183       if (runp->aiocbp->aiocb.aio_fildes != fildes)
 184         runp = NULL;
 185       else
 186         while (runp != NULL && runp->aiocbp != elem)
 187           runp = runp->next_prio;
 188     }
 189
 190   return runp;
 191 }
 192
 193
 194 struct requestlist *
 195 internal_function
 196 __aio_find_req_fd (int fildes)
 197 {
 198   struct requestlist *runp = requests;
 199
 200   while (runp != NULL && runp->aiocbp->aiocb.aio_fildes < fildes)
 201     runp = runp->next_fd;
 202
 203   return (runp != NULL && runp->aiocbp->aiocb.aio_fildes == fildes
 204           ? runp : NULL);
 205 }
 206
 207
 208 void
 209 internal_function
 210 __aio_remove_request (struct requestlist *last, struct requestlist *req,
 211                       int all)
 212 {
 213   if (last != NULL)
 214     last->next_prio = req->next_prio;
 215   else
 216     {
 217       if (all || req->next_prio == NULL)
 218         {
 219           if (req->last_fd != NULL)
 220             req->last_fd->next_fd = req->next_fd;
 221           else
 222             requests = req->next_fd;
 223           if (req->next_fd != NULL)
 224             req->next_fd->last_fd = req->last_fd;
 225         }
 226       else
 227         {
 228           if (req->last_fd != NULL)
 229             req->last_fd->next_fd = req->next_prio;
 230           else
 231             requests = req->next_prio;
 232
 233           if (req->next_fd != NULL)
 234             req->next_fd->last_fd = req->next_prio;
 235
 236           req->next_prio->last_fd = req->last_fd;
 237           req->next_prio->next_fd = req->next_fd;
 238
 239           /* Mark this entry as runnable.  */
 240           req->next_prio->running = yes;
 241         }
 242
 243       if (req->running == yes)
 244         {
 245           struct requestlist *runp = runlist;
 246
 247           last = NULL;
 248           while (runp != NULL)
 249             {
 250               if (runp == req)
 251                 {
 252                   if (last == NULL)
 253                     runlist = runp->next_run;
 254                   else
 255                     last->next_run = runp->next_run;
 256                   break;
 257                 }
 258               last = runp;
 259               runp = runp->next_run;
 260             }
 261         }
 262     }
 263 }
 264
 265
 266 /* The thread handler.  */
 267 static void *handle_fildes_io (void *arg);
 268
 269
 270 /* User optimization.  */
 271 void
 272 __aio_init (const struct aioinit *init)
 273 {
 274   /* Get the mutex.  */
 275   pthread_mutex_lock (&__aio_requests_mutex);
 276
 277   /* Only allow writing new values if the table is not yet allocated.  */
 278   if (pool == NULL)
 279     {
 280       optim.aio_threads = init->aio_threads < 1 ? 1 : init->aio_threads;
 281       optim.aio_num = (init->aio_num < ENTRIES_PER_ROW
 282                        ? ENTRIES_PER_ROW
 283                        : init->aio_num & ~ENTRIES_PER_ROW);
 284     }
 285
 286   if (init->aio_idle_time != 0)
 287     optim.aio_idle_time = init->aio_idle_time;
 288
 289   /* Release the mutex.  */
 290   pthread_mutex_unlock (&__aio_requests_mutex);
 291 }
 292 weak_alias (__aio_init, aio_init)
 293
 294
 295 /* The main function of the async I/O handling.  It enqueues requests
 296    and if necessary starts and handles threads.  */
 297 struct requestlist *
 298 internal_function
 299 __aio_enqueue_request (aiocb_union *aiocbp, int operation)
 300 {
 301   int result = 0;
 302   int policy, prio;
 303   struct sched_param param;
 304   struct requestlist *last, *runp, *newp;
 305   int running = no;
 306
 307   if (operation == LIO_SYNC || operation == LIO_DSYNC)
 308     aiocbp->aiocb.aio_reqprio = 0;
 309   else if (aiocbp->aiocb.aio_reqprio < 0
 310            || aiocbp->aiocb.aio_reqprio > AIO_PRIO_DELTA_MAX)
 311     {
 312       /* Invalid priority value.  */
 313       __set_errno (EINVAL);
 314       aiocbp->aiocb.__error_code = EINVAL;
 315       aiocbp->aiocb.__return_value = -1;
 316       return NULL;
 317     }
 318
 319   /* Compute priority for this request.  */
 320   pthread_getschedparam (pthread_self (), &policy, &param);
 321   prio = param.sched_priority - aiocbp->aiocb.aio_reqprio;
 322
 323   /* Get the mutex.  */
 324   pthread_mutex_lock (&__aio_requests_mutex);
 325
 326   last = NULL;
 327   runp = requests;
 328   /* First look whether the current file descriptor is currently
 329      worked with.  */
 330   while (runp != NULL
 331          && runp->aiocbp->aiocb.aio_fildes < aiocbp->aiocb.aio_fildes)
 332     {
 333       last = runp;
 334       runp = runp->next_fd;
 335     }
 336
 337   /* Get a new element for the waiting list.  */
 338   newp = get_elem ();
 339   if (newp == NULL)
 340     {
 341       pthread_mutex_unlock (&__aio_requests_mutex);
 342       __set_errno (EAGAIN);
 343       return NULL;
 344     }
 345   newp->aiocbp = aiocbp;
 346   newp->caller_pid = (aiocbp->aiocb.aio_sigevent.sigev_notify == SIGEV_SIGNAL
 347                       ? getpid () : 0);
 348   newp->waiting = NULL;
 349
 350   aiocbp->aiocb.__abs_prio = prio;
 351   aiocbp->aiocb.__policy = policy;
 352   aiocbp->aiocb.aio_lio_opcode = operation;
 353   aiocbp->aiocb.__error_code = EINPROGRESS;
 354   aiocbp->aiocb.__return_value = 0;
 355
 356   if (runp != NULL
 357       && runp->aiocbp->aiocb.aio_fildes == aiocbp->aiocb.aio_fildes)
 358     {
 359       /* The current file descriptor is worked on.  It makes no sense
 360          to start another thread since this new thread would fight
 361          with the running thread for the resources.  But we also cannot
 362          say that the thread processing this desriptor shall immediately
 363          after finishing the current job process this request if there
 364          are other threads in the running queue which have a higher
 365          priority.  */
 366
 367       /* Simply enqueue it after the running one according to the
 368          priority.  */
 369       while (runp->next_prio != NULL
 370              && runp->next_prio->aiocbp->aiocb.__abs_prio >= prio)
 371         runp = runp->next_prio;
 372
 373       newp->next_prio = runp->next_prio;
 374       runp->next_prio = newp;
 375
 376       running = queued;
 377     }
 378   else
 379     {
 380       running = yes;
 381       /* Enqueue this request for a new descriptor.  */
 382       if (last == NULL)
 383         {
 384           newp->last_fd = NULL;
 385           newp->next_fd = requests;
 386           if (requests != NULL)
 387             requests->last_fd = newp;
 388           requests = newp;
 389         }
 390       else
 391         {
 392           newp->next_fd = last->next_fd;
 393           newp->last_fd = last;
 394           last->next_fd = newp;
 395           if (newp->next_fd != NULL)
 396             newp->next_fd->last_fd = newp;
 397         }
 398
 399       newp->next_prio = NULL;
 400     }
 401
 402   if (running == yes)
 403     {
 404       /* We try to create a new thread for this file descriptor.  The
 405          function which gets called will handle all available requests
 406          for this descriptor and when all are processed it will
 407          terminate.
 408
 409          If no new thread can be created or if the specified limit of
 410          threads for AIO is reached we queue the request.  */
 411
 412       /* See if we need to and are able to create a thread.  */
 413       if (nthreads < optim.aio_threads && idle_thread_count == 0)
 414         {
 415           pthread_t thid;
 416           pthread_attr_t attr;
 417
 418           /* Make sure the thread is created detached.  */
 419           pthread_attr_init (&attr);
 420           pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
 421
 422           /* Now try to start a thread.  */
 423           if (pthread_create (&thid, &attr, handle_fildes_io, newp) == 0)
 424             {
 425               /* We managed to enqueue the request.  All errors which can
 426                  happen now can be recognized by calls to `aio_return' and
 427                  `aio_error'.  */
 428               running = allocated;
 429               ++nthreads;
 430             }
 431           else if (nthreads == 0)
 432             /* We cannot create a thread in the moment and there is
 433                also no thread running.  This is a problem.  `errno' is
 434                set to EAGAIN if this is only a temporary problem.  */
 435             result = -1;
 436         }
 437     }
 438
 439   /* Enqueue the request in the run queue if it is not yet running.  */
 440   if (running == yes && result == 0)
 441     {
 442       add_request_to_runlist (newp);
 443
 444       /* If there is a thread waiting for work, then let it know that we
 445          have just given it something to do. */
 446       if (idle_thread_count > 0)
 447         pthread_cond_signal (&__aio_new_request_notification);
 448     }
 449
 450   if (result == 0)
 451     newp->running = running;
 452   else
 453     {
 454       /* Something went wrong.  */
 455       __aio_free_request (newp);
 456       newp = NULL;
 457     }
 458
 459   /* Release the mutex.  */
 460   pthread_mutex_unlock (&__aio_requests_mutex);
 461
 462   return newp;
 463 }
 464
 465
 466 static void *
 467 handle_fildes_io (void *arg)
 468 {
 469   pthread_t self = pthread_self ();
 470   struct sched_param param;
 471   struct requestlist *runp = (struct requestlist *) arg;
 472   aiocb_union *aiocbp;
 473   int policy;
 474   int fildes;
 475
 476   pthread_getschedparam (self, &policy, &param);
 477
 478   do
 479     {
 480       /* If runp is NULL, then we were created to service the work queue
 481          in general, not to handle any particular request. In that case we
 482          skip the "do work" stuff on the first pass, and go directly to the
 483          "get work off the work queue" part of this loop, which is near the
 484          end. */
 485       if (runp == NULL)
 486         pthread_mutex_lock (&__aio_requests_mutex);
 487       else
 488         {
 489           /* Update our variables.  */
 490           aiocbp = runp->aiocbp;
 491           fildes = aiocbp->aiocb.aio_fildes;
 492
 493           /* Change the priority to the requested value (if necessary).  */
 494           if (aiocbp->aiocb.__abs_prio != param.sched_priority
 495               || aiocbp->aiocb.__policy != policy)
 496             {
 497               param.sched_priority = aiocbp->aiocb.__abs_prio;
 498               policy = aiocbp->aiocb.__policy;
 499               pthread_setschedparam (self, policy, &param);
 500             }
 501
 502           /* Process request pointed to by RUNP.  We must not be disturbed
 503              by signals.  */
 504           if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_READ)
 505             {
 506               if (aiocbp->aiocb.aio_lio_opcode & 128)
 507                 aiocbp->aiocb.__return_value =
 508                   TEMP_FAILURE_RETRY (__pread64 (fildes, (void *)
 509                                                  aiocbp->aiocb64.aio_buf,
 510                                                  aiocbp->aiocb64.aio_nbytes,
 511                                                  aiocbp->aiocb64.aio_offset));
 512               else
 513                 aiocbp->aiocb.__return_value =
 514                   TEMP_FAILURE_RETRY (pread (fildes,
 515                                              (void *) aiocbp->aiocb.aio_buf,
 516                                              aiocbp->aiocb.aio_nbytes,
 517                                              aiocbp->aiocb.aio_offset));
 518
 519               if (aiocbp->aiocb.__return_value == -1 && errno == ESPIPE)
 520                 /* The Linux kernel is different from others.  It returns
 521                    ESPIPE if using pread on a socket.  Other platforms
 522                    simply ignore the offset parameter and behave like
 523                    read.  */
 524                 aiocbp->aiocb.__return_value =
 525                   TEMP_FAILURE_RETRY (read (fildes,
 526                                             (void *) aiocbp->aiocb64.aio_buf,
 527                                             aiocbp->aiocb64.aio_nbytes));
 528             }
 529           else if ((aiocbp->aiocb.aio_lio_opcode & 127) == LIO_WRITE)
 530             {
 531               if (aiocbp->aiocb.aio_lio_opcode & 128)
 532                 aiocbp->aiocb.__return_value =
 533                   TEMP_FAILURE_RETRY (__pwrite64 (fildes, (const void *)
 534                                                   aiocbp->aiocb64.aio_buf,
 535                                                   aiocbp->aiocb64.aio_nbytes,
 536                                                   aiocbp->aiocb64.aio_offset));
 537               else
 538                 aiocbp->aiocb.__return_value =
 539                   TEMP_FAILURE_RETRY (pwrite (fildes, (const void *)
 540                                               aiocbp->aiocb.aio_buf,
 541                                               aiocbp->aiocb.aio_nbytes,
 542                                               aiocbp->aiocb.aio_offset));
 543
 544               if (aiocbp->aiocb.__return_value == -1 && errno == ESPIPE)
 545                 /* The Linux kernel is different from others.  It returns
 546                    ESPIPE if using pwrite on a socket.  Other platforms
 547                    simply ignore the offset parameter and behave like
 548                    write.  */
 549                 aiocbp->aiocb.__return_value =
 550                   TEMP_FAILURE_RETRY (write (fildes,
 551                                              (void *) aiocbp->aiocb64.aio_buf,
 552                                              aiocbp->aiocb64.aio_nbytes));
 553             }
 554           else if (aiocbp->aiocb.aio_lio_opcode == LIO_DSYNC)
 555             aiocbp->aiocb.__return_value =
 556               TEMP_FAILURE_RETRY (fdatasync (fildes));
 557           else if (aiocbp->aiocb.aio_lio_opcode == LIO_SYNC)
 558             aiocbp->aiocb.__return_value =
 559               TEMP_FAILURE_RETRY (fsync (fildes));
 560           else
 561             {
 562               /* This is an invalid opcode.  */
 563               aiocbp->aiocb.__return_value = -1;
 564               __set_errno (EINVAL);
 565             }
 566
 567           /* Get the mutex.  */
 568           pthread_mutex_lock (&__aio_requests_mutex);
 569
 570           /* In theory we would need here a write memory barrier since the
 571              callers test using aio_error() whether the request finished
 572              and once this value != EINPROGRESS the field __return_value
 573              must be committed to memory.
 574
 575              But since the pthread_mutex_lock call involves write memory
 576              barriers as well it is not necessary.  */
 577
 578           if (aiocbp->aiocb.__return_value == -1)
 579             aiocbp->aiocb.__error_code = errno;
 580           else
 581             aiocbp->aiocb.__error_code = 0;
 582
 583           /* Send the signal to notify about finished processing of the
 584              request.  */
 585           __aio_notify (runp);
 586
 587           /* Now dequeue the current request.  */
 588           __aio_remove_request (NULL, runp, 0);
 589           if (runp->next_prio != NULL)
 590             add_request_to_runlist (runp->next_prio);
 591
 592           /* Free the old element.  */
 593           __aio_free_request (runp);
 594         }
 595
 596       runp = runlist;
 597
 598       /* If the runlist is empty, then we sleep for a while, waiting for
 599          something to arrive in it. */
 600       if (runp == NULL && optim.aio_idle_time >= 0)
 601         {
 602           struct timeval now;
 603           struct timespec wakeup_time;
 604
 605           ++idle_thread_count;
 606           gettimeofday (&now, NULL);
 607           wakeup_time.tv_sec = now.tv_sec + optim.aio_idle_time;
 608           wakeup_time.tv_nsec = now.tv_usec * 1000;
 609           if (wakeup_time.tv_nsec > 1000000000)
 610             {
 611               wakeup_time.tv_nsec -= 1000000000;
 612               ++wakeup_time.tv_sec;
 613             }
 614           pthread_cond_timedwait (&__aio_new_request_notification,
 615                                   &__aio_requests_mutex,
 616                                   &wakeup_time);
 617           --idle_thread_count;
 618           runp = runlist;
 619         }
 620
 621       if (runp == NULL)
 622         --nthreads;
 623       else
 624         {
 625           assert (runp->running == yes);
 626           runp->running = allocated;
 627           runlist = runp->next_run;
 628
 629           /* If we have a request to process, and there's still another in
 630              the run list, then we need to either wake up or create a new
 631              thread to service the request that is still in the run list. */
 632           if (runlist != NULL)
 633             {
 634               /* There are at least two items in the work queue to work on.
 635                  If there are other idle threads, then we should wake them
 636                  up for these other work elements; otherwise, we should try
 637                  to create a new thread. */
 638               if (idle_thread_count > 0)
 639                 pthread_cond_signal (&__aio_new_request_notification);
 640               else if (nthreads < optim.aio_threads)
 641                 {
 642                   pthread_t thid;
 643                   pthread_attr_t attr;
 644
 645                   /* Make sure the thread is created detached.  */
 646                   pthread_attr_init (&attr);
 647                   pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
 648
 649                   /* Now try to start a thread. If we fail, no big deal,
 650                      because we know that there is at least one thread (us)
 651                      that is working on AIO operations. */
 652                   if (pthread_create (&thid, &attr, handle_fildes_io, NULL)
 653                       == 0)
 654                     ++nthreads;
 655                 }
 656             }
 657         }
 658
 659       /* Release the mutex.  */
 660       pthread_mutex_unlock (&__aio_requests_mutex);
 661     }
 662   while (runp != NULL);
 663
 664   pthread_exit (NULL);
 665 }
 666
 667
 668 /* Free allocated resources.  */
 669 static void
 670 __attribute__ ((unused))
 671 free_res (void)
 672 {
 673   size_t row;
 674
 675   /* The first block of rows as specified in OPTIM is allocated in
 676      one chunk.  */
 677   free (pool[0]);
 678
 679   for (row = optim.aio_num / ENTRIES_PER_ROW; row < pool_tab_size; ++row)
 680     free (pool[row]);
 681
 682   free (pool);
 683 }
 684 text_set_element (__libc_subfreeres, free_res);
 685
 686
 687 /* Add newrequest to the runlist. The __abs_prio flag of newrequest must
 688    be correctly set to do this. Also, you had better set newrequest's
 689    "running" flag to "yes" before you release your lock or you'll throw an
 690    assertion. */
 691 static void
 692 add_request_to_runlist (struct requestlist *newrequest)
 693 {
 694   int prio = newrequest->aiocbp->aiocb.__abs_prio;
 695   struct requestlist *runp;
 696
 697   if (runlist == NULL || runlist->aiocbp->aiocb.__abs_prio < prio)
 698     {
 699       newrequest->next_run = runlist;
 700       runlist = newrequest;
 701     }
 702   else
 703     {
 704       runp = runlist;
 705
 706       while (runp->next_run != NULL
 707              && runp->next_run->aiocbp->aiocb.__abs_prio >= prio)
 708         runp = runp->next_run;
 709
 710       newrequest->next_run = runp->next_run;
 711       runp->next_run = newrequest;
 712     }
 713 }