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.57 by root, Mon Oct 23 18:38:15 2006 UTC vs.
Revision 1.71 by root, Tue Oct 24 16:35:04 2006 UTC

1/* solaris */
2#define _POSIX_PTHREAD_SEMANTICS 1
3
4#if __linux
5# define _GNU_SOURCE
6#endif
7
1#define _REENTRANT 1 8#define _REENTRANT 1
9
2#include <errno.h> 10#include <errno.h>
3 11
4#include "EXTERN.h" 12#include "EXTERN.h"
5#include "perl.h" 13#include "perl.h"
6#include "XSUB.h" 14#include "XSUB.h"
… …
41# define NAME_MAX 4096 49# define NAME_MAX 4096
42#endif 50#endif
43 51
44#if __ia64 52#if __ia64
45# define STACKSIZE 65536 53# define STACKSIZE 65536
54#elif __i386 || __x86_64 /* 16k is unreasonably high :( */
55# define STACKSIZE PTHREAD_STACK_MIN
46#else 56#else
47# define STACKSIZE 8192 57# define STACKSIZE 16384
48#endif 58#endif
59
60/* buffer size for various temporary buffers */
61#define AIO_BUFSIZE 65536
62
63#define dBUF \
64 char *aio_buf; \
65 LOCK (wrklock); \
66 self->dbuf = aio_buf = malloc (AIO_BUFSIZE); \
67 UNLOCK (wrklock); \
68 if (!aio_buf) \
69 return -1;
49 70
50enum { 71enum {
51 REQ_QUIT, 72 REQ_QUIT,
52 REQ_OPEN, REQ_CLOSE, 73 REQ_OPEN, REQ_CLOSE,
53 REQ_READ, REQ_WRITE, REQ_READAHEAD, 74 REQ_READ, REQ_WRITE, REQ_READAHEAD,
… …
56 REQ_FSYNC, REQ_FDATASYNC, 77 REQ_FSYNC, REQ_FDATASYNC,
57 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 78 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
58 REQ_READDIR, 79 REQ_READDIR,
59 REQ_LINK, REQ_SYMLINK, 80 REQ_LINK, REQ_SYMLINK,
60 REQ_GROUP, REQ_NOP, 81 REQ_GROUP, REQ_NOP,
61 REQ_SLEEP, 82 REQ_BUSY,
62}; 83};
63 84
64#define AIO_REQ_KLASS "IO::AIO::REQ" 85#define AIO_REQ_KLASS "IO::AIO::REQ"
65#define AIO_GRP_KLASS "IO::AIO::GRP" 86#define AIO_GRP_KLASS "IO::AIO::GRP"
66 87
67typedef struct aio_cb 88typedef struct aio_cb
68{ 89{
69 struct aio_cb *volatile next; 90 struct aio_cb *volatile next;
70
71 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
72
73 SV *self; /* the perl counterpart of this request, if any */
74 91
75 SV *data, *callback; 92 SV *data, *callback;
76 SV *fh, *fh2; 93 SV *fh, *fh2;
77 void *dataptr, *data2ptr; 94 void *dataptr, *data2ptr;
78 Stat_t *statdata; 95 Stat_t *statdata;
79 off_t offset; 96 off_t offset;
80 size_t length; 97 size_t length;
81 ssize_t result; 98 ssize_t result;
82 99
100 STRLEN dataoffset;
83 int type; 101 int type;
84 int fd, fd2; 102 int fd, fd2;
85 int errorno; 103 int errorno;
86 STRLEN dataoffset;
87 mode_t mode; /* open */ 104 mode_t mode; /* open */
105
88 unsigned char cancelled; 106 unsigned char flags;
107 unsigned char pri;
108
109 SV *self; /* the perl counterpart of this request, if any */
110 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
89} aio_cb; 111} aio_cb;
112
113enum {
114 FLAG_CANCELLED = 0x01,
115};
90 116
91typedef aio_cb *aio_req; 117typedef aio_cb *aio_req;
92typedef aio_cb *aio_req_ornot; 118typedef aio_cb *aio_req_ornot;
119
120enum {
121 PRI_MIN = -4,
122 PRI_MAX = 4,
123
124 DEFAULT_PRI = 0,
125 PRI_BIAS = -PRI_MIN,
126 NUM_PRI = PRI_MAX + PRI_BIAS + 1,
127};
128
129static int next_pri = DEFAULT_PRI + PRI_BIAS;
93 130
94static int started, wanted; 131static int started, wanted;
95static volatile int nreqs; 132static volatile int nreqs;
96static int max_outstanding = 1<<30; 133static int max_outstanding = 1<<30;
97static int respipe [2]; 134static int respipe [2];
98 135
136#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
137# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
138#else
139# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
140#endif
141
142#define LOCK(mutex) pthread_mutex_lock (&(mutex))
143#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
144
145/* worker threasd management */
146static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
147
148typedef struct worker {
149 /* locked by wrklock */
150 struct worker *prev, *next;
151
152 pthread_t tid;
153
154 /* locked by reslock, reqlock or wrklock */
155 aio_req req; /* currently processed request */
156 void *dbuf;
157 DIR *dirp;
158} worker;
159
160static worker wrk_first = { &wrk_first, &wrk_first, 0 };
161
162static void worker_clear (worker *wrk)
163{
164 if (wrk->dirp)
165 {
166 closedir (wrk->dirp);
167 wrk->dirp = 0;
168 }
169
170 if (wrk->dbuf)
171 {
172 free (wrk->dbuf);
173 wrk->dbuf = 0;
174 }
175}
176
177static void worker_free (worker *wrk)
178{
179 wrk->next->prev = wrk->prev;
180 wrk->prev->next = wrk->next;
181
182 free (wrk);
183}
184
99static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; 185static pthread_mutex_t reslock = AIO_MUTEX_INIT;
100static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; 186static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 187static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
102 188
103static volatile aio_req reqs, reqe; /* queue start, queue end */ 189/*
104static volatile aio_req ress, rese; /* queue start, queue end */ 190 * a somewhat faster data structure might be nice, but
191 * with 8 priorities this actually needs <20 insns
192 * per shift, the most expensive operation.
193 */
194typedef struct {
195 aio_req qs[NUM_PRI], qe[NUM_PRI]; /* qstart, qend */
196 int size;
197} reqq;
198
199static reqq req_queue;
200static reqq res_queue;
201
202int reqq_push (reqq *q, aio_req req)
203{
204 int pri = req->pri;
205 req->next = 0;
206
207 if (q->qe[pri])
208 {
209 q->qe[pri]->next = req;
210 q->qe[pri] = req;
211 }
212 else
213 q->qe[pri] = q->qs[pri] = req;
214
215 return q->size++;
216}
217
218aio_req reqq_shift (reqq *q)
219{
220 int pri;
221
222 if (!q->size)
223 return 0;
224
225 --q->size;
226
227 for (pri = NUM_PRI; pri--; )
228 {
229 aio_req req = q->qs[pri];
230
231 if (req)
232 {
233 if (!(q->qs[pri] = req->next))
234 q->qe[pri] = 0;
235
236 return req;
237 }
238 }
239
240 abort ();
241}
105 242
106static void req_invoke (aio_req req); 243static void req_invoke (aio_req req);
107static void req_free (aio_req req); 244static void req_free (aio_req req);
108 245
109/* must be called at most once */ 246/* must be called at most once */
… …
130 return mg ? (aio_req)mg->mg_ptr : 0; 267 return mg ? (aio_req)mg->mg_ptr : 0;
131} 268}
132 269
133static void aio_grp_feed (aio_req grp) 270static void aio_grp_feed (aio_req grp)
134{ 271{
135 while (grp->length < grp->fd2 && !grp->cancelled) 272 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
136 { 273 {
137 int old_len = grp->length; 274 int old_len = grp->length;
138 275
139 if (grp->fh2 && SvOK (grp->fh2)) 276 if (grp->fh2 && SvOK (grp->fh2))
140 { 277 {
… …
143 ENTER; 280 ENTER;
144 SAVETMPS; 281 SAVETMPS;
145 PUSHMARK (SP); 282 PUSHMARK (SP);
146 XPUSHs (req_sv (grp, AIO_GRP_KLASS)); 283 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
147 PUTBACK; 284 PUTBACK;
148 call_sv (grp->fh2, G_VOID | G_EVAL); 285 call_sv (grp->fh2, G_VOID | G_EVAL | G_KEEPERR);
149 SPAGAIN; 286 SPAGAIN;
150 FREETMPS; 287 FREETMPS;
151 LEAVE; 288 LEAVE;
152 } 289 }
153 290
… …
176 } 313 }
177} 314}
178 315
179static void poll_wait () 316static void poll_wait ()
180{ 317{
181 if (nreqs && !ress)
182 {
183 fd_set rfd; 318 fd_set rfd;
319
320 while (nreqs)
321 {
322 int size;
323#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
324 LOCK (reslock);
325#endif
326 size = res_queue.size;
327#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
328 UNLOCK (reslock);
329#endif
330
331 if (size)
332 return;
333
184 FD_ZERO(&rfd); 334 FD_ZERO(&rfd);
185 FD_SET(respipe [0], &rfd); 335 FD_SET(respipe [0], &rfd);
186 336
187 select (respipe [0] + 1, &rfd, 0, 0, 0); 337 select (respipe [0] + 1, &rfd, 0, 0, 0);
188 } 338 }
189} 339}
190 340
191static void req_invoke (aio_req req) 341static void req_invoke (aio_req req)
192{ 342{
193 dSP; 343 dSP;
194 int errorno = errno;
195 344
196 if (req->cancelled || !SvOK (req->callback)) 345 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
197 return; 346 {
198
199 errno = req->errorno; 347 errno = req->errorno;
200 348
201 ENTER; 349 ENTER;
202 SAVETMPS; 350 SAVETMPS;
203 PUSHMARK (SP); 351 PUSHMARK (SP);
204 EXTEND (SP, 1); 352 EXTEND (SP, 1);
205 353
206 switch (req->type) 354 switch (req->type)
207 {
208 case REQ_READDIR:
209 { 355 {
210 SV *rv = &PL_sv_undef; 356 case REQ_READDIR:
211
212 if (req->result >= 0)
213 { 357 {
214 char *buf = req->data2ptr; 358 SV *rv = &PL_sv_undef;
215 AV *av = newAV ();
216 359
217 while (req->result) 360 if (req->result >= 0)
218 { 361 {
362 int i;
363 char *buf = req->data2ptr;
364 AV *av = newAV ();
365
366 av_extend (av, req->result - 1);
367
368 for (i = 0; i < req->result; ++i)
369 {
219 SV *sv = newSVpv (buf, 0); 370 SV *sv = newSVpv (buf, 0);
220 371
221 av_push (av, sv); 372 av_store (av, i, sv);
222 buf += SvCUR (sv) + 1; 373 buf += SvCUR (sv) + 1;
223 req->result--; 374 }
375
376 rv = sv_2mortal (newRV_noinc ((SV *)av));
224 } 377 }
225 378
226 rv = sv_2mortal (newRV_noinc ((SV *)av)); 379 PUSHs (rv);
227 } 380 }
381 break;
228 382
229 PUSHs (rv); 383 case REQ_OPEN:
384 {
385 /* convert fd to fh */
386 SV *fh;
387
388 PUSHs (sv_2mortal (newSViv (req->result)));
389 PUTBACK;
390 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
391 SPAGAIN;
392
393 fh = SvREFCNT_inc (POPs);
394
395 PUSHMARK (SP);
396 XPUSHs (sv_2mortal (fh));
397 }
398 break;
399
400 case REQ_GROUP:
401 req->fd = 2; /* mark group as finished */
402
403 if (req->data)
404 {
405 int i;
406 AV *av = (AV *)req->data;
407
408 EXTEND (SP, AvFILL (av) + 1);
409 for (i = 0; i <= AvFILL (av); ++i)
410 PUSHs (*av_fetch (av, i, 0));
411 }
412 break;
413
414 case REQ_NOP:
415 case REQ_BUSY:
416 break;
417
418 default:
419 PUSHs (sv_2mortal (newSViv (req->result)));
420 break;
230 } 421 }
231 break;
232 422
233 case REQ_OPEN:
234 {
235 /* convert fd to fh */
236 SV *fh;
237 423
238 PUSHs (sv_2mortal (newSViv (req->result)));
239 PUTBACK; 424 PUTBACK;
240 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
241 SPAGAIN;
242
243 fh = SvREFCNT_inc (POPs);
244
245 PUSHMARK (SP);
246 XPUSHs (sv_2mortal (fh));
247 }
248 break;
249
250 case REQ_GROUP:
251 req->fd = 2; /* mark group as finished */
252
253 if (req->data)
254 {
255 int i;
256 AV *av = (AV *)req->data;
257
258 EXTEND (SP, AvFILL (av) + 1);
259 for (i = 0; i <= AvFILL (av); ++i)
260 PUSHs (*av_fetch (av, i, 0));
261 }
262 break;
263
264 case REQ_NOP:
265 case REQ_SLEEP:
266 break;
267
268 default:
269 PUSHs (sv_2mortal (newSViv (req->result)));
270 break;
271 }
272
273
274 PUTBACK;
275 call_sv (req->callback, G_VOID | G_EVAL); 425 call_sv (req->callback, G_VOID | G_EVAL);
276 SPAGAIN; 426 SPAGAIN;
277 427
278 FREETMPS; 428 FREETMPS;
279 LEAVE; 429 LEAVE;
280
281 errno = errorno;
282
283 if (SvTRUE (ERRSV))
284 { 430 }
285 req_free (req);
286 croak (0);
287 }
288}
289 431
290static void req_free (aio_req req)
291{
292 if (req->grp) 432 if (req->grp)
293 { 433 {
294 aio_req grp = req->grp; 434 aio_req grp = req->grp;
295 435
296 /* unlink request */ 436 /* unlink request */
… …
301 grp->grp_first = req->grp_next; 441 grp->grp_first = req->grp_next;
302 442
303 aio_grp_dec (grp); 443 aio_grp_dec (grp);
304 } 444 }
305 445
446 if (SvTRUE (ERRSV))
447 {
448 req_free (req);
449 croak (0);
450 }
451}
452
453static void req_free (aio_req req)
454{
306 if (req->self) 455 if (req->self)
307 { 456 {
308 sv_unmagic (req->self, PERL_MAGIC_ext); 457 sv_unmagic (req->self, PERL_MAGIC_ext);
309 SvREFCNT_dec (req->self); 458 SvREFCNT_dec (req->self);
310 } 459 }
… …
313 SvREFCNT_dec (req->fh); 462 SvREFCNT_dec (req->fh);
314 SvREFCNT_dec (req->fh2); 463 SvREFCNT_dec (req->fh2);
315 SvREFCNT_dec (req->callback); 464 SvREFCNT_dec (req->callback);
316 Safefree (req->statdata); 465 Safefree (req->statdata);
317 466
318 if (req->type == REQ_READDIR && req->result >= 0) 467 if (req->type == REQ_READDIR)
319 free (req->data2ptr); 468 free (req->data2ptr);
320 469
321 Safefree (req); 470 Safefree (req);
322} 471}
323 472
324static void req_cancel (aio_req req) 473static void req_cancel (aio_req req)
325{ 474{
326 req->cancelled = 1; 475 req->flags |= FLAG_CANCELLED;
327 476
328 if (req->type == REQ_GROUP) 477 if (req->type == REQ_GROUP)
329 { 478 {
330 aio_req sub; 479 aio_req sub;
331 480
… …
341 int do_croak = 0; 490 int do_croak = 0;
342 aio_req req; 491 aio_req req;
343 492
344 for (;;) 493 for (;;)
345 { 494 {
346 pthread_mutex_lock (&reslock); 495 LOCK (reslock);
347 req = ress; 496 req = reqq_shift (&res_queue);
348 497
349 if (req) 498 if (req)
350 { 499 {
351 ress = req->next;
352
353 if (!ress) 500 if (!res_queue.size)
354 { 501 {
355 /* read any signals sent by the worker threads */ 502 /* read any signals sent by the worker threads */
356 char buf [32]; 503 char buf [32];
357 while (read (respipe [0], buf, 32) == 32) 504 while (read (respipe [0], buf, 32) == 32)
358 ; 505 ;
359
360 rese = 0;
361 } 506 }
362 } 507 }
363 508
364 pthread_mutex_unlock (&reslock); 509 UNLOCK (reslock);
365 510
366 if (!req) 511 if (!req)
367 break; 512 break;
368 513
369 --nreqs; 514 --nreqs;
… …
403 548
404static void *aio_proc(void *arg); 549static void *aio_proc(void *arg);
405 550
406static void start_thread (void) 551static void start_thread (void)
407{ 552{
553 worker *wrk = calloc (1, sizeof (worker));
554
555 if (!wrk)
556 croak ("unable to allocate worker thread data");
557
408 sigset_t fullsigset, oldsigset; 558 sigset_t fullsigset, oldsigset;
409 pthread_t tid;
410 pthread_attr_t attr; 559 pthread_attr_t attr;
411 560
412 pthread_attr_init (&attr); 561 pthread_attr_init (&attr);
413 pthread_attr_setstacksize (&attr, STACKSIZE); 562 pthread_attr_setstacksize (&attr, STACKSIZE);
414 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED); 563 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
415 564
416 sigfillset (&fullsigset); 565 sigfillset (&fullsigset);
566
567 LOCK (wrklock);
417 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset); 568 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
418 569
419 if (pthread_create (&tid, &attr, aio_proc, 0) == 0) 570 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
571 {
572 wrk->prev = &wrk_first;
573 wrk->next = wrk_first.next;
574 wrk_first.next->prev = wrk;
575 wrk_first.next = wrk;
420 started++; 576 started++;
577 }
578 else
579 free (wrk);
421 580
422 sigprocmask (SIG_SETMASK, &oldsigset, 0); 581 sigprocmask (SIG_SETMASK, &oldsigset, 0);
582 UNLOCK (wrklock);
423} 583}
424 584
425static void req_send (aio_req req) 585static void req_send (aio_req req)
426{ 586{
427 while (started < wanted && nreqs >= started) 587 while (started < wanted && nreqs >= started)
428 start_thread (); 588 start_thread ();
429 589
430 ++nreqs; 590 ++nreqs;
431 591
432 pthread_mutex_lock (&reqlock); 592 LOCK (reqlock);
433 593 reqq_push (&req_queue, req);
434 req->next = 0;
435
436 if (reqe)
437 {
438 reqe->next = req;
439 reqe = req;
440 }
441 else
442 reqe = reqs = req;
443
444 pthread_cond_signal (&reqwait); 594 pthread_cond_signal (&reqwait);
445 pthread_mutex_unlock (&reqlock); 595 UNLOCK (reqlock);
446 596
447 if (nreqs > max_outstanding) 597 if (nreqs > max_outstanding)
448 for (;;) 598 for (;;)
449 { 599 {
450 poll_cb (); 600 poll_cb ();
… …
457} 607}
458 608
459static void end_thread (void) 609static void end_thread (void)
460{ 610{
461 aio_req req; 611 aio_req req;
612
462 Newz (0, req, 1, aio_cb); 613 Newz (0, req, 1, aio_cb);
614
463 req->type = REQ_QUIT; 615 req->type = REQ_QUIT;
616 req->pri = PRI_MAX + PRI_BIAS;
464 617
465 req_send (req); 618 req_send (req);
466} 619}
467 620
468static void min_parallel (int nthreads) 621static void min_parallel (int nthreads)
… …
520static ssize_t pread (int fd, void *buf, size_t count, off_t offset) 673static ssize_t pread (int fd, void *buf, size_t count, off_t offset)
521{ 674{
522 ssize_t res; 675 ssize_t res;
523 off_t ooffset; 676 off_t ooffset;
524 677
525 pthread_mutex_lock (&preadwritelock); 678 LOCK (preadwritelock);
526 ooffset = lseek (fd, 0, SEEK_CUR); 679 ooffset = lseek (fd, 0, SEEK_CUR);
527 lseek (fd, offset, SEEK_SET); 680 lseek (fd, offset, SEEK_SET);
528 res = read (fd, buf, count); 681 res = read (fd, buf, count);
529 lseek (fd, ooffset, SEEK_SET); 682 lseek (fd, ooffset, SEEK_SET);
530 pthread_mutex_unlock (&preadwritelock); 683 UNLOCK (preadwritelock);
531 684
532 return res; 685 return res;
533} 686}
534 687
535static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset) 688static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset)
536{ 689{
537 ssize_t res; 690 ssize_t res;
538 off_t ooffset; 691 off_t ooffset;
539 692
540 pthread_mutex_lock (&preadwritelock); 693 LOCK (preadwritelock);
541 ooffset = lseek (fd, 0, SEEK_CUR); 694 ooffset = lseek (fd, 0, SEEK_CUR);
542 lseek (fd, offset, SEEK_SET); 695 lseek (fd, offset, SEEK_SET);
543 res = write (fd, buf, count); 696 res = write (fd, buf, count);
544 lseek (fd, offset, SEEK_SET); 697 lseek (fd, offset, SEEK_SET);
545 pthread_mutex_unlock (&preadwritelock); 698 UNLOCK (preadwritelock);
546 699
547 return res; 700 return res;
548} 701}
549#endif 702#endif
550 703
… …
555#if !HAVE_READAHEAD 708#if !HAVE_READAHEAD
556# define readahead aio_readahead 709# define readahead aio_readahead
557 710
558static ssize_t readahead (int fd, off_t offset, size_t count) 711static ssize_t readahead (int fd, off_t offset, size_t count)
559{ 712{
560 char readahead_buf[4096]; 713 dBUF;
561 714
562 while (count > 0) 715 while (count > 0)
563 { 716 {
564 size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf); 717 size_t len = count < AIO_BUFSIZE ? count : AIO_BUFSIZE;
565 718
566 pread (fd, readahead_buf, len, offset); 719 pread (fd, aio_buf, len, offset);
567 offset += len; 720 offset += len;
568 count -= len; 721 count -= len;
569 } 722 }
570 723
571 errno = 0; 724 errno = 0;
… …
580static int readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res) 733static int readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res)
581{ 734{
582 struct dirent *e; 735 struct dirent *e;
583 int errorno; 736 int errorno;
584 737
585 pthread_mutex_lock (&readdirlock); 738 LOCK (readdirlock);
586 739
587 e = readdir (dirp); 740 e = readdir (dirp);
588 errorno = errno; 741 errorno = errno;
589 742
590 if (e) 743 if (e)
… …
593 strcpy (ent->d_name, e->d_name); 746 strcpy (ent->d_name, e->d_name);
594 } 747 }
595 else 748 else
596 *res = 0; 749 *res = 0;
597 750
598 pthread_mutex_unlock (&readdirlock); 751 UNLOCK (readdirlock);
599 752
600 errno = errorno; 753 errno = errorno;
601 return e ? 0 : -1; 754 return e ? 0 : -1;
602} 755}
603#endif 756#endif
604 757
605/* sendfile always needs emulation */ 758/* sendfile always needs emulation */
606static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count) 759static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count, worker *self)
607{ 760{
608 ssize_t res; 761 ssize_t res;
609 762
610 if (!count) 763 if (!count)
611 return 0; 764 return 0;
… …
660#endif 813#endif
661 ) 814 )
662 ) 815 )
663 { 816 {
664 /* emulate sendfile. this is a major pain in the ass */ 817 /* emulate sendfile. this is a major pain in the ass */
665 char buf[4096]; 818 dBUF;
819
666 res = 0; 820 res = 0;
667 821
668 while (count) 822 while (count)
669 { 823 {
670 ssize_t cnt; 824 ssize_t cnt;
671 825
672 cnt = pread (ifd, buf, count > 4096 ? 4096 : count, offset); 826 cnt = pread (ifd, aio_buf, count > AIO_BUFSIZE ? AIO_BUFSIZE : count, offset);
673 827
674 if (cnt <= 0) 828 if (cnt <= 0)
675 { 829 {
676 if (cnt && !res) res = -1; 830 if (cnt && !res) res = -1;
677 break; 831 break;
678 } 832 }
679 833
680 cnt = write (ofd, buf, cnt); 834 cnt = write (ofd, aio_buf, cnt);
681 835
682 if (cnt <= 0) 836 if (cnt <= 0)
683 { 837 {
684 if (cnt && !res) res = -1; 838 if (cnt && !res) res = -1;
685 break; 839 break;
… …
693 847
694 return res; 848 return res;
695} 849}
696 850
697/* read a full directory */ 851/* read a full directory */
698static int scandir_ (const char *path, void **namesp) 852static void scandir_ (aio_req req, worker *self)
699{ 853{
700 DIR *dirp = opendir (path); 854 DIR *dirp;
701 union 855 union
702 { 856 {
703 struct dirent d; 857 struct dirent d;
704 char b [offsetof (struct dirent, d_name) + NAME_MAX + 1]; 858 char b [offsetof (struct dirent, d_name) + NAME_MAX + 1];
705 } u; 859 } *u;
706 struct dirent *entp; 860 struct dirent *entp;
707 char *name, *names; 861 char *name, *names;
708 int memlen = 4096; 862 int memlen = 4096;
709 int memofs = 0; 863 int memofs = 0;
710 int res = 0; 864 int res = 0;
711 int errorno; 865 int errorno;
712 866
713 if (!dirp) 867 LOCK (wrklock);
714 return -1; 868 self->dirp = dirp = opendir (req->dataptr);
869 self->dbuf = u = malloc (sizeof (*u));
870 UNLOCK (wrklock);
715 871
716 names = malloc (memlen); 872 req->data2ptr = names = malloc (memlen);
717 873
874 if (dirp && u && names)
718 for (;;) 875 for (;;)
719 { 876 {
877 errno = 0;
720 errno = 0, readdir_r (dirp, &u.d, &entp); 878 readdir_r (dirp, &u->d, &entp);
721 879
722 if (!entp) 880 if (!entp)
723 break; 881 break;
724 882
725 name = entp->d_name; 883 name = entp->d_name;
726 884
727 if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2]))) 885 if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2])))
728 { 886 {
729 int len = strlen (name) + 1; 887 int len = strlen (name) + 1;
730 888
731 res++; 889 res++;
732 890
733 while (memofs + len > memlen) 891 while (memofs + len > memlen)
734 { 892 {
735 memlen *= 2; 893 memlen *= 2;
894 LOCK (wrklock);
736 names = realloc (names, memlen); 895 req->data2ptr = names = realloc (names, memlen);
896 UNLOCK (wrklock);
897
737 if (!names) 898 if (!names)
738 break; 899 break;
739 } 900 }
740 901
741 memcpy (names + memofs, name, len); 902 memcpy (names + memofs, name, len);
742 memofs += len; 903 memofs += len;
743 } 904 }
744 } 905 }
745 906
746 errorno = errno;
747 closedir (dirp);
748
749 if (errorno) 907 if (errno)
750 {
751 free (names);
752 errno = errorno;
753 res = -1; 908 res = -1;
754 } 909
755 910 req->result = res;
756 *namesp = (void *)names;
757 return res;
758} 911}
759 912
760/*****************************************************************************/ 913/*****************************************************************************/
761 914
762static void *aio_proc (void *thr_arg) 915static void *aio_proc (void *thr_arg)
763{ 916{
764 aio_req req; 917 aio_req req;
765 int type; 918 int type;
919 worker *self = (worker *)thr_arg;
766 920
767 do 921 do
768 { 922 {
769 pthread_mutex_lock (&reqlock); 923 LOCK (reqlock);
770 924
771 for (;;) 925 for (;;)
772 { 926 {
773 req = reqs; 927 self->req = req = reqq_shift (&req_queue);
774
775 if (reqs)
776 {
777 reqs = reqs->next;
778 if (!reqs) reqe = 0;
779 }
780 928
781 if (req) 929 if (req)
782 break; 930 break;
783 931
784 pthread_cond_wait (&reqwait, &reqlock); 932 pthread_cond_wait (&reqwait, &reqlock);
785 } 933 }
786 934
787 pthread_mutex_unlock (&reqlock); 935 UNLOCK (reqlock);
788 936
789 errno = 0; /* strictly unnecessary */ 937 errno = 0; /* strictly unnecessary */
790
791 if (!req->cancelled)
792 switch (type = req->type) /* remember type for QUIT check */ 938 type = req->type; /* remember type for QUIT check */
939
940 if (!(req->flags & FLAG_CANCELLED))
941 switch (type)
793 { 942 {
794 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 943 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
795 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 944 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
796 945
797 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 946 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
798 case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break; 947 case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length, self); break;
799 948
800 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break; 949 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;
801 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break; 950 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;
802 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break; 951 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;
803 952
… …
809 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break; 958 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
810 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break; 959 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
811 960
812 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 961 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
813 case REQ_FSYNC: req->result = fsync (req->fd); break; 962 case REQ_FSYNC: req->result = fsync (req->fd); break;
814 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break; 963 case REQ_READDIR: scandir_ (req, self); break;
815 964
816 case REQ_SLEEP: 965 case REQ_BUSY:
817 { 966 {
818 struct timeval tv; 967 struct timeval tv;
819 968
820 tv.tv_sec = req->fd; 969 tv.tv_sec = req->fd;
821 tv.tv_usec = req->fd2; 970 tv.tv_usec = req->fd2;
… …
833 break; 982 break;
834 } 983 }
835 984
836 req->errorno = errno; 985 req->errorno = errno;
837 986
838 pthread_mutex_lock (&reslock); 987 LOCK (reslock);
839 988
840 req->next = 0; 989 if (!reqq_push (&res_queue, req))
841
842 if (rese)
843 {
844 rese->next = req;
845 rese = req;
846 }
847 else
848 {
849 rese = ress = req;
850
851 /* write a dummy byte to the pipe so fh becomes ready */ 990 /* write a dummy byte to the pipe so fh becomes ready */
852 write (respipe [1], &respipe, 1); 991 write (respipe [1], &respipe, 1);
853 }
854 992
855 pthread_mutex_unlock (&reslock); 993 self->req = 0;
994 worker_clear (self);
995
996 UNLOCK (reslock);
856 } 997 }
857 while (type != REQ_QUIT); 998 while (type != REQ_QUIT);
858 999
1000 LOCK (wrklock);
1001 worker_free (self);
1002 UNLOCK (wrklock);
1003
859 return 0; 1004 return 0;
860} 1005}
861 1006
862/*****************************************************************************/ 1007/*****************************************************************************/
863 1008
864static void atfork_prepare (void) 1009static void atfork_prepare (void)
865{ 1010{
866 pthread_mutex_lock (&reqlock); 1011 LOCK (wrklock);
867 pthread_mutex_lock (&reslock); 1012 LOCK (reqlock);
1013 LOCK (reslock);
868#if !HAVE_PREADWRITE 1014#if !HAVE_PREADWRITE
869 pthread_mutex_lock (&preadwritelock); 1015 LOCK (preadwritelock);
870#endif 1016#endif
871#if !HAVE_READDIR_R 1017#if !HAVE_READDIR_R
872 pthread_mutex_lock (&readdirlock); 1018 LOCK (readdirlock);
873#endif 1019#endif
874} 1020}
875 1021
876static void atfork_parent (void) 1022static void atfork_parent (void)
877{ 1023{
878#if !HAVE_READDIR_R 1024#if !HAVE_READDIR_R
879 pthread_mutex_unlock (&readdirlock); 1025 UNLOCK (readdirlock);
880#endif 1026#endif
881#if !HAVE_PREADWRITE 1027#if !HAVE_PREADWRITE
882 pthread_mutex_unlock (&preadwritelock); 1028 UNLOCK (preadwritelock);
883#endif 1029#endif
884 pthread_mutex_unlock (&reslock); 1030 UNLOCK (reslock);
885 pthread_mutex_unlock (&reqlock); 1031 UNLOCK (reqlock);
1032 UNLOCK (wrklock);
886} 1033}
887 1034
888static void atfork_child (void) 1035static void atfork_child (void)
889{ 1036{
890 aio_req prv; 1037 aio_req prv;
891 1038
1039 while (prv = reqq_shift (&req_queue))
1040 req_free (prv);
1041
1042 while (prv = reqq_shift (&res_queue))
1043 req_free (prv);
1044
1045 while (wrk_first.next != &wrk_first)
1046 {
1047 worker *wrk = wrk_first.next;
1048
1049 if (wrk->req)
1050 req_free (wrk->req);
1051
1052 worker_clear (wrk);
1053 worker_free (wrk);
1054 }
1055
892 started = 0; 1056 started = 0;
893 1057 nreqs = 0;
894 while (reqs)
895 {
896 prv = reqs;
897 reqs = prv->next;
898 req_free (prv);
899 }
900
901 reqs = reqe = 0;
902
903 while (ress)
904 {
905 prv = ress;
906 ress = prv->next;
907 req_free (prv);
908 }
909
910 ress = rese = 0;
911 1058
912 close (respipe [0]); 1059 close (respipe [0]);
913 close (respipe [1]); 1060 close (respipe [1]);
914 create_pipe (); 1061 create_pipe ();
915 1062
916 atfork_parent (); 1063 atfork_parent ();
917} 1064}
918 1065
919#define dREQ \ 1066#define dREQ \
920 aio_req req; \ 1067 aio_req req; \
1068 int req_pri = next_pri; \
1069 next_pri = DEFAULT_PRI + PRI_BIAS; \
921 \ 1070 \
922 if (SvOK (callback) && !SvROK (callback)) \ 1071 if (SvOK (callback) && !SvROK (callback)) \
923 croak ("callback must be undef or of reference type"); \ 1072 croak ("callback must be undef or of reference type"); \
924 \ 1073 \
925 Newz (0, req, 1, aio_cb); \ 1074 Newz (0, req, 1, aio_cb); \
926 if (!req) \ 1075 if (!req) \
927 croak ("out of memory during aio_req allocation"); \ 1076 croak ("out of memory during aio_req allocation"); \
928 \ 1077 \
929 req->callback = newSVsv (callback) 1078 req->callback = newSVsv (callback); \
1079 req->pri = req_pri
930 1080
931#define REQ_SEND \ 1081#define REQ_SEND \
932 req_send (req); \ 1082 req_send (req); \
933 \ 1083 \
934 if (GIMME_V != G_VOID) \ 1084 if (GIMME_V != G_VOID) \
… …
1188 1338
1189 REQ_SEND; 1339 REQ_SEND;
1190} 1340}
1191 1341
1192void 1342void
1193aio_sleep (delay,callback=&PL_sv_undef) 1343aio_busy (delay,callback=&PL_sv_undef)
1194 double delay 1344 double delay
1195 SV * callback 1345 SV * callback
1196 PPCODE: 1346 PPCODE:
1197{ 1347{
1198 dREQ; 1348 dREQ;
1199 1349
1200 req->type = REQ_SLEEP; 1350 req->type = REQ_BUSY;
1201 req->fd = delay < 0. ? 0 : delay; 1351 req->fd = delay < 0. ? 0 : delay;
1202 req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd); 1352 req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);
1203 1353
1204 REQ_SEND; 1354 REQ_SEND;
1205} 1355}
… …
1209 SV * callback 1359 SV * callback
1210 PROTOTYPE: ;$ 1360 PROTOTYPE: ;$
1211 PPCODE: 1361 PPCODE:
1212{ 1362{
1213 dREQ; 1363 dREQ;
1364
1214 req->type = REQ_GROUP; 1365 req->type = REQ_GROUP;
1215 req_send (req); 1366 req_send (req);
1367
1216 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1368 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1217} 1369}
1218 1370
1219void 1371void
1220aio_nop (callback=&PL_sv_undef) 1372aio_nop (callback=&PL_sv_undef)
… …
1225 1377
1226 req->type = REQ_NOP; 1378 req->type = REQ_NOP;
1227 1379
1228 REQ_SEND; 1380 REQ_SEND;
1229} 1381}
1382
1383void
1384aioreq_pri (int pri = DEFAULT_PRI)
1385 CODE:
1386 if (pri < PRI_MIN) pri = PRI_MIN;
1387 if (pri > PRI_MAX) pri = PRI_MAX;
1388 next_pri = pri + PRI_BIAS;
1389
1390void
1391aioreq_nice (int nice = 0)
1392 CODE:
1393 nice = next_pri - nice;
1394 if (nice < PRI_MIN) nice = PRI_MIN;
1395 if (nice > PRI_MAX) nice = PRI_MAX;
1396 next_pri = nice + PRI_BIAS;
1230 1397
1231void 1398void
1232flush () 1399flush ()
1233 PROTOTYPE: 1400 PROTOTYPE:
1234 CODE: 1401 CODE:
… …
1283 1450
1284MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1451MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1285 1452
1286void 1453void
1287cancel (aio_req_ornot req) 1454cancel (aio_req_ornot req)
1288 PROTOTYPE:
1289 CODE: 1455 CODE:
1290 req_cancel (req); 1456 req_cancel (req);
1291 1457
1292void 1458void
1293cb (aio_req req, SV *callback=&PL_sv_undef) 1459cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1294 CODE: 1460 CODE:
1295 SvREFCNT_dec (req->callback); 1461 SvREFCNT_dec (req->callback);
1296 req->callback = newSVsv (callback); 1462 req->callback = newSVsv (callback);
1297 1463
1298MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1464MODULE = IO::AIO PACKAGE = IO::AIO::GRP
… …
1343 SvREFCNT_dec (grp->data); 1509 SvREFCNT_dec (grp->data);
1344 grp->data = (SV *)av; 1510 grp->data = (SV *)av;
1345} 1511}
1346 1512
1347void 1513void
1348feed_limit (aio_req grp, int limit) 1514limit (aio_req grp, int limit)
1349 CODE: 1515 CODE:
1350 grp->fd2 = limit; 1516 grp->fd2 = limit;
1351 aio_grp_feed (grp); 1517 aio_grp_feed (grp);
1352 1518
1353void 1519void

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines