ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.xs
(Generate patch)

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.52 by root, Sun Oct 22 22:14:54 2006 UTC vs.
Revision 1.103 by root, Sun Jul 8 09:09:34 2007 UTC

1#define _REENTRANT 1 1#include "xthread.h"
2
2#include <errno.h> 3#include <errno.h>
3 4
4#include "EXTERN.h" 5#include "EXTERN.h"
5#include "perl.h" 6#include "perl.h"
6#include "XSUB.h" 7#include "XSUB.h"
7 8
8#include "autoconf/config.h"
9
10#include <pthread.h>
11
12#include <stddef.h> 9#include <stddef.h>
10#include <stdlib.h>
13#include <errno.h> 11#include <errno.h>
14#include <sys/time.h>
15#include <sys/select.h>
16#include <sys/types.h> 12#include <sys/types.h>
17#include <sys/stat.h> 13#include <sys/stat.h>
18#include <limits.h> 14#include <limits.h>
19#include <unistd.h>
20#include <fcntl.h> 15#include <fcntl.h>
21#include <signal.h>
22#include <sched.h> 16#include <sched.h>
17
18#ifdef _WIN32
19
20# define SIGIO 0
21 typedef Direntry_t X_DIRENT;
22#undef malloc
23#undef free
24
25// perl overrides all those nice win32 functions
26# undef open
27# undef read
28# undef write
29# undef stat
30# undef fstat
31# define lstat stat
32# undef truncate
33# undef ftruncate
34# undef open
35# undef close
36# undef unlink
37# undef rmdir
38# undef rename
39# undef lseek
40
41# define chown(a,b,c) (errno = ENOSYS, -1)
42# define fchown(a,b,c) (errno = ENOSYS, -1)
43# define fchmod(a,b) (errno = ENOSYS, -1)
44# define symlink(a,b) (errno = ENOSYS, -1)
45# define readlink(a,b,c) (errno = ENOSYS, -1)
46# define mknod(a,b,c) (errno = ENOSYS, -1)
47# define truncate(a,b) (errno = ENOSYS, -1)
48# define ftruncate(fd,o) chsize ((fd), (o))
49# define fsync(fd) _commit (fd)
50# define opendir(fd) (errno = ENOSYS, 0)
51# define readdir(fd) (errno = ENOSYS, -1)
52# define closedir(fd) (errno = ENOSYS, -1)
53# define mkdir(a,b) mkdir (a)
54
55#else
56
57# include "autoconf/config.h"
58# include <sys/time.h>
59# include <sys/select.h>
60# include <unistd.h>
61# include <utime.h>
62# include <signal.h>
63 typedef struct dirent X_DIRENT;
64
65#endif
23 66
24#if HAVE_SENDFILE 67#if HAVE_SENDFILE
25# if __linux 68# if __linux
26# include <sys/sendfile.h> 69# include <sys/sendfile.h>
27# elif __freebsd 70# elif __freebsd
34# else 77# else
35# error sendfile support requested but not available 78# error sendfile support requested but not available
36# endif 79# endif
37#endif 80#endif
38 81
82/* number of seconds after which idle threads exit */
83#define IDLE_TIMEOUT 10
84
39/* used for struct dirent, AIX doesn't provide it */ 85/* used for struct dirent, AIX doesn't provide it */
40#ifndef NAME_MAX 86#ifndef NAME_MAX
41# define NAME_MAX 4096 87# define NAME_MAX 4096
42#endif 88#endif
43 89
44#if __ia64 90/* buffer size for various temporary buffers */
45# define STACKSIZE 65536 91#define AIO_BUFSIZE 65536
92
93/* use NV for 32 bit perls as it allows larger offsets */
94#if IVSIZE >= 8
95# define SvVAL64 SvIV
46#else 96#else
47# define STACKSIZE 8192 97# define SvVAL64 SvNV
48#endif 98#endif
99
100#define dBUF \
101 char *aio_buf; \
102 X_LOCK (wrklock); \
103 self->dbuf = aio_buf = malloc (AIO_BUFSIZE); \
104 X_UNLOCK (wrklock); \
105 if (!aio_buf) \
106 return -1;
107
108typedef SV SV8; /* byte-sv, used for argument-checking */
49 109
50enum { 110enum {
51 REQ_QUIT, 111 REQ_QUIT,
52 REQ_OPEN, REQ_CLOSE, 112 REQ_OPEN, REQ_CLOSE,
53 REQ_READ, REQ_WRITE, REQ_READAHEAD, 113 REQ_READ, REQ_WRITE,
54 REQ_SENDFILE, 114 REQ_READAHEAD, REQ_SENDFILE,
55 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 115 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
116 REQ_TRUNCATE, REQ_FTRUNCATE,
117 REQ_UTIME, REQ_FUTIME,
118 REQ_CHMOD, REQ_FCHMOD,
119 REQ_CHOWN, REQ_FCHOWN,
56 REQ_FSYNC, REQ_FDATASYNC, 120 REQ_FSYNC, REQ_FDATASYNC,
57 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 121 REQ_UNLINK, REQ_RMDIR, REQ_MKDIR, REQ_RENAME,
58 REQ_READDIR, 122 REQ_MKNOD, REQ_READDIR,
59 REQ_LINK, REQ_SYMLINK, 123 REQ_LINK, REQ_SYMLINK, REQ_READLINK,
60 REQ_SLEEP, 124 REQ_GROUP, REQ_NOP,
61 REQ_GROUP, 125 REQ_BUSY,
62}; 126};
63 127
64#define AIO_REQ_KLASS "IO::AIO::REQ" 128#define AIO_REQ_KLASS "IO::AIO::REQ"
65#define AIO_GRP_KLASS "IO::AIO::GRP" 129#define AIO_GRP_KLASS "IO::AIO::GRP"
66 130
67typedef struct aio_cb 131typedef struct aio_cb
68{ 132{
69 struct aio_cb *volatile next; 133 struct aio_cb *volatile next;
70 134
135 SV *callback;
136 SV *sv1, *sv2;
137 void *ptr1, *ptr2;
138 off_t offs;
139 size_t size;
140 ssize_t result;
141 double nv1, nv2;
142
143 STRLEN stroffset;
144 int type;
145 int int1, int2, int3;
146 int errorno;
147 mode_t mode; /* open */
148
149 unsigned char flags;
150 unsigned char pri;
151
152 SV *self; /* the perl counterpart of this request, if any */
71 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first; 153 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
72
73 SV *self; /* the perl counterpart of this request, if any */
74
75 SV *data, *callback;
76 SV *fh, *fh2;
77 void *dataptr, *data2ptr;
78 Stat_t *statdata;
79 off_t offset;
80 size_t length;
81 ssize_t result;
82
83 int type;
84 int fd, fd2;
85 int errorno;
86 STRLEN dataoffset;
87 mode_t mode; /* open */
88 unsigned char cancelled;
89} aio_cb; 154} aio_cb;
155
156enum {
157 FLAG_CANCELLED = 0x01, /* request was cancelled */
158 FLAG_SV2_RO_OFF = 0x40, /* data was set readonly */
159 FLAG_PTR2_FREE = 0x80, /* need to free(ptr2) */
160};
90 161
91typedef aio_cb *aio_req; 162typedef aio_cb *aio_req;
92typedef aio_cb *aio_req_ornot; 163typedef aio_cb *aio_req_ornot;
93 164
165enum {
166 PRI_MIN = -4,
167 PRI_MAX = 4,
168
169 DEFAULT_PRI = 0,
170 PRI_BIAS = -PRI_MIN,
171 NUM_PRI = PRI_MAX + PRI_BIAS + 1,
172};
173
174#define AIO_TICKS ((1000000 + 1023) >> 10)
175
176static unsigned int max_poll_time = 0;
177static unsigned int max_poll_reqs = 0;
178
179/* calculcate time difference in ~1/AIO_TICKS of a second */
180static int tvdiff (struct timeval *tv1, struct timeval *tv2)
181{
182 return (tv2->tv_sec - tv1->tv_sec ) * AIO_TICKS
183 + ((tv2->tv_usec - tv1->tv_usec) >> 10);
184}
185
186static thread_t main_tid;
187static int main_sig;
188static int block_sig_level;
189
190void block_sig ()
191{
192 sigset_t ss;
193
194 if (block_sig_level++)
195 return;
196
197 if (!main_sig)
198 return;
199
200 sigemptyset (&ss);
201 sigaddset (&ss, main_sig);
202 pthread_sigmask (SIG_BLOCK, &ss, 0);
203}
204
205void unblock_sig ()
206{
207 sigset_t ss;
208
209 if (--block_sig_level)
210 return;
211
212 if (!main_sig)
213 return;
214
215 sigemptyset (&ss);
216 sigaddset (&ss, main_sig);
217 pthread_sigmask (SIG_UNBLOCK, &ss, 0);
218}
219
220static int next_pri = DEFAULT_PRI + PRI_BIAS;
221
94static int started, wanted; 222static unsigned int started, idle, wanted;
95static volatile int nreqs; 223
96static int max_outstanding = 1<<30; 224/* worker threads management */
225static mutex_t wrklock = X_MUTEX_INIT;
226
227typedef struct worker {
228 /* locked by wrklock */
229 struct worker *prev, *next;
230
231 thread_t tid;
232
233 /* locked by reslock, reqlock or wrklock */
234 aio_req req; /* currently processed request */
235 void *dbuf;
236 DIR *dirp;
237} worker;
238
239static worker wrk_first = { &wrk_first, &wrk_first, 0 };
240
241static void worker_clear (worker *wrk)
242{
243 if (wrk->dirp)
244 {
245 closedir (wrk->dirp);
246 wrk->dirp = 0;
247 }
248
249 if (wrk->dbuf)
250 {
251 free (wrk->dbuf);
252 wrk->dbuf = 0;
253 }
254}
255
256static void worker_free (worker *wrk)
257{
258 wrk->next->prev = wrk->prev;
259 wrk->prev->next = wrk->next;
260
261 free (wrk);
262}
263
264static volatile unsigned int nreqs, nready, npending;
265static volatile unsigned int max_idle = 4;
266static volatile unsigned int max_outstanding = 0xffffffff;
97static int respipe [2]; 267static int respipe [2];
98 268
99static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; 269static mutex_t reslock = X_MUTEX_INIT;
100static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; 270static mutex_t reqlock = X_MUTEX_INIT;
101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 271static cond_t reqwait = X_COND_INIT;
102 272
103static volatile aio_req reqs, reqe; /* queue start, queue end */ 273#if WORDACCESS_UNSAFE
104static volatile aio_req ress, rese; /* queue start, queue end */
105 274
275static unsigned int get_nready ()
276{
277 unsigned int retval;
278
279 X_LOCK (reqlock);
280 retval = nready;
281 X_UNLOCK (reqlock);
282
283 return retval;
284}
285
286static unsigned int get_npending ()
287{
288 unsigned int retval;
289
290 X_LOCK (reslock);
291 retval = npending;
292 X_UNLOCK (reslock);
293
294 return retval;
295}
296
297static unsigned int get_nthreads ()
298{
299 unsigned int retval;
300
301 X_LOCK (wrklock);
302 retval = started;
303 X_UNLOCK (wrklock);
304
305 return retval;
306}
307
308#else
309
310# define get_nready() nready
311# define get_npending() npending
312# define get_nthreads() started
313
314#endif
315
316/*
317 * a somewhat faster data structure might be nice, but
318 * with 8 priorities this actually needs <20 insns
319 * per shift, the most expensive operation.
320 */
321typedef struct {
322 aio_req qs[NUM_PRI], qe[NUM_PRI]; /* qstart, qend */
323 int size;
324} reqq;
325
326static reqq req_queue;
327static reqq res_queue;
328
329int reqq_push (reqq *q, aio_req req)
330{
331 int pri = req->pri;
332 req->next = 0;
333
334 if (q->qe[pri])
335 {
336 q->qe[pri]->next = req;
337 q->qe[pri] = req;
338 }
339 else
340 q->qe[pri] = q->qs[pri] = req;
341
342 return q->size++;
343}
344
345aio_req reqq_shift (reqq *q)
346{
347 int pri;
348
349 if (!q->size)
350 return 0;
351
352 --q->size;
353
354 for (pri = NUM_PRI; pri--; )
355 {
356 aio_req req = q->qs[pri];
357
358 if (req)
359 {
360 if (!(q->qs[pri] = req->next))
361 q->qe[pri] = 0;
362
363 return req;
364 }
365 }
366
367 abort ();
368}
369
370static int poll_cb ();
106static void req_invoke (aio_req req); 371static int req_invoke (aio_req req);
372static void req_destroy (aio_req req);
107static void req_free (aio_req req); 373static void req_cancel (aio_req req);
108 374
109/* must be called at most once */ 375/* must be called at most once */
110static SV *req_sv (aio_req req, const char *klass) 376static SV *req_sv (aio_req req, const char *klass)
111{ 377{
112 if (!req->self) 378 if (!req->self)
118 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1))); 384 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
119} 385}
120 386
121static aio_req SvAIO_REQ (SV *sv) 387static aio_req SvAIO_REQ (SV *sv)
122{ 388{
389 MAGIC *mg;
390
123 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) 391 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
124 croak ("object of class " AIO_REQ_KLASS " expected"); 392 croak ("object of class " AIO_REQ_KLASS " expected");
125 393
126 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 394 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
127 395
128 return mg ? (aio_req)mg->mg_ptr : 0; 396 return mg ? (aio_req)mg->mg_ptr : 0;
129} 397}
130 398
131static void aio_grp_feed (aio_req grp) 399static void aio_grp_feed (aio_req grp)
132{ 400{
133 while (grp->length < grp->fd2) 401 block_sig ();
402
403 while (grp->size < grp->int2 && !(grp->flags & FLAG_CANCELLED))
134 { 404 {
135 int old_len = grp->length; 405 int old_len = grp->size;
136 406
137 if (grp->fh2 && SvOK (grp->fh2)) 407 if (grp->sv2 && SvOK (grp->sv2))
138 { 408 {
139 dSP; 409 dSP;
140 410
141 ENTER; 411 ENTER;
142 SAVETMPS; 412 SAVETMPS;
143 PUSHMARK (SP); 413 PUSHMARK (SP);
144 XPUSHs (req_sv (grp, AIO_GRP_KLASS)); 414 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
145 PUTBACK; 415 PUTBACK;
146 call_sv (grp->fh2, G_VOID | G_EVAL); 416 call_sv (grp->sv2, G_VOID | G_EVAL | G_KEEPERR);
147 SPAGAIN; 417 SPAGAIN;
148 FREETMPS; 418 FREETMPS;
149 LEAVE; 419 LEAVE;
150 } 420 }
151 421
152 /* stop if no progress has been made */ 422 /* stop if no progress has been made */
153 if (old_len == grp->length) 423 if (old_len == grp->size)
154 { 424 {
155 SvREFCNT_dec (grp->fh2); 425 SvREFCNT_dec (grp->sv2);
156 grp->fh2 = 0; 426 grp->sv2 = 0;
157 break; 427 break;
158 } 428 }
159 } 429 }
430
431 unblock_sig ();
160} 432}
161 433
162static void aio_grp_dec (aio_req grp) 434static void aio_grp_dec (aio_req grp)
163{ 435{
164 --grp->length; 436 --grp->size;
165 437
166 /* call feeder, if applicable */ 438 /* call feeder, if applicable */
167 aio_grp_feed (grp); 439 aio_grp_feed (grp);
168 440
169 /* finish, if done */ 441 /* finish, if done */
170 if (!grp->length && grp->fd) 442 if (!grp->size && grp->int1)
171 { 443 {
444 block_sig ();
445
172 req_invoke (grp); 446 if (!req_invoke (grp))
447 {
448 req_destroy (grp);
449 unblock_sig ();
450 croak (0);
451 }
452
173 req_free (grp); 453 req_destroy (grp);
454 unblock_sig ();
174 } 455 }
175} 456}
176 457
177static void poll_wait () 458static int req_invoke (aio_req req)
178{ 459{
179 if (nreqs && !ress) 460 dSP;
461
462 if (req->flags & FLAG_SV2_RO_OFF)
463 SvREADONLY_off (req->sv2);
464
465 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
180 { 466 {
181 fd_set rfd; 467 ENTER;
182 FD_ZERO(&rfd); 468 SAVETMPS;
183 FD_SET(respipe [0], &rfd); 469 PUSHMARK (SP);
470 EXTEND (SP, 1);
184 471
185 select (respipe [0] + 1, &rfd, 0, 0, 0); 472 switch (req->type)
473 {
474 case REQ_READDIR:
475 {
476 SV *rv = &PL_sv_undef;
477
478 if (req->result >= 0)
479 {
480 int i;
481 char *buf = req->ptr2;
482 AV *av = newAV ();
483
484 av_extend (av, req->result - 1);
485
486 for (i = 0; i < req->result; ++i)
487 {
488 SV *sv = newSVpv (buf, 0);
489
490 av_store (av, i, sv);
491 buf += SvCUR (sv) + 1;
492 }
493
494 rv = sv_2mortal (newRV_noinc ((SV *)av));
495 }
496
497 PUSHs (rv);
498 }
499 break;
500
501 case REQ_OPEN:
502 {
503 /* convert fd to fh */
504 SV *fh;
505
506 PUSHs (sv_2mortal (newSViv (req->result)));
507 PUTBACK;
508 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
509 SPAGAIN;
510
511 fh = POPs;
512 PUSHMARK (SP);
513 XPUSHs (fh);
514 }
515 break;
516
517 case REQ_GROUP:
518 req->int1 = 2; /* mark group as finished */
519
520 if (req->sv1)
521 {
522 int i;
523 AV *av = (AV *)req->sv1;
524
525 EXTEND (SP, AvFILL (av) + 1);
526 for (i = 0; i <= AvFILL (av); ++i)
527 PUSHs (*av_fetch (av, i, 0));
528 }
529 break;
530
531 case REQ_NOP:
532 case REQ_BUSY:
533 break;
534
535 case REQ_READLINK:
536 if (req->result > 0)
537 {
538 SvCUR_set (req->sv2, req->result);
539 *SvEND (req->sv2) = 0;
540 PUSHs (req->sv2);
541 }
542 break;
543
544 case REQ_STAT:
545 case REQ_LSTAT:
546 case REQ_FSTAT:
547 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
548 PL_laststatval = req->result;
549 PL_statcache = *(Stat_t *)(req->ptr2);
550 PUSHs (sv_2mortal (newSViv (req->result)));
551 break;
552
553 case REQ_READ:
554 SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0));
555 *SvEND (req->sv2) = 0;
556 PUSHs (sv_2mortal (newSViv (req->result)));
557 break;
558
559 default:
560 PUSHs (sv_2mortal (newSViv (req->result)));
561 break;
562 }
563
564 errno = req->errorno;
565
566 PUTBACK;
567 call_sv (req->callback, G_VOID | G_EVAL);
568 SPAGAIN;
569
570 FREETMPS;
571 LEAVE;
186 } 572 }
187}
188 573
189static void req_invoke (aio_req req)
190{
191 dSP;
192 int errorno = errno;
193
194 if (req->cancelled || !SvOK (req->callback))
195 return;
196
197 errno = req->errorno;
198
199 ENTER;
200 SAVETMPS;
201 PUSHMARK (SP);
202 EXTEND (SP, 1);
203
204 switch (req->type)
205 {
206 case REQ_READDIR:
207 {
208 SV *rv = &PL_sv_undef;
209
210 if (req->result >= 0)
211 {
212 char *buf = req->data2ptr;
213 AV *av = newAV ();
214
215 while (req->result)
216 {
217 SV *sv = newSVpv (buf, 0);
218
219 av_push (av, sv);
220 buf += SvCUR (sv) + 1;
221 req->result--;
222 }
223
224 rv = sv_2mortal (newRV_noinc ((SV *)av));
225 }
226
227 PUSHs (rv);
228 }
229 break;
230
231 case REQ_OPEN:
232 {
233 /* convert fd to fh */
234 SV *fh;
235
236 PUSHs (sv_2mortal (newSViv (req->result)));
237 PUTBACK;
238 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
239 SPAGAIN;
240
241 fh = SvREFCNT_inc (POPs);
242
243 PUSHMARK (SP);
244 XPUSHs (sv_2mortal (fh));
245 }
246 break;
247
248 case REQ_GROUP:
249 req->fd = 2; /* mark group as finished */
250
251 if (req->data)
252 {
253 int i;
254 AV *av = (AV *)req->data;
255
256 EXTEND (SP, AvFILL (av) + 1);
257 for (i = 0; i <= AvFILL (av); ++i)
258 PUSHs (*av_fetch (av, i, 0));
259 }
260 break;
261
262 case REQ_SLEEP:
263 break;
264
265 default:
266 PUSHs (sv_2mortal (newSViv (req->result)));
267 break;
268 }
269
270
271 PUTBACK;
272 call_sv (req->callback, G_VOID | G_EVAL);
273 SPAGAIN;
274
275 FREETMPS;
276 LEAVE;
277
278 errno = errorno;
279
280 if (SvTRUE (ERRSV))
281 {
282 req_free (req);
283 croak (0);
284 }
285}
286
287static void req_free (aio_req req)
288{
289 if (req->grp) 574 if (req->grp)
290 { 575 {
291 aio_req grp = req->grp; 576 aio_req grp = req->grp;
292 577
293 /* unlink request */ 578 /* unlink request */
298 grp->grp_first = req->grp_next; 583 grp->grp_first = req->grp_next;
299 584
300 aio_grp_dec (grp); 585 aio_grp_dec (grp);
301 } 586 }
302 587
588 return !SvTRUE (ERRSV);
589}
590
591static void req_destroy (aio_req req)
592{
303 if (req->self) 593 if (req->self)
304 { 594 {
305 sv_unmagic (req->self, PERL_MAGIC_ext); 595 sv_unmagic (req->self, PERL_MAGIC_ext);
306 SvREFCNT_dec (req->self); 596 SvREFCNT_dec (req->self);
307 } 597 }
308 598
309 SvREFCNT_dec (req->data);
310 SvREFCNT_dec (req->fh); 599 SvREFCNT_dec (req->sv1);
311 SvREFCNT_dec (req->fh2); 600 SvREFCNT_dec (req->sv2);
312 SvREFCNT_dec (req->callback); 601 SvREFCNT_dec (req->callback);
313 Safefree (req->statdata);
314 602
315 if (req->type == REQ_READDIR && req->result >= 0) 603 if (req->flags & FLAG_PTR2_FREE)
316 free (req->data2ptr); 604 free (req->ptr2);
317 605
318 Safefree (req); 606 Safefree (req);
319} 607}
320 608
609static void req_cancel_subs (aio_req grp)
610{
611 aio_req sub;
612
613 if (grp->type != REQ_GROUP)
614 return;
615
616 SvREFCNT_dec (grp->sv2);
617 grp->sv2 = 0;
618
619 for (sub = grp->grp_first; sub; sub = sub->grp_next)
620 req_cancel (sub);
621}
622
321static void req_cancel (aio_req req) 623static void req_cancel (aio_req req)
322{ 624{
323 req->cancelled = 1; 625 req->flags |= FLAG_CANCELLED;
324 626
325 if (req->type == REQ_GROUP) 627 req_cancel_subs (req);
628}
629
630X_THREAD_PROC (aio_proc);
631
632static void start_thread (void)
633{
634 worker *wrk = calloc (1, sizeof (worker));
635
636 if (!wrk)
637 croak ("unable to allocate worker thread data");
638
639 X_LOCK (wrklock);
640
641 if (thread_create (&wrk->tid, aio_proc, (void *)wrk))
326 { 642 {
327 aio_req sub; 643 wrk->prev = &wrk_first;
644 wrk->next = wrk_first.next;
645 wrk_first.next->prev = wrk;
646 wrk_first.next = wrk;
647 ++started;
648 }
649 else
650 free (wrk);
328 651
329 for (sub = req->grp_first; sub; sub = sub->grp_next) 652 X_UNLOCK (wrklock);
330 req_cancel (sub); 653}
654
655static void maybe_start_thread ()
656{
657 if (get_nthreads () >= wanted)
658 return;
659
660 /* todo: maybe use idle here, but might be less exact */
661 if (0 <= (int)get_nthreads () + (int)get_npending () - (int)nreqs)
662 return;
663
664 start_thread ();
665}
666
667static void req_send (aio_req req)
668{
669 block_sig ();
670
671 ++nreqs;
672
673 X_LOCK (reqlock);
674 ++nready;
675 reqq_push (&req_queue, req);
676 X_COND_SIGNAL (reqwait);
677 X_UNLOCK (reqlock);
678
679 unblock_sig ();
680
681 maybe_start_thread ();
682}
683
684static void end_thread (void)
685{
686 aio_req req;
687
688 Newz (0, req, 1, aio_cb);
689
690 req->type = REQ_QUIT;
691 req->pri = PRI_MAX + PRI_BIAS;
692
693 X_LOCK (reqlock);
694 reqq_push (&req_queue, req);
695 X_COND_SIGNAL (reqwait);
696 X_UNLOCK (reqlock);
697
698 X_LOCK (wrklock);
699 --started;
700 X_UNLOCK (wrklock);
701}
702
703static void set_max_idle (int nthreads)
704{
705 if (WORDACCESS_UNSAFE) X_LOCK (reqlock);
706 max_idle = nthreads <= 0 ? 1 : nthreads;
707 if (WORDACCESS_UNSAFE) X_UNLOCK (reqlock);
708}
709
710static void min_parallel (int nthreads)
711{
712 if (wanted < nthreads)
713 wanted = nthreads;
714}
715
716static void max_parallel (int nthreads)
717{
718 if (wanted > nthreads)
719 wanted = nthreads;
720
721 while (started > wanted)
722 end_thread ();
723}
724
725static void poll_wait ()
726{
727 fd_set rfd;
728
729 while (nreqs)
730 {
731 int size;
732 if (WORDACCESS_UNSAFE) X_LOCK (reslock);
733 size = res_queue.size;
734 if (WORDACCESS_UNSAFE) X_UNLOCK (reslock);
735
736 if (size)
737 return;
738
739 maybe_start_thread ();
740
741 FD_ZERO(&rfd);
742 FD_SET(respipe [0], &rfd);
743
744 select (respipe [0] + 1, &rfd, 0, 0, 0);
331 } 745 }
332} 746}
333 747
334static int poll_cb () 748static int poll_cb ()
335{ 749{
336 dSP; 750 dSP;
337 int count = 0; 751 int count = 0;
752 int maxreqs = max_poll_reqs;
338 int do_croak = 0; 753 int do_croak = 0;
754 struct timeval tv_start, tv_now;
339 aio_req req; 755 aio_req req;
756
757 if (max_poll_time)
758 gettimeofday (&tv_start, 0);
759
760 block_sig ();
340 761
341 for (;;) 762 for (;;)
342 { 763 {
343 pthread_mutex_lock (&reslock); 764 for (;;)
344 req = ress;
345
346 if (req)
347 { 765 {
348 ress = req->next; 766 maybe_start_thread ();
349 767
768 X_LOCK (reslock);
769 req = reqq_shift (&res_queue);
770
350 if (!ress) 771 if (req)
351 { 772 {
773 --npending;
774
775 if (!res_queue.size)
776 {
352 /* read any signals sent by the worker threads */ 777 /* read any signals sent by the worker threads */
353 char buf [32]; 778 char buf [4];
354 while (read (respipe [0], buf, 32) == 32) 779 while (read (respipe [0], buf, 4) == 4)
780 ;
355 ; 781 }
782 }
356 783
784 X_UNLOCK (reslock);
785
786 if (!req)
787 break;
788
789 --nreqs;
790
791 if (req->type == REQ_GROUP && req->size)
792 {
793 req->int1 = 1; /* mark request as delayed */
794 continue;
795 }
796 else
797 {
798 if (!req_invoke (req))
799 {
800 req_destroy (req);
801 unblock_sig ();
802 croak (0);
803 }
804
805 count++;
806 }
807
808 req_destroy (req);
809
810 if (maxreqs && !--maxreqs)
811 break;
812
813 if (max_poll_time)
814 {
815 gettimeofday (&tv_now, 0);
816
817 if (tvdiff (&tv_start, &tv_now) >= max_poll_time)
357 rese = 0; 818 break;
358 } 819 }
359 } 820 }
360 821
361 pthread_mutex_unlock (&reslock); 822 if (nreqs <= max_outstanding)
362
363 if (!req)
364 break; 823 break;
365 824
366 --nreqs; 825 poll_wait ();
367 826
368 if (req->type == REQ_QUIT) 827 ++maxreqs;
369 started--;
370 else if (req->type == REQ_GROUP && req->length)
371 {
372 req->fd = 1; /* mark request as delayed */
373 continue;
374 }
375 else
376 {
377 if (req->type == REQ_READ)
378 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
379
380 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
381 SvREADONLY_off (req->data);
382
383 if (req->statdata)
384 {
385 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
386 PL_laststatval = req->result;
387 PL_statcache = *(req->statdata);
388 }
389
390 req_invoke (req);
391
392 count++;
393 }
394
395 req_free (req);
396 } 828 }
397 829
830 unblock_sig ();
398 return count; 831 return count;
399}
400
401static void *aio_proc(void *arg);
402
403static void start_thread (void)
404{
405 sigset_t fullsigset, oldsigset;
406 pthread_t tid;
407 pthread_attr_t attr;
408
409 pthread_attr_init (&attr);
410 pthread_attr_setstacksize (&attr, STACKSIZE);
411 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
412
413 sigfillset (&fullsigset);
414 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
415
416 if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
417 started++;
418
419 sigprocmask (SIG_SETMASK, &oldsigset, 0);
420}
421
422static void req_send (aio_req req)
423{
424 while (started < wanted && nreqs >= started)
425 start_thread ();
426
427 ++nreqs;
428
429 pthread_mutex_lock (&reqlock);
430
431 req->next = 0;
432
433 if (reqe)
434 {
435 reqe->next = req;
436 reqe = req;
437 }
438 else
439 reqe = reqs = req;
440
441 pthread_cond_signal (&reqwait);
442 pthread_mutex_unlock (&reqlock);
443
444 if (nreqs > max_outstanding)
445 for (;;)
446 {
447 poll_cb ();
448
449 if (nreqs <= max_outstanding)
450 break;
451
452 poll_wait ();
453 }
454}
455
456static void end_thread (void)
457{
458 aio_req req;
459 Newz (0, req, 1, aio_cb);
460 req->type = REQ_QUIT;
461
462 req_send (req);
463}
464
465static void min_parallel (int nthreads)
466{
467 if (wanted < nthreads)
468 wanted = nthreads;
469}
470
471static void max_parallel (int nthreads)
472{
473 int cur = started;
474
475 if (wanted > nthreads)
476 wanted = nthreads;
477
478 while (cur > wanted)
479 {
480 end_thread ();
481 cur--;
482 }
483
484 while (started > wanted)
485 {
486 poll_wait ();
487 poll_cb ();
488 }
489}
490
491static void create_pipe ()
492{
493 if (pipe (respipe))
494 croak ("unable to initialize result pipe");
495
496 if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
497 croak ("cannot set result pipe to nonblocking mode");
498
499 if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
500 croak ("cannot set result pipe to nonblocking mode");
501} 832}
502 833
503/*****************************************************************************/ 834/*****************************************************************************/
504/* work around various missing functions */ 835/* work around various missing functions */
505 836
510/* 841/*
511 * make our pread/pwrite safe against themselves, but not against 842 * make our pread/pwrite safe against themselves, but not against
512 * normal read/write by using a mutex. slows down execution a lot, 843 * normal read/write by using a mutex. slows down execution a lot,
513 * but that's your problem, not mine. 844 * but that's your problem, not mine.
514 */ 845 */
515static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER; 846static mutex_t preadwritelock = X_MUTEX_INIT;
516 847
517static ssize_t pread (int fd, void *buf, size_t count, off_t offset) 848static ssize_t pread (int fd, void *buf, size_t count, off_t offset)
518{ 849{
519 ssize_t res; 850 ssize_t res;
520 off_t ooffset; 851 off_t ooffset;
521 852
522 pthread_mutex_lock (&preadwritelock); 853 X_LOCK (preadwritelock);
523 ooffset = lseek (fd, 0, SEEK_CUR); 854 ooffset = lseek (fd, 0, SEEK_CUR);
524 lseek (fd, offset, SEEK_SET); 855 lseek (fd, offset, SEEK_SET);
525 res = read (fd, buf, count); 856 res = read (fd, buf, count);
526 lseek (fd, ooffset, SEEK_SET); 857 lseek (fd, ooffset, SEEK_SET);
527 pthread_mutex_unlock (&preadwritelock); 858 X_UNLOCK (preadwritelock);
528 859
529 return res; 860 return res;
530} 861}
531 862
532static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset) 863static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset)
533{ 864{
534 ssize_t res; 865 ssize_t res;
535 off_t ooffset; 866 off_t ooffset;
536 867
537 pthread_mutex_lock (&preadwritelock); 868 X_LOCK (preadwritelock);
538 ooffset = lseek (fd, 0, SEEK_CUR); 869 ooffset = lseek (fd, 0, SEEK_CUR);
539 lseek (fd, offset, SEEK_SET); 870 lseek (fd, offset, SEEK_SET);
540 res = write (fd, buf, count); 871 res = write (fd, buf, count);
541 lseek (fd, offset, SEEK_SET); 872 lseek (fd, offset, SEEK_SET);
542 pthread_mutex_unlock (&preadwritelock); 873 X_UNLOCK (preadwritelock);
543 874
544 return res; 875 return res;
545} 876}
877#endif
878
879#ifndef HAVE_FUTIMES
880
881# define utimes(path,times) aio_utimes (path, times)
882# define futimes(fd,times) aio_futimes (fd, times)
883
884int aio_utimes (const char *filename, const struct timeval times[2])
885{
886 if (times)
887 {
888 struct utimbuf buf;
889
890 buf.actime = times[0].tv_sec;
891 buf.modtime = times[1].tv_sec;
892
893 return utime (filename, &buf);
894 }
895 else
896 return utime (filename, 0);
897}
898
899int aio_futimes (int fd, const struct timeval tv[2])
900{
901 errno = ENOSYS;
902 return -1;
903}
904
546#endif 905#endif
547 906
548#if !HAVE_FDATASYNC 907#if !HAVE_FDATASYNC
549# define fdatasync fsync 908# define fdatasync fsync
550#endif 909#endif
551 910
552#if !HAVE_READAHEAD 911#if !HAVE_READAHEAD
553# define readahead aio_readahead 912# define readahead(fd,offset,count) aio_readahead (fd, offset, count, self)
554 913
555static ssize_t readahead (int fd, off_t offset, size_t count) 914static ssize_t aio_readahead (int fd, off_t offset, size_t count, worker *self)
556{ 915{
557 char readahead_buf[4096]; 916 size_t todo = count;
917 dBUF;
558 918
559 while (count > 0) 919 while (todo > 0)
560 { 920 {
561 size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf); 921 size_t len = todo < AIO_BUFSIZE ? todo : AIO_BUFSIZE;
562 922
563 pread (fd, readahead_buf, len, offset); 923 pread (fd, aio_buf, len, offset);
564 offset += len; 924 offset += len;
565 count -= len; 925 todo -= len;
566 } 926 }
567 927
568 errno = 0; 928 errno = 0;
929 return count;
569} 930}
931
570#endif 932#endif
571 933
572#if !HAVE_READDIR_R 934#if !HAVE_READDIR_R
573# define readdir_r aio_readdir_r 935# define readdir_r aio_readdir_r
574 936
575static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER; 937static mutex_t readdirlock = X_MUTEX_INIT;
576 938
577static int readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res) 939static int readdir_r (DIR *dirp, X_DIRENT *ent, X_DIRENT **res)
578{ 940{
579 struct dirent *e; 941 X_DIRENT *e;
580 int errorno; 942 int errorno;
581 943
582 pthread_mutex_lock (&readdirlock); 944 X_LOCK (readdirlock);
583 945
584 e = readdir (dirp); 946 e = readdir (dirp);
585 errorno = errno; 947 errorno = errno;
586 948
587 if (e) 949 if (e)
590 strcpy (ent->d_name, e->d_name); 952 strcpy (ent->d_name, e->d_name);
591 } 953 }
592 else 954 else
593 *res = 0; 955 *res = 0;
594 956
595 pthread_mutex_unlock (&readdirlock); 957 X_UNLOCK (readdirlock);
596 958
597 errno = errorno; 959 errno = errorno;
598 return e ? 0 : -1; 960 return e ? 0 : -1;
599} 961}
600#endif 962#endif
601 963
602/* sendfile always needs emulation */ 964/* sendfile always needs emulation */
603static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count) 965static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count, worker *self)
604{ 966{
605 ssize_t res; 967 ssize_t res;
606 968
607 if (!count) 969 if (!count)
608 return 0; 970 return 0;
619 { 981 {
620 off_t sbytes; 982 off_t sbytes;
621 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0); 983 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
622 984
623 if (res < 0 && sbytes) 985 if (res < 0 && sbytes)
624 /* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */ 986 /* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
625 res = sbytes; 987 res = sbytes;
626 } 988 }
627 989
628# elif __hpux 990# elif __hpux
629 res = sendfile (ofd, ifd, offset, count, 0, 0); 991 res = sendfile (ofd, ifd, offset, count, 0, 0);
657#endif 1019#endif
658 ) 1020 )
659 ) 1021 )
660 { 1022 {
661 /* emulate sendfile. this is a major pain in the ass */ 1023 /* emulate sendfile. this is a major pain in the ass */
662 char buf[4096]; 1024 dBUF;
1025
663 res = 0; 1026 res = 0;
664 1027
665 while (count) 1028 while (count)
666 { 1029 {
667 ssize_t cnt; 1030 ssize_t cnt;
668 1031
669 cnt = pread (ifd, buf, count > 4096 ? 4096 : count, offset); 1032 cnt = pread (ifd, aio_buf, count > AIO_BUFSIZE ? AIO_BUFSIZE : count, offset);
670 1033
671 if (cnt <= 0) 1034 if (cnt <= 0)
672 { 1035 {
673 if (cnt && !res) res = -1; 1036 if (cnt && !res) res = -1;
674 break; 1037 break;
675 } 1038 }
676 1039
677 cnt = write (ofd, buf, cnt); 1040 cnt = write (ofd, aio_buf, cnt);
678 1041
679 if (cnt <= 0) 1042 if (cnt <= 0)
680 { 1043 {
681 if (cnt && !res) res = -1; 1044 if (cnt && !res) res = -1;
682 break; 1045 break;
690 1053
691 return res; 1054 return res;
692} 1055}
693 1056
694/* read a full directory */ 1057/* read a full directory */
695static int scandir_ (const char *path, void **namesp) 1058static void scandir_ (aio_req req, worker *self)
696{ 1059{
697 DIR *dirp = opendir (path); 1060 DIR *dirp;
698 union 1061 union
699 { 1062 {
700 struct dirent d; 1063 X_DIRENT d;
701 char b [offsetof (struct dirent, d_name) + NAME_MAX + 1]; 1064 char b [offsetof (X_DIRENT, d_name) + NAME_MAX + 1];
702 } u; 1065 } *u;
703 struct dirent *entp; 1066 X_DIRENT *entp;
704 char *name, *names; 1067 char *name, *names;
705 int memlen = 4096; 1068 int memlen = 4096;
706 int memofs = 0; 1069 int memofs = 0;
707 int res = 0; 1070 int res = 0;
708 int errorno;
709 1071
710 if (!dirp) 1072 X_LOCK (wrklock);
711 return -1; 1073 self->dirp = dirp = opendir (req->ptr1);
712 1074 self->dbuf = u = malloc (sizeof (*u));
1075 req->flags |= FLAG_PTR2_FREE;
713 names = malloc (memlen); 1076 req->ptr2 = names = malloc (memlen);
1077 X_UNLOCK (wrklock);
1078
1079 if (dirp && u && names)
1080 for (;;)
1081 {
1082 errno = 0;
1083 readdir_r (dirp, &u->d, &entp);
1084
1085 if (!entp)
1086 break;
1087
1088 name = entp->d_name;
1089
1090 if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
1091 {
1092 int len = strlen (name) + 1;
1093
1094 res++;
1095
1096 while (memofs + len > memlen)
1097 {
1098 memlen *= 2;
1099 X_LOCK (wrklock);
1100 req->ptr2 = names = realloc (names, memlen);
1101 X_UNLOCK (wrklock);
1102
1103 if (!names)
1104 break;
1105 }
1106
1107 memcpy (names + memofs, name, len);
1108 memofs += len;
1109 }
1110 }
1111
1112 if (errno)
1113 res = -1;
1114
1115 req->result = res;
1116}
1117
1118/*****************************************************************************/
1119
1120X_THREAD_PROC (aio_proc)
1121{
1122 {//D
1123 aio_req req;
1124 struct timespec ts;
1125 worker *self = (worker *)thr_arg;
1126
1127 /* try to distribute timeouts somewhat randomly */
1128 ts.tv_nsec = ((unsigned long)self & 1023UL) * (1000000000UL / 1024UL);
714 1129
715 for (;;) 1130 for (;;)
716 { 1131 {
717 errno = 0, readdir_r (dirp, &u.d, &entp); 1132 ts.tv_sec = time (0) + IDLE_TIMEOUT;
718 1133
719 if (!entp) 1134 X_LOCK (reqlock);
720 break;
721
722 name = entp->d_name;
723
724 if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
725 {
726 int len = strlen (name) + 1;
727
728 res++;
729
730 while (memofs + len > memlen)
731 {
732 memlen *= 2;
733 names = realloc (names, memlen);
734 if (!names)
735 break;
736 }
737
738 memcpy (names + memofs, name, len);
739 memofs += len;
740 }
741 }
742
743 errorno = errno;
744 closedir (dirp);
745
746 if (errorno)
747 {
748 free (names);
749 errno = errorno;
750 res = -1;
751 }
752
753 *namesp = (void *)names;
754 return res;
755}
756
757/*****************************************************************************/
758
759static void *aio_proc (void *thr_arg)
760{
761 aio_req req;
762 int type;
763
764 do
765 {
766 pthread_mutex_lock (&reqlock);
767 1135
768 for (;;) 1136 for (;;)
769 { 1137 {
770 req = reqs; 1138 self->req = req = reqq_shift (&req_queue);
771
772 if (reqs)
773 {
774 reqs = reqs->next;
775 if (!reqs) reqe = 0;
776 }
777 1139
778 if (req) 1140 if (req)
779 break; 1141 break;
780 1142
781 pthread_cond_wait (&reqwait, &reqlock); 1143 ++idle;
1144
1145 if (X_COND_TIMEDWAIT (reqwait, reqlock, ts)
1146 == ETIMEDOUT)
1147 {
1148 if (idle > max_idle)
1149 {
1150 --idle;
1151 X_UNLOCK (reqlock);
1152 X_LOCK (wrklock);
1153 --started;
1154 X_UNLOCK (wrklock);
1155 goto quit;
1156 }
1157
1158 /* we are allowed to idle, so do so without any timeout */
1159 X_COND_WAIT (reqwait, reqlock);
1160 ts.tv_sec = time (0) + IDLE_TIMEOUT;
1161 }
1162
1163 --idle;
782 } 1164 }
783 1165
784 pthread_mutex_unlock (&reqlock); 1166 --nready;
1167
1168 X_UNLOCK (reqlock);
785 1169
786 errno = 0; /* strictly unnecessary */ 1170 errno = 0; /* strictly unnecessary */
787 1171
788 if (!req->cancelled) 1172 if (!(req->flags & FLAG_CANCELLED))
789 switch (type = req->type) /* remember type for QUIT check */ 1173 switch (req->type)
790 { 1174 {
791 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 1175 case REQ_READ: req->result = req->offs >= 0
792 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 1176 ? pread (req->int1, req->ptr1, req->size, req->offs)
1177 : read (req->int1, req->ptr1, req->size); break;
1178 case REQ_WRITE: req->result = req->offs >= 0
1179 ? pwrite (req->int1, req->ptr1, req->size, req->offs)
1180 : write (req->int1, req->ptr1, req->size); break;
793 1181
794 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 1182 case REQ_READAHEAD: req->result = readahead (req->int1, req->offs, req->size); break;
795 case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break; 1183 case REQ_SENDFILE: req->result = sendfile_ (req->int1, req->int2, req->offs, req->size, self); break;
796 1184
797 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break; 1185 case REQ_STAT: req->result = stat (req->ptr1, (Stat_t *)req->ptr2); break;
798 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break; 1186 case REQ_LSTAT: req->result = lstat (req->ptr1, (Stat_t *)req->ptr2); break;
799 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break; 1187 case REQ_FSTAT: req->result = fstat (req->int1, (Stat_t *)req->ptr2); break;
800 1188
1189 case REQ_CHOWN: req->result = chown (req->ptr1, req->int2, req->int3); break;
1190 case REQ_FCHOWN: req->result = fchown (req->int1, req->int2, req->int3); break;
1191 case REQ_CHMOD: req->result = chmod (req->ptr1, req->mode); break;
1192 case REQ_FCHMOD: req->result = fchmod (req->int1, req->mode); break;
1193 case REQ_TRUNCATE: req->result = truncate (req->ptr1, req->offs); break;
1194 case REQ_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break;
1195
801 case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break; 1196 case REQ_OPEN: req->result = open (req->ptr1, req->int1, req->mode); break;
802 case REQ_CLOSE: req->result = close (req->fd); break; 1197 case REQ_CLOSE: req->result = close (req->int1); break;
803 case REQ_UNLINK: req->result = unlink (req->dataptr); break; 1198 case REQ_UNLINK: req->result = unlink (req->ptr1); break;
804 case REQ_RMDIR: req->result = rmdir (req->dataptr); break; 1199 case REQ_RMDIR: req->result = rmdir (req->ptr1); break;
1200 case REQ_MKDIR: req->result = mkdir (req->ptr1, req->mode); break;
805 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break; 1201 case REQ_RENAME: req->result = rename (req->ptr2, req->ptr1); break;
806 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break; 1202 case REQ_LINK: req->result = link (req->ptr2, req->ptr1); break;
807 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break; 1203 case REQ_SYMLINK: req->result = symlink (req->ptr2, req->ptr1); break;
1204 case REQ_MKNOD: req->result = mknod (req->ptr2, req->mode, (dev_t)req->offs); break;
1205 case REQ_READLINK: req->result = readlink (req->ptr2, req->ptr1, NAME_MAX); break;
808 1206
809 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 1207 case REQ_FDATASYNC: req->result = fdatasync (req->int1); break;
810 case REQ_FSYNC: req->result = fsync (req->fd); break; 1208 case REQ_FSYNC: req->result = fsync (req->int1); break;
811 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break; 1209 case REQ_READDIR: scandir_ (req, self); break;
812 1210
813 case REQ_SLEEP: 1211 case REQ_BUSY:
1212#ifdef _WIN32
1213 Sleep (req->nv1 * 1000.);
1214#else
814 { 1215 {
815 struct timeval tv; 1216 struct timeval tv;
816 1217
817 tv.tv_sec = req->fd; 1218 tv.tv_sec = req->nv1;
818 tv.tv_usec = req->fd2; 1219 tv.tv_usec = (req->nv1 - tv.tv_sec) * 1000000.;
819 1220
820 req->result = select (0, 0, 0, 0, &tv); 1221 req->result = select (0, 0, 0, 0, &tv);
821 } 1222 }
1223#endif
1224 break;
1225
1226 case REQ_UTIME:
1227 case REQ_FUTIME:
1228 {
1229 struct timeval tv[2];
1230 struct timeval *times;
1231
1232 if (req->nv1 != -1. || req->nv2 != -1.)
1233 {
1234 tv[0].tv_sec = req->nv1;
1235 tv[0].tv_usec = (req->nv1 - tv[0].tv_sec) * 1000000.;
1236 tv[1].tv_sec = req->nv2;
1237 tv[1].tv_usec = (req->nv2 - tv[1].tv_sec) * 1000000.;
1238
1239 times = tv;
1240 }
1241 else
1242 times = 0;
1243
1244
1245 req->result = req->type == REQ_FUTIME
1246 ? futimes (req->int1, times)
1247 : utimes (req->ptr1, times);
1248 }
1249
1250 case REQ_GROUP:
1251 case REQ_NOP:
1252 break;
822 1253
823 case REQ_QUIT: 1254 case REQ_QUIT:
824 break; 1255 goto quit;
825 1256
826 default: 1257 default:
827 req->result = ENOSYS; 1258 req->result = -1;
828 break; 1259 break;
829 } 1260 }
830 1261
831 req->errorno = errno; 1262 req->errorno = errno;
832 1263
833 pthread_mutex_lock (&reslock); 1264 X_LOCK (reslock);
834 1265
835 req->next = 0; 1266 ++npending;
836 1267
837 if (rese) 1268 if (!reqq_push (&res_queue, req))
838 { 1269 {
839 rese->next = req;
840 rese = req;
841 }
842 else
843 {
844 rese = ress = req;
845
846 /* write a dummy byte to the pipe so fh becomes ready */ 1270 /* write a dummy byte to the pipe so fh becomes ready */
847 write (respipe [1], &respipe, 1); 1271 write (respipe [1], &respipe, 1);
1272
1273 /* optionally signal the main thread asynchronously */
1274 if (main_sig)
1275 pthread_kill (main_tid, main_sig);
848 } 1276 }
849 1277
850 pthread_mutex_unlock (&reslock); 1278 self->req = 0;
1279 worker_clear (self);
1280
1281 X_UNLOCK (reslock);
851 } 1282 }
852 while (type != REQ_QUIT); 1283
1284quit:
1285 X_LOCK (wrklock);
1286 worker_free (self);
1287 X_UNLOCK (wrklock);
853 1288
854 return 0; 1289 return 0;
1290 }//D
855} 1291}
856 1292
857/*****************************************************************************/ 1293/*****************************************************************************/
858 1294
859static void atfork_prepare (void) 1295static void atfork_prepare (void)
860{ 1296{
861 pthread_mutex_lock (&reqlock); 1297 X_LOCK (wrklock);
862 pthread_mutex_lock (&reslock); 1298 X_LOCK (reqlock);
1299 X_LOCK (reslock);
863#if !HAVE_PREADWRITE 1300#if !HAVE_PREADWRITE
864 pthread_mutex_lock (&preadwritelock); 1301 X_LOCK (preadwritelock);
865#endif 1302#endif
866#if !HAVE_READDIR_R 1303#if !HAVE_READDIR_R
867 pthread_mutex_lock (&readdirlock); 1304 X_LOCK (readdirlock);
868#endif 1305#endif
869} 1306}
870 1307
871static void atfork_parent (void) 1308static void atfork_parent (void)
872{ 1309{
873#if !HAVE_READDIR_R 1310#if !HAVE_READDIR_R
874 pthread_mutex_unlock (&readdirlock); 1311 X_UNLOCK (readdirlock);
875#endif 1312#endif
876#if !HAVE_PREADWRITE 1313#if !HAVE_PREADWRITE
877 pthread_mutex_unlock (&preadwritelock); 1314 X_UNLOCK (preadwritelock);
878#endif 1315#endif
879 pthread_mutex_unlock (&reslock); 1316 X_UNLOCK (reslock);
880 pthread_mutex_unlock (&reqlock); 1317 X_UNLOCK (reqlock);
1318 X_UNLOCK (wrklock);
881} 1319}
882 1320
883static void atfork_child (void) 1321static void atfork_child (void)
884{ 1322{
885 aio_req prv; 1323 aio_req prv;
886 1324
887 started = 0; 1325 while (prv = reqq_shift (&req_queue))
1326 req_destroy (prv);
888 1327
889 while (reqs) 1328 while (prv = reqq_shift (&res_queue))
1329 req_destroy (prv);
1330
1331 while (wrk_first.next != &wrk_first)
890 { 1332 {
891 prv = reqs; 1333 worker *wrk = wrk_first.next;
892 reqs = prv->next; 1334
1335 if (wrk->req)
1336 req_destroy (wrk->req);
1337
1338 worker_clear (wrk);
893 req_free (prv); 1339 worker_free (wrk);
894 } 1340 }
895 1341
896 reqs = reqe = 0; 1342 started = 0;
897 1343 idle = 0;
898 while (ress) 1344 nreqs = 0;
899 { 1345 nready = 0;
900 prv = ress; 1346 npending = 0;
901 ress = prv->next;
902 req_free (prv);
903 }
904
905 ress = rese = 0;
906 1347
907 close (respipe [0]); 1348 close (respipe [0]);
908 close (respipe [1]); 1349 close (respipe [1]);
909 create_pipe (); 1350
1351 if (!create_pipe (respipe))
1352 croak ("cannot set result pipe to nonblocking mode");
910 1353
911 atfork_parent (); 1354 atfork_parent ();
912} 1355}
913 1356
914#define dREQ \ 1357#define dREQ \
915 aio_req req; \ 1358 aio_req req; \
1359 int req_pri = next_pri; \
1360 next_pri = DEFAULT_PRI + PRI_BIAS; \
916 \ 1361 \
917 if (SvOK (callback) && !SvROK (callback)) \ 1362 if (SvOK (callback) && !SvROK (callback)) \
918 croak ("callback must be undef or of reference type"); \ 1363 croak ("callback must be undef or of reference type"); \
919 \ 1364 \
920 Newz (0, req, 1, aio_cb); \ 1365 Newz (0, req, 1, aio_cb); \
921 if (!req) \ 1366 if (!req) \
922 croak ("out of memory during aio_req allocation"); \ 1367 croak ("out of memory during aio_req allocation"); \
923 \ 1368 \
924 req->callback = newSVsv (callback) 1369 req->callback = newSVsv (callback); \
1370 req->pri = req_pri
925 1371
926#define REQ_SEND \ 1372#define REQ_SEND \
927 req_send (req); \ 1373 req_send (req); \
928 \ 1374 \
929 if (GIMME_V != G_VOID) \ 1375 if (GIMME_V != G_VOID) \
934PROTOTYPES: ENABLE 1380PROTOTYPES: ENABLE
935 1381
936BOOT: 1382BOOT:
937{ 1383{
938 HV *stash = gv_stashpv ("IO::AIO", 1); 1384 HV *stash = gv_stashpv ("IO::AIO", 1);
1385
939 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); 1386 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
940 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1387 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
941 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1388 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1389 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
1390 newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
1391#ifdef _WIN32
1392 X_MUTEX_CHECK (wrklock);
1393 X_MUTEX_CHECK (reslock);
1394 X_MUTEX_CHECK (reqlock);
1395 X_MUTEX_CHECK (reqwait);
1396 X_MUTEX_CHECK (preadwritelock);
1397 X_MUTEX_CHECK (readdirlock);
1398#else
1399 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1400 newCONSTSUB (stash, "SIGIO", newSViv (SIGIO));
1401#endif
942 1402
943 create_pipe (); 1403 if (!create_pipe (respipe))
1404 croak ("cannot set result pipe to nonblocking mode");
1405
944 pthread_atfork (atfork_prepare, atfork_parent, atfork_child); 1406 X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child);
945} 1407}
946 1408
947void 1409void
948min_parallel (nthreads) 1410max_poll_reqs (int nreqs)
949 int nthreads
950 PROTOTYPE: $ 1411 PROTOTYPE: $
1412 CODE:
1413 max_poll_reqs = nreqs;
951 1414
952void 1415void
953max_parallel (nthreads) 1416max_poll_time (double nseconds)
954 int nthreads
955 PROTOTYPE: $ 1417 PROTOTYPE: $
1418 CODE:
1419 max_poll_time = nseconds * AIO_TICKS;
1420
1421void
1422min_parallel (int nthreads)
1423 PROTOTYPE: $
1424
1425void
1426max_parallel (int nthreads)
1427 PROTOTYPE: $
1428
1429void
1430max_idle (int nthreads)
1431 PROTOTYPE: $
1432 CODE:
1433 set_max_idle (nthreads);
956 1434
957int 1435int
958max_outstanding (nreqs) 1436max_outstanding (int maxreqs)
959 int nreqs 1437 PROTOTYPE: $
960 PROTOTYPE: $
961 CODE: 1438 CODE:
962 RETVAL = max_outstanding; 1439 RETVAL = max_outstanding;
963 max_outstanding = nreqs; 1440 max_outstanding = maxreqs;
1441 OUTPUT:
1442 RETVAL
964 1443
965void 1444void
966aio_open (pathname,flags,mode,callback=&PL_sv_undef) 1445aio_open (SV8 *pathname, int flags, int mode, SV *callback=&PL_sv_undef)
967 SV * pathname
968 int flags
969 int mode
970 SV * callback
971 PROTOTYPE: $$$;$ 1446 PROTOTYPE: $$$;$
972 PPCODE: 1447 PPCODE:
973{ 1448{
974 dREQ; 1449 dREQ;
975 1450
976 req->type = REQ_OPEN; 1451 req->type = REQ_OPEN;
977 req->data = newSVsv (pathname); 1452 req->sv1 = newSVsv (pathname);
978 req->dataptr = SvPVbyte_nolen (req->data); 1453 req->ptr1 = SvPVbyte_nolen (req->sv1);
979 req->fd = flags; 1454 req->int1 = flags;
980 req->mode = mode; 1455 req->mode = mode;
981 1456
982 REQ_SEND; 1457 REQ_SEND;
983} 1458}
984 1459
985void 1460void
986aio_close (fh,callback=&PL_sv_undef) 1461aio_close (SV *fh, SV *callback=&PL_sv_undef)
987 SV * fh
988 SV * callback
989 PROTOTYPE: $;$ 1462 PROTOTYPE: $;$
990 ALIAS: 1463 ALIAS:
991 aio_close = REQ_CLOSE 1464 aio_close = REQ_CLOSE
992 aio_fsync = REQ_FSYNC 1465 aio_fsync = REQ_FSYNC
993 aio_fdatasync = REQ_FDATASYNC 1466 aio_fdatasync = REQ_FDATASYNC
994 PPCODE: 1467 PPCODE:
995{ 1468{
996 dREQ; 1469 dREQ;
997 1470
998 req->type = ix; 1471 req->type = ix;
999 req->fh = newSVsv (fh); 1472 req->sv1 = newSVsv (fh);
1000 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh))); 1473 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
1001 1474
1002 REQ_SEND (req); 1475 REQ_SEND (req);
1003} 1476}
1004 1477
1005void 1478void
1006aio_read (fh,offset,length,data,dataoffset,callback=&PL_sv_undef) 1479aio_read (SV *fh, SV *offset, SV *length, SV8 *data, IV dataoffset, SV *callback=&PL_sv_undef)
1007 SV * fh
1008 UV offset
1009 UV length
1010 SV * data
1011 UV dataoffset
1012 SV * callback
1013 ALIAS: 1480 ALIAS:
1014 aio_read = REQ_READ 1481 aio_read = REQ_READ
1015 aio_write = REQ_WRITE 1482 aio_write = REQ_WRITE
1016 PROTOTYPE: $$$$$;$ 1483 PROTOTYPE: $$$$$;$
1017 PPCODE: 1484 PPCODE:
1018{ 1485{
1019 aio_req req;
1020 STRLEN svlen; 1486 STRLEN svlen;
1021 char *svptr = SvPVbyte (data, svlen); 1487 char *svptr = SvPVbyte (data, svlen);
1488 UV len = SvUV (length);
1022 1489
1023 SvUPGRADE (data, SVt_PV); 1490 SvUPGRADE (data, SVt_PV);
1024 SvPOK_on (data); 1491 SvPOK_on (data);
1025 1492
1026 if (dataoffset < 0) 1493 if (dataoffset < 0)
1027 dataoffset += svlen; 1494 dataoffset += svlen;
1028 1495
1029 if (dataoffset < 0 || dataoffset > svlen) 1496 if (dataoffset < 0 || dataoffset > svlen)
1030 croak ("data offset outside of string"); 1497 croak ("dataoffset outside of data scalar");
1031 1498
1032 if (ix == REQ_WRITE) 1499 if (ix == REQ_WRITE)
1033 { 1500 {
1034 /* write: check length and adjust. */ 1501 /* write: check length and adjust. */
1035 if (length < 0 || length + dataoffset > svlen) 1502 if (!SvOK (length) || len + dataoffset > svlen)
1036 length = svlen - dataoffset; 1503 len = svlen - dataoffset;
1037 } 1504 }
1038 else 1505 else
1039 { 1506 {
1040 /* read: grow scalar as necessary */ 1507 /* read: grow scalar as necessary */
1041 svptr = SvGROW (data, length + dataoffset); 1508 svptr = SvGROW (data, len + dataoffset + 1);
1042 } 1509 }
1043 1510
1044 if (length < 0) 1511 if (len < 0)
1045 croak ("length must not be negative"); 1512 croak ("length must not be negative");
1046 1513
1047 { 1514 {
1048 dREQ; 1515 dREQ;
1049 1516
1050 req->type = ix; 1517 req->type = ix;
1051 req->fh = newSVsv (fh); 1518 req->sv1 = newSVsv (fh);
1052 req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh)) 1519 req->int1 = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
1053 : IoOFP (sv_2io (fh))); 1520 : IoOFP (sv_2io (fh)));
1521 req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
1054 req->offset = offset; 1522 req->size = len;
1055 req->length = length;
1056 req->data = SvREFCNT_inc (data); 1523 req->sv2 = SvREFCNT_inc (data);
1057 req->dataptr = (char *)svptr + dataoffset; 1524 req->ptr1 = (char *)svptr + dataoffset;
1525 req->stroffset = dataoffset;
1058 1526
1059 if (!SvREADONLY (data)) 1527 if (!SvREADONLY (data))
1060 { 1528 {
1061 SvREADONLY_on (data); 1529 SvREADONLY_on (data);
1062 req->data2ptr = (void *)data; 1530 req->flags |= FLAG_SV2_RO_OFF;
1063 } 1531 }
1064 1532
1065 REQ_SEND; 1533 REQ_SEND;
1066 } 1534 }
1067} 1535}
1068 1536
1069void 1537void
1538aio_readlink (SV8 *path, SV *callback=&PL_sv_undef)
1539 PROTOTYPE: $$;$
1540 PPCODE:
1541{
1542 SV *data;
1543 dREQ;
1544
1545 data = newSV (NAME_MAX);
1546 SvPOK_on (data);
1547
1548 req->type = REQ_READLINK;
1549 req->sv1 = newSVsv (path);
1550 req->ptr2 = SvPVbyte_nolen (req->sv1);
1551 req->sv2 = data;
1552 req->ptr1 = SvPVbyte_nolen (data);
1553
1554 REQ_SEND;
1555}
1556
1557void
1070aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef) 1558aio_sendfile (SV *out_fh, SV *in_fh, SV *in_offset, UV length, SV *callback=&PL_sv_undef)
1071 SV * out_fh
1072 SV * in_fh
1073 UV in_offset
1074 UV length
1075 SV * callback
1076 PROTOTYPE: $$$$;$ 1559 PROTOTYPE: $$$$;$
1077 PPCODE: 1560 PPCODE:
1078{ 1561{
1079 dREQ; 1562 dREQ;
1080 1563
1081 req->type = REQ_SENDFILE; 1564 req->type = REQ_SENDFILE;
1082 req->fh = newSVsv (out_fh); 1565 req->sv1 = newSVsv (out_fh);
1083 req->fd = PerlIO_fileno (IoIFP (sv_2io (out_fh))); 1566 req->int1 = PerlIO_fileno (IoIFP (sv_2io (out_fh)));
1084 req->fh2 = newSVsv (in_fh); 1567 req->sv2 = newSVsv (in_fh);
1085 req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh))); 1568 req->int2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));
1086 req->offset = in_offset; 1569 req->offs = SvVAL64 (in_offset);
1087 req->length = length; 1570 req->size = length;
1088 1571
1089 REQ_SEND; 1572 REQ_SEND;
1090} 1573}
1091 1574
1092void 1575void
1093aio_readahead (fh,offset,length,callback=&PL_sv_undef) 1576aio_readahead (SV *fh, SV *offset, IV length, SV *callback=&PL_sv_undef)
1094 SV * fh
1095 UV offset
1096 IV length
1097 SV * callback
1098 PROTOTYPE: $$$;$ 1577 PROTOTYPE: $$$;$
1099 PPCODE: 1578 PPCODE:
1100{ 1579{
1101 dREQ; 1580 dREQ;
1102 1581
1103 req->type = REQ_READAHEAD; 1582 req->type = REQ_READAHEAD;
1104 req->fh = newSVsv (fh); 1583 req->sv1 = newSVsv (fh);
1105 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh))); 1584 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
1106 req->offset = offset; 1585 req->offs = SvVAL64 (offset);
1107 req->length = length; 1586 req->size = length;
1108 1587
1109 REQ_SEND; 1588 REQ_SEND;
1110} 1589}
1111 1590
1112void 1591void
1113aio_stat (fh_or_path,callback=&PL_sv_undef) 1592aio_stat (SV8 *fh_or_path, SV *callback=&PL_sv_undef)
1114 SV * fh_or_path
1115 SV * callback
1116 ALIAS: 1593 ALIAS:
1117 aio_stat = REQ_STAT 1594 aio_stat = REQ_STAT
1118 aio_lstat = REQ_LSTAT 1595 aio_lstat = REQ_LSTAT
1119 PPCODE: 1596 PPCODE:
1120{ 1597{
1121 dREQ; 1598 dREQ;
1122 1599
1123 New (0, req->statdata, 1, Stat_t); 1600 req->ptr2 = malloc (sizeof (Stat_t));
1124 if (!req->statdata) 1601 if (!req->ptr2)
1125 { 1602 {
1126 req_free (req); 1603 req_destroy (req);
1127 croak ("out of memory during aio_req->statdata allocation"); 1604 croak ("out of memory during aio_stat statdata allocation");
1128 } 1605 }
1606
1607 req->flags |= FLAG_PTR2_FREE;
1608 req->sv1 = newSVsv (fh_or_path);
1129 1609
1130 if (SvPOK (fh_or_path)) 1610 if (SvPOK (fh_or_path))
1131 { 1611 {
1132 req->type = ix; 1612 req->type = ix;
1133 req->data = newSVsv (fh_or_path);
1134 req->dataptr = SvPVbyte_nolen (req->data); 1613 req->ptr1 = SvPVbyte_nolen (req->sv1);
1135 } 1614 }
1136 else 1615 else
1137 { 1616 {
1138 req->type = REQ_FSTAT; 1617 req->type = REQ_FSTAT;
1139 req->fh = newSVsv (fh_or_path);
1140 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); 1618 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
1141 } 1619 }
1142 1620
1143 REQ_SEND; 1621 REQ_SEND;
1144} 1622}
1145 1623
1146void 1624void
1625aio_utime (SV8 *fh_or_path, SV *atime, SV *mtime, SV *callback=&PL_sv_undef)
1626 PPCODE:
1627{
1628 dREQ;
1629
1630 req->nv1 = SvOK (atime) ? SvNV (atime) : -1.;
1631 req->nv2 = SvOK (mtime) ? SvNV (mtime) : -1.;
1632 req->sv1 = newSVsv (fh_or_path);
1633
1634 if (SvPOK (fh_or_path))
1635 {
1636 req->type = REQ_UTIME;
1637 req->ptr1 = SvPVbyte_nolen (req->sv1);
1638 }
1639 else
1640 {
1641 req->type = REQ_FUTIME;
1642 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
1643 }
1644
1645 REQ_SEND;
1646}
1647
1648void
1649aio_truncate (SV8 *fh_or_path, SV *offset, SV *callback=&PL_sv_undef)
1650 PPCODE:
1651{
1652 dREQ;
1653
1654 req->sv1 = newSVsv (fh_or_path);
1655 req->offs = SvOK (offset) ? SvVAL64 (offset) : -1;
1656
1657 if (SvPOK (fh_or_path))
1658 {
1659 req->type = REQ_TRUNCATE;
1660 req->ptr1 = SvPVbyte_nolen (req->sv1);
1661 }
1662 else
1663 {
1664 req->type = REQ_FTRUNCATE;
1665 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
1666 }
1667
1668 REQ_SEND;
1669}
1670
1671void
1672aio_chmod (SV8 *fh_or_path, int mode, SV *callback=&PL_sv_undef)
1673 PPCODE:
1674{
1675 dREQ;
1676
1677 req->mode = mode;
1678 req->sv1 = newSVsv (fh_or_path);
1679
1680 if (SvPOK (fh_or_path))
1681 {
1682 req->type = REQ_CHMOD;
1683 req->ptr1 = SvPVbyte_nolen (req->sv1);
1684 }
1685 else
1686 {
1687 req->type = REQ_FCHMOD;
1688 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
1689 }
1690
1691 REQ_SEND;
1692}
1693
1694void
1695aio_chown (SV8 *fh_or_path, SV *uid, SV *gid, SV *callback=&PL_sv_undef)
1696 PPCODE:
1697{
1698 dREQ;
1699
1700 req->int2 = SvOK (uid) ? SvIV (uid) : -1;
1701 req->int3 = SvOK (gid) ? SvIV (gid) : -1;
1702 req->sv1 = newSVsv (fh_or_path);
1703
1704 if (SvPOK (fh_or_path))
1705 {
1706 req->type = REQ_CHOWN;
1707 req->ptr1 = SvPVbyte_nolen (req->sv1);
1708 }
1709 else
1710 {
1711 req->type = REQ_FCHOWN;
1712 req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
1713 }
1714
1715 REQ_SEND;
1716}
1717
1718void
1147aio_unlink (pathname,callback=&PL_sv_undef) 1719aio_unlink (SV8 *pathname, SV *callback=&PL_sv_undef)
1148 SV * pathname
1149 SV * callback
1150 ALIAS: 1720 ALIAS:
1151 aio_unlink = REQ_UNLINK 1721 aio_unlink = REQ_UNLINK
1152 aio_rmdir = REQ_RMDIR 1722 aio_rmdir = REQ_RMDIR
1153 aio_readdir = REQ_READDIR 1723 aio_readdir = REQ_READDIR
1154 PPCODE: 1724 PPCODE:
1155{ 1725{
1156 dREQ; 1726 dREQ;
1157 1727
1158 req->type = ix; 1728 req->type = ix;
1159 req->data = newSVsv (pathname); 1729 req->sv1 = newSVsv (pathname);
1160 req->dataptr = SvPVbyte_nolen (req->data); 1730 req->ptr1 = SvPVbyte_nolen (req->sv1);
1731
1732 REQ_SEND;
1733}
1734
1735void
1736aio_mkdir (SV8 *pathname, int mode, SV *callback=&PL_sv_undef)
1737 PPCODE:
1738{
1739 dREQ;
1161 1740
1741 req->type = REQ_MKDIR;
1742 req->sv1 = newSVsv (pathname);
1743 req->ptr1 = SvPVbyte_nolen (req->sv1);
1744 req->mode = mode;
1745
1162 REQ_SEND; 1746 REQ_SEND;
1163} 1747}
1164 1748
1165void 1749void
1166aio_link (oldpath,newpath,callback=&PL_sv_undef) 1750aio_link (SV8 *oldpath, SV8 *newpath, SV *callback=&PL_sv_undef)
1167 SV * oldpath
1168 SV * newpath
1169 SV * callback
1170 ALIAS: 1751 ALIAS:
1171 aio_link = REQ_LINK 1752 aio_link = REQ_LINK
1172 aio_symlink = REQ_SYMLINK 1753 aio_symlink = REQ_SYMLINK
1173 aio_rename = REQ_RENAME 1754 aio_rename = REQ_RENAME
1174 PPCODE: 1755 PPCODE:
1175{ 1756{
1176 dREQ; 1757 dREQ;
1177 1758
1178 req->type = ix; 1759 req->type = ix;
1179 req->fh = newSVsv (oldpath); 1760 req->sv2 = newSVsv (oldpath);
1180 req->data2ptr = SvPVbyte_nolen (req->fh); 1761 req->ptr2 = SvPVbyte_nolen (req->sv2);
1181 req->data = newSVsv (newpath); 1762 req->sv1 = newSVsv (newpath);
1182 req->dataptr = SvPVbyte_nolen (req->data); 1763 req->ptr1 = SvPVbyte_nolen (req->sv1);
1183 1764
1184 REQ_SEND; 1765 REQ_SEND;
1185} 1766}
1186 1767
1187void 1768void
1188aio_sleep (delay,callback=&PL_sv_undef) 1769aio_mknod (SV8 *pathname, int mode, UV dev, SV *callback=&PL_sv_undef)
1189 double delay
1190 SV * callback
1191 PPCODE: 1770 PPCODE:
1192{ 1771{
1193 dREQ; 1772 dREQ;
1194 1773
1195 req->type = REQ_SLEEP; 1774 req->type = REQ_MKNOD;
1196 req->fd = delay < 0. ? 0 : delay; 1775 req->sv1 = newSVsv (pathname);
1197 req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd); 1776 req->ptr1 = SvPVbyte_nolen (req->sv1);
1198 1777 req->mode = (mode_t)mode;
1778 req->offs = dev;
1779
1199 REQ_SEND; 1780 REQ_SEND;
1200} 1781}
1201 1782
1202void 1783void
1784aio_busy (double delay, SV *callback=&PL_sv_undef)
1785 PPCODE:
1786{
1787 dREQ;
1788
1789 req->type = REQ_BUSY;
1790 req->nv1 = delay < 0. ? 0. : delay;
1791
1792 REQ_SEND;
1793}
1794
1795void
1203aio_group (callback=&PL_sv_undef) 1796aio_group (SV *callback=&PL_sv_undef)
1204 SV * callback
1205 PROTOTYPE: ;$ 1797 PROTOTYPE: ;$
1206 PPCODE: 1798 PPCODE:
1207{ 1799{
1208 dREQ; 1800 dREQ;
1801
1209 req->type = REQ_GROUP; 1802 req->type = REQ_GROUP;
1803
1210 req_send (req); 1804 req_send (req);
1211 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1805 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1212} 1806}
1807
1808void
1809aio_nop (SV *callback=&PL_sv_undef)
1810 PPCODE:
1811{
1812 dREQ;
1813
1814 req->type = REQ_NOP;
1815
1816 REQ_SEND;
1817}
1818
1819int
1820aioreq_pri (int pri = 0)
1821 PROTOTYPE: ;$
1822 CODE:
1823 RETVAL = next_pri - PRI_BIAS;
1824 if (items > 0)
1825 {
1826 if (pri < PRI_MIN) pri = PRI_MIN;
1827 if (pri > PRI_MAX) pri = PRI_MAX;
1828 next_pri = pri + PRI_BIAS;
1829 }
1830 OUTPUT:
1831 RETVAL
1832
1833void
1834aioreq_nice (int nice = 0)
1835 CODE:
1836 nice = next_pri - nice;
1837 if (nice < PRI_MIN) nice = PRI_MIN;
1838 if (nice > PRI_MAX) nice = PRI_MAX;
1839 next_pri = nice + PRI_BIAS;
1213 1840
1214void 1841void
1215flush () 1842flush ()
1216 PROTOTYPE: 1843 PROTOTYPE:
1217 CODE: 1844 CODE:
1219 { 1846 {
1220 poll_wait (); 1847 poll_wait ();
1221 poll_cb (); 1848 poll_cb ();
1222 } 1849 }
1223 1850
1224void 1851int
1225poll() 1852poll()
1226 PROTOTYPE: 1853 PROTOTYPE:
1227 CODE: 1854 CODE:
1228 if (nreqs)
1229 {
1230 poll_wait (); 1855 poll_wait ();
1231 poll_cb (); 1856 RETVAL = poll_cb ();
1232 } 1857 OUTPUT:
1858 RETVAL
1233 1859
1234int 1860int
1235poll_fileno() 1861poll_fileno()
1236 PROTOTYPE: 1862 PROTOTYPE:
1237 CODE: 1863 CODE:
1249 1875
1250void 1876void
1251poll_wait() 1877poll_wait()
1252 PROTOTYPE: 1878 PROTOTYPE:
1253 CODE: 1879 CODE:
1254 if (nreqs)
1255 poll_wait (); 1880 poll_wait ();
1881
1882void
1883setsig (int signum = SIGIO)
1884 PROTOTYPE: ;$
1885 CODE:
1886{
1887 if (block_sig_level)
1888 croak ("cannot call IO::AIO::setsig from within aio_block/callback");
1889
1890 X_LOCK (reslock);
1891 main_tid = pthread_self ();
1892 main_sig = signum;
1893 X_UNLOCK (reslock);
1894
1895 if (main_sig && npending)
1896 pthread_kill (main_tid, main_sig);
1897}
1898
1899void
1900aio_block (SV *cb)
1901 PROTOTYPE: &
1902 PPCODE:
1903{
1904 int count;
1905
1906 block_sig ();
1907 PUSHMARK (SP);
1908 PUTBACK;
1909 count = call_sv (cb, GIMME_V | G_NOARGS | G_EVAL);
1910 SPAGAIN;
1911 unblock_sig ();
1912
1913 if (SvTRUE (ERRSV))
1914 croak (0);
1915
1916 XSRETURN (count);
1917}
1256 1918
1257int 1919int
1258nreqs() 1920nreqs()
1259 PROTOTYPE: 1921 PROTOTYPE:
1260 CODE: 1922 CODE:
1261 RETVAL = nreqs; 1923 RETVAL = nreqs;
1262 OUTPUT: 1924 OUTPUT:
1263 RETVAL 1925 RETVAL
1264 1926
1265PROTOTYPES: DISABLE 1927int
1266 1928nready()
1267MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1268
1269void
1270cancel (aio_req_ornot req)
1271 PROTOTYPE: 1929 PROTOTYPE:
1272 CODE: 1930 CODE:
1931 RETVAL = get_nready ();
1932 OUTPUT:
1933 RETVAL
1934
1935int
1936npending()
1937 PROTOTYPE:
1938 CODE:
1939 RETVAL = get_npending ();
1940 OUTPUT:
1941 RETVAL
1942
1943int
1944nthreads()
1945 PROTOTYPE:
1946 CODE:
1947 if (WORDACCESS_UNSAFE) X_LOCK (wrklock);
1948 RETVAL = started;
1949 if (WORDACCESS_UNSAFE) X_UNLOCK (wrklock);
1950 OUTPUT:
1951 RETVAL
1952
1953PROTOTYPES: DISABLE
1954
1955MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1956
1957void
1958cancel (aio_req_ornot req)
1959 CODE:
1273 req_cancel (req); 1960 req_cancel (req);
1961
1962void
1963cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1964 CODE:
1965 SvREFCNT_dec (req->callback);
1966 req->callback = newSVsv (callback);
1274 1967
1275MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1968MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1276 1969
1277void 1970void
1278add (aio_req grp, ...) 1971add (aio_req grp, ...)
1279 PPCODE: 1972 PPCODE:
1280{ 1973{
1281 int i; 1974 int i;
1975 aio_req req;
1282 1976
1977 if (main_sig && !block_sig_level)
1978 croak ("aio_group->add called outside aio_block/callback context while IO::AIO::setsig is in use");
1979
1283 if (grp->fd == 2) 1980 if (grp->int1 == 2)
1284 croak ("cannot add requests to IO::AIO::GRP after the group finished"); 1981 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1285 1982
1286 for (i = 1; i < items; ++i ) 1983 for (i = 1; i < items; ++i )
1287 { 1984 {
1288 if (GIMME_V != G_VOID) 1985 if (GIMME_V != G_VOID)
1289 XPUSHs (sv_2mortal (newSVsv (ST (i)))); 1986 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1290 1987
1291 aio_req req = SvAIO_REQ (ST (i)); 1988 req = SvAIO_REQ (ST (i));
1292 1989
1293 if (req) 1990 if (req)
1294 { 1991 {
1295 ++grp->length; 1992 ++grp->size;
1296 req->grp = grp; 1993 req->grp = grp;
1297 1994
1298 req->grp_prev = 0; 1995 req->grp_prev = 0;
1299 req->grp_next = grp->grp_first; 1996 req->grp_next = grp->grp_first;
1300 1997
1305 } 2002 }
1306 } 2003 }
1307} 2004}
1308 2005
1309void 2006void
2007cancel_subs (aio_req_ornot req)
2008 CODE:
2009 req_cancel_subs (req);
2010
2011void
1310result (aio_req grp, ...) 2012result (aio_req grp, ...)
1311 CODE: 2013 CODE:
1312{ 2014{
1313 int i; 2015 int i;
2016 AV *av;
2017
2018 grp->errorno = errno;
2019
1314 AV *av = newAV (); 2020 av = newAV ();
1315 2021
1316 for (i = 1; i < items; ++i ) 2022 for (i = 1; i < items; ++i )
1317 av_push (av, newSVsv (ST (i))); 2023 av_push (av, newSVsv (ST (i)));
1318 2024
1319 SvREFCNT_dec (grp->data); 2025 SvREFCNT_dec (grp->sv1);
1320 grp->data = (SV *)av; 2026 grp->sv1 = (SV *)av;
1321} 2027}
1322 2028
1323void 2029void
1324lock (aio_req grp) 2030errno (aio_req grp, int errorno = errno)
2031 CODE:
2032 grp->errorno = errorno;
2033
2034void
2035limit (aio_req grp, int limit)
1325 CODE: 2036 CODE:
1326 ++grp->length; 2037 grp->int2 = limit;
2038 aio_grp_feed (grp);
1327 2039
1328void 2040void
1329unlock (aio_req grp) 2041feed (aio_req grp, SV *callback=&PL_sv_undef)
1330 CODE: 2042 CODE:
1331 aio_grp_dec (grp); 2043{
2044 SvREFCNT_dec (grp->sv2);
2045 grp->sv2 = newSVsv (callback);
1332 2046
1333void 2047 if (grp->int2 <= 0)
1334feeder_limit (aio_req grp, int limit) 2048 grp->int2 = 2;
1335 CODE: 2049
1336 grp->fd2 = limit;
1337 aio_grp_feed (grp); 2050 aio_grp_feed (grp);
1338
1339void
1340set_feeder (aio_req grp, SV *callback=&PL_sv_undef)
1341 CODE:
1342{
1343 SvREFCNT_dec (grp->fh2);
1344 grp->fh2 = newSVsv (callback);
1345
1346 if (grp->fd2 <= 0)
1347 grp->fd2 = 2;
1348
1349 aio_grp_feed (grp);
1350} 2051}
1351 2052

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines