ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/libeio/eio.c
Revision: 1.40
Committed: Sat Jun 13 14:58:33 2009 UTC (14 years, 11 months ago) by root
Content type: text/plain
Branch: MAIN
CVS Tags: rel-3_23
Changes since 1.39: +4 -1 lines
Log Message:
3.23

File Contents

# User Rev Content
1 root 1.14 /*
2     * libeio implementation
3     *
4 root 1.31 * Copyright (c) 2007,2008,2009 Marc Alexander Lehmann <libeio@schmorp.de>
5 root 1.14 * 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 root 1.1 #include "eio.h"
41 root 1.29
42     #ifdef EIO_STACKSIZE
43     # define XTHREAD_STACKSIZE EIO_STACKSIZE
44     #endif
45 root 1.1 #include "xthread.h"
46    
47     #include <errno.h>
48     #include <stddef.h>
49     #include <stdlib.h>
50 root 1.3 #include <string.h>
51 root 1.1 #include <errno.h>
52     #include <sys/types.h>
53     #include <sys/stat.h>
54     #include <limits.h>
55     #include <fcntl.h>
56 root 1.8 #include <assert.h>
57 root 1.1
58     #ifndef EIO_FINISH
59     # define EIO_FINISH(req) ((req)->finish) && !EIO_CANCELLED (req) ? (req)->finish (req) : 0
60     #endif
61    
62     #ifndef EIO_DESTROY
63     # define EIO_DESTROY(req) do { if ((req)->destroy) (req)->destroy (req); } while (0)
64     #endif
65    
66     #ifndef EIO_FEED
67     # define EIO_FEED(req) do { if ((req)->feed ) (req)->feed (req); } while (0)
68     #endif
69    
70     #ifdef _WIN32
71    
72     /*doh*/
73     #else
74    
75     # include "config.h"
76     # include <sys/time.h>
77     # include <sys/select.h>
78 root 1.37 # include <sys/mman.h>
79 root 1.1 # include <unistd.h>
80     # include <utime.h>
81     # include <signal.h>
82 root 1.4 # include <dirent.h>
83 root 1.1
84 root 1.31 /* POSIX_SOURCE is useless on bsd's, and XOPEN_SOURCE is unreliable there, too */
85     # if defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__)
86 root 1.33 # define _DIRENT_HAVE_D_TYPE /* sigh */
87 root 1.31 # define D_INO(de) (de)->d_fileno
88 root 1.33 # define D_NAMLEN(de) (de)->d_namlen
89 root 1.31 # elif defined(__linux) || defined(d_ino) || _XOPEN_SOURCE >= 600
90     # define D_INO(de) (de)->d_ino
91     # endif
92    
93 root 1.33 #ifdef _D_EXACT_NAMLEN
94     # undef D_NAMLEN
95     # define D_NAMLEN(de) _D_EXACT_NAMLEN (de)
96     #endif
97    
98 root 1.31 # ifdef _DIRENT_HAVE_D_TYPE
99     # define D_TYPE(de) (de)->d_type
100     # endif
101    
102 root 1.1 # ifndef EIO_STRUCT_DIRENT
103     # define EIO_STRUCT_DIRENT struct dirent
104     # endif
105    
106     #endif
107    
108     #if HAVE_SENDFILE
109     # if __linux
110     # include <sys/sendfile.h>
111     # elif __freebsd
112     # include <sys/socket.h>
113     # include <sys/uio.h>
114     # elif __hpux
115     # include <sys/socket.h>
116     # elif __solaris /* not yet */
117     # include <sys/sendfile.h>
118     # else
119     # error sendfile support requested but not available
120     # endif
121     #endif
122    
123 root 1.31 #ifndef D_TYPE
124     # define D_TYPE(de) 0
125     #endif
126     #ifndef D_INO
127     # define D_INO(de) 0
128     #endif
129 root 1.33 #ifndef D_NAMLEN
130     # define D_NAMLEN(de) strlen ((de)->d_name)
131     #endif
132 root 1.31
133 root 1.1 /* number of seconds after which an idle threads exit */
134     #define IDLE_TIMEOUT 10
135    
136     /* used for struct dirent, AIX doesn't provide it */
137     #ifndef NAME_MAX
138     # define NAME_MAX 4096
139     #endif
140    
141     /* buffer size for various temporary buffers */
142     #define EIO_BUFSIZE 65536
143    
144     #define dBUF \
145     char *eio_buf; \
146 root 1.14 ETP_WORKER_LOCK (self); \
147 root 1.1 self->dbuf = eio_buf = malloc (EIO_BUFSIZE); \
148 root 1.14 ETP_WORKER_UNLOCK (self); \
149 root 1.9 errno = ENOMEM; \
150 root 1.1 if (!eio_buf) \
151     return -1;
152    
153     #define EIO_TICKS ((1000000 + 1023) >> 10)
154    
155 root 1.13 /*****************************************************************************/
156 root 1.1
157 root 1.24 #if __GNUC__ >= 3
158     # define expect(expr,value) __builtin_expect ((expr),(value))
159     #else
160     # define expect(expr,value) (expr)
161     #endif
162    
163     #define expect_false(expr) expect ((expr) != 0, 0)
164     #define expect_true(expr) expect ((expr) != 0, 1)
165    
166     /*****************************************************************************/
167    
168 root 1.14 #define ETP_PRI_MIN EIO_PRI_MIN
169     #define ETP_PRI_MAX EIO_PRI_MAX
170    
171 root 1.15 struct etp_worker;
172    
173 root 1.14 #define ETP_REQ eio_req
174     #define ETP_DESTROY(req) eio_destroy (req)
175     static int eio_finish (eio_req *req);
176     #define ETP_FINISH(req) eio_finish (req)
177 root 1.15 static void eio_execute (struct etp_worker *self, eio_req *req);
178     #define ETP_EXECUTE(wrk,req) eio_execute (wrk,req)
179 root 1.14
180     #define ETP_WORKER_CLEAR(req) \
181     if (wrk->dbuf) \
182     { \
183     free (wrk->dbuf); \
184     wrk->dbuf = 0; \
185     } \
186     \
187     if (wrk->dirp) \
188     { \
189     closedir (wrk->dirp); \
190     wrk->dirp = 0; \
191     }
192 root 1.32
193 root 1.14 #define ETP_WORKER_COMMON \
194     void *dbuf; \
195     DIR *dirp;
196    
197     /*****************************************************************************/
198    
199     #define ETP_NUM_PRI (ETP_PRI_MAX - ETP_PRI_MIN + 1)
200    
201 root 1.1 /* calculcate time difference in ~1/EIO_TICKS of a second */
202     static int tvdiff (struct timeval *tv1, struct timeval *tv2)
203     {
204     return (tv2->tv_sec - tv1->tv_sec ) * EIO_TICKS
205     + ((tv2->tv_usec - tv1->tv_usec) >> 10);
206     }
207    
208 root 1.9 static unsigned int started, idle, wanted = 4;
209 root 1.1
210 root 1.14 static void (*want_poll_cb) (void);
211     static void (*done_poll_cb) (void);
212 root 1.13
213 root 1.14 static unsigned int max_poll_time; /* reslock */
214     static unsigned int max_poll_reqs; /* reslock */
215 root 1.13
216 root 1.14 static volatile unsigned int nreqs; /* reqlock */
217     static volatile unsigned int nready; /* reqlock */
218     static volatile unsigned int npending; /* reqlock */
219 root 1.13 static volatile unsigned int max_idle = 4;
220    
221 root 1.14 static mutex_t wrklock = X_MUTEX_INIT;
222 root 1.13 static mutex_t reslock = X_MUTEX_INIT;
223     static mutex_t reqlock = X_MUTEX_INIT;
224     static cond_t reqwait = X_COND_INIT;
225 root 1.1
226 root 1.22 #if !HAVE_PREADWRITE
227     /*
228     * make our pread/pwrite emulation safe against themselves, but not against
229     * normal read/write by using a mutex. slows down execution a lot,
230     * but that's your problem, not mine.
231     */
232     static mutex_t preadwritelock = X_MUTEX_INIT;
233     #endif
234    
235 root 1.14 typedef struct etp_worker
236 root 1.9 {
237 root 1.14 /* locked by wrklock */
238     struct etp_worker *prev, *next;
239 root 1.1
240     thread_t tid;
241    
242 root 1.14 /* locked by reslock, reqlock or wrklock */
243     ETP_REQ *req; /* currently processed request */
244    
245     ETP_WORKER_COMMON
246     } etp_worker;
247    
248     static etp_worker wrk_first = { &wrk_first, &wrk_first, 0 }; /* NOT etp */
249 root 1.1
250 root 1.14 #define ETP_WORKER_LOCK(wrk) X_LOCK (wrklock)
251     #define ETP_WORKER_UNLOCK(wrk) X_UNLOCK (wrklock)
252 root 1.13
253     /* worker threads management */
254 root 1.1
255 root 1.14 static void etp_worker_clear (etp_worker *wrk)
256 root 1.1 {
257 root 1.14 ETP_WORKER_CLEAR (wrk);
258 root 1.1 }
259    
260 root 1.14 static void etp_worker_free (etp_worker *wrk)
261 root 1.1 {
262     wrk->next->prev = wrk->prev;
263     wrk->prev->next = wrk->next;
264    
265     free (wrk);
266     }
267    
268 root 1.14 static unsigned int etp_nreqs (void)
269 root 1.1 {
270 root 1.14 int retval;
271     if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
272     retval = nreqs;
273     if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
274     return retval;
275 root 1.1 }
276    
277 root 1.14 static unsigned int etp_nready (void)
278 root 1.1 {
279     unsigned int retval;
280    
281     if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
282     retval = nready;
283     if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
284    
285     return retval;
286     }
287    
288 root 1.14 static unsigned int etp_npending (void)
289 root 1.1 {
290     unsigned int retval;
291    
292     if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
293     retval = npending;
294     if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
295    
296     return retval;
297     }
298    
299 root 1.14 static unsigned int etp_nthreads (void)
300 root 1.1 {
301     unsigned int retval;
302    
303     if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
304     retval = started;
305     if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
306    
307     return retval;
308     }
309    
310     /*
311     * a somewhat faster data structure might be nice, but
312     * with 8 priorities this actually needs <20 insns
313     * per shift, the most expensive operation.
314     */
315     typedef struct {
316 root 1.14 ETP_REQ *qs[ETP_NUM_PRI], *qe[ETP_NUM_PRI]; /* qstart, qend */
317 root 1.1 int size;
318 root 1.14 } etp_reqq;
319 root 1.1
320 root 1.14 static etp_reqq req_queue;
321     static etp_reqq res_queue;
322 root 1.1
323 root 1.14 static int reqq_push (etp_reqq *q, ETP_REQ *req)
324 root 1.1 {
325     int pri = req->pri;
326     req->next = 0;
327    
328     if (q->qe[pri])
329     {
330     q->qe[pri]->next = req;
331     q->qe[pri] = req;
332     }
333     else
334     q->qe[pri] = q->qs[pri] = req;
335    
336     return q->size++;
337     }
338    
339 root 1.14 static ETP_REQ *reqq_shift (etp_reqq *q)
340 root 1.1 {
341     int pri;
342    
343     if (!q->size)
344     return 0;
345    
346     --q->size;
347    
348 root 1.14 for (pri = ETP_NUM_PRI; pri--; )
349 root 1.1 {
350     eio_req *req = q->qs[pri];
351    
352     if (req)
353     {
354     if (!(q->qs[pri] = (eio_req *)req->next))
355     q->qe[pri] = 0;
356    
357     return req;
358     }
359     }
360    
361     abort ();
362     }
363    
364 root 1.13 static void etp_atfork_prepare (void)
365     {
366 root 1.14 X_LOCK (wrklock);
367 root 1.13 X_LOCK (reqlock);
368     X_LOCK (reslock);
369     #if !HAVE_PREADWRITE
370     X_LOCK (preadwritelock);
371     #endif
372     }
373    
374     static void etp_atfork_parent (void)
375     {
376     #if !HAVE_PREADWRITE
377     X_UNLOCK (preadwritelock);
378     #endif
379     X_UNLOCK (reslock);
380     X_UNLOCK (reqlock);
381 root 1.14 X_UNLOCK (wrklock);
382 root 1.13 }
383    
384     static void etp_atfork_child (void)
385     {
386 root 1.14 ETP_REQ *prv;
387 root 1.13
388 root 1.21 while ((prv = reqq_shift (&req_queue)))
389 root 1.14 ETP_DESTROY (prv);
390 root 1.13
391 root 1.21 while ((prv = reqq_shift (&res_queue)))
392 root 1.14 ETP_DESTROY (prv);
393 root 1.13
394     while (wrk_first.next != &wrk_first)
395     {
396 root 1.14 etp_worker *wrk = wrk_first.next;
397 root 1.13
398     if (wrk->req)
399 root 1.14 ETP_DESTROY (wrk->req);
400 root 1.13
401 root 1.14 etp_worker_clear (wrk);
402     etp_worker_free (wrk);
403 root 1.13 }
404    
405     started = 0;
406     idle = 0;
407     nreqs = 0;
408     nready = 0;
409     npending = 0;
410    
411     etp_atfork_parent ();
412     }
413    
414     static void
415     etp_once_init (void)
416     {
417     X_THREAD_ATFORK (etp_atfork_prepare, etp_atfork_parent, etp_atfork_child);
418     }
419    
420     static int
421 root 1.14 etp_init (void (*want_poll)(void), void (*done_poll)(void))
422 root 1.13 {
423     static pthread_once_t doinit = PTHREAD_ONCE_INIT;
424    
425     pthread_once (&doinit, etp_once_init);
426    
427 root 1.14 want_poll_cb = want_poll;
428     done_poll_cb = done_poll;
429 root 1.19
430     return 0;
431 root 1.14 }
432    
433     X_THREAD_PROC (etp_proc);
434    
435     static void etp_start_thread (void)
436     {
437     etp_worker *wrk = calloc (1, sizeof (etp_worker));
438    
439     /*TODO*/
440     assert (("unable to allocate worker thread data", wrk));
441    
442     X_LOCK (wrklock);
443    
444     if (thread_create (&wrk->tid, etp_proc, (void *)wrk))
445     {
446     wrk->prev = &wrk_first;
447     wrk->next = wrk_first.next;
448     wrk_first.next->prev = wrk;
449     wrk_first.next = wrk;
450     ++started;
451     }
452     else
453     free (wrk);
454    
455     X_UNLOCK (wrklock);
456     }
457    
458     static void etp_maybe_start_thread (void)
459     {
460 root 1.24 if (expect_true (etp_nthreads () >= wanted))
461 root 1.14 return;
462    
463     /* todo: maybe use idle here, but might be less exact */
464 root 1.24 if (expect_true (0 <= (int)etp_nthreads () + (int)etp_npending () - (int)etp_nreqs ()))
465 root 1.14 return;
466    
467     etp_start_thread ();
468     }
469    
470     static void etp_end_thread (void)
471     {
472     eio_req *req = calloc (1, sizeof (eio_req));
473    
474     req->type = -1;
475     req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
476    
477     X_LOCK (reqlock);
478     reqq_push (&req_queue, req);
479     X_COND_SIGNAL (reqwait);
480     X_UNLOCK (reqlock);
481    
482     X_LOCK (wrklock);
483     --started;
484     X_UNLOCK (wrklock);
485     }
486    
487     static int etp_poll (void)
488     {
489     unsigned int maxreqs;
490     unsigned int maxtime;
491     struct timeval tv_start, tv_now;
492    
493     X_LOCK (reslock);
494     maxreqs = max_poll_reqs;
495     maxtime = max_poll_time;
496     X_UNLOCK (reslock);
497    
498     if (maxtime)
499     gettimeofday (&tv_start, 0);
500    
501     for (;;)
502     {
503     ETP_REQ *req;
504    
505     etp_maybe_start_thread ();
506    
507     X_LOCK (reslock);
508     req = reqq_shift (&res_queue);
509    
510     if (req)
511     {
512     --npending;
513    
514     if (!res_queue.size && done_poll_cb)
515     done_poll_cb ();
516     }
517    
518     X_UNLOCK (reslock);
519    
520     if (!req)
521     return 0;
522    
523     X_LOCK (reqlock);
524     --nreqs;
525     X_UNLOCK (reqlock);
526    
527 root 1.24 if (expect_false (req->type == EIO_GROUP && req->size))
528 root 1.14 {
529     req->int1 = 1; /* mark request as delayed */
530     continue;
531     }
532     else
533     {
534     int res = ETP_FINISH (req);
535 root 1.24 if (expect_false (res))
536 root 1.14 return res;
537     }
538    
539 root 1.24 if (expect_false (maxreqs && !--maxreqs))
540 root 1.14 break;
541    
542     if (maxtime)
543     {
544     gettimeofday (&tv_now, 0);
545    
546     if (tvdiff (&tv_start, &tv_now) >= maxtime)
547     break;
548     }
549     }
550    
551     errno = EAGAIN;
552     return -1;
553     }
554    
555     static void etp_cancel (ETP_REQ *req)
556     {
557     X_LOCK (wrklock);
558     req->flags |= EIO_FLAG_CANCELLED;
559     X_UNLOCK (wrklock);
560    
561     eio_grp_cancel (req);
562     }
563    
564     static void etp_submit (ETP_REQ *req)
565     {
566     req->pri -= ETP_PRI_MIN;
567    
568 root 1.24 if (expect_false (req->pri < ETP_PRI_MIN - ETP_PRI_MIN)) req->pri = ETP_PRI_MIN - ETP_PRI_MIN;
569     if (expect_false (req->pri > ETP_PRI_MAX - ETP_PRI_MIN)) req->pri = ETP_PRI_MAX - ETP_PRI_MIN;
570    
571     if (expect_false (req->type == EIO_GROUP))
572     {
573     /* I hope this is worth it :/ */
574     X_LOCK (reqlock);
575     ++nreqs;
576     X_UNLOCK (reqlock);
577    
578     X_LOCK (reslock);
579    
580     ++npending;
581    
582     if (!reqq_push (&res_queue, req) && want_poll_cb)
583     want_poll_cb ();
584 root 1.14
585 root 1.24 X_UNLOCK (reslock);
586     }
587     else
588     {
589     X_LOCK (reqlock);
590     ++nreqs;
591     ++nready;
592     reqq_push (&req_queue, req);
593     X_COND_SIGNAL (reqwait);
594     X_UNLOCK (reqlock);
595 root 1.14
596 root 1.24 etp_maybe_start_thread ();
597     }
598 root 1.14 }
599    
600     static void etp_set_max_poll_time (double nseconds)
601     {
602     if (WORDACCESS_UNSAFE) X_LOCK (reslock);
603     max_poll_time = nseconds;
604     if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
605     }
606    
607     static void etp_set_max_poll_reqs (unsigned int maxreqs)
608     {
609     if (WORDACCESS_UNSAFE) X_LOCK (reslock);
610     max_poll_reqs = maxreqs;
611     if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
612     }
613 root 1.13
614 root 1.14 static void etp_set_max_idle (unsigned int nthreads)
615     {
616     if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
617     max_idle = nthreads <= 0 ? 1 : nthreads;
618     if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
619 root 1.13 }
620    
621 root 1.14 static void etp_set_min_parallel (unsigned int nthreads)
622     {
623     if (wanted < nthreads)
624     wanted = nthreads;
625     }
626    
627     static void etp_set_max_parallel (unsigned int nthreads)
628     {
629     if (wanted > nthreads)
630     wanted = nthreads;
631    
632     while (started > wanted)
633     etp_end_thread ();
634     }
635 root 1.13
636     /*****************************************************************************/
637    
638 root 1.2 static void grp_try_feed (eio_req *grp)
639 root 1.1 {
640     while (grp->size < grp->int2 && !EIO_CANCELLED (grp))
641     {
642 root 1.26 grp->flags &= ~EIO_FLAG_GROUPADD;
643 root 1.1
644     EIO_FEED (grp);
645    
646     /* stop if no progress has been made */
647 root 1.26 if (!(grp->flags & EIO_FLAG_GROUPADD))
648 root 1.1 {
649     grp->feed = 0;
650 root 1.2 break;
651 root 1.1 }
652     }
653     }
654    
655     static int grp_dec (eio_req *grp)
656     {
657     --grp->size;
658    
659     /* call feeder, if applicable */
660 root 1.2 grp_try_feed (grp);
661 root 1.1
662     /* finish, if done */
663     if (!grp->size && grp->int1)
664 root 1.2 return eio_finish (grp);
665 root 1.1 else
666     return 0;
667     }
668    
669     void eio_destroy (eio_req *req)
670     {
671 root 1.6 if ((req)->flags & EIO_FLAG_PTR1_FREE) free (req->ptr1);
672     if ((req)->flags & EIO_FLAG_PTR2_FREE) free (req->ptr2);
673 root 1.1
674     EIO_DESTROY (req);
675     }
676    
677 root 1.2 static int eio_finish (eio_req *req)
678 root 1.1 {
679     int res = EIO_FINISH (req);
680    
681     if (req->grp)
682     {
683     int res2;
684     eio_req *grp = req->grp;
685    
686     /* unlink request */
687     if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
688     if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
689    
690     if (grp->grp_first == req)
691     grp->grp_first = req->grp_next;
692    
693     res2 = grp_dec (grp);
694    
695     if (!res && res2)
696     res = res2;
697     }
698    
699     eio_destroy (req);
700    
701     return res;
702     }
703    
704     void eio_grp_cancel (eio_req *grp)
705     {
706     for (grp = grp->grp_first; grp; grp = grp->grp_next)
707     eio_cancel (grp);
708     }
709    
710     void eio_cancel (eio_req *req)
711     {
712 root 1.14 etp_cancel (req);
713     }
714 root 1.1
715 root 1.14 void eio_submit (eio_req *req)
716     {
717     etp_submit (req);
718 root 1.1 }
719    
720 root 1.14 unsigned int eio_nreqs (void)
721 root 1.1 {
722 root 1.14 return etp_nreqs ();
723 root 1.1 }
724    
725 root 1.14 unsigned int eio_nready (void)
726 root 1.1 {
727 root 1.14 return etp_nready ();
728 root 1.1 }
729    
730 root 1.14 unsigned int eio_npending (void)
731 root 1.1 {
732 root 1.14 return etp_npending ();
733 root 1.1 }
734    
735 root 1.14 unsigned int eio_nthreads (void)
736 root 1.1 {
737 root 1.14 return etp_nthreads ();
738 root 1.1 }
739    
740     void eio_set_max_poll_time (double nseconds)
741     {
742 root 1.14 etp_set_max_poll_time (nseconds);
743 root 1.1 }
744    
745     void eio_set_max_poll_reqs (unsigned int maxreqs)
746     {
747 root 1.14 etp_set_max_poll_reqs (maxreqs);
748 root 1.1 }
749    
750     void eio_set_max_idle (unsigned int nthreads)
751     {
752 root 1.14 etp_set_max_idle (nthreads);
753 root 1.1 }
754    
755     void eio_set_min_parallel (unsigned int nthreads)
756     {
757 root 1.14 etp_set_min_parallel (nthreads);
758 root 1.1 }
759    
760     void eio_set_max_parallel (unsigned int nthreads)
761     {
762 root 1.14 etp_set_max_parallel (nthreads);
763 root 1.1 }
764    
765     int eio_poll (void)
766     {
767 root 1.14 return etp_poll ();
768 root 1.1 }
769    
770     /*****************************************************************************/
771     /* work around various missing functions */
772    
773     #if !HAVE_PREADWRITE
774 root 1.27 # undef pread
775     # undef pwrite
776 root 1.9 # define pread eio__pread
777     # define pwrite eio__pwrite
778 root 1.1
779 root 1.9 static ssize_t
780     eio__pread (int fd, void *buf, size_t count, off_t offset)
781 root 1.1 {
782     ssize_t res;
783     off_t ooffset;
784    
785     X_LOCK (preadwritelock);
786     ooffset = lseek (fd, 0, SEEK_CUR);
787     lseek (fd, offset, SEEK_SET);
788     res = read (fd, buf, count);
789     lseek (fd, ooffset, SEEK_SET);
790     X_UNLOCK (preadwritelock);
791    
792     return res;
793     }
794    
795 root 1.9 static ssize_t
796     eio__pwrite (int fd, void *buf, size_t count, off_t offset)
797 root 1.1 {
798     ssize_t res;
799     off_t ooffset;
800    
801     X_LOCK (preadwritelock);
802     ooffset = lseek (fd, 0, SEEK_CUR);
803     lseek (fd, offset, SEEK_SET);
804     res = write (fd, buf, count);
805 root 1.30 lseek (fd, ooffset, SEEK_SET);
806 root 1.1 X_UNLOCK (preadwritelock);
807    
808     return res;
809     }
810     #endif
811    
812     #ifndef HAVE_FUTIMES
813    
814 root 1.27 # undef utimes
815     # undef futimes
816 root 1.9 # define utimes(path,times) eio__utimes (path, times)
817     # define futimes(fd,times) eio__futimes (fd, times)
818 root 1.1
819 root 1.9 static int
820     eio__utimes (const char *filename, const struct timeval times[2])
821 root 1.1 {
822     if (times)
823     {
824     struct utimbuf buf;
825    
826     buf.actime = times[0].tv_sec;
827     buf.modtime = times[1].tv_sec;
828    
829     return utime (filename, &buf);
830     }
831     else
832     return utime (filename, 0);
833     }
834    
835 root 1.9 static int eio__futimes (int fd, const struct timeval tv[2])
836 root 1.1 {
837     errno = ENOSYS;
838     return -1;
839     }
840    
841     #endif
842    
843     #if !HAVE_FDATASYNC
844 root 1.27 # undef fdatasync
845     # define fdatasync(fd) fsync (fd)
846 root 1.1 #endif
847    
848 root 1.27 /* sync_file_range always needs emulation */
849     int
850     eio__sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags)
851     {
852     #if HAVE_SYNC_FILE_RANGE
853     int res;
854    
855     if (EIO_SYNC_FILE_RANGE_WAIT_BEFORE != SYNC_FILE_RANGE_WAIT_BEFORE
856     || EIO_SYNC_FILE_RANGE_WRITE != SYNC_FILE_RANGE_WRITE
857     || EIO_SYNC_FILE_RANGE_WAIT_AFTER != SYNC_FILE_RANGE_WAIT_AFTER)
858     {
859     flags = 0
860     | (flags & EIO_SYNC_FILE_RANGE_WAIT_BEFORE ? SYNC_FILE_RANGE_WAIT_BEFORE : 0)
861     | (flags & EIO_SYNC_FILE_RANGE_WRITE ? SYNC_FILE_RANGE_WRITE : 0)
862     | (flags & EIO_SYNC_FILE_RANGE_WAIT_AFTER ? SYNC_FILE_RANGE_WAIT_AFTER : 0);
863     }
864    
865     res = sync_file_range (fd, offset, nbytes, flags);
866    
867 root 1.28 if (!res || errno != ENOSYS)
868 root 1.27 return res;
869     #endif
870    
871     /* even though we could play tricks with the flags, it's better to always
872     * call fdatasync, as thta matches the expectation of it's users best */
873     return fdatasync (fd);
874     }
875    
876 root 1.1 #if !HAVE_READAHEAD
877 root 1.27 # undef readahead
878 root 1.9 # define readahead(fd,offset,count) eio__readahead (fd, offset, count, self)
879 root 1.1
880 root 1.9 static ssize_t
881 root 1.17 eio__readahead (int fd, off_t offset, size_t count, etp_worker *self)
882 root 1.1 {
883     size_t todo = count;
884     dBUF;
885    
886     while (todo > 0)
887     {
888     size_t len = todo < EIO_BUFSIZE ? todo : EIO_BUFSIZE;
889    
890 root 1.3 pread (fd, eio_buf, len, offset);
891 root 1.1 offset += len;
892     todo -= len;
893     }
894    
895     errno = 0;
896     return count;
897     }
898    
899     #endif
900    
901     /* sendfile always needs emulation */
902 root 1.9 static ssize_t
903 root 1.14 eio__sendfile (int ofd, int ifd, off_t offset, size_t count, etp_worker *self)
904 root 1.1 {
905     ssize_t res;
906    
907     if (!count)
908     return 0;
909    
910     #if HAVE_SENDFILE
911     # if __linux
912     res = sendfile (ofd, ifd, &offset, count);
913    
914     # elif __freebsd
915     /*
916     * Of course, the freebsd sendfile is a dire hack with no thoughts
917     * wasted on making it similar to other I/O functions.
918     */
919     {
920     off_t sbytes;
921     res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
922    
923     if (res < 0 && sbytes)
924     /* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
925     res = sbytes;
926     }
927    
928     # elif __hpux
929     res = sendfile (ofd, ifd, offset, count, 0, 0);
930    
931     # elif __solaris
932     {
933     struct sendfilevec vec;
934     size_t sbytes;
935    
936     vec.sfv_fd = ifd;
937     vec.sfv_flag = 0;
938     vec.sfv_off = offset;
939     vec.sfv_len = count;
940    
941     res = sendfilev (ofd, &vec, 1, &sbytes);
942    
943     if (res < 0 && sbytes)
944     res = sbytes;
945     }
946    
947     # endif
948     #else
949     res = -1;
950     errno = ENOSYS;
951     #endif
952    
953     if (res < 0
954     && (errno == ENOSYS || errno == EINVAL || errno == ENOTSOCK
955     #if __solaris
956     || errno == EAFNOSUPPORT || errno == EPROTOTYPE
957     #endif
958     )
959     )
960     {
961     /* emulate sendfile. this is a major pain in the ass */
962     dBUF;
963    
964     res = 0;
965    
966     while (count)
967     {
968     ssize_t cnt;
969    
970     cnt = pread (ifd, eio_buf, count > EIO_BUFSIZE ? EIO_BUFSIZE : count, offset);
971    
972     if (cnt <= 0)
973     {
974     if (cnt && !res) res = -1;
975     break;
976     }
977    
978     cnt = write (ofd, eio_buf, cnt);
979    
980     if (cnt <= 0)
981     {
982     if (cnt && !res) res = -1;
983     break;
984     }
985    
986     offset += cnt;
987     res += cnt;
988     count -= cnt;
989     }
990     }
991    
992     return res;
993     }
994    
995 root 1.37 static signed char
996     eio_dent_cmp (const eio_dirent *a, const eio_dirent *b)
997     {
998 root 1.39 return b->score - a->score ? b->score - a->score /* works because our signed char is always 0..100 */
999     : a->inode < b->inode ? -1 : a->inode > b->inode ? 1 : 0;
1000 root 1.37 }
1001    
1002 root 1.39 #define EIO_DENT_CMP(i,op,j) eio_dent_cmp (&i, &j) op 0
1003    
1004     #define EIO_QSORT_CUTOFF 30 /* quite high, but performs well on many filesystems */
1005     #define EIO_QSORT_SKIP 60 /* when to skip qsort completely */
1006 root 1.37
1007     static void
1008     eio_dent_sort (eio_dirent *dents, int size)
1009 root 1.31 {
1010 root 1.37 int i, j;
1011    
1012 root 1.39 if (size <= 1)
1013     return; /* our insertion sort relies on size > 0 */
1014    
1015     if (size > EIO_QSORT_SKIP) /* skip quicksort for small directories */
1016 root 1.37 {
1017 root 1.38 /* first, use quicksort */
1018     /* should be good for 2**31 entries */
1019     struct rng { int l, r; } rng [32];
1020    
1021     i = 0;
1022     rng[0].l = 0;
1023     rng[0].r = size;
1024 root 1.37
1025 root 1.38 while (expect_true (i >= 0))
1026 root 1.37 {
1027 root 1.38 int L = rng [i].l;
1028     int R = rng [i].r - 1;
1029 root 1.37
1030 root 1.38 if (expect_false (L + EIO_QSORT_CUTOFF < R))
1031 root 1.37 {
1032 root 1.38 eio_dirent piv = dents [L];
1033    
1034     while (L < R)
1035     {
1036 root 1.39 while (EIO_DENT_CMP (dents [R], >=, piv) && L < R)
1037 root 1.38 --R;
1038 root 1.37
1039 root 1.38 if (L < R)
1040     dents [L++] = dents [R];
1041 root 1.37
1042 root 1.39 while (EIO_DENT_CMP (dents [L], <=, piv) && L < R)
1043 root 1.38 ++L;
1044 root 1.37
1045 root 1.38 if (L < R)
1046     dents [R--] = dents [L];
1047     }
1048 root 1.31
1049 root 1.38 dents [L] = piv;
1050 root 1.37
1051 root 1.38 ++i;
1052     rng [i].l = L + 1;
1053     rng [i].r = rng [i - 1].r;
1054     rng [i - 1].r = L;
1055 root 1.37
1056 root 1.38 if (rng [i].r - rng [i].l > rng [i - 1].r - rng [i - 1].l)
1057     {
1058     struct rng t;
1059 root 1.37
1060 root 1.38 t = rng [i]; rng [i] = rng [i - 1]; rng [i - 1] = t;
1061     }
1062 root 1.37 }
1063 root 1.38 else
1064     --i;
1065 root 1.37 }
1066     }
1067    
1068 root 1.39 /* use an insertion sort after qsort, or for small arrays */
1069     /* first move the smallest element to the front, to act as a sentinel */
1070     {
1071     int min = 0;
1072    
1073 root 1.40 for (i = size > EIO_QSORT_SKIP ? EIO_QSORT_CUTOFF + 1 : size; --i; )
1074 root 1.39 if (EIO_DENT_CMP (dents [i], <, dents [min]))
1075     min = i;
1076    
1077     /* swap elements 0 and j (minimum) */
1078     {
1079     eio_dirent tmp = dents [0]; dents [0] = dents [min]; dents [min] = tmp;
1080     }
1081     }
1082    
1083     /* then do standard insertion sort */
1084 root 1.37 for (i = 1; i < size; ++i)
1085     {
1086     eio_dirent value = dents [i];
1087    
1088 root 1.39 for (j = i - 1; EIO_DENT_CMP (dents [j], >, value); --j)
1089 root 1.40 {
1090     assert (j >= 0);
1091 root 1.37 dents [j + 1] = dents [j];
1092 root 1.40 }
1093 root 1.37
1094     dents [j + 1] = value;
1095     }
1096 root 1.31 }
1097    
1098 root 1.1 /* read a full directory */
1099 root 1.9 static void
1100 root 1.14 eio__scandir (eio_req *req, etp_worker *self)
1101 root 1.1 {
1102     DIR *dirp;
1103     EIO_STRUCT_DIRENT *entp;
1104 root 1.37 char *name, *names;
1105 root 1.31 int namesalloc = 4096;
1106     int namesoffs = 0;
1107     int flags = req->int1;
1108     eio_dirent *dents = 0;
1109     int dentalloc = 128;
1110     int dentoffs = 0;
1111    
1112     req->result = -1;
1113    
1114     if (!(flags & EIO_READDIR_DENTS))
1115     flags &= ~(EIO_READDIR_DIRS_FIRST | EIO_READDIR_STAT_ORDER);
1116 root 1.1
1117 root 1.14 X_LOCK (wrklock);
1118 root 1.20 /* the corresponding closedir is in ETP_WORKER_CLEAR */
1119 root 1.1 self->dirp = dirp = opendir (req->ptr1);
1120 root 1.31 req->flags |= EIO_FLAG_PTR1_FREE | EIO_FLAG_PTR2_FREE;
1121     req->ptr1 = names = malloc (namesalloc);
1122     req->ptr2 = dents = flags ? malloc (dentalloc * sizeof (eio_dirent)) : 0;
1123 root 1.14 X_UNLOCK (wrklock);
1124 root 1.1
1125 root 1.31 if (dirp && names && (!flags || dents))
1126 root 1.1 for (;;)
1127     {
1128     errno = 0;
1129 root 1.16 entp = readdir (dirp);
1130 root 1.1
1131     if (!entp)
1132 root 1.31 {
1133     if (errno)
1134     break;
1135    
1136     /* sort etc. */
1137     req->int1 = flags;
1138     req->result = dentoffs;
1139    
1140 root 1.32 if (dents)
1141     {
1142     eio_dirent *ent = dents + dentoffs;
1143    
1144     while (ent > dents)
1145     (--ent)->name = names + (size_t)ent->name;
1146     }
1147    
1148     if (flags & EIO_READDIR_STAT_ORDER
1149     || !(~flags & (EIO_READDIR_DIRS_FIRST | EIO_READDIR_FOUND_UNKNOWN)))
1150 root 1.37 eio_dent_sort (dents, dentoffs); /* score depends of DIRS_FIRST */
1151 root 1.32 else if (flags & EIO_READDIR_DIRS_FIRST)
1152 root 1.31 {
1153     /* in this case, all is known, and we just put dirs first and sort them */
1154     eio_dirent *ent = dents + dentoffs;
1155     eio_dirent *dir = dents;
1156    
1157 root 1.35 /* now move dirs to the front, and non-dirs to the back */
1158 root 1.37 /* by walking from both sides and swapping if necessary */
1159 root 1.31 while (ent > dir)
1160     {
1161     if (dir->type == DT_DIR)
1162     ++dir;
1163     else
1164     {
1165     --ent;
1166    
1167     if (ent->type == DT_DIR)
1168     {
1169     eio_dirent tmp = *dir;
1170     *dir = *ent;
1171     *ent = tmp;
1172    
1173     ++dir;
1174     }
1175     }
1176     }
1177    
1178     /* now sort the dirs only */
1179 root 1.37 eio_dent_sort (dents, dir - dents);
1180 root 1.31 }
1181    
1182 root 1.32 /* only provide the names array unless DENTS is specified */
1183     if (!(flags & EIO_READDIR_DENTS))
1184     {
1185     X_LOCK (wrklock);
1186     assert (!dents);
1187     req->ptr1 = 0;
1188     req->ptr2 = names;
1189     X_UNLOCK (wrklock);
1190     }
1191 root 1.31
1192     break;
1193     }
1194 root 1.1
1195 root 1.32 /* now add the entry to our list(s) */
1196 root 1.1 name = entp->d_name;
1197    
1198 root 1.32 /* skip . and .. entries */
1199 root 1.1 if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
1200     {
1201 root 1.33 int len = D_NAMLEN (entp) + 1;
1202 root 1.1
1203 root 1.31 while (expect_false (namesoffs + len > namesalloc))
1204 root 1.1 {
1205 root 1.31 namesalloc *= 2;
1206 root 1.14 X_LOCK (wrklock);
1207 root 1.31 req->ptr1 = names = realloc (names, namesalloc);
1208 root 1.14 X_UNLOCK (wrklock);
1209 root 1.1
1210     if (!names)
1211     break;
1212     }
1213    
1214 root 1.31 memcpy (names + namesoffs, name, len);
1215    
1216     if (dents)
1217     {
1218     struct eio_dirent *ent;
1219    
1220     if (expect_false (dentoffs == dentalloc))
1221     {
1222     dentalloc *= 2;
1223     X_LOCK (wrklock);
1224     req->ptr2 = dents = realloc (dents, dentalloc * sizeof (eio_dirent));
1225     X_UNLOCK (wrklock);
1226    
1227     if (!dents)
1228     break;
1229     }
1230    
1231     ent = dents + dentoffs;
1232    
1233 root 1.32 ent->name = (char *)(size_t)namesoffs; /* rather dirtily we store the offset in the pointer */
1234 root 1.31 ent->namelen = len - 1;
1235     ent->inode = D_INO (entp);
1236    
1237     switch (D_TYPE (entp))
1238     {
1239     default:
1240     ent->type = EIO_DT_UNKNOWN;
1241     flags |= EIO_READDIR_FOUND_UNKNOWN;
1242     break;
1243    
1244     #ifdef DT_FIFO
1245     case DT_FIFO: ent->type = EIO_DT_FIFO; break;
1246     #endif
1247     #ifdef DT_CHR
1248     case DT_CHR: ent->type = EIO_DT_CHR; break;
1249     #endif
1250 root 1.33 #ifdef DT_MPC
1251     case DT_MPC: ent->type = EIO_DT_MPC; break;
1252     #endif
1253 root 1.32 #ifdef DT_DIR
1254 root 1.31 case DT_DIR: ent->type = EIO_DT_DIR; break;
1255     #endif
1256 root 1.33 #ifdef DT_NAM
1257     case DT_NAM: ent->type = EIO_DT_NAM; break;
1258     #endif
1259 root 1.32 #ifdef DT_BLK
1260 root 1.31 case DT_BLK: ent->type = EIO_DT_BLK; break;
1261     #endif
1262 root 1.33 #ifdef DT_MPB
1263     case DT_MPB: ent->type = EIO_DT_MPB; break;
1264     #endif
1265 root 1.32 #ifdef DT_REG
1266 root 1.31 case DT_REG: ent->type = EIO_DT_REG; break;
1267     #endif
1268 root 1.33 #ifdef DT_NWK
1269     case DT_NWK: ent->type = EIO_DT_NWK; break;
1270     #endif
1271     #ifdef DT_CMP
1272     case DT_CMP: ent->type = EIO_DT_CMP; break;
1273     #endif
1274 root 1.32 #ifdef DT_LNK
1275 root 1.31 case DT_LNK: ent->type = EIO_DT_LNK; break;
1276     #endif
1277     #ifdef DT_SOCK
1278     case DT_SOCK: ent->type = EIO_DT_SOCK; break;
1279     #endif
1280 root 1.33 #ifdef DT_DOOR
1281     case DT_DOOR: ent->type = EIO_DT_DOOR; break;
1282     #endif
1283 root 1.31 #ifdef DT_WHT
1284     case DT_WHT: ent->type = EIO_DT_WHT; break;
1285     #endif
1286     }
1287    
1288     ent->score = 0;
1289    
1290     if (flags & EIO_READDIR_DIRS_FIRST)
1291     {
1292     if (ent->type == EIO_DT_UNKNOWN)
1293     {
1294     if (*name == '.') /* leading dots are likely directories, and, in any case, rare */
1295     ent->score = 98;
1296     else if (!strchr (name, '.')) /* absense of dots indicate likely dirs */
1297 root 1.34 ent->score = len <= 2 ? len + 6 : len <= 4 ? 5 : len <= 7 ? 4 : 1; /* shorter == more likely dir, but avoid too many classes */
1298 root 1.31 }
1299 root 1.33 else if (ent->type == EIO_DT_DIR)
1300 root 1.31 ent->score = 100;
1301     }
1302     }
1303    
1304     namesoffs += len;
1305     ++dentoffs;
1306 root 1.1 }
1307 root 1.36
1308     if (EIO_CANCELLED (req))
1309     {
1310     errno = ECANCELED;
1311     break;
1312     }
1313 root 1.1 }
1314     }
1315    
1316 root 1.23 #if !(_POSIX_MAPPED_FILES && _POSIX_SYNCHRONIZED_IO)
1317 root 1.27 # undef msync
1318 root 1.28 # define msync(a,b,c) ((errno = ENOSYS), -1)
1319 root 1.23 #endif
1320    
1321     int
1322     eio__mtouch (void *mem, size_t len, int flags)
1323     {
1324     intptr_t addr = (intptr_t)mem;
1325     intptr_t end = addr + len;
1326     #ifdef PAGESIZE
1327     const intptr_t page = PAGESIZE;
1328     #else
1329     static intptr_t page;
1330    
1331     if (!page)
1332     page = sysconf (_SC_PAGESIZE);
1333     #endif
1334    
1335     addr &= ~(page - 1); /* assume page size is always a power of two */
1336    
1337     if (addr < end)
1338     if (flags) /* modify */
1339     do { *((volatile sig_atomic_t *)addr) |= 0; } while ((addr += page) < len);
1340     else
1341     do { *((volatile sig_atomic_t *)addr) ; } while ((addr += page) < len);
1342    
1343     return 0;
1344     }
1345    
1346 root 1.1 /*****************************************************************************/
1347    
1348 root 1.6 #define ALLOC(len) \
1349     if (!req->ptr2) \
1350     { \
1351 root 1.14 X_LOCK (wrklock); \
1352 root 1.7 req->flags |= EIO_FLAG_PTR2_FREE; \
1353 root 1.14 X_UNLOCK (wrklock); \
1354 root 1.7 req->ptr2 = malloc (len); \
1355     if (!req->ptr2) \
1356     { \
1357     errno = ENOMEM; \
1358     req->result = -1; \
1359     break; \
1360     } \
1361 root 1.6 }
1362    
1363 root 1.14 X_THREAD_PROC (etp_proc)
1364 root 1.1 {
1365 root 1.14 ETP_REQ *req;
1366 root 1.1 struct timespec ts;
1367 root 1.14 etp_worker *self = (etp_worker *)thr_arg;
1368 root 1.1
1369     /* try to distribute timeouts somewhat randomly */
1370     ts.tv_nsec = ((unsigned long)self & 1023UL) * (1000000000UL / 1024UL);
1371    
1372     for (;;)
1373     {
1374     X_LOCK (reqlock);
1375    
1376     for (;;)
1377     {
1378     self->req = req = reqq_shift (&req_queue);
1379    
1380     if (req)
1381     break;
1382    
1383     ++idle;
1384    
1385 root 1.14 ts.tv_sec = time (0) + IDLE_TIMEOUT;
1386 root 1.1 if (X_COND_TIMEDWAIT (reqwait, reqlock, ts) == ETIMEDOUT)
1387     {
1388     if (idle > max_idle)
1389     {
1390     --idle;
1391     X_UNLOCK (reqlock);
1392 root 1.14 X_LOCK (wrklock);
1393 root 1.1 --started;
1394 root 1.14 X_UNLOCK (wrklock);
1395 root 1.1 goto quit;
1396     }
1397    
1398     /* we are allowed to idle, so do so without any timeout */
1399     X_COND_WAIT (reqwait, reqlock);
1400     }
1401    
1402     --idle;
1403     }
1404    
1405     --nready;
1406    
1407     X_UNLOCK (reqlock);
1408    
1409 root 1.15 if (req->type < 0)
1410     goto quit;
1411 root 1.1
1412     if (!EIO_CANCELLED (req))
1413 root 1.15 ETP_EXECUTE (self, req);
1414 root 1.1
1415     X_LOCK (reslock);
1416    
1417     ++npending;
1418    
1419 root 1.14 if (!reqq_push (&res_queue, req) && want_poll_cb)
1420     want_poll_cb ();
1421 root 1.1
1422     self->req = 0;
1423 root 1.14 etp_worker_clear (self);
1424 root 1.1
1425     X_UNLOCK (reslock);
1426     }
1427    
1428     quit:
1429 root 1.14 X_LOCK (wrklock);
1430     etp_worker_free (self);
1431     X_UNLOCK (wrklock);
1432 root 1.1
1433     return 0;
1434     }
1435    
1436     /*****************************************************************************/
1437    
1438     int eio_init (void (*want_poll)(void), void (*done_poll)(void))
1439     {
1440 root 1.19 return etp_init (want_poll, done_poll);
1441 root 1.1 }
1442    
1443 root 1.6 static void eio_api_destroy (eio_req *req)
1444     {
1445     free (req);
1446     }
1447    
1448     #define REQ(rtype) \
1449     eio_req *req; \
1450     \
1451     req = (eio_req *)calloc (1, sizeof *req); \
1452     if (!req) \
1453     return 0; \
1454     \
1455 root 1.11 req->type = rtype; \
1456     req->pri = pri; \
1457     req->finish = cb; \
1458     req->data = data; \
1459 root 1.6 req->destroy = eio_api_destroy;
1460    
1461     #define SEND eio_submit (req); return req
1462    
1463 root 1.9 #define PATH \
1464     req->flags |= EIO_FLAG_PTR1_FREE; \
1465     req->ptr1 = strdup (path); \
1466     if (!req->ptr1) \
1467     { \
1468     eio_api_destroy (req); \
1469     return 0; \
1470     }
1471    
1472 root 1.15 static void eio_execute (etp_worker *self, eio_req *req)
1473     {
1474     errno = 0;
1475    
1476     switch (req->type)
1477     {
1478     case EIO_READ: ALLOC (req->size);
1479     req->result = req->offs >= 0
1480     ? pread (req->int1, req->ptr2, req->size, req->offs)
1481     : read (req->int1, req->ptr2, req->size); break;
1482     case EIO_WRITE: req->result = req->offs >= 0
1483     ? pwrite (req->int1, req->ptr2, req->size, req->offs)
1484     : write (req->int1, req->ptr2, req->size); break;
1485    
1486     case EIO_READAHEAD: req->result = readahead (req->int1, req->offs, req->size); break;
1487     case EIO_SENDFILE: req->result = eio__sendfile (req->int1, req->int2, req->offs, req->size, self); break;
1488    
1489     case EIO_STAT: ALLOC (sizeof (EIO_STRUCT_STAT));
1490     req->result = stat (req->ptr1, (EIO_STRUCT_STAT *)req->ptr2); break;
1491     case EIO_LSTAT: ALLOC (sizeof (EIO_STRUCT_STAT));
1492     req->result = lstat (req->ptr1, (EIO_STRUCT_STAT *)req->ptr2); break;
1493     case EIO_FSTAT: ALLOC (sizeof (EIO_STRUCT_STAT));
1494     req->result = fstat (req->int1, (EIO_STRUCT_STAT *)req->ptr2); break;
1495    
1496     case EIO_CHOWN: req->result = chown (req->ptr1, req->int2, req->int3); break;
1497     case EIO_FCHOWN: req->result = fchown (req->int1, req->int2, req->int3); break;
1498     case EIO_CHMOD: req->result = chmod (req->ptr1, (mode_t)req->int2); break;
1499     case EIO_FCHMOD: req->result = fchmod (req->int1, (mode_t)req->int2); break;
1500     case EIO_TRUNCATE: req->result = truncate (req->ptr1, req->offs); break;
1501     case EIO_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break;
1502    
1503     case EIO_OPEN: req->result = open (req->ptr1, req->int1, (mode_t)req->int2); break;
1504     case EIO_CLOSE: req->result = close (req->int1); break;
1505     case EIO_DUP2: req->result = dup2 (req->int1, req->int2); break;
1506     case EIO_UNLINK: req->result = unlink (req->ptr1); break;
1507     case EIO_RMDIR: req->result = rmdir (req->ptr1); break;
1508     case EIO_MKDIR: req->result = mkdir (req->ptr1, (mode_t)req->int2); break;
1509     case EIO_RENAME: req->result = rename (req->ptr1, req->ptr2); break;
1510     case EIO_LINK: req->result = link (req->ptr1, req->ptr2); break;
1511     case EIO_SYMLINK: req->result = symlink (req->ptr1, req->ptr2); break;
1512 root 1.18 case EIO_MKNOD: req->result = mknod (req->ptr1, (mode_t)req->int2, (dev_t)req->int3); break;
1513 root 1.15
1514     case EIO_READLINK: ALLOC (NAME_MAX);
1515     req->result = readlink (req->ptr1, req->ptr2, NAME_MAX); break;
1516    
1517     case EIO_SYNC: req->result = 0; sync (); break;
1518     case EIO_FSYNC: req->result = fsync (req->int1); break;
1519     case EIO_FDATASYNC: req->result = fdatasync (req->int1); break;
1520 root 1.23 case EIO_MSYNC: req->result = msync (req->ptr2, req->size, req->int1); break;
1521     case EIO_MTOUCH: req->result = eio__mtouch (req->ptr2, req->size, req->int1); break;
1522 root 1.27 case EIO_SYNC_FILE_RANGE: req->result = eio__sync_file_range (req->int1, req->offs, req->size, req->int2); break;
1523 root 1.15
1524     case EIO_READDIR: eio__scandir (req, self); break;
1525    
1526     case EIO_BUSY:
1527     #ifdef _WIN32
1528     Sleep (req->nv1 * 1000.);
1529     #else
1530     {
1531     struct timeval tv;
1532    
1533     tv.tv_sec = req->nv1;
1534     tv.tv_usec = (req->nv1 - tv.tv_sec) * 1000000.;
1535    
1536     req->result = select (0, 0, 0, 0, &tv);
1537     }
1538     #endif
1539     break;
1540    
1541     case EIO_UTIME:
1542     case EIO_FUTIME:
1543     {
1544     struct timeval tv[2];
1545     struct timeval *times;
1546    
1547     if (req->nv1 != -1. || req->nv2 != -1.)
1548     {
1549     tv[0].tv_sec = req->nv1;
1550     tv[0].tv_usec = (req->nv1 - tv[0].tv_sec) * 1000000.;
1551     tv[1].tv_sec = req->nv2;
1552     tv[1].tv_usec = (req->nv2 - tv[1].tv_sec) * 1000000.;
1553    
1554     times = tv;
1555     }
1556     else
1557     times = 0;
1558    
1559    
1560     req->result = req->type == EIO_FUTIME
1561     ? futimes (req->int1, times)
1562     : utimes (req->ptr1, times);
1563     }
1564 root 1.25 break;
1565 root 1.15
1566     case EIO_GROUP:
1567 root 1.24 abort (); /* handled in eio_request */
1568    
1569 root 1.15 case EIO_NOP:
1570     req->result = 0;
1571     break;
1572    
1573     case EIO_CUSTOM:
1574 root 1.23 ((void (*)(eio_req *))req->feed) (req);
1575 root 1.15 break;
1576    
1577     default:
1578     req->result = -1;
1579     break;
1580     }
1581    
1582     req->errorno = errno;
1583     }
1584    
1585 root 1.12 #ifndef EIO_NO_WRAPPERS
1586    
1587 root 1.10 eio_req *eio_nop (int pri, eio_cb cb, void *data)
1588 root 1.9 {
1589     REQ (EIO_NOP); SEND;
1590     }
1591    
1592 root 1.10 eio_req *eio_busy (double delay, int pri, eio_cb cb, void *data)
1593 root 1.9 {
1594     REQ (EIO_BUSY); req->nv1 = delay; SEND;
1595     }
1596    
1597 root 1.10 eio_req *eio_sync (int pri, eio_cb cb, void *data)
1598 root 1.9 {
1599     REQ (EIO_SYNC); SEND;
1600     }
1601 root 1.6
1602 root 1.10 eio_req *eio_fsync (int fd, int pri, eio_cb cb, void *data)
1603 root 1.6 {
1604 root 1.9 REQ (EIO_FSYNC); req->int1 = fd; SEND;
1605 root 1.6 }
1606    
1607 root 1.23 eio_req *eio_msync (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1608     {
1609     REQ (EIO_MSYNC); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1610     }
1611    
1612     eio_req *eio_mtouch (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1613     {
1614     REQ (EIO_MTOUCH); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1615     }
1616    
1617 root 1.27 eio_req *eio_sync_file_range (int fd, off_t offset, size_t nbytes, unsigned int flags, int pri, eio_cb cb, void *data)
1618     {
1619     REQ (EIO_SYNC_FILE_RANGE); req->int1 = fd; req->offs = offset; req->size = nbytes; req->int2 = flags; SEND;
1620     }
1621    
1622 root 1.10 eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data)
1623 root 1.6 {
1624 root 1.9 REQ (EIO_FDATASYNC); req->int1 = fd; SEND;
1625     }
1626    
1627 root 1.10 eio_req *eio_close (int fd, int pri, eio_cb cb, void *data)
1628 root 1.9 {
1629     REQ (EIO_CLOSE); req->int1 = fd; SEND;
1630 root 1.6 }
1631    
1632 root 1.10 eio_req *eio_readahead (int fd, off_t offset, size_t length, int pri, eio_cb cb, void *data)
1633 root 1.6 {
1634 root 1.9 REQ (EIO_READAHEAD); req->int1 = fd; req->offs = offset; req->size = length; SEND;
1635 root 1.6 }
1636    
1637 root 1.10 eio_req *eio_read (int fd, void *buf, size_t length, off_t offset, int pri, eio_cb cb, void *data)
1638 root 1.6 {
1639 root 1.10 REQ (EIO_READ); req->int1 = fd; req->offs = offset; req->size = length; req->ptr2 = buf; SEND;
1640 root 1.6 }
1641    
1642 root 1.10 eio_req *eio_write (int fd, void *buf, size_t length, off_t offset, int pri, eio_cb cb, void *data)
1643 root 1.6 {
1644 root 1.10 REQ (EIO_WRITE); req->int1 = fd; req->offs = offset; req->size = length; req->ptr2 = buf; SEND;
1645 root 1.6 }
1646    
1647 root 1.10 eio_req *eio_fstat (int fd, int pri, eio_cb cb, void *data)
1648 root 1.6 {
1649 root 1.9 REQ (EIO_FSTAT); req->int1 = fd; SEND;
1650 root 1.6 }
1651    
1652 root 1.10 eio_req *eio_futime (int fd, double atime, double mtime, int pri, eio_cb cb, void *data)
1653 root 1.6 {
1654 root 1.9 REQ (EIO_FUTIME); req->int1 = fd; req->nv1 = atime; req->nv2 = mtime; SEND;
1655 root 1.6 }
1656    
1657 root 1.10 eio_req *eio_ftruncate (int fd, off_t offset, int pri, eio_cb cb, void *data)
1658 root 1.6 {
1659 root 1.9 REQ (EIO_FTRUNCATE); req->int1 = fd; req->offs = offset; SEND;
1660 root 1.6 }
1661    
1662 root 1.10 eio_req *eio_fchmod (int fd, mode_t mode, int pri, eio_cb cb, void *data)
1663 root 1.6 {
1664 root 1.9 REQ (EIO_FCHMOD); req->int1 = fd; req->int2 = (long)mode; SEND;
1665 root 1.6 }
1666    
1667 root 1.10 eio_req *eio_fchown (int fd, uid_t uid, gid_t gid, int pri, eio_cb cb, void *data)
1668 root 1.6 {
1669 root 1.9 REQ (EIO_FCHOWN); req->int1 = fd; req->int2 = (long)uid; req->int3 = (long)gid; SEND;
1670 root 1.6 }
1671    
1672 root 1.10 eio_req *eio_dup2 (int fd, int fd2, int pri, eio_cb cb, void *data)
1673 root 1.6 {
1674 root 1.9 REQ (EIO_DUP2); req->int1 = fd; req->int2 = fd2; SEND;
1675 root 1.6 }
1676    
1677 root 1.10 eio_req *eio_sendfile (int out_fd, int in_fd, off_t in_offset, size_t length, int pri, eio_cb cb, void *data)
1678 root 1.6 {
1679 root 1.9 REQ (EIO_SENDFILE); req->int1 = out_fd; req->int2 = in_fd; req->offs = in_offset; req->size = length; SEND;
1680 root 1.6 }
1681    
1682 root 1.10 eio_req *eio_open (const char *path, int flags, mode_t mode, int pri, eio_cb cb, void *data)
1683 root 1.6 {
1684 root 1.9 REQ (EIO_OPEN); PATH; req->int1 = flags; req->int2 = (long)mode; SEND;
1685 root 1.6 }
1686    
1687 root 1.10 eio_req *eio_utime (const char *path, double atime, double mtime, int pri, eio_cb cb, void *data)
1688 root 1.6 {
1689 root 1.9 REQ (EIO_UTIME); PATH; req->nv1 = atime; req->nv2 = mtime; SEND;
1690 root 1.6 }
1691    
1692 root 1.10 eio_req *eio_truncate (const char *path, off_t offset, int pri, eio_cb cb, void *data)
1693 root 1.6 {
1694 root 1.9 REQ (EIO_TRUNCATE); PATH; req->offs = offset; SEND;
1695 root 1.6 }
1696    
1697 root 1.10 eio_req *eio_chown (const char *path, uid_t uid, gid_t gid, int pri, eio_cb cb, void *data)
1698 root 1.6 {
1699 root 1.9 REQ (EIO_CHOWN); PATH; req->int2 = (long)uid; req->int3 = (long)gid; SEND;
1700 root 1.6 }
1701    
1702 root 1.10 eio_req *eio_chmod (const char *path, mode_t mode, int pri, eio_cb cb, void *data)
1703 root 1.6 {
1704 root 1.9 REQ (EIO_CHMOD); PATH; req->int2 = (long)mode; SEND;
1705 root 1.6 }
1706    
1707 root 1.10 eio_req *eio_mkdir (const char *path, mode_t mode, int pri, eio_cb cb, void *data)
1708 root 1.6 {
1709 root 1.9 REQ (EIO_MKDIR); PATH; req->int2 = (long)mode; SEND;
1710 root 1.6 }
1711    
1712 root 1.9 static eio_req *
1713 root 1.10 eio__1path (int type, const char *path, int pri, eio_cb cb, void *data)
1714 root 1.6 {
1715 root 1.9 REQ (type); PATH; SEND;
1716 root 1.6 }
1717    
1718 root 1.10 eio_req *eio_readlink (const char *path, int pri, eio_cb cb, void *data)
1719 root 1.6 {
1720 root 1.10 return eio__1path (EIO_READLINK, path, pri, cb, data);
1721 root 1.6 }
1722    
1723 root 1.10 eio_req *eio_stat (const char *path, int pri, eio_cb cb, void *data)
1724 root 1.6 {
1725 root 1.10 return eio__1path (EIO_STAT, path, pri, cb, data);
1726 root 1.6 }
1727    
1728 root 1.10 eio_req *eio_lstat (const char *path, int pri, eio_cb cb, void *data)
1729 root 1.6 {
1730 root 1.10 return eio__1path (EIO_LSTAT, path, pri, cb, data);
1731 root 1.6 }
1732    
1733 root 1.10 eio_req *eio_unlink (const char *path, int pri, eio_cb cb, void *data)
1734 root 1.6 {
1735 root 1.10 return eio__1path (EIO_UNLINK, path, pri, cb, data);
1736 root 1.6 }
1737    
1738 root 1.10 eio_req *eio_rmdir (const char *path, int pri, eio_cb cb, void *data)
1739 root 1.6 {
1740 root 1.10 return eio__1path (EIO_RMDIR, path, pri, cb, data);
1741 root 1.6 }
1742    
1743 root 1.31 eio_req *eio_readdir (const char *path, int flags, int pri, eio_cb cb, void *data)
1744 root 1.1 {
1745 root 1.31 REQ (EIO_READDIR); PATH; req->int1 = flags; SEND;
1746 root 1.1 }
1747    
1748 root 1.10 eio_req *eio_mknod (const char *path, mode_t mode, dev_t dev, int pri, eio_cb cb, void *data)
1749 root 1.1 {
1750 root 1.18 REQ (EIO_MKNOD); PATH; req->int2 = (long)mode; req->int3 = (long)dev; SEND;
1751 root 1.1 }
1752    
1753 root 1.9 static eio_req *
1754 root 1.10 eio__2path (int type, const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1755 root 1.1 {
1756 root 1.9 REQ (type); PATH;
1757 root 1.1
1758 root 1.9 req->flags |= EIO_FLAG_PTR2_FREE;
1759     req->ptr2 = strdup (new_path);
1760     if (!req->ptr2)
1761     {
1762     eio_api_destroy (req);
1763     return 0;
1764     }
1765 root 1.1
1766 root 1.9 SEND;
1767 root 1.1 }
1768    
1769 root 1.10 eio_req *eio_link (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1770 root 1.1 {
1771 root 1.10 return eio__2path (EIO_LINK, path, new_path, pri, cb, data);
1772 root 1.1 }
1773    
1774 root 1.10 eio_req *eio_symlink (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1775 root 1.1 {
1776 root 1.10 return eio__2path (EIO_SYMLINK, path, new_path, pri, cb, data);
1777 root 1.1 }
1778    
1779 root 1.10 eio_req *eio_rename (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1780 root 1.1 {
1781 root 1.10 return eio__2path (EIO_RENAME, path, new_path, pri, cb, data);
1782     }
1783    
1784 root 1.15 eio_req *eio_custom (eio_cb execute, int pri, eio_cb cb, void *data)
1785     {
1786 root 1.23 REQ (EIO_CUSTOM); req->feed = (void (*)(eio_req *))execute; SEND;
1787 root 1.15 }
1788    
1789 root 1.12 #endif
1790    
1791 root 1.10 eio_req *eio_grp (eio_cb cb, void *data)
1792     {
1793     const int pri = EIO_PRI_MAX;
1794    
1795     REQ (EIO_GROUP); SEND;
1796 root 1.1 }
1797    
1798 root 1.9 #undef REQ
1799     #undef PATH
1800     #undef SEND
1801 root 1.1
1802 root 1.9 /*****************************************************************************/
1803     /* grp functions */
1804 root 1.1
1805 root 1.2 void eio_grp_feed (eio_req *grp, void (*feed)(eio_req *req), int limit)
1806 root 1.1 {
1807     grp->int2 = limit;
1808     grp->feed = feed;
1809 root 1.2
1810     grp_try_feed (grp);
1811     }
1812    
1813     void eio_grp_limit (eio_req *grp, int limit)
1814     {
1815     grp->int2 = limit;
1816    
1817     grp_try_feed (grp);
1818 root 1.1 }
1819    
1820     void eio_grp_add (eio_req *grp, eio_req *req)
1821     {
1822     assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2));
1823    
1824 root 1.26 grp->flags |= EIO_FLAG_GROUPADD;
1825    
1826 root 1.1 ++grp->size;
1827     req->grp = grp;
1828    
1829     req->grp_prev = 0;
1830     req->grp_next = grp->grp_first;
1831    
1832     if (grp->grp_first)
1833     grp->grp_first->grp_prev = req;
1834    
1835     grp->grp_first = req;
1836     }
1837    
1838 root 1.9 /*****************************************************************************/
1839     /* misc garbage */
1840    
1841     ssize_t eio_sendfile_sync (int ofd, int ifd, off_t offset, size_t count)
1842     {
1843 root 1.14 etp_worker wrk;
1844 root 1.37 ssize_t ret;
1845 root 1.9
1846     wrk.dbuf = 0;
1847    
1848 root 1.37 ret = eio__sendfile (ofd, ifd, offset, count, &wrk);
1849 root 1.9
1850     if (wrk.dbuf)
1851     free (wrk.dbuf);
1852 root 1.37
1853     return ret;
1854 root 1.9 }
1855 root 1.1