ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/libeio/eio.c
Revision: 1.43
Committed: Wed Jun 17 18:29:15 2009 UTC (14 years, 11 months ago) by root
Content type: text/plain
Branch: MAIN
CVS Tags: rel-3_31, rel-3_3, rel-3_261, rel-3_26, rel-3_25
Changes since 1.42: +1 -1 lines
Log Message:
*** empty log message ***

File Contents

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