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