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