ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/libeio/eio.c
Revision: 1.39
Committed: Sat Jun 13 03:59:04 2009 UTC (14 years, 11 months ago) by root
Content type: text/plain
Branch: MAIN
CVS Tags: rel-3_22
Changes since 1.38: +29 -8 lines
Log Message:
*** empty log message ***

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     for (i = size > EIO_QSORT_SKIP ? EIO_QSORT_CUTOFF : size; --i; )
1074     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.37 dents [j + 1] = dents [j];
1090    
1091     dents [j + 1] = value;
1092     }
1093 root 1.31 }
1094    
1095 root 1.1 /* read a full directory */
1096 root 1.9 static void
1097 root 1.14 eio__scandir (eio_req *req, etp_worker *self)
1098 root 1.1 {
1099     DIR *dirp;
1100     EIO_STRUCT_DIRENT *entp;
1101 root 1.37 char *name, *names;
1102 root 1.31 int namesalloc = 4096;
1103     int namesoffs = 0;
1104     int flags = req->int1;
1105     eio_dirent *dents = 0;
1106     int dentalloc = 128;
1107     int dentoffs = 0;
1108    
1109     req->result = -1;
1110    
1111     if (!(flags & EIO_READDIR_DENTS))
1112     flags &= ~(EIO_READDIR_DIRS_FIRST | EIO_READDIR_STAT_ORDER);
1113 root 1.1
1114 root 1.14 X_LOCK (wrklock);
1115 root 1.20 /* the corresponding closedir is in ETP_WORKER_CLEAR */
1116 root 1.1 self->dirp = dirp = opendir (req->ptr1);
1117 root 1.31 req->flags |= EIO_FLAG_PTR1_FREE | EIO_FLAG_PTR2_FREE;
1118     req->ptr1 = names = malloc (namesalloc);
1119     req->ptr2 = dents = flags ? malloc (dentalloc * sizeof (eio_dirent)) : 0;
1120 root 1.14 X_UNLOCK (wrklock);
1121 root 1.1
1122 root 1.31 if (dirp && names && (!flags || dents))
1123 root 1.1 for (;;)
1124     {
1125     errno = 0;
1126 root 1.16 entp = readdir (dirp);
1127 root 1.1
1128     if (!entp)
1129 root 1.31 {
1130     if (errno)
1131     break;
1132    
1133     /* sort etc. */
1134     req->int1 = flags;
1135     req->result = dentoffs;
1136    
1137 root 1.32 if (dents)
1138     {
1139     eio_dirent *ent = dents + dentoffs;
1140    
1141     while (ent > dents)
1142     (--ent)->name = names + (size_t)ent->name;
1143     }
1144    
1145     if (flags & EIO_READDIR_STAT_ORDER
1146     || !(~flags & (EIO_READDIR_DIRS_FIRST | EIO_READDIR_FOUND_UNKNOWN)))
1147 root 1.37 eio_dent_sort (dents, dentoffs); /* score depends of DIRS_FIRST */
1148 root 1.32 else if (flags & EIO_READDIR_DIRS_FIRST)
1149 root 1.31 {
1150     /* in this case, all is known, and we just put dirs first and sort them */
1151     eio_dirent *ent = dents + dentoffs;
1152     eio_dirent *dir = dents;
1153    
1154 root 1.35 /* now move dirs to the front, and non-dirs to the back */
1155 root 1.37 /* by walking from both sides and swapping if necessary */
1156 root 1.31 while (ent > dir)
1157     {
1158     if (dir->type == DT_DIR)
1159     ++dir;
1160     else
1161     {
1162     --ent;
1163    
1164     if (ent->type == DT_DIR)
1165     {
1166     eio_dirent tmp = *dir;
1167     *dir = *ent;
1168     *ent = tmp;
1169    
1170     ++dir;
1171     }
1172     }
1173     }
1174    
1175     /* now sort the dirs only */
1176 root 1.37 eio_dent_sort (dents, dir - dents);
1177 root 1.31 }
1178    
1179 root 1.32 /* only provide the names array unless DENTS is specified */
1180     if (!(flags & EIO_READDIR_DENTS))
1181     {
1182     X_LOCK (wrklock);
1183     assert (!dents);
1184     req->ptr1 = 0;
1185     req->ptr2 = names;
1186     X_UNLOCK (wrklock);
1187     }
1188 root 1.31
1189     break;
1190     }
1191 root 1.1
1192 root 1.32 /* now add the entry to our list(s) */
1193 root 1.1 name = entp->d_name;
1194    
1195 root 1.32 /* skip . and .. entries */
1196 root 1.1 if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
1197     {
1198 root 1.33 int len = D_NAMLEN (entp) + 1;
1199 root 1.1
1200 root 1.31 while (expect_false (namesoffs + len > namesalloc))
1201 root 1.1 {
1202 root 1.31 namesalloc *= 2;
1203 root 1.14 X_LOCK (wrklock);
1204 root 1.31 req->ptr1 = names = realloc (names, namesalloc);
1205 root 1.14 X_UNLOCK (wrklock);
1206 root 1.1
1207     if (!names)
1208     break;
1209     }
1210    
1211 root 1.31 memcpy (names + namesoffs, name, len);
1212    
1213     if (dents)
1214     {
1215     struct eio_dirent *ent;
1216    
1217     if (expect_false (dentoffs == dentalloc))
1218     {
1219     dentalloc *= 2;
1220     X_LOCK (wrklock);
1221     req->ptr2 = dents = realloc (dents, dentalloc * sizeof (eio_dirent));
1222     X_UNLOCK (wrklock);
1223    
1224     if (!dents)
1225     break;
1226     }
1227    
1228     ent = dents + dentoffs;
1229    
1230 root 1.32 ent->name = (char *)(size_t)namesoffs; /* rather dirtily we store the offset in the pointer */
1231 root 1.31 ent->namelen = len - 1;
1232     ent->inode = D_INO (entp);
1233    
1234     switch (D_TYPE (entp))
1235     {
1236     default:
1237     ent->type = EIO_DT_UNKNOWN;
1238     flags |= EIO_READDIR_FOUND_UNKNOWN;
1239     break;
1240    
1241     #ifdef DT_FIFO
1242     case DT_FIFO: ent->type = EIO_DT_FIFO; break;
1243     #endif
1244     #ifdef DT_CHR
1245     case DT_CHR: ent->type = EIO_DT_CHR; break;
1246     #endif
1247 root 1.33 #ifdef DT_MPC
1248     case DT_MPC: ent->type = EIO_DT_MPC; break;
1249     #endif
1250 root 1.32 #ifdef DT_DIR
1251 root 1.31 case DT_DIR: ent->type = EIO_DT_DIR; break;
1252     #endif
1253 root 1.33 #ifdef DT_NAM
1254     case DT_NAM: ent->type = EIO_DT_NAM; break;
1255     #endif
1256 root 1.32 #ifdef DT_BLK
1257 root 1.31 case DT_BLK: ent->type = EIO_DT_BLK; break;
1258     #endif
1259 root 1.33 #ifdef DT_MPB
1260     case DT_MPB: ent->type = EIO_DT_MPB; break;
1261     #endif
1262 root 1.32 #ifdef DT_REG
1263 root 1.31 case DT_REG: ent->type = EIO_DT_REG; break;
1264     #endif
1265 root 1.33 #ifdef DT_NWK
1266     case DT_NWK: ent->type = EIO_DT_NWK; break;
1267     #endif
1268     #ifdef DT_CMP
1269     case DT_CMP: ent->type = EIO_DT_CMP; break;
1270     #endif
1271 root 1.32 #ifdef DT_LNK
1272 root 1.31 case DT_LNK: ent->type = EIO_DT_LNK; break;
1273     #endif
1274     #ifdef DT_SOCK
1275     case DT_SOCK: ent->type = EIO_DT_SOCK; break;
1276     #endif
1277 root 1.33 #ifdef DT_DOOR
1278     case DT_DOOR: ent->type = EIO_DT_DOOR; break;
1279     #endif
1280 root 1.31 #ifdef DT_WHT
1281     case DT_WHT: ent->type = EIO_DT_WHT; break;
1282     #endif
1283     }
1284    
1285     ent->score = 0;
1286    
1287     if (flags & EIO_READDIR_DIRS_FIRST)
1288     {
1289     if (ent->type == EIO_DT_UNKNOWN)
1290     {
1291     if (*name == '.') /* leading dots are likely directories, and, in any case, rare */
1292     ent->score = 98;
1293     else if (!strchr (name, '.')) /* absense of dots indicate likely dirs */
1294 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 */
1295 root 1.31 }
1296 root 1.33 else if (ent->type == EIO_DT_DIR)
1297 root 1.31 ent->score = 100;
1298     }
1299     }
1300    
1301     namesoffs += len;
1302     ++dentoffs;
1303 root 1.1 }
1304 root 1.36
1305     if (EIO_CANCELLED (req))
1306     {
1307     errno = ECANCELED;
1308     break;
1309     }
1310 root 1.1 }
1311     }
1312    
1313 root 1.23 #if !(_POSIX_MAPPED_FILES && _POSIX_SYNCHRONIZED_IO)
1314 root 1.27 # undef msync
1315 root 1.28 # define msync(a,b,c) ((errno = ENOSYS), -1)
1316 root 1.23 #endif
1317    
1318     int
1319     eio__mtouch (void *mem, size_t len, int flags)
1320     {
1321     intptr_t addr = (intptr_t)mem;
1322     intptr_t end = addr + len;
1323     #ifdef PAGESIZE
1324     const intptr_t page = PAGESIZE;
1325     #else
1326     static intptr_t page;
1327    
1328     if (!page)
1329     page = sysconf (_SC_PAGESIZE);
1330     #endif
1331    
1332     addr &= ~(page - 1); /* assume page size is always a power of two */
1333    
1334     if (addr < end)
1335     if (flags) /* modify */
1336     do { *((volatile sig_atomic_t *)addr) |= 0; } while ((addr += page) < len);
1337     else
1338     do { *((volatile sig_atomic_t *)addr) ; } while ((addr += page) < len);
1339    
1340     return 0;
1341     }
1342    
1343 root 1.1 /*****************************************************************************/
1344    
1345 root 1.6 #define ALLOC(len) \
1346     if (!req->ptr2) \
1347     { \
1348 root 1.14 X_LOCK (wrklock); \
1349 root 1.7 req->flags |= EIO_FLAG_PTR2_FREE; \
1350 root 1.14 X_UNLOCK (wrklock); \
1351 root 1.7 req->ptr2 = malloc (len); \
1352     if (!req->ptr2) \
1353     { \
1354     errno = ENOMEM; \
1355     req->result = -1; \
1356     break; \
1357     } \
1358 root 1.6 }
1359    
1360 root 1.14 X_THREAD_PROC (etp_proc)
1361 root 1.1 {
1362 root 1.14 ETP_REQ *req;
1363 root 1.1 struct timespec ts;
1364 root 1.14 etp_worker *self = (etp_worker *)thr_arg;
1365 root 1.1
1366     /* try to distribute timeouts somewhat randomly */
1367     ts.tv_nsec = ((unsigned long)self & 1023UL) * (1000000000UL / 1024UL);
1368    
1369     for (;;)
1370     {
1371     X_LOCK (reqlock);
1372    
1373     for (;;)
1374     {
1375     self->req = req = reqq_shift (&req_queue);
1376    
1377     if (req)
1378     break;
1379    
1380     ++idle;
1381    
1382 root 1.14 ts.tv_sec = time (0) + IDLE_TIMEOUT;
1383 root 1.1 if (X_COND_TIMEDWAIT (reqwait, reqlock, ts) == ETIMEDOUT)
1384     {
1385     if (idle > max_idle)
1386     {
1387     --idle;
1388     X_UNLOCK (reqlock);
1389 root 1.14 X_LOCK (wrklock);
1390 root 1.1 --started;
1391 root 1.14 X_UNLOCK (wrklock);
1392 root 1.1 goto quit;
1393     }
1394    
1395     /* we are allowed to idle, so do so without any timeout */
1396     X_COND_WAIT (reqwait, reqlock);
1397     }
1398    
1399     --idle;
1400     }
1401    
1402     --nready;
1403    
1404     X_UNLOCK (reqlock);
1405    
1406 root 1.15 if (req->type < 0)
1407     goto quit;
1408 root 1.1
1409     if (!EIO_CANCELLED (req))
1410 root 1.15 ETP_EXECUTE (self, req);
1411 root 1.1
1412     X_LOCK (reslock);
1413    
1414     ++npending;
1415    
1416 root 1.14 if (!reqq_push (&res_queue, req) && want_poll_cb)
1417     want_poll_cb ();
1418 root 1.1
1419     self->req = 0;
1420 root 1.14 etp_worker_clear (self);
1421 root 1.1
1422     X_UNLOCK (reslock);
1423     }
1424    
1425     quit:
1426 root 1.14 X_LOCK (wrklock);
1427     etp_worker_free (self);
1428     X_UNLOCK (wrklock);
1429 root 1.1
1430     return 0;
1431     }
1432    
1433     /*****************************************************************************/
1434    
1435     int eio_init (void (*want_poll)(void), void (*done_poll)(void))
1436     {
1437 root 1.19 return etp_init (want_poll, done_poll);
1438 root 1.1 }
1439    
1440 root 1.6 static void eio_api_destroy (eio_req *req)
1441     {
1442     free (req);
1443     }
1444    
1445     #define REQ(rtype) \
1446     eio_req *req; \
1447     \
1448     req = (eio_req *)calloc (1, sizeof *req); \
1449     if (!req) \
1450     return 0; \
1451     \
1452 root 1.11 req->type = rtype; \
1453     req->pri = pri; \
1454     req->finish = cb; \
1455     req->data = data; \
1456 root 1.6 req->destroy = eio_api_destroy;
1457    
1458     #define SEND eio_submit (req); return req
1459    
1460 root 1.9 #define PATH \
1461     req->flags |= EIO_FLAG_PTR1_FREE; \
1462     req->ptr1 = strdup (path); \
1463     if (!req->ptr1) \
1464     { \
1465     eio_api_destroy (req); \
1466     return 0; \
1467     }
1468    
1469 root 1.15 static void eio_execute (etp_worker *self, eio_req *req)
1470     {
1471     errno = 0;
1472    
1473     switch (req->type)
1474     {
1475     case EIO_READ: ALLOC (req->size);
1476     req->result = req->offs >= 0
1477     ? pread (req->int1, req->ptr2, req->size, req->offs)
1478     : read (req->int1, req->ptr2, req->size); break;
1479     case EIO_WRITE: req->result = req->offs >= 0
1480     ? pwrite (req->int1, req->ptr2, req->size, req->offs)
1481     : write (req->int1, req->ptr2, req->size); break;
1482    
1483     case EIO_READAHEAD: req->result = readahead (req->int1, req->offs, req->size); break;
1484     case EIO_SENDFILE: req->result = eio__sendfile (req->int1, req->int2, req->offs, req->size, self); break;
1485    
1486     case EIO_STAT: ALLOC (sizeof (EIO_STRUCT_STAT));
1487     req->result = stat (req->ptr1, (EIO_STRUCT_STAT *)req->ptr2); break;
1488     case EIO_LSTAT: ALLOC (sizeof (EIO_STRUCT_STAT));
1489     req->result = lstat (req->ptr1, (EIO_STRUCT_STAT *)req->ptr2); break;
1490     case EIO_FSTAT: ALLOC (sizeof (EIO_STRUCT_STAT));
1491     req->result = fstat (req->int1, (EIO_STRUCT_STAT *)req->ptr2); break;
1492    
1493     case EIO_CHOWN: req->result = chown (req->ptr1, req->int2, req->int3); break;
1494     case EIO_FCHOWN: req->result = fchown (req->int1, req->int2, req->int3); break;
1495     case EIO_CHMOD: req->result = chmod (req->ptr1, (mode_t)req->int2); break;
1496     case EIO_FCHMOD: req->result = fchmod (req->int1, (mode_t)req->int2); break;
1497     case EIO_TRUNCATE: req->result = truncate (req->ptr1, req->offs); break;
1498     case EIO_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break;
1499    
1500     case EIO_OPEN: req->result = open (req->ptr1, req->int1, (mode_t)req->int2); break;
1501     case EIO_CLOSE: req->result = close (req->int1); break;
1502     case EIO_DUP2: req->result = dup2 (req->int1, req->int2); break;
1503     case EIO_UNLINK: req->result = unlink (req->ptr1); break;
1504     case EIO_RMDIR: req->result = rmdir (req->ptr1); break;
1505     case EIO_MKDIR: req->result = mkdir (req->ptr1, (mode_t)req->int2); break;
1506     case EIO_RENAME: req->result = rename (req->ptr1, req->ptr2); break;
1507     case EIO_LINK: req->result = link (req->ptr1, req->ptr2); break;
1508     case EIO_SYMLINK: req->result = symlink (req->ptr1, req->ptr2); break;
1509 root 1.18 case EIO_MKNOD: req->result = mknod (req->ptr1, (mode_t)req->int2, (dev_t)req->int3); break;
1510 root 1.15
1511     case EIO_READLINK: ALLOC (NAME_MAX);
1512     req->result = readlink (req->ptr1, req->ptr2, NAME_MAX); break;
1513    
1514     case EIO_SYNC: req->result = 0; sync (); break;
1515     case EIO_FSYNC: req->result = fsync (req->int1); break;
1516     case EIO_FDATASYNC: req->result = fdatasync (req->int1); break;
1517 root 1.23 case EIO_MSYNC: req->result = msync (req->ptr2, req->size, req->int1); break;
1518     case EIO_MTOUCH: req->result = eio__mtouch (req->ptr2, req->size, req->int1); break;
1519 root 1.27 case EIO_SYNC_FILE_RANGE: req->result = eio__sync_file_range (req->int1, req->offs, req->size, req->int2); break;
1520 root 1.15
1521     case EIO_READDIR: eio__scandir (req, self); break;
1522    
1523     case EIO_BUSY:
1524     #ifdef _WIN32
1525     Sleep (req->nv1 * 1000.);
1526     #else
1527     {
1528     struct timeval tv;
1529    
1530     tv.tv_sec = req->nv1;
1531     tv.tv_usec = (req->nv1 - tv.tv_sec) * 1000000.;
1532    
1533     req->result = select (0, 0, 0, 0, &tv);
1534     }
1535     #endif
1536     break;
1537    
1538     case EIO_UTIME:
1539     case EIO_FUTIME:
1540     {
1541     struct timeval tv[2];
1542     struct timeval *times;
1543    
1544     if (req->nv1 != -1. || req->nv2 != -1.)
1545     {
1546     tv[0].tv_sec = req->nv1;
1547     tv[0].tv_usec = (req->nv1 - tv[0].tv_sec) * 1000000.;
1548     tv[1].tv_sec = req->nv2;
1549     tv[1].tv_usec = (req->nv2 - tv[1].tv_sec) * 1000000.;
1550    
1551     times = tv;
1552     }
1553     else
1554     times = 0;
1555    
1556    
1557     req->result = req->type == EIO_FUTIME
1558     ? futimes (req->int1, times)
1559     : utimes (req->ptr1, times);
1560     }
1561 root 1.25 break;
1562 root 1.15
1563     case EIO_GROUP:
1564 root 1.24 abort (); /* handled in eio_request */
1565    
1566 root 1.15 case EIO_NOP:
1567     req->result = 0;
1568     break;
1569    
1570     case EIO_CUSTOM:
1571 root 1.23 ((void (*)(eio_req *))req->feed) (req);
1572 root 1.15 break;
1573    
1574     default:
1575     req->result = -1;
1576     break;
1577     }
1578    
1579     req->errorno = errno;
1580     }
1581    
1582 root 1.12 #ifndef EIO_NO_WRAPPERS
1583    
1584 root 1.10 eio_req *eio_nop (int pri, eio_cb cb, void *data)
1585 root 1.9 {
1586     REQ (EIO_NOP); SEND;
1587     }
1588    
1589 root 1.10 eio_req *eio_busy (double delay, int pri, eio_cb cb, void *data)
1590 root 1.9 {
1591     REQ (EIO_BUSY); req->nv1 = delay; SEND;
1592     }
1593    
1594 root 1.10 eio_req *eio_sync (int pri, eio_cb cb, void *data)
1595 root 1.9 {
1596     REQ (EIO_SYNC); SEND;
1597     }
1598 root 1.6
1599 root 1.10 eio_req *eio_fsync (int fd, int pri, eio_cb cb, void *data)
1600 root 1.6 {
1601 root 1.9 REQ (EIO_FSYNC); req->int1 = fd; SEND;
1602 root 1.6 }
1603    
1604 root 1.23 eio_req *eio_msync (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1605     {
1606     REQ (EIO_MSYNC); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1607     }
1608    
1609     eio_req *eio_mtouch (void *addr, size_t length, int flags, int pri, eio_cb cb, void *data)
1610     {
1611     REQ (EIO_MTOUCH); req->ptr2 = addr; req->size = length; req->int1 = flags; SEND;
1612     }
1613    
1614 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)
1615     {
1616     REQ (EIO_SYNC_FILE_RANGE); req->int1 = fd; req->offs = offset; req->size = nbytes; req->int2 = flags; SEND;
1617     }
1618    
1619 root 1.10 eio_req *eio_fdatasync (int fd, int pri, eio_cb cb, void *data)
1620 root 1.6 {
1621 root 1.9 REQ (EIO_FDATASYNC); req->int1 = fd; SEND;
1622     }
1623    
1624 root 1.10 eio_req *eio_close (int fd, int pri, eio_cb cb, void *data)
1625 root 1.9 {
1626     REQ (EIO_CLOSE); req->int1 = fd; SEND;
1627 root 1.6 }
1628    
1629 root 1.10 eio_req *eio_readahead (int fd, off_t offset, size_t length, int pri, eio_cb cb, void *data)
1630 root 1.6 {
1631 root 1.9 REQ (EIO_READAHEAD); req->int1 = fd; req->offs = offset; req->size = length; SEND;
1632 root 1.6 }
1633    
1634 root 1.10 eio_req *eio_read (int fd, void *buf, size_t length, off_t offset, int pri, eio_cb cb, void *data)
1635 root 1.6 {
1636 root 1.10 REQ (EIO_READ); req->int1 = fd; req->offs = offset; req->size = length; req->ptr2 = buf; SEND;
1637 root 1.6 }
1638    
1639 root 1.10 eio_req *eio_write (int fd, void *buf, size_t length, off_t offset, int pri, eio_cb cb, void *data)
1640 root 1.6 {
1641 root 1.10 REQ (EIO_WRITE); req->int1 = fd; req->offs = offset; req->size = length; req->ptr2 = buf; SEND;
1642 root 1.6 }
1643    
1644 root 1.10 eio_req *eio_fstat (int fd, int pri, eio_cb cb, void *data)
1645 root 1.6 {
1646 root 1.9 REQ (EIO_FSTAT); req->int1 = fd; SEND;
1647 root 1.6 }
1648    
1649 root 1.10 eio_req *eio_futime (int fd, double atime, double mtime, int pri, eio_cb cb, void *data)
1650 root 1.6 {
1651 root 1.9 REQ (EIO_FUTIME); req->int1 = fd; req->nv1 = atime; req->nv2 = mtime; SEND;
1652 root 1.6 }
1653    
1654 root 1.10 eio_req *eio_ftruncate (int fd, off_t offset, int pri, eio_cb cb, void *data)
1655 root 1.6 {
1656 root 1.9 REQ (EIO_FTRUNCATE); req->int1 = fd; req->offs = offset; SEND;
1657 root 1.6 }
1658    
1659 root 1.10 eio_req *eio_fchmod (int fd, mode_t mode, int pri, eio_cb cb, void *data)
1660 root 1.6 {
1661 root 1.9 REQ (EIO_FCHMOD); req->int1 = fd; req->int2 = (long)mode; SEND;
1662 root 1.6 }
1663    
1664 root 1.10 eio_req *eio_fchown (int fd, uid_t uid, gid_t gid, int pri, eio_cb cb, void *data)
1665 root 1.6 {
1666 root 1.9 REQ (EIO_FCHOWN); req->int1 = fd; req->int2 = (long)uid; req->int3 = (long)gid; SEND;
1667 root 1.6 }
1668    
1669 root 1.10 eio_req *eio_dup2 (int fd, int fd2, int pri, eio_cb cb, void *data)
1670 root 1.6 {
1671 root 1.9 REQ (EIO_DUP2); req->int1 = fd; req->int2 = fd2; SEND;
1672 root 1.6 }
1673    
1674 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)
1675 root 1.6 {
1676 root 1.9 REQ (EIO_SENDFILE); req->int1 = out_fd; req->int2 = in_fd; req->offs = in_offset; req->size = length; SEND;
1677 root 1.6 }
1678    
1679 root 1.10 eio_req *eio_open (const char *path, int flags, mode_t mode, int pri, eio_cb cb, void *data)
1680 root 1.6 {
1681 root 1.9 REQ (EIO_OPEN); PATH; req->int1 = flags; req->int2 = (long)mode; SEND;
1682 root 1.6 }
1683    
1684 root 1.10 eio_req *eio_utime (const char *path, double atime, double mtime, int pri, eio_cb cb, void *data)
1685 root 1.6 {
1686 root 1.9 REQ (EIO_UTIME); PATH; req->nv1 = atime; req->nv2 = mtime; SEND;
1687 root 1.6 }
1688    
1689 root 1.10 eio_req *eio_truncate (const char *path, off_t offset, int pri, eio_cb cb, void *data)
1690 root 1.6 {
1691 root 1.9 REQ (EIO_TRUNCATE); PATH; req->offs = offset; SEND;
1692 root 1.6 }
1693    
1694 root 1.10 eio_req *eio_chown (const char *path, uid_t uid, gid_t gid, int pri, eio_cb cb, void *data)
1695 root 1.6 {
1696 root 1.9 REQ (EIO_CHOWN); PATH; req->int2 = (long)uid; req->int3 = (long)gid; SEND;
1697 root 1.6 }
1698    
1699 root 1.10 eio_req *eio_chmod (const char *path, mode_t mode, int pri, eio_cb cb, void *data)
1700 root 1.6 {
1701 root 1.9 REQ (EIO_CHMOD); PATH; req->int2 = (long)mode; SEND;
1702 root 1.6 }
1703    
1704 root 1.10 eio_req *eio_mkdir (const char *path, mode_t mode, int pri, eio_cb cb, void *data)
1705 root 1.6 {
1706 root 1.9 REQ (EIO_MKDIR); PATH; req->int2 = (long)mode; SEND;
1707 root 1.6 }
1708    
1709 root 1.9 static eio_req *
1710 root 1.10 eio__1path (int type, const char *path, int pri, eio_cb cb, void *data)
1711 root 1.6 {
1712 root 1.9 REQ (type); PATH; SEND;
1713 root 1.6 }
1714    
1715 root 1.10 eio_req *eio_readlink (const char *path, int pri, eio_cb cb, void *data)
1716 root 1.6 {
1717 root 1.10 return eio__1path (EIO_READLINK, path, pri, cb, data);
1718 root 1.6 }
1719    
1720 root 1.10 eio_req *eio_stat (const char *path, int pri, eio_cb cb, void *data)
1721 root 1.6 {
1722 root 1.10 return eio__1path (EIO_STAT, path, pri, cb, data);
1723 root 1.6 }
1724    
1725 root 1.10 eio_req *eio_lstat (const char *path, int pri, eio_cb cb, void *data)
1726 root 1.6 {
1727 root 1.10 return eio__1path (EIO_LSTAT, path, pri, cb, data);
1728 root 1.6 }
1729    
1730 root 1.10 eio_req *eio_unlink (const char *path, int pri, eio_cb cb, void *data)
1731 root 1.6 {
1732 root 1.10 return eio__1path (EIO_UNLINK, path, pri, cb, data);
1733 root 1.6 }
1734    
1735 root 1.10 eio_req *eio_rmdir (const char *path, int pri, eio_cb cb, void *data)
1736 root 1.6 {
1737 root 1.10 return eio__1path (EIO_RMDIR, path, pri, cb, data);
1738 root 1.6 }
1739    
1740 root 1.31 eio_req *eio_readdir (const char *path, int flags, int pri, eio_cb cb, void *data)
1741 root 1.1 {
1742 root 1.31 REQ (EIO_READDIR); PATH; req->int1 = flags; SEND;
1743 root 1.1 }
1744    
1745 root 1.10 eio_req *eio_mknod (const char *path, mode_t mode, dev_t dev, int pri, eio_cb cb, void *data)
1746 root 1.1 {
1747 root 1.18 REQ (EIO_MKNOD); PATH; req->int2 = (long)mode; req->int3 = (long)dev; SEND;
1748 root 1.1 }
1749    
1750 root 1.9 static eio_req *
1751 root 1.10 eio__2path (int type, const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1752 root 1.1 {
1753 root 1.9 REQ (type); PATH;
1754 root 1.1
1755 root 1.9 req->flags |= EIO_FLAG_PTR2_FREE;
1756     req->ptr2 = strdup (new_path);
1757     if (!req->ptr2)
1758     {
1759     eio_api_destroy (req);
1760     return 0;
1761     }
1762 root 1.1
1763 root 1.9 SEND;
1764 root 1.1 }
1765    
1766 root 1.10 eio_req *eio_link (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1767 root 1.1 {
1768 root 1.10 return eio__2path (EIO_LINK, path, new_path, pri, cb, data);
1769 root 1.1 }
1770    
1771 root 1.10 eio_req *eio_symlink (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1772 root 1.1 {
1773 root 1.10 return eio__2path (EIO_SYMLINK, path, new_path, pri, cb, data);
1774 root 1.1 }
1775    
1776 root 1.10 eio_req *eio_rename (const char *path, const char *new_path, int pri, eio_cb cb, void *data)
1777 root 1.1 {
1778 root 1.10 return eio__2path (EIO_RENAME, path, new_path, pri, cb, data);
1779     }
1780    
1781 root 1.15 eio_req *eio_custom (eio_cb execute, int pri, eio_cb cb, void *data)
1782     {
1783 root 1.23 REQ (EIO_CUSTOM); req->feed = (void (*)(eio_req *))execute; SEND;
1784 root 1.15 }
1785    
1786 root 1.12 #endif
1787    
1788 root 1.10 eio_req *eio_grp (eio_cb cb, void *data)
1789     {
1790     const int pri = EIO_PRI_MAX;
1791    
1792     REQ (EIO_GROUP); SEND;
1793 root 1.1 }
1794    
1795 root 1.9 #undef REQ
1796     #undef PATH
1797     #undef SEND
1798 root 1.1
1799 root 1.9 /*****************************************************************************/
1800     /* grp functions */
1801 root 1.1
1802 root 1.2 void eio_grp_feed (eio_req *grp, void (*feed)(eio_req *req), int limit)
1803 root 1.1 {
1804     grp->int2 = limit;
1805     grp->feed = feed;
1806 root 1.2
1807     grp_try_feed (grp);
1808     }
1809    
1810     void eio_grp_limit (eio_req *grp, int limit)
1811     {
1812     grp->int2 = limit;
1813    
1814     grp_try_feed (grp);
1815 root 1.1 }
1816    
1817     void eio_grp_add (eio_req *grp, eio_req *req)
1818     {
1819     assert (("cannot add requests to IO::AIO::GRP after the group finished", grp->int1 != 2));
1820    
1821 root 1.26 grp->flags |= EIO_FLAG_GROUPADD;
1822    
1823 root 1.1 ++grp->size;
1824     req->grp = grp;
1825    
1826     req->grp_prev = 0;
1827     req->grp_next = grp->grp_first;
1828    
1829     if (grp->grp_first)
1830     grp->grp_first->grp_prev = req;
1831    
1832     grp->grp_first = req;
1833     }
1834    
1835 root 1.9 /*****************************************************************************/
1836     /* misc garbage */
1837    
1838     ssize_t eio_sendfile_sync (int ofd, int ifd, off_t offset, size_t count)
1839     {
1840 root 1.14 etp_worker wrk;
1841 root 1.37 ssize_t ret;
1842 root 1.9
1843     wrk.dbuf = 0;
1844    
1845 root 1.37 ret = eio__sendfile (ofd, ifd, offset, count, &wrk);
1846 root 1.9
1847     if (wrk.dbuf)
1848     free (wrk.dbuf);
1849 root 1.37
1850     return ret;
1851 root 1.9 }
1852 root 1.1