deps/libeio/eio.c

   1 /*
   2  * libeio implementation
   3  *
   4  * Copyright (c) 2007,2008,2009,2010 Marc Alexander Lehmann <libeio@schmorp.de>
   5  * All rights reserved.
   6  *
   7  * Redistribution and use in source and binary forms, with or without modifica-
   8  * tion, are permitted provided that the following conditions are met:
   9  *
  10  *   1.  Redistributions of source code must retain the above copyright notice,
  11  *       this list of conditions and the following disclaimer.
  12  *
  13  *   2.  Redistributions in binary form must reproduce the above copyright
  14  *       notice, this list of conditions and the following disclaimer in the
  15  *       documentation and/or other materials provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
  18  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MER-
  19  * CHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO
  20  * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPE-
  21  * CIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  22  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  23  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  24  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTH-
  25  * ERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
  26  * OF THE POSSIBILITY OF SUCH DAMAGE.
  27  *
  28  * Alternatively, the contents of this file may be used under the terms of
  29  * the GNU General Public License ("GPL") version 2 or any later version,
  30  * in which case the provisions of the GPL are applicable instead of
  31  * the above. If you wish to allow the use of your version of this file
  32  * only under the terms of the GPL and not to allow others to use your
  33  * version of this file under the BSD license, indicate your decision
  34  * by deleting the provisions above and replace them with the notice
  35  * and other provisions required by the GPL. If you do not delete the
  36  * provisions above, a recipient may use your version of this file under
  37  * either the BSD or the GPL.
  38  */
  39
  40 #include "eio.h"
  41
  42 #ifdef EIO_STACKSIZE
  43 # define XTHREAD_STACKSIZE EIO_STACKSIZE
  44 #endif
  45
  46 // For statically-linked pthreads-w32, use:
  47 // #ifdef _WIN32
  48 // # define PTW32_STATIC_LIB 1
  49 // #endif
  50 #include "xthread.h"
  51
  52 #include <errno.h>
  53 #include <stddef.h>
  54 #include <stdlib.h>
  55 #include <string.h>
  56 #include <errno.h>
  57 #include <sys/types.h>
  58 #include <sys/stat.h>
  59 #include <limits.h>
  60 #include <fcntl.h>
  61 #include <assert.h>
  62
  63 #ifndef _WIN32
  64 #include <sys/statvfs.h>
  65 #endif
  66
  67 #ifndef EIO_FINISH
  68 # define EIO_FINISH(req)  ((req)->finish) && !EIO_CANCELLED (req) ? (req)->finish (req) : 0
  69 #endif
  70
  71 #ifndef EIO_DESTROY
  72 # define EIO_DESTROY(req) do { if ((req)->destroy) (req)->destroy (req); } while (0)
  73 #endif
  74
  75 #ifndef EIO_FEED
  76 # define EIO_FEED(req)    do { if ((req)->feed   ) (req)->feed    (req); } while (0)
  77 #endif
  78
  79 #ifdef _WIN32
  80
  81 # include <errno.h>
  82 # include <sys/time.h>
  83 # include <unistd.h>
  84 # include <utime.h>
  85 # include <signal.h>
  86 # include <dirent.h>
  87 # include <windows.h>
  88
  89 # define ENOTSOCK WSAENOTSOCK
  90 # define EOPNOTSUPP WSAEOPNOTSUPP
  91 # define ECANCELED 140
  92
  93 # ifndef EIO_STRUCT_DIRENT
  94 #  define EIO_STRUCT_DIRENT struct dirent
  95 # endif
  96
  97 #else
  98
  99 # include "config.h"
 100 # include <sys/time.h>
 101 # include <sys/select.h>
 102 # include <unistd.h>
 103 # include <utime.h>
 104 # include <signal.h>
 105 # include <dirent.h>
 106
 107 #if _POSIX_MEMLOCK || _POSIX_MEMLOCK_RANGE || _POSIX_MAPPED_FILES
 108 # include <sys/mman.h>
 109 #endif
 110
 111 /* POSIX_SOURCE is useless on bsd's, and XOPEN_SOURCE is unreliable there, too */
 112 # if __FreeBSD__ || defined __NetBSD__ || defined __OpenBSD__
 113 #  define _DIRENT_HAVE_D_TYPE /* sigh */
 114 #  define D_INO(de) (de)->d_fileno
 115 #  define D_NAMLEN(de) (de)->d_namlen
 116 # elif __linux || defined d_ino || _XOPEN_SOURCE >= 600
 117 #  define D_INO(de) (de)->d_ino
 118 # endif
 119
 120 #ifdef _D_EXACT_NAMLEN
 121 # undef D_NAMLEN
 122 # define D_NAMLEN(de) _D_EXACT_NAMLEN (de)
 123 #endif
 124
 125 # ifdef _DIRENT_HAVE_D_TYPE
 126 #  define D_TYPE(de) (de)->d_type
 127 # endif
 128
 129 # ifndef EIO_STRUCT_DIRENT
 130 #  define EIO_STRUCT_DIRENT struct dirent
 131 # endif
 132
 133 #endif
 134
 135 #if HAVE_SENDFILE
 136 # if __linux
 137 #  include <sys/sendfile.h>
 138 # elif __FreeBSD__ || defined __APPLE__
 139 #  include <sys/socket.h>
 140 #  include <sys/uio.h>
 141 # elif __hpux
 142 #  include <sys/socket.h>
 143 # elif __solaris
 144 #  include <sys/sendfile.h>
 145 # else
 146 #  error sendfile support requested but not available
 147 # endif
 148 #endif
 149
 150 #ifndef D_TYPE
 151 # define D_TYPE(de) 0
 152 #endif
 153 #ifndef D_INO
 154 # define D_INO(de) 0
 155 #endif
 156 #ifndef D_NAMLEN
 157 # define D_NAMLEN(de) strlen ((de)->d_name)
 158 #endif
 159
 160 /* number of seconds after which an idle threads exit */
 161 #define IDLE_TIMEOUT 10
 162
 163 /* used for struct dirent, AIX doesn't provide it */
 164 #ifndef NAME_MAX
 165 # define NAME_MAX 4096
 166 #endif
 167
 168 /* used for readlink etc. */
 169 #ifndef PATH_MAX
 170 # define PATH_MAX 4096
 171 #endif
 172
 173 /* buffer size for various temporary buffers */
 174 #define EIO_BUFSIZE 65536
 175
 176 #define dBUF                                    \
 177   char *eio_buf;                                \
 178   ETP_WORKER_LOCK (self);                       \
 179   self->dbuf = eio_buf = malloc (EIO_BUFSIZE);  \
 180   ETP_WORKER_UNLOCK (self);                     \
 181   errno = ENOMEM;                               \
 182   if (!eio_buf)                                 \
 183     return -1;
 184
 185 #define EIO_TICKS ((1000000 + 1023) >> 10)
 186
 187 /*****************************************************************************/
 188
 189 #if __GNUC__ >= 3
 190 # define expect(expr,value) __builtin_expect ((expr),(value))
 191 #else
 192 # define expect(expr,value) (expr)
 193 #endif
 194
 195 #define expect_false(expr) expect ((expr) != 0, 0)
 196 #define expect_true(expr)  expect ((expr) != 0, 1)
 197
 198 /*****************************************************************************/
 199
 200 #define ETP_PRI_MIN EIO_PRI_MIN
 201 #define ETP_PRI_MAX EIO_PRI_MAX
 202
 203 struct etp_worker;
 204
 205 #define ETP_REQ eio_req
 206 #define ETP_DESTROY(req) eio_destroy (req)
 207 static int eio_finish (eio_req *req);
 208 #define ETP_FINISH(req)  eio_finish (req)
 209 static void eio_execute (struct etp_worker *self, eio_req *req);
 210 #define ETP_EXECUTE(wrk,req) eio_execute (wrk,req)
 211
 212 #define ETP_WORKER_CLEAR(req)   \
 213   if (wrk->dbuf)                \
 214     {                           \
 215       free (wrk->dbuf);         \
 216       wrk->dbuf = 0;            \
 217     }                           \
 218                                 \
 219   if (wrk->dirp)                \
 220     {                           \
 221       closedir (wrk->dirp);     \
 222       wrk->dirp = 0;            \
 223     }
 224
 225 #define ETP_WORKER_COMMON \
 226   void *dbuf;   \
 227   DIR *dirp;
 228
 229 /*****************************************************************************/
 230
 231 #define ETP_NUM_PRI (ETP_PRI_MAX - ETP_PRI_MIN + 1)
 232
 233 /* calculate time difference in ~1/EIO_TICKS of a second */
 234 static int tvdiff (struct timeval *tv1, struct timeval *tv2)
 235 {
 236   return  (tv2->tv_sec  - tv1->tv_sec ) * EIO_TICKS
 237        + ((tv2->tv_usec - tv1->tv_usec) >> 10);
 238 }
 239
 240 static unsigned int started, idle, wanted = 4;
 241
 242 static void (*want_poll_cb) (void);
 243 static void (*done_poll_cb) (void);
 244
 245 static unsigned int max_poll_time;     /* reslock */
 246 static unsigned int max_poll_reqs;     /* reslock */
 247
 248 static volatile unsigned int nreqs;    /* reqlock */
 249 static volatile unsigned int nready;   /* reqlock */
 250 static volatile unsigned int npending; /* reqlock */
 251 static volatile unsigned int max_idle = 4;
 252
 253 static xmutex_t wrklock = X_MUTEX_INIT;
 254 static xmutex_t reslock = X_MUTEX_INIT;
 255 static xmutex_t reqlock = X_MUTEX_INIT;
 256 static xcond_t  reqwait = X_COND_INIT;
 257
 258 #if defined (__APPLE__)
 259 static xmutex_t apple_bug_writelock = X_MUTEX_INIT;
 260 #endif
 261
 262 #if !HAVE_PREADWRITE
 263 /*
 264  * make our pread/pwrite emulation safe against themselves, but not against
 265  * normal read/write by using a mutex. slows down execution a lot,
 266  * but that's your problem, not mine.
 267  */
 268 static xmutex_t preadwritelock = X_MUTEX_INIT;
 269 #endif
 270
 271 typedef struct etp_worker
 272 {
 273   /* locked by wrklock */
 274   struct etp_worker *prev, *next;
 275
 276   xthread_t tid;
 277
 278   /* locked by reslock, reqlock or wrklock */
 279   ETP_REQ *req; /* currently processed request */
 280
 281   ETP_WORKER_COMMON
 282 } etp_worker;
 283
 284 static etp_worker wrk_first = { &wrk_first, &wrk_first, 0 }; /* NOT etp */
 285
 286 #define ETP_WORKER_LOCK(wrk)   X_LOCK   (wrklock)
 287 #define ETP_WORKER_UNLOCK(wrk) X_UNLOCK (wrklock)
 288
 289 /* worker threads management */
 290
 291 static void etp_worker_clear (etp_worker *wrk)
 292 {
 293   ETP_WORKER_CLEAR (wrk);
 294 }
 295
 296 static void etp_worker_free (etp_worker *wrk)
 297 {
 298   wrk->next->prev = wrk->prev;
 299   wrk->prev->next = wrk->next;
 300
 301   free (wrk);
 302 }
 303
 304 static unsigned int etp_nreqs (void)
 305 {
 306   int retval;
 307   if (WORDACCESS_UNSAFE) X_LOCK   (reqlock);
 308   retval = nreqs;
 309   if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
 310   return retval;
 311 }
 312
 313 static unsigned int etp_nready (void)
 314 {
 315   unsigned int retval;
 316
 317   if (WORDACCESS_UNSAFE) X_LOCK   (reqlock);
 318   retval = nready;
 319   if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
 320
 321   return retval;
 322 }
 323
 324 static unsigned int etp_npending (void)
 325 {
 326   unsigned int retval;
 327
 328   if (WORDACCESS_UNSAFE) X_LOCK   (reqlock);
 329   retval = npending;
 330   if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
 331
 332   return retval;
 333 }
 334
 335 static unsigned int etp_nthreads (void)
 336 {
 337   unsigned int retval;
 338
 339   if (WORDACCESS_UNSAFE) X_LOCK   (reqlock);
 340   retval = started;
 341   if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
 342
 343   return retval;
 344 }
 345
 346 /*
 347  * a somewhat faster data structure might be nice, but
 348  * with 8 priorities this actually needs <20 insns
 349  * per shift, the most expensive operation.
 350  */
 351 typedef struct {
 352   ETP_REQ *qs[ETP_NUM_PRI], *qe[ETP_NUM_PRI]; /* qstart, qend */
 353   int size;
 354 } etp_reqq;
 355
 356 static etp_reqq req_queue;
 357 static etp_reqq res_queue;
 358
 359 static int reqq_push (etp_reqq *q, ETP_REQ *req)
 360 {
 361   int pri = req->pri;
 362   req->next = 0;
 363
 364   if (q->qe[pri])
 365     {
 366       q->qe[pri]->next = req;
 367       q->qe[pri] = req;
 368     }
 369   else
 370     q->qe[pri] = q->qs[pri] = req;
 371
 372   return q->size++;
 373 }
 374
 375 static ETP_REQ *reqq_shift (etp_reqq *q)
 376 {
 377   int pri;
 378
 379   if (!q->size)
 380     return 0;
 381
 382   --q->size;
 383
 384   for (pri = ETP_NUM_PRI; pri--; )
 385     {
 386       eio_req *req = q->qs[pri];
 387
 388       if (req)
 389         {
 390           if (!(q->qs[pri] = (eio_req *)req->next))
 391             q->qe[pri] = 0;
 392
 393           return req;
 394         }
 395     }
 396
 397   abort ();
 398 }
 399
 400 static void etp_atfork_prepare (void)
 401 {
 402   X_LOCK (wrklock);
 403   X_LOCK (reqlock);
 404   X_LOCK (reslock);
 405 #if !HAVE_PREADWRITE
 406   X_LOCK (preadwritelock);
 407 #endif
 408 }
 409
 410 static void etp_atfork_parent (void)
 411 {
 412 #if !HAVE_PREADWRITE
 413   X_UNLOCK (preadwritelock);
 414 #endif
 415   X_UNLOCK (reslock);
 416   X_UNLOCK (reqlock);
 417   X_UNLOCK (wrklock);
 418 }
 419
 420 static void etp_atfork_child (void)
 421 {
 422   ETP_REQ *prv;
 423
 424   while ((prv = reqq_shift (&req_queue)))
 425     ETP_DESTROY (prv);
 426
 427   while ((prv = reqq_shift (&res_queue)))
 428     ETP_DESTROY (prv);
 429
 430   while (wrk_first.next != &wrk_first)
 431     {
 432       etp_worker *wrk = wrk_first.next;
 433
 434       if (wrk->req)
 435         ETP_DESTROY (wrk->req);
 436
 437       etp_worker_clear (wrk);
 438       etp_worker_free (wrk);
 439     }
 440
 441   started  = 0;
 442   idle     = 0;
 443   nreqs    = 0;
 444   nready   = 0;
 445   npending = 0;
 446
 447   etp_atfork_parent ();
 448 }
 449
 450 static void
 451 etp_once_init (void)
 452 {
 453   X_THREAD_ATFORK (etp_atfork_prepare, etp_atfork_parent, etp_atfork_child);
 454 }
 455
 456 static int
 457 etp_init (void (*want_poll)(void), void (*done_poll)(void))
 458 {
 459   static pthread_once_t doinit = PTHREAD_ONCE_INIT;
 460
 461   pthread_once (&doinit, etp_once_init);
 462
 463   want_poll_cb = want_poll;
 464   done_poll_cb = done_poll;
 465
 466   return 0;
 467 }
 468
 469 X_THREAD_PROC (etp_proc);
 470
 471 static void etp_start_thread (void)
 472 {
 473   etp_worker *wrk = calloc (1, sizeof (etp_worker));
 474
 475   /*TODO*/
 476   assert (("unable to allocate worker thread data", wrk));
 477
 478   X_LOCK (wrklock);
 479
 480   if (thread_create (&wrk->tid, etp_proc, (void *)wrk))
 481     {
 482       wrk->prev = &wrk_first;
 483       wrk->next = wrk_first.next;
 484       wrk_first.next->prev = wrk;
 485       wrk_first.next = wrk;
 486       ++started;
 487     }
 488   else
 489     free (wrk);
 490
 491   X_UNLOCK (wrklock);
 492 }
 493
 494 static void etp_maybe_start_thread (void)
 495 {
 496   if (expect_true (etp_nthreads () >= wanted))
 497     return;
 498
 499   /* todo: maybe use idle here, but might be less exact */
 500   if (expect_true (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()))
 501     return;
 502
 503   etp_start_thread ();
 504 }
 505
 506 static void etp_end_thread (void)
 507 {
 508   eio_req *req = calloc (1, sizeof (eio_req));
 509
 510   req->type = -1;
 511   req->pri  = ETP_PRI_MAX - ETP_PRI_MIN;
 512
 513   X_LOCK (reqlock);
 514   reqq_push (&req_queue, req);
 515   X_COND_SIGNAL (reqwait);
 516   X_UNLOCK (reqlock);
 517
 518   X_LOCK (wrklock);
 519   --started;
 520   X_UNLOCK (wrklock);
 521 }
 522
 523 static int etp_poll (void)
 524 {
 525   unsigned int maxreqs;
 526   unsigned int maxtime;
 527   struct timeval tv_start, tv_now;
 528
 529   X_LOCK (reslock);
 530   maxreqs = max_poll_reqs;
 531   maxtime = max_poll_time;
 532   X_UNLOCK (reslock);
 533
 534   if (maxtime)
 535     gettimeofday (&tv_start, 0);
 536
 537   for (;;)
 538     {
 539       ETP_REQ *req;
 540
 541       etp_maybe_start_thread ();
 542
 543       X_LOCK (reslock);
 544       req = reqq_shift (&res_queue);
 545
 546       if (req)
 547         {
 548           --npending;
 549
 550           if (!res_queue.size && done_poll_cb)
 551             done_poll_cb ();
 552         }
 553
 554       X_UNLOCK (reslock);
 555
 556       if (!req)
 557         return 0;
 558
 559       X_LOCK (reqlock);
 560       --nreqs;
 561       X_UNLOCK (reqlock);
 562
 563       if (expect_false (req->type == EIO_GROUP && req->size))
 564         {
 565           req->int1 = 1; /* mark request as delayed */
 566           continue;
 567         }
 568       else
 569         {
 570           int res = ETP_FINISH (req);
 571           if (expect_false (res))
 572             return res;
 573         }
 574
 575       if (expect_false (maxreqs && !--maxreqs))
 576         break;
 577
 578       if (maxtime)
 579         {
 580           gettimeofday (&tv_now, 0);
 581
 582           if (tvdiff (&tv_start, &tv_now) >= maxtime)
 583             break;
 584         }
 585     }
 586
 587   errno = EAGAIN;
 588   return -1;
 589 }
 590
 591 static void etp_cancel (ETP_REQ *req)
 592 {
 593   X_LOCK   (wrklock);
 594   req->flags |= EIO_FLAG_CANCELLED;
 595   X_UNLOCK (wrklock);
 596
 597   eio_grp_cancel (req);
 598 }
 599
 600 static void etp_submit (ETP_REQ *req)
 601 {
 602   req->pri -= ETP_PRI_MIN;
 603
 604   if (expect_false (req->pri < ETP_PRI_MIN - ETP_PRI_MIN)) req->pri = ETP_PRI_MIN - ETP_PRI_MIN;
 605   if (expect_false (req->pri > ETP_PRI_MAX - ETP_PRI_MIN)) req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
 606
 607   if (expect_false (req->type == EIO_GROUP))
 608     {
 609       /* I hope this is worth it :/ */
 610       X_LOCK (reqlock);
 611       ++nreqs;
 612       X_UNLOCK (reqlock);
 613
 614       X_LOCK (reslock);
 615
 616       ++npending;
 617
 618       if (!reqq_push (&res_queue, req) && want_poll_cb)
 619         want_poll_cb ();
 620
 621       X_UNLOCK (reslock);
 622     }
 623   else
 624     {
 625       X_LOCK (reqlock);
 626       ++nreqs;
 627       ++nready;
 628       reqq_push (&req_queue, req);
 629       X_COND_SIGNAL (reqwait);
 630       X_UNLOCK (reqlock);
 631
 632       etp_maybe_start_thread ();
 633     }
 634 }
 635
 636 static void etp_set_max_poll_time (double nseconds)
 637 {
 638   if (WORDACCESS_UNSAFE) X_LOCK   (reslock);
 639   max_poll_time = nseconds * EIO_TICKS;
 640   if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
 641 }
 642
 643 static void etp_set_max_poll_reqs (unsigned int maxreqs)
 644 {
 645   if (WORDACCESS_UNSAFE) X_LOCK   (reslock);
 646   max_poll_reqs = maxreqs;
 647   if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
 648 }
 649
 650 static void etp_set_max_idle (unsigned int nthreads)
 651 {
 652   if (WORDACCESS_UNSAFE) X_LOCK   (reqlock);
 653   max_idle = nthreads <= 0 ? 1 : nthreads;
 654   if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
 655 }
 656
 657 static void etp_set_min_parallel (unsigned int nthreads)
 658 {
 659   if (wanted < nthreads)
 660     wanted = nthreads;
 661 }
 662
 663 static void etp_set_max_parallel (unsigned int nthreads)
 664 {
 665   if (wanted > nthreads)
 666     wanted = nthreads;
 667
 668   while (started > wanted)
 669     etp_end_thread ();
 670 }
 671
 672 /*****************************************************************************/
 673
 674 static void grp_try_feed (eio_req *grp)
 675 {
 676   while (grp->size < grp->int2 && !EIO_CANCELLED (grp))
 677     {
 678       grp->flags &= ~EIO_FLAG_GROUPADD;
 679
 680       EIO_FEED (grp);
 681
 682       /* stop if no progress has been made */
 683       if (!(grp->flags & EIO_FLAG_GROUPADD))
 684         {
 685           grp->feed = 0;
 686           break;
 687         }
 688     }
 689 }
 690
 691 static int grp_dec (eio_req *grp)
 692 {
 693   --grp->size;
 694
 695   /* call feeder, if applicable */
 696   grp_try_feed (grp);
 697
 698   /* finish, if done */
 699   if (!grp->size && grp->int1)
 700     return eio_finish (grp);
 701   else
 702     return 0;
 703 }
 704
 705 void eio_destroy (eio_req *req)
 706 {
 707   if ((req)->flags & EIO_FLAG_PTR1_FREE) free (req->ptr1);
 708   if ((req)->flags & EIO_FLAG_PTR2_FREE) free (req->ptr2);
 709
 710   EIO_DESTROY (req);
 711 }
 712
 713 static int eio_finish (eio_req *req)
 714 {
 715   int res = EIO_FINISH (req);
 716
 717   if (req->grp)
 718     {
 719       int res2;
 720       eio_req *grp = req->grp;
 721
 722       /* unlink request */
 723       if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
 724       if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
 725
 726       if (grp->grp_first == req)
 727         grp->grp_first = req->grp_next;
 728
 729       res2 = grp_dec (grp);
 730
 731       if (!res && res2)
 732         res = res2;
 733     }
 734
 735   eio_destroy (req);
 736
 737   return res;
 738 }
 739
 740 void eio_grp_cancel (eio_req *grp)
 741 {
 742   for (grp = grp->grp_first; grp; grp = grp->grp_next)
 743     eio_cancel (grp);
 744 }
 745
 746 void eio_cancel (eio_req *req)
 747 {
 748   etp_cancel (req);
 749 }
 750
 751 void eio_submit (eio_req *req)
 752 {
 753   etp_submit (req);
 754 }
 755
 756 unsigned int eio_nreqs (void)
 757 {
 758   return etp_nreqs ();
 759 }
 760
 761 unsigned int eio_nready (void)
 762 {
 763   return etp_nready ();
 764 }
 765
 766 unsigned int eio_npending (void)
 767 {
 768   return etp_npending ();
 769 }
 770
 771 unsigned int eio_nthreads (void)
 772 {
 773   return etp_nthreads ();
 774 }
 775
 776 void eio_set_max_poll_time (double nseconds)
 777 {
 778   etp_set_max_poll_time (nseconds);
 779 }
 780
 781 void eio_set_max_poll_reqs (unsigned int maxreqs)
 782 {
 783   etp_set_max_poll_reqs (maxreqs);
 784 }
 785
 786 void eio_set_max_idle (unsigned int nthreads)
 787 {
 788   etp_set_max_idle (nthreads);
 789 }
 790
 791 void eio_set_min_parallel (unsigned int nthreads)
 792 {
 793   etp_set_min_parallel (nthreads);
 794 }
 795
 796 void eio_set_max_parallel (unsigned int nthreads)
 797 {
 798   etp_set_max_parallel (nthreads);
 799 }
 800
 801 int eio_poll (void)
 802 {
 803   return etp_poll ();
 804 }
 805
 806 /*****************************************************************************/
 807 /* work around various missing functions */
 808
 809 #if !HAVE_PREADWRITE
 810 # undef pread
 811 # undef pwrite
 812 # define pread  eio__pread
 813 # define pwrite eio__pwrite
 814
 815 ssize_t
 816 eio__pread (int fd, void *buf, size_t count, off_t offset)
 817 {
 818   ssize_t res;
 819   off_t ooffset;
 820
 821   X_LOCK (preadwritelock);
 822   ooffset = lseek (fd, 0, SEEK_CUR);
 823   lseek (fd, offset, SEEK_SET);
 824   res = read (fd, buf, count);
 825   lseek (fd, ooffset, SEEK_SET);
 826   X_UNLOCK (preadwritelock);
 827
 828   return res;
 829 }
 830
 831 ssize_t
 832 eio__pwrite (int fd, void *buf, size_t count, off_t offset)
 833 {
 834   ssize_t res;
 835   off_t ooffset;
 836
 837   X_LOCK (preadwritelock);
 838   ooffset = lseek (fd, 0, SEEK_CUR);
 839   lseek (fd, offset, SEEK_SET);
 840   res = write (fd, buf, count);
 841   lseek (fd, ooffset, SEEK_SET);
 842   X_UNLOCK (preadwritelock);
 843
 844   return res;
 845 }
 846 #endif
 847
 848 #ifndef HAVE_UTIMES
 849
 850 # undef utimes
 851 # define utimes(path,times)  eio__utimes (path, times)
 852
 853 static int
 854 eio__utimes (const char *filename, const struct timeval times[2])
 855 {
 856   if (times)
 857     {
 858       struct utimbuf buf;
 859
 860       buf.actime  = times[0].tv_sec;
 861       buf.modtime = times[1].tv_sec;
 862
 863       return utime (filename, &buf);
 864     }
 865   else
 866     return utime (filename, 0);
 867 }
 868
 869 #endif
 870
 871 #ifndef HAVE_FUTIMES
 872
 873 # undef futimes
 874 # define futimes(fd,times) eio__futimes (fd, times)
 875
 876 static int eio__futimes (int fd, const struct timeval tv[2])
 877 {
 878   errno = ENOSYS;
 879   return -1;
 880 }
 881
 882 #endif
 883
 884 #ifdef _WIN32
 885 # define fsync(fd) (FlushFileBuffers((HANDLE)_get_osfhandle(fd)) ? 0 : -1)
 886 #endif
 887
 888 #if !HAVE_FDATASYNC
 889 # undef fdatasync
 890 # define fdatasync(fd) fsync (fd)
 891 #endif
 892
 893 // Use unicode and big file aware stat on windows
 894 #ifdef _WIN32
 895 # undef stat
 896 # undef fstat
 897 # define stat  _stati64
 898 # define fstat _fstati64
 899 #endif
 900
 901 /* sync_file_range always needs emulation */
 902 int
 903 eio__sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags)
 904 {
 905 #if HAVE_SYNC_FILE_RANGE
 906   int res;
 907
 908   if (EIO_SYNC_FILE_RANGE_WAIT_BEFORE   != SYNC_FILE_RANGE_WAIT_BEFORE
 909       || EIO_SYNC_FILE_RANGE_WRITE      != SYNC_FILE_RANGE_WRITE
 910       || EIO_SYNC_FILE_RANGE_WAIT_AFTER != SYNC_FILE_RANGE_WAIT_AFTER)
 911     {
 912       flags = 0
 913          | (flags & EIO_SYNC_FILE_RANGE_WAIT_BEFORE ? SYNC_FILE_RANGE_WAIT_BEFORE : 0)
 914          | (flags & EIO_SYNC_FILE_RANGE_WRITE       ? SYNC_FILE_RANGE_WRITE       : 0)
 915          | (flags & EIO_SYNC_FILE_RANGE_WAIT_AFTER  ? SYNC_FILE_RANGE_WAIT_AFTER  : 0);
 916     }
 917
 918   res = sync_file_range (fd, offset, nbytes, flags);
 919
 920   if (!res || errno != ENOSYS)
 921     return res;
 922 #endif
 923
 924   /* even though we could play tricks with the flags, it's better to always
 925    * call fdatasync, as that matches the expectation of its users best */
 926   return fdatasync (fd);
 927 }
 928
 929 #if !HAVE_READAHEAD
 930 # undef readahead
 931 # define readahead(fd,offset,count) eio__readahead (fd, offset, count, self)
 932
 933 static ssize_t
 934 eio__readahead (int fd, off_t offset, size_t count, etp_worker *self)
 935 {
 936   size_t todo = count;
 937   dBUF;
 938
 939   while (todo > 0)
 940     {
 941       size_t len = todo < EIO_BUFSIZE ? todo : EIO_BUFSIZE;
 942
 943       pread (fd, eio_buf, len, offset);
 944       offset += len;
 945       todo   -= len;
 946     }
 947
 948   errno = 0;
 949   return count;
 950 }
 951
 952 #endif
 953
 954 /* sendfile always needs emulation */
 955 static ssize_t
 956 eio__sendfile (int ofd, int ifd, off_t offset, size_t count, etp_worker *self)
 957 {
 958   ssize_t res;
 959
 960   if (!count)
 961     return 0;
 962
 963 #if HAVE_SENDFILE
 964 # if __linux
 965   res = sendfile (ofd, ifd, &offset, count);
 966
 967 # elif __FreeBSD__
 968   /*
 969    * Of course, the freebsd sendfile is a dire hack with no thoughts
 970    * wasted on making it similar to other I/O functions.
 971    */
 972   {
 973     off_t sbytes;
 974     res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
 975
 976     #if 0 /* according to the manpage, this is correct, but broken behaviour */
 977     /* freebsd' sendfile will return 0 on success */
 978     /* freebsd 8 documents it as only setting *sbytes on EINTR and EAGAIN, but */
 979     /* not on e.g. EIO or EPIPE - sounds broken */
 980     if ((res < 0 && (errno == EAGAIN || errno == EINTR) && sbytes) || res == 0)
 981       res = sbytes;
 982     #endif
 983
 984     /* according to source inspection, this is correct, and useful behaviour */
 985     if (sbytes)
 986       res = sbytes;
 987   }
 988
 989 # elif defined (__APPLE__)
 990
 991   {
 992     off_t sbytes = count;
 993     res = sendfile (ifd, ofd, offset, &sbytes, 0, 0);
 994
 995     /* according to the manpage, sbytes is always valid */
 996     if (sbytes)
 997       res = sbytes;
 998   }
 999
1000 # elif __hpux
1001   res = sendfile (ofd, ifd, offset, count, 0, 0);
1002
1003 # elif __solaris
1004   {
1005     struct sendfilevec vec;
1006     size_t sbytes;
1007
1008     vec.sfv_fd   = ifd;
1009     vec.sfv_flag = 0;
1010     vec.sfv_off  = offset;
1011     vec.sfv_len  = count;
1012
1013     res = sendfilev (ofd, &vec, 1, &sbytes);
1014
1015     if (res < 0 && sbytes)
1016       res = sbytes;
1017   }
1018
1019 # endif
1020
1021 //#elif defined (_WIN32)
1022 //
1023 //  /* does not work, just for documentation of what would need to be done */
1024 //  {
1025 //    HANDLE h = TO_SOCKET (ifd);
1026 //    SetFilePointer (h, offset, 0, FILE_BEGIN);
1027 //    res = TransmitFile (TO_SOCKET (ofd), h, count, 0, 0, 0, 0);
1028 //  }
1029
1030 #else
1031   res = -1;
1032   errno = ENOSYS;
1033 #endif
1034
1035   if (res <  0
1036       && (errno == ENOSYS || errno == EINVAL || errno == ENOTSOCK
1037           /* BSDs */
1038 #ifdef ENOTSUP /* sigh, if the steenking pile called openbsd would only try to at least compile posix code... */
1039           || errno == ENOTSUP
1040 #endif
1041           || errno == EOPNOTSUPP /* BSDs */
1042 #if __solaris
1043           || errno == EAFNOSUPPORT || errno == EPROTOTYPE
1044 #endif
1045          )
1046       )
1047     {
1048       /* emulate sendfile. this is a major pain in the ass */
1049       dBUF;
1050
1051       res = 0;
1052
1053       while (count)
1054         {
1055           ssize_t cnt;
1056
1057           cnt = pread (ifd, eio_buf, count > EIO_BUFSIZE ? EIO_BUFSIZE : count, offset);
1058
1059           if (cnt <= 0)
1060             {
1061               if (cnt && !res) res = -1;
1062               break;
1063             }
1064
1065           cnt = write (ofd, eio_buf, cnt);
1066
1067           if (cnt <= 0)
1068             {
1069               if (cnt && !res) res = -1;
1070               break;
1071             }
1072
1073           offset += cnt;
1074           res    += cnt;
1075           count  -= cnt;
1076         }
1077     }
1078
1079   return res;
1080 }
1081
1082 static signed char
1083 eio_dent_cmp (const eio_dirent *a, const eio_dirent *b)
1084 {
1085     return a->score - b->score ? a->score - b->score /* works because our signed char is always 0..100 */
1086               : a->inode < b->inode ? -1 : a->inode > b->inode ? 1 : 0;
1087 }
1088
1089 #define EIO_DENT_CMP(i,op,j) eio_dent_cmp (&i, &j) op 0
1090
1091 #define EIO_SORT_CUTOFF 30 /* quite high, but performs well on many filesystems */
1092 #define EIO_SORT_FAST   60 /* when to only use insertion sort */
1093
1094 static void
1095 eio_dent_radix_sort (eio_dirent *dents, int size, signed char score_bits, ino_t inode_bits)
1096 {
1097   unsigned char bits [9 + sizeof (ino_t) * 8];
1098   unsigned char *bit = bits;
1099
1100   assert (CHAR_BIT == 8);
1101   assert (sizeof (eio_dirent) * 8 < 256);
1102   assert (offsetof (eio_dirent, inode)); /* we use 0 as sentinel */
1103   assert (offsetof (eio_dirent, score)); /* we use 0 as sentinel */
1104
1105   if (size <= EIO_SORT_FAST)
1106     return;
1107
1108   /* first prepare an array of bits to test in our radix sort */
1109   /* try to take endianness into account, as well as differences in ino_t sizes */
1110   /* inode_bits must contain all inodes ORed together */
1111   /* which is used to skip bits that are 0 everywhere, which is very common */
1112   {
1113     ino_t endianness;
1114     int i, j;
1115
1116     /* we store the byte offset of byte n into byte n of "endianness" */
1117     for (i = 0; i < sizeof (ino_t); ++i)
1118       ((unsigned char *)&endianness)[i] = i;
1119
1120     *bit++ = 0;
1121
1122     for (i = 0; i < sizeof (ino_t); ++i)
1123       {
1124         /* shifting off the byte offsets out of "endianness" */
1125         int offs = (offsetof (eio_dirent, inode) + (endianness & 0xff)) * 8;
1126         endianness >>= 8;
1127
1128         for (j = 0; j < 8; ++j)
1129           if (inode_bits & (((ino_t)1) << (i * 8 + j)))
1130             *bit++ = offs + j;
1131       }
1132
1133     for (j = 0; j < 8; ++j)
1134       if (score_bits & (1 << j))
1135         *bit++ = offsetof (eio_dirent, score) * 8 + j;
1136   }
1137
1138   /* now actually do the sorting (a variant of MSD radix sort) */
1139   {
1140     eio_dirent    *base_stk [9 + sizeof (ino_t) * 8], *base;
1141     eio_dirent    *end_stk  [9 + sizeof (ino_t) * 8], *end;
1142     unsigned char *bit_stk  [9 + sizeof (ino_t) * 8];
1143     int stk_idx = 0;
1144
1145     base_stk [stk_idx] = dents;
1146     end_stk  [stk_idx] = dents + size;
1147     bit_stk  [stk_idx] = bit - 1;
1148
1149     do
1150       {
1151         base = base_stk [stk_idx];
1152         end  = end_stk  [stk_idx];
1153         bit  = bit_stk  [stk_idx];
1154
1155         for (;;)
1156           {
1157             unsigned char O = *bit >> 3;
1158             unsigned char M = 1 << (*bit & 7);
1159
1160             eio_dirent *a = base;
1161             eio_dirent *b = end;
1162
1163             if (b - a < EIO_SORT_CUTOFF)
1164               break;
1165
1166             /* now bit-partition the array on the bit */
1167             /* this ugly asymmetric loop seems to perform much better than typical */
1168             /* partition algos found in the literature */
1169             do
1170               if (!(((unsigned char *)a)[O] & M))
1171                 ++a;
1172               else if (!(((unsigned char *)--b)[O] & M))
1173                 {
1174                   eio_dirent tmp = *a; *a = *b; *b = tmp;
1175                   ++a;
1176                 }
1177             while (b > a);
1178
1179             /* next bit, or stop, if no bits left in this path */
1180             if (!*--bit)
1181               break;
1182
1183             base_stk [stk_idx] = a;
1184             end_stk  [stk_idx] = end;
1185             bit_stk  [stk_idx] = bit;
1186             ++stk_idx;
1187
1188             end = a;
1189           }
1190       }
1191     while (stk_idx--);
1192   }
1193 }
1194
1195 static void
1196 eio_dent_insertion_sort (eio_dirent *dents, int size)
1197 {
1198   /* first move the smallest element to the front, to act as a sentinel */
1199   {
1200     int i;
1201     eio_dirent *min = dents;
1202
1203     /* the radix pre-pass ensures that the minimum element is in the first EIO_SORT_CUTOFF + 1 elements */
1204     for (i = size > EIO_SORT_FAST ? EIO_SORT_CUTOFF + 1 : size; --i; )
1205       if (EIO_DENT_CMP (dents [i], <, *min))
1206         min = &dents [i];
1207
1208     /* swap elements 0 and j (minimum) */
1209     {
1210       eio_dirent tmp = *dents; *dents = *min; *min = tmp;
1211     }
1212   }
1213
1214   /* then do standard insertion sort, assuming that all elements are >= dents [0] */
1215   {
1216     eio_dirent *i, *j;
1217
1218     for (i = dents + 1; i < dents + size; ++i)
1219       {
1220         eio_dirent value = *i;
1221
1222         for (j = i - 1; EIO_DENT_CMP (*j, >, value); --j)
1223           j [1] = j [0];
1224
1225         j [1] = value;
1226       }
1227   }
1228 }
1229
1230 static void
1231 eio_dent_sort (eio_dirent *dents, int size, signed char score_bits, ino_t inode_bits)
1232 {
1233   if (size <= 1)
1234     return; /* our insertion sort relies on size > 0 */
1235
1236   /* first we use a radix sort, but only for dirs >= EIO_SORT_FAST */
1237   /* and stop sorting when the partitions are <= EIO_SORT_CUTOFF */
1238   eio_dent_radix_sort (dents, size, score_bits, inode_bits);
1239
1240   /* use an insertion sort at the end, or for small arrays, */
1241   /* as insertion sort is more efficient for small partitions */
1242   eio_dent_insertion_sort (dents, size);
1243 }
1244
1245 /* read a full directory */
1246 static void
1247 eio__scandir (eio_req *req, etp_worker *self)
1248 {
1249   DIR *dirp;
1250   EIO_STRUCT_DIRENT *entp;
1251   char *name, *names;
1252   int namesalloc = 4096;
1253   int namesoffs = 0;
1254   int flags = req->int1;
1255   eio_dirent *dents = 0;
1256   int dentalloc = 128;
1257   int dentoffs = 0;
1258   ino_t inode_bits = 0;
1259
1260   req->result = -1;
1261
1262   if (!(flags & EIO_READDIR_DENTS))
1263     flags &= ~(EIO_READDIR_DIRS_FIRST | EIO_READDIR_STAT_ORDER);
1264
1265   X_LOCK (wrklock);
1266   /* the corresponding closedir is in ETP_WORKER_CLEAR */
1267   self->dirp = dirp = opendir (req->ptr1);
1268   req->flags |= EIO_FLAG_PTR1_FREE | EIO_FLAG_PTR2_FREE;
1269   req->ptr1 = dents = flags ? malloc (dentalloc * sizeof (eio_dirent)) : 0;
1270   req->ptr2 = names = malloc (namesalloc);
1271   X_UNLOCK (wrklock);
1272
1273   if (dirp && names && (!flags || dents))
1274     for (;;)
1275       {
1276         errno = 0;
1277         entp = readdir (dirp);
1278
1279         if (!entp)
1280           {
1281             if (errno)
1282               break;
1283
1284             /* sort etc. */
1285             req->int1   = flags;
1286             req->result = dentoffs;
1287
1288             if (flags & EIO_READDIR_STAT_ORDER)
1289               eio_dent_sort (dents, dentoffs, 0, inode_bits); /* sort by inode exclusively */
1290             else if (flags & EIO_READDIR_DIRS_FIRST)
1291               if (flags & EIO_READDIR_FOUND_UNKNOWN)
1292                 eio_dent_sort (dents, dentoffs, 7, inode_bits); /* sort by score and inode */
1293               else
1294                 {
1295                   /* in this case, all is known, and we just put dirs first and sort them */
1296                   eio_dirent *oth = dents + dentoffs;
1297                   eio_dirent *dir = dents;
1298
1299                   /* now partition dirs to the front, and non-dirs to the back */
1300                   /* by walking from both sides and swapping if necessary */
1301                   /* also clear score, so it doesn't influence sorting */
1302                   while (oth > dir)
1303                     {
1304                       if (dir->type == EIO_DT_DIR)
1305                         ++dir;
1306                       else if ((--oth)->type == EIO_DT_DIR)
1307                         {
1308                           eio_dirent tmp = *dir; *dir = *oth; *oth = tmp;
1309
1310                           ++dir;
1311                         }
1312                     }
1313
1314                   /* now sort the dirs only */
1315                   eio_dent_sort (dents, dir - dents, 0, inode_bits);
1316                 }
1317
1318             break;
1319           }
1320
1321         /* now add the entry to our list(s) */
1322         name = entp->d_name;
1323
1324         /* skip . and .. entries */
1325         if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
1326           {
1327             int len = D_NAMLEN (entp) + 1;
1328
1329             while (expect_false (namesoffs + len > namesalloc))
1330               {
1331                 namesalloc *= 2;
1332                 X_LOCK (wrklock);
1333                 req->ptr2 = names = realloc (names, namesalloc);
1334                 X_UNLOCK (wrklock);
1335
1336                 if (!names)
1337                   break;
1338               }
1339
1340             memcpy (names + namesoffs, name, len);
1341
1342             if (dents)
1343               {
1344                 struct eio_dirent *ent;
1345
1346                 if (expect_false (dentoffs == dentalloc))
1347                   {
1348                     dentalloc *= 2;
1349                     X_LOCK (wrklock);
1350                     req->ptr1 = dents = realloc (dents, dentalloc * sizeof (eio_dirent));
1351                     X_UNLOCK (wrklock);
1352
1353                     if (!dents)
1354                       break;
1355                   }
1356
1357                 ent = dents + dentoffs;
1358
1359                 ent->nameofs = namesoffs; /* rather dirtily we store the offset in the pointer */
1360                 ent->namelen = len - 1;
1361                 ent->inode   = D_INO (entp);
1362
1363                 inode_bits |= ent->inode;
1364
1365                 switch (D_TYPE (entp))
1366                   {
1367                     default:
1368                       ent->type = EIO_DT_UNKNOWN;
1369                       flags |= EIO_READDIR_FOUND_UNKNOWN;
1370                       break;
1371
1372                     #ifdef DT_FIFO
1373                       case DT_FIFO: ent->type = EIO_DT_FIFO; break;
1374                     #endif
1375                     #ifdef DT_CHR
1376                       case DT_CHR:  ent->type = EIO_DT_CHR;  break;
1377                     #endif
1378                     #ifdef DT_MPC
1379                       case DT_MPC:  ent->type = EIO_DT_MPC;  break;
1380                     #endif
1381                     #ifdef DT_DIR
1382                       case DT_DIR:  ent->type = EIO_DT_DIR;  break;
1383                     #endif
1384                     #ifdef DT_NAM
1385                       case DT_NAM:  ent->type = EIO_DT_NAM;  break;
1386                     #endif
1387                     #ifdef DT_BLK
1388                       case DT_BLK:  ent->type = EIO_DT_BLK;  break;
1389                     #endif
1390                     #ifdef DT_MPB
1391                       case DT_MPB:  ent->type = EIO_DT_MPB;  break;
1392                     #endif
1393                     #ifdef DT_REG
1394                       case DT_REG:  ent->type = EIO_DT_REG;  break;
1395                     #endif
1396                     #ifdef DT_NWK
1397                       case DT_NWK:  ent->type = EIO_DT_NWK;  break;
1398                     #endif
1399                     #ifdef DT_CMP
1400                       case DT_CMP:  ent->type = EIO_DT_CMP;  break;
1401                     #endif
1402                     #ifdef DT_LNK
1403                       case DT_LNK:  ent->type = EIO_DT_LNK;  break;
1404                     #endif
1405                     #ifdef DT_SOCK
1406                       case DT_SOCK: ent->type = EIO_DT_SOCK; break;
1407                     #endif
1408                     #ifdef DT_DOOR
1409                       case DT_DOOR: ent->type = EIO_DT_DOOR; break;
1410                     #endif
1411                     #ifdef DT_WHT
1412                       case DT_WHT:  ent->type = EIO_DT_WHT;  break;
1413                     #endif
1414                   }
1415
1416                 ent->score = 7;
1417
1418                 if (flags & EIO_READDIR_DIRS_FIRST)
1419                   {
1420                     if (ent->type == EIO_DT_UNKNOWN)
1421                       {
1422                         if (*name == '.') /* leading dots are likely directories, and, in any case, rare */
1423                           ent->score = 1;
1424                         else if (!strchr (name, '.')) /* absense of dots indicate likely dirs */
1425                           ent->score = len <= 2 ? 4 - len : len <= 4 ? 4 : len <= 7 ? 5 : 6; /* shorter == more likely dir, but avoid too many classes */
1426                       }
1427                     else if (ent->type == EIO_DT_DIR)
1428                       ent->score = 0;
1429                   }
1430               }
1431
1432             namesoffs += len;
1433             ++dentoffs;
1434           }
1435
1436         if (EIO_CANCELLED (req))
1437           {
1438             errno = ECANCELED;
1439             break;
1440           }
1441       }
1442 }
1443
1444 #ifdef PAGESIZE
1445 # define eio_pagesize() PAGESIZE
1446
1447 #elif defined(_WIN32)
1448   /* Windows */
1449   static intptr_t
1450   eio_pagesize (void)
1451   {
1452     SYSTEM_INFO si;
1453     GetSystemInfo(&si);
1454     return si.dwPageSize;
1455   }
1456
1457 #else
1458   /* POSIX */
1459   static intptr_t
1460   eio_pagesize (void)
1461   {
1462     static intptr_t page;
1463
1464     if (!page)
1465       page = sysconf (_SC_PAGESIZE);
1466
1467     return page;
1468   }
1469 #endif
1470
1471 static void
1472 eio_page_align (void **addr, size_t *length)
1473 {
1474   intptr_t mask = eio_pagesize () - 1;
1475
1476   /* round down addr */
1477   intptr_t adj = mask & (intptr_t)*addr;
1478
1479   *addr   = (void *)((intptr_t)*addr - adj);
1480   *length += adj;
1481
1482   /* round up length */
1483   *length = (*length + mask) & ~mask;
1484 }
1485
1486 #if !_POSIX_MEMLOCK
1487 # define eio__mlockall(a) ((errno = ENOSYS), -1)
1488 #else
1489
1490 static int
1491 eio__mlockall (int flags)
1492 {
1493   #if __GLIBC__ == 2 && __GLIBC_MINOR__ <= 7
1494     extern int mallopt (int, int);
1495     mallopt (-6, 238); /* http://bugs.debian.org/cgi-bin/bugreport.cgi?bug=473812 */
1496   #endif
1497
1498   if (EIO_MCL_CURRENT   != MCL_CURRENT
1499       || EIO_MCL_FUTURE != MCL_FUTURE)
1500     {
1501       flags = 0
1502          | (flags & EIO_MCL_CURRENT ? MCL_CURRENT : 0)
1503          | (flags & EIO_MCL_FUTURE  ? MCL_FUTURE : 0);
1504     }
1505
1506   return mlockall (flags);
1507 }
1508 #endif
1509
1510 #if !_POSIX_MEMLOCK_RANGE
1511 # define eio__mlock(a,b) ((errno = ENOSYS), -1)
1512 #else
1513
1514 static int
1515 eio__mlock (void *addr, size_t length)
1516 {
1517   eio_page_align (&addr, &length);
1518
1519   return mlock (addr, length);
1520 }
1521
1522 #endif
1523
1524 #if !(_POSIX_MAPPED_FILES && _POSIX_SYNCHRONIZED_IO)
1525 # define eio__msync(a,b,c) ((errno = ENOSYS), -1)
1526 #else
1527
1528 int
1529 eio__msync (void *mem, size_t len, int flags)
1530 {
1531   eio_page_align (&mem, &len);
1532
1533   if (EIO_MS_ASYNC         != MS_SYNC
1534       || EIO_MS_INVALIDATE != MS_INVALIDATE
1535       || EIO_MS_SYNC       != MS_SYNC)
1536     {
1537       flags = 0
1538          | (flags & EIO_MS_ASYNC      ? MS_ASYNC : 0)
1539          | (flags & EIO_MS_INVALIDATE ? MS_INVALIDATE : 0)
1540          | (flags & EIO_MS_SYNC       ? MS_SYNC : 0);
1541     }
1542
1543   return msync (mem, len, flags);
1544 }
1545
1546 #endif
1547
1548 int
1549 eio__mtouch (void *mem, size_t len, int flags)
1550 {
1551   eio_page_align (&mem, &len);
1552
1553   {
1554     intptr_t addr = (intptr_t)mem;
1555     intptr_t end = addr + len;
1556     intptr_t page = eio_pagesize ();
1557
1558     if (addr < end)
1559       if (flags & EIO_MT_MODIFY) /* modify */
1560         do { *((volatile sig_atomic_t *)addr) |= 0; } while ((addr += page) < len);
1561       else
1562         do { *((volatile sig_atomic_t *)addr)     ; } while ((addr += page) < len);
1563   }
1564
1565   return 0;
1566 }
1567
1568 /*****************************************************************************/
1569
1570 #define ALLOC(len)                              \
1571   if (!req->ptr2)                               \
1572     {                                           \
1573       X_LOCK (wrklock);                         \
1574       req->flags |= EIO_FLAG_PTR2_FREE;         \
1575       X_UNLOCK (wrklock);                       \
1576       req->ptr2 = malloc (len);                 \
1577       if (!req->ptr2)                           \
1578         {                                       \
1579           errno       = ENOMEM;                 \
1580           req->result = -1;                     \
1581           break;                                \
1582         }                                       \
1583     }
1584
1585 X_THREAD_PROC (etp_proc)
1586 {
1587   ETP_REQ *req;
1588   struct timespec ts;
1589   etp_worker *self = (etp_worker *)thr_arg;
1590
1591   /* try to distribute timeouts somewhat randomly */
1592   ts.tv_nsec = ((unsigned long)self & 1023UL) * (1000000000UL / 1024UL);
1593
1594   for (;;)
1595     {
1596       X_LOCK (reqlock);
1597
1598       for (;;)
1599         {
1600           self->req = req = reqq_shift (&req_queue);
1601
1602           if (req)
1603             break;
1604
1605           ++idle;
1606
1607           ts.tv_sec = time (0) + IDLE_TIMEOUT;
1608           if (X_COND_TIMEDWAIT (reqwait, reqlock, ts) == ETIMEDOUT)
1609             {
1610               if (idle > max_idle)
1611                 {
1612                   --idle;
1613                   X_UNLOCK (reqlock);
1614                   X_LOCK (wrklock);
1615                   --started;
1616                   X_UNLOCK (wrklock);
1617                   goto quit;
1618                 }
1619
1620               /* we are allowed to idle, so do so without any timeout */
1621               X_COND_WAIT (reqwait, reqlock);
1622             }
1623
1624           --idle;
1625         }
1626
1627       --nready;
1628
1629       X_UNLOCK (reqlock);
1630
1631       if (req->type < 0)
1632         goto quit;
1633
1634       if (!EIO_CANCELLED (req))
1635         ETP_EXECUTE (self, req);
1636
1637       X_LOCK (reslock);
1638
1639       ++npending;
1640
1641       if (!reqq_push (&res_queue, req) && want_poll_cb)
1642         want_poll_cb ();
1643
1644       self->req = 0;
1645       etp_worker_clear (self);
1646
1647       X_UNLOCK (reslock);
1648     }
1649
1650 quit:
1651   X_LOCK (wrklock);
1652   etp_worker_free (self);
1653   X_UNLOCK (wrklock);
1654
1655   return 0;
1656 }
1657
1658 /*****************************************************************************/
1659
1660 int eio_init (void (*want_poll)(void), void (*done_poll)(void))
1661 {
1662   return etp_init (want_poll, done_poll);
1663 }
1664
1665 static void eio_api_destroy (eio_req *req)
1666 {
1667   free (req);
1668 }
1669
1670 #define REQ(rtype)                                              \
1671   eio_req *req;                                                 \
1672                                                                 \
1673   req = (eio_req *)calloc (1, sizeof *req);                     \
1674   if (!req)                                                     \
1675     return 0;                                                   \
1676                                                                 \
1677   req->type    = rtype;                                         \
1678   req->pri     = pri;                                           \
1679   req->finish  = cb;                                            \
1680   req->data    = data;                                          \
1681   req->destroy = eio_api_destroy;
1682
1683 #define SEND eio_submit (req); return req
1684
1685 #define PATH                                                    \
1686   req->flags |= EIO_FLAG_PTR1_FREE;                             \
1687   req->ptr1 = strdup (path);                                    \
1688   if (!req->ptr1)                                               \
1689     {                                                           \
1690       eio_api_destroy (req);                                    \
1691       return 0;                                                 \
1692     }
1693
1694 static void eio_execute (etp_worker *self, eio_req *req)
1695 {
1696   switch (req->type)
1697     {
1698       case EIO_READ:      ALLOC (req->size);
1699                           req->result = req->offs >= 0
1700                                       ? pread     (req->int1, req->ptr2, req->size, req->offs)
1701                                       : read      (req->int1, req->ptr2, req->size); break;
1702       case EIO_WRITE:
1703 #if defined (__APPLE__)
1704                           pthread_mutex_lock (&apple_bug_writelock);
1705 #endif
1706
1707                           req->result = req->offs >= 0
1708                                       ? pwrite    (req->int1, req->ptr2, req->size, req->offs)
1709                                       : write     (req->int1, req->ptr2, req->size);
1710
1711 #if defined (__APPLE__)
1712                           pthread_mutex_unlock (&apple_bug_writelock);
1713 #endif
1714                           break;
1715
1716       case EIO_READAHEAD: req->result = readahead     (req->int1, req->offs, req->size); break;
1717       case EIO_SENDFILE:  req->result = eio__sendfile (req->int1, req->int2, req->offs, req->size, self); break;
1718
1719       case EIO_STAT:      ALLOC (sizeof (EIO_STRUCT_STAT));
1720                           req->result = stat      (req->ptr1, (EIO_STRUCT_STAT *)req->ptr2); break;
1721 #ifndef _WIN32
1722       case EIO_LSTAT:     ALLOC (sizeof (EIO_STRUCT_STAT));
1723                           req->result = lstat     (req->ptr1, (EIO_STRUCT_STAT *)req->ptr2); break;
1724 #endif
1725       case EIO_FSTAT:     ALLOC (sizeof (EIO_STRUCT_STAT));
1726                           req->result = fstat     (req->int1, (EIO_STRUCT_STAT *)req->ptr2); break;
1727
1728 #ifndef _WIN32
1729       case EIO_STATVFS:   ALLOC (sizeof (EIO_STRUCT_STATVFS));
1730                           req->result = statvfs   (req->ptr1, (EIO_STRUCT_STATVFS *)req->ptr2); break;
1731       case EIO_FSTATVFS:  ALLOC (sizeof (EIO_STRUCT_STATVFS));
1732                           req->result = fstatvfs  (req->int1, (EIO_STRUCT_STATVFS *)req->ptr2); break;
1733
1734       case EIO_CHOWN:     req->result = chown     (req->ptr1, req->int2, req->int3); break;
1735       case EIO_FCHOWN:    req->result = fchown    (req->int1, req->int2, req->int3); break;
1736 #endif
1737       case EIO_CHMOD:     req->result = chmod     (req->ptr1, (mode_t)req->int2); break;
1738 #ifndef _WIN32
1739       case EIO_FCHMOD:    req->result = fchmod    (req->int1, (mode_t)req->int2); break;
1740       case EIO_TRUNCATE:  req->result = truncate  (req->ptr1, req->offs); break;
1741 #endif
1742       case EIO_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break;
1743
1744       case EIO_OPEN:      req->result = open      (req->ptr1, req->int1, (mode_t)req->int2); break;
1745       case EIO_CLOSE:     req->result = close     (req->int1); break;
1746       case EIO_DUP2:      req->result = dup2      (req->int1, req->int2); break;
1747       case EIO_UNLINK:    req->result = unlink    (req->ptr1); break;
1748       case EIO_RMDIR:     req->result = rmdir     (req->ptr1); break;
1749 #ifdef _WIN32
1750       case EIO_MKDIR:     req->result = mkdir     (req->ptr1); break;
1751 #else
1752       case EIO_MKDIR:     req->result = mkdir     (req->ptr1, (mode_t)req->int2); break;
1753 #endif
1754       case EIO_RENAME:    req->result = rename    (req->ptr1, req->ptr2); break;
1755 #ifndef _WIN32
1756       case EIO_LINK:      req->result = link      (req->ptr1, req->ptr2); break;
1757       case EIO_SYMLINK:   req->result = symlink   (req->ptr1, req->ptr2); break;
1758       case EIO_MKNOD:     req->result = mknod     (req->ptr1, (mode_t)req->int2, (dev_t)req->int3); break;
1759 #endif
1760
1761 #ifndef _WIN32
1762       case EIO_READLINK:  ALLOC (PATH_MAX);
1763                           req->result = readlink  (req->ptr1, req->ptr2, PATH_MAX); break;
1764 #endif
1765
1766 #ifndef _WIN32
1767       case EIO_SYNC:      req->result = 0; sync (); break;
1768 #endif
1769       case EIO_FSYNC:     req->result = fsync     (req->int1); break;
1770       case EIO_FDATASYNC: req->result = fdatasync (req->int1); break;
1771       case EIO_MSYNC:     req->result = eio__msync (req->ptr2, req->size, req->int1); break;
1772       case EIO_MTOUCH:    req->result = eio__mtouch (req->ptr2, req->size, req->int1); break;
1773       case EIO_MLOCK:     req->result = eio__mlock (req->ptr2, req->size); break;
1774       case EIO_MLOCKALL:  req->result = eio__mlockall (req->int1); break;
1775       case EIO_SYNC_FILE_RANGE: req->result = eio__sync_file_range (req->int1, req->offs, req->size, req->int2); break;
1776
1777       case EIO_READDIR:   eio__scandir (req, self); break;
1778
1779       case EIO_BUSY:
1780 #ifdef _WIN32
1781         Sleep (req->nv1 * 1e3);
1782 #else
1783         {
1784           struct timeval tv;
1785
1786           tv.tv_sec  = req->nv1;
1787           tv.tv_usec = (req->nv1 - tv.tv_sec) * 1e6;
1788
1789           req->result = select (0, 0, 0, 0, &tv);
1790         }
1791 #endif
1792         break;
1793
1794       case EIO_UTIME:
1795       case EIO_FUTIME:
1796         {
1797           struct timeval tv[2];
1798           struct timeval *times;
1799
1800           if (req->nv1 != -1. || req->nv2 != -1.)
1801             {
1802               tv[0].tv_sec  = req->nv1;
1803               tv[0].tv_usec = (req->nv1 - tv[0].tv_sec) * 1000000.;
1804               tv[1].tv_sec  = req->nv2;
1805               tv[1].tv_usec = (req->nv2 - tv[1].tv_sec) * 1000000.;
1806
1807               times = tv;
1808             }
1809           else
1810             times = 0;
1811
1812           req->result = req->type == EIO_FUTIME
1813                         ? futimes (req->int1, times)
1814                         : utimes  (req->ptr1, times);
1815         }
1816         break;
1817
1818       case EIO_GROUP:
1819         abort (); /* handled in eio_request */
1820
1821       case EIO_NOP:
1822         req->result = 0;
1823         break;
1824
1825       case EIO_CUSTOM:
1826         ((void (*)(eio_req *))req->feed) (req);
1827         break;
1828
1829       default:
1830         errno = ENOSYS;
1831         req->result = -1;
1832         break;
1833     }
1834
1835   req->errorno = errno;
1836 }
1837
1838 #ifndef EIO_NO_WRAPPERS
1839
1840 eio_req *eio_nop (int pri, eio_cb cb, void *data)
1841 {
1842   REQ (EIO_NOP); SEND;
1843 }
1844
1845 eio_req *eio_busy (double delay, int pri, eio_cb cb, void *data)
1846 {
1847   REQ (EIO_BUSY); req->nv1 = delay; SEND;
1848 }
1849
1850 eio_req *eio_sync (int pri, eio_cb cb, void *data)
1851 {
1852   REQ (EIO_SYNC); SEND;
1853 }
1854
1855 eio_req *eio_fsync (int fd, int pri, eio_cb cb, void *data)
1856 {
1857   REQ (EIO_FSYNC); req->int1 = fd; SEND;
1858 }
1859
1860 eio_req *eio_msync (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1861 {
1862   REQ (EIO_MSYNC); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1863 }
1864
1865 eio_req *eio_mtouch (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1866 {
1867   REQ (EIO_MTOUCH); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1868 }
1869
1870 eio_req *eio_mlock (void *addr, size_t length, int pri, eio_cb cb, void *data)
1871 {
1872   REQ (EIO_MLOCK); req->ptr2 = addr; req->size = length; SEND;
1873 }
1874
1875 eio_req *eio_mlockall (int flags, int pri, eio_cb cb, void *data)
1876 {
1877   REQ (EIO_MLOCKALL); req->int1 = flags; SEND;
1878 }
1879
1880 eio_req *eio_sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags, int pri, eio_cb cb, void *data)
1881 {
1882   REQ (EIO_SYNC_FILE_RANGE); req->int1 = fd; req->offs = offset; req->size = nbytes; req->int2 = flags; SEND;
1883 }
1884
1885 eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data)
1886 {
1887   REQ (EIO_FDATASYNC); req->int1 = fd; SEND;
1888 }
1889
1890 eio_req *eio_close (int fd, int pri, eio_cb cb, void *data)
1891 {
1892   REQ (EIO_CLOSE); req->int1 = fd; SEND;
1893 }
1894
1895 eio_req *eio_readahead (int fd, off_t offset, size_t length, int pri, eio_cb cb, void *data)
1896 {
1897   REQ (EIO_READAHEAD); req->int1 = fd; req->offs = offset; req->size = length; SEND;
1898 }
1899
1900 eio_req *eio_read (int fd, void *buf, size_t length, off_t offset, int pri, eio_cb cb, void *data)
1901 {
1902   REQ (EIO_READ); req->int1 = fd; req->offs = offset; req->size = length; req->ptr2 = buf; SEND;
1903 }
1904
1905 eio_req *eio_write (int fd, void *buf, size_t length, off_t offset, int pri, eio_cb cb, void *data)
1906 {
1907   REQ (EIO_WRITE); req->int1 = fd; req->offs = offset; req->size = length; req->ptr2 = buf; SEND;
1908 }
1909
1910 eio_req *eio_fstat (int fd, int pri, eio_cb cb, void *data)
1911 {
1912   REQ (EIO_FSTAT); req->int1 = fd; SEND;
1913 }
1914
1915 eio_req *eio_fstatvfs (int fd, int pri, eio_cb cb, void *data)
1916 {
1917   REQ (EIO_FSTATVFS); req->int1 = fd; SEND;
1918 }
1919
1920 eio_req *eio_futime (int fd, double atime, double mtime, int pri, eio_cb cb, void *data)
1921 {
1922   REQ (EIO_FUTIME); req->int1 = fd; req->nv1 = atime; req->nv2 = mtime; SEND;
1923 }
1924
1925 eio_req *eio_ftruncate (int fd, off_t offset, int pri, eio_cb cb, void *data)
1926 {
1927   REQ (EIO_FTRUNCATE); req->int1 = fd; req->offs = offset; SEND;
1928 }
1929
1930 eio_req *eio_fchmod (int fd, mode_t mode, int pri, eio_cb cb, void *data)
1931 {
1932   REQ (EIO_FCHMOD); req->int1 = fd; req->int2 = (long)mode; SEND;
1933 }
1934
1935 eio_req *eio_fchown (int fd, uid_t uid, gid_t gid, int pri, eio_cb cb, void *data)
1936 {
1937   REQ (EIO_FCHOWN); req->int1 = fd; req->int2 = (long)uid; req->int3 = (long)gid; SEND;
1938 }
1939
1940 eio_req *eio_dup2 (int fd, int fd2, int pri, eio_cb cb, void *data)
1941 {
1942   REQ (EIO_DUP2); req->int1 = fd; req->int2 = fd2; SEND;
1943 }
1944
1945 eio_req *eio_sendfile (int out_fd, int in_fd, off_t in_offset, size_t length, int pri, eio_cb cb, void *data)
1946 {
1947   REQ (EIO_SENDFILE); req->int1 = out_fd; req->int2 = in_fd; req->offs = in_offset; req->size = length; SEND;
1948 }
1949
1950 eio_req *eio_open (const char *path, int flags, mode_t mode, int pri, eio_cb cb, void *data)
1951 {
1952   REQ (EIO_OPEN); PATH; req->int1 = flags; req->int2 = (long)mode; SEND;
1953 }
1954
1955 eio_req *eio_utime (const char *path, double atime, double mtime, int pri, eio_cb cb, void *data)
1956 {
1957   REQ (EIO_UTIME); PATH; req->nv1 = atime; req->nv2 = mtime; SEND;
1958 }
1959
1960 eio_req *eio_truncate (const char *path, off_t offset, int pri, eio_cb cb, void *data)
1961 {
1962   REQ (EIO_TRUNCATE); PATH; req->offs = offset; SEND;
1963 }
1964
1965 eio_req *eio_chown (const char *path, uid_t uid, gid_t gid, int pri, eio_cb cb, void *data)
1966 {
1967   REQ (EIO_CHOWN); PATH; req->int2 = (long)uid; req->int3 = (long)gid; SEND;
1968 }
1969
1970 eio_req *eio_chmod (const char *path, mode_t mode, int pri, eio_cb cb, void *data)
1971 {
1972   REQ (EIO_CHMOD); PATH; req->int2 = (long)mode; SEND;
1973 }
1974
1975 eio_req *eio_mkdir (const char *path, mode_t mode, int pri, eio_cb cb, void *data)
1976 {
1977   REQ (EIO_MKDIR); PATH; req->int2 = (long)mode; SEND;
1978 }
1979
1980 static eio_req *
1981 eio__1path (int type, const char *path, int pri, eio_cb cb, void *data)
1982 {
1983   REQ (type); PATH; SEND;
1984 }
1985
1986 eio_req *eio_readlink (const char *path, int pri, eio_cb cb, void *data)
1987 {
1988   return eio__1path (EIO_READLINK, path, pri, cb, data);
1989 }
1990
1991 eio_req *eio_stat (const char *path, int pri, eio_cb cb, void *data)
1992 {
1993   return eio__1path (EIO_STAT, path, pri, cb, data);
1994 }
1995
1996 eio_req *eio_lstat (const char *path, int pri, eio_cb cb, void *data)
1997 {
1998   return eio__1path (EIO_LSTAT, path, pri, cb, data);
1999 }
2000
2001 eio_req *eio_statvfs (const char *path, int pri, eio_cb cb, void *data)
2002 {
2003   return eio__1path (EIO_STATVFS, path, pri, cb, data);
2004 }
2005
2006 eio_req *eio_unlink (const char *path, int pri, eio_cb cb, void *data)
2007 {
2008   return eio__1path (EIO_UNLINK, path, pri, cb, data);
2009 }
2010
2011 eio_req *eio_rmdir (const char *path, int pri, eio_cb cb, void *data)
2012 {
2013   return eio__1path (EIO_RMDIR, path, pri, cb, data);
2014 }
2015
2016 eio_req *eio_readdir (const char *path, int flags, int pri, eio_cb cb, void *data)
2017 {
2018   REQ (EIO_READDIR); PATH; req->int1 = flags; SEND;
2019 }
2020
2021 eio_req *eio_mknod (const char *path, mode_t mode, dev_t dev, int pri, eio_cb cb, void *data)
2022 {
2023   REQ (EIO_MKNOD); PATH; req->int2 = (long)mode; req->int3 = (long)dev; SEND;
2024 }
2025
2026 static eio_req *
2027 eio__2path (int type, const char *path, const char *new_path, int pri, eio_cb cb, void *data)
2028 {
2029   REQ (type); PATH;
2030
2031   req->flags |= EIO_FLAG_PTR2_FREE;
2032   req->ptr2 = strdup (new_path);
2033   if (!req->ptr2)
2034     {
2035       eio_api_destroy (req);
2036       return 0;
2037     }
2038
2039   SEND;
2040 }
2041
2042 eio_req *eio_link (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
2043 {
2044   return eio__2path (EIO_LINK, path, new_path, pri, cb, data);
2045 }
2046
2047 eio_req *eio_symlink (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
2048 {
2049   return eio__2path (EIO_SYMLINK, path, new_path, pri, cb, data);
2050 }
2051
2052 eio_req *eio_rename (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
2053 {
2054   return eio__2path (EIO_RENAME, path, new_path, pri, cb, data);
2055 }
2056
2057 eio_req *eio_custom (eio_cb execute, int pri, eio_cb cb, void *data)
2058 {
2059   REQ (EIO_CUSTOM); req->feed = (void (*)(eio_req *))execute; SEND;
2060 }
2061
2062 #endif
2063
2064 eio_req *eio_grp (eio_cb cb, void *data)
2065 {
2066   const int pri = EIO_PRI_MAX;
2067
2068   REQ (EIO_GROUP); SEND;
2069 }
2070
2071 #undef REQ
2072 #undef PATH
2073 #undef SEND
2074
2075 /*****************************************************************************/
2076 /* grp functions */
2077
2078 void eio_grp_feed (eio_req *grp, void (*feed)(eio_req *req), int limit)
2079 {
2080   grp->int2 = limit;
2081   grp->feed = feed;
2082
2083   grp_try_feed (grp);
2084 }
2085
2086 void eio_grp_limit (eio_req *grp, int limit)
2087 {
2088   grp->int2 = limit;
2089
2090   grp_try_feed (grp);
2091 }
2092
2093 void eio_grp_add (eio_req *grp, eio_req *req)
2094 {
2095   assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2));
2096
2097   grp->flags |= EIO_FLAG_GROUPADD;
2098
2099   ++grp->size;
2100   req->grp = grp;
2101
2102   req->grp_prev = 0;
2103   req->grp_next = grp->grp_first;
2104
2105   if (grp->grp_first)
2106     grp->grp_first->grp_prev = req;
2107
2108   grp->grp_first = req;
2109 }
2110
2111 /*****************************************************************************/
2112 /* misc garbage */
2113
2114 ssize_t eio_sendfile_sync (int ofd, int ifd, off_t offset, size_t count)
2115 {
2116   etp_worker wrk;
2117   ssize_t ret;
2118
2119   wrk.dbuf = 0;
2120
2121   ret = eio__sendfile (ofd, ifd, offset, count, &wrk);
2122
2123   if (wrk.dbuf)
2124     free (wrk.dbuf);
2125
2126   return ret;
2127 }
2128