ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/libeio/eio.c
Revision: 1.45
Committed: Sat Jan 2 12:50:22 2010 UTC (14 years, 4 months ago) by root
Content type: text/plain
Branch: MAIN
Changes since 1.44: +25 -4 lines
Log Message:
*** empty log message ***

File Contents

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