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.74 by root, Wed Oct 25 17:57:30 2006 UTC vs.
Revision 1.83 by root, Sat Oct 28 00:17:30 2006 UTC

1/* solaris */ 1/* solaris */
2#define _POSIX_PTHREAD_SEMANTICS 1 2#define _POSIX_PTHREAD_SEMANTICS 1
3 3
4#if __linux 4#if __linux && !defined(_GNU_SOURCE)
5# define _GNU_SOURCE 5# define _GNU_SOURCE
6#endif 6#endif
7 7
8/* just in case */
8#define _REENTRANT 1 9#define _REENTRANT 1
9 10
10#include <errno.h> 11#include <errno.h>
11 12
12#include "EXTERN.h" 13#include "EXTERN.h"
… …
47/* used for struct dirent, AIX doesn't provide it */ 48/* used for struct dirent, AIX doesn't provide it */
48#ifndef NAME_MAX 49#ifndef NAME_MAX
49# define NAME_MAX 4096 50# define NAME_MAX 4096
50#endif 51#endif
51 52
53#ifndef PTHREAD_STACK_MIN
54/* care for broken platforms, e.g. windows */
55# define PTHREAD_STACK_MIN 16384
56#endif
57
52#if __ia64 58#if __ia64
53# define STACKSIZE 65536 59# define STACKSIZE 65536
54#elif __i386 || __x86_64 /* 16k is unreasonably high :( */ 60#elif __i386 || __x86_64 /* 16k is unreasonably high :( */
55# define STACKSIZE PTHREAD_STACK_MIN 61# define STACKSIZE PTHREAD_STACK_MIN
56#else 62#else
57# define STACKSIZE 16384 63# define STACKSIZE 16384
64#endif
65
66/* wether word reads are potentially non-atomic.
67 * this is conservatice, likely most arches this runs
68 * on have atomic word read/writes.
69 */
70#ifndef WORDREAD_UNSAFE
71# if __i386 || __x86_64
72# define WORDREAD_UNSAFE 0
73# else
74# define WORDREAD_UNSAFE 1
75# endif
58#endif 76#endif
59 77
60/* buffer size for various temporary buffers */ 78/* buffer size for various temporary buffers */
61#define AIO_BUFSIZE 65536 79#define AIO_BUFSIZE 65536
62 80
… …
74 REQ_READ, REQ_WRITE, REQ_READAHEAD, 92 REQ_READ, REQ_WRITE, REQ_READAHEAD,
75 REQ_SENDFILE, 93 REQ_SENDFILE,
76 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 94 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
77 REQ_FSYNC, REQ_FDATASYNC, 95 REQ_FSYNC, REQ_FDATASYNC,
78 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 96 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
79 REQ_READDIR, 97 REQ_MKNOD, REQ_READDIR,
80 REQ_LINK, REQ_SYMLINK, 98 REQ_LINK, REQ_SYMLINK,
81 REQ_GROUP, REQ_NOP, 99 REQ_GROUP, REQ_NOP,
82 REQ_BUSY, 100 REQ_BUSY,
83}; 101};
84 102
… …
126 NUM_PRI = PRI_MAX + PRI_BIAS + 1, 144 NUM_PRI = PRI_MAX + PRI_BIAS + 1,
127}; 145};
128 146
129static int next_pri = DEFAULT_PRI + PRI_BIAS; 147static int next_pri = DEFAULT_PRI + PRI_BIAS;
130 148
131static int started, wanted; 149static unsigned int started, wanted;
132static volatile int nreqs;
133static int respipe [2];
134 150
135#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP) 151#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
136# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP 152# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
137#else 153#else
138# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER 154# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
139#endif 155#endif
140 156
141#define LOCK(mutex) pthread_mutex_lock (&(mutex)) 157#define LOCK(mutex) pthread_mutex_lock (&(mutex))
142#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex)) 158#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
143 159
144/* worker threasd management */ 160/* worker threads management */
145static pthread_mutex_t wrklock = AIO_MUTEX_INIT; 161static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
146 162
147typedef struct worker { 163typedef struct worker {
148 /* locked by wrklock */ 164 /* locked by wrklock */
149 struct worker *prev, *next; 165 struct worker *prev, *next;
… …
179 wrk->prev->next = wrk->next; 195 wrk->prev->next = wrk->next;
180 196
181 free (wrk); 197 free (wrk);
182} 198}
183 199
200static volatile unsigned int nreqs, nready, npending;
201static volatile unsigned int max_outstanding = 0xffffffff;
202static int respipe [2];
203
184static pthread_mutex_t reslock = AIO_MUTEX_INIT; 204static pthread_mutex_t reslock = AIO_MUTEX_INIT;
185static pthread_mutex_t reqlock = AIO_MUTEX_INIT; 205static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
186static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 206static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
207
208#if WORDREAD_UNSAFE
209
210static unsigned int get_nready ()
211{
212 unsigned int retval;
213
214 LOCK (reqlock);
215 retval = nready;
216 UNLOCK (reqlock);
217
218 return retval;
219}
220
221static unsigned int get_npending ()
222{
223 unsigned int retval;
224
225 LOCK (reslock);
226 retval = npending;
227 UNLOCK (reslock);
228
229 return retval;
230}
231
232#else
233
234# define get_nready() nready
235# define get_npending() npending
236
237#endif
187 238
188/* 239/*
189 * a somewhat faster data structure might be nice, but 240 * a somewhat faster data structure might be nice, but
190 * with 8 priorities this actually needs <20 insns 241 * with 8 priorities this actually needs <20 insns
191 * per shift, the most expensive operation. 242 * per shift, the most expensive operation.
… …
237 } 288 }
238 289
239 abort (); 290 abort ();
240} 291}
241 292
242static int poll_cb (); 293static int poll_cb (int max);
243static void req_invoke (aio_req req); 294static void req_invoke (aio_req req);
244static void req_free (aio_req req); 295static void req_free (aio_req req);
245static void req_cancel (aio_req req); 296static void req_cancel (aio_req req);
246 297
247/* must be called at most once */ 298/* must be called at most once */
… …
312 req_invoke (grp); 363 req_invoke (grp);
313 req_free (grp); 364 req_free (grp);
314 } 365 }
315} 366}
316 367
317static void poll_wait ()
318{
319 fd_set rfd;
320
321 while (nreqs)
322 {
323 int size;
324#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
325 LOCK (reslock);
326#endif
327 size = res_queue.size;
328#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
329 UNLOCK (reslock);
330#endif
331
332 if (size)
333 return;
334
335 FD_ZERO(&rfd);
336 FD_SET(respipe [0], &rfd);
337
338 select (respipe [0] + 1, &rfd, 0, 0, 0);
339 }
340}
341
342static void req_invoke (aio_req req) 368static void req_invoke (aio_req req)
343{ 369{
344 dSP; 370 dSP;
345 371
346 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback)) 372 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
347 { 373 {
348 errno = req->errorno;
349
350 ENTER; 374 ENTER;
351 SAVETMPS; 375 SAVETMPS;
352 PUSHMARK (SP); 376 PUSHMARK (SP);
353 EXTEND (SP, 1); 377 EXTEND (SP, 1);
354 378
… …
419 default: 443 default:
420 PUSHs (sv_2mortal (newSViv (req->result))); 444 PUSHs (sv_2mortal (newSViv (req->result)));
421 break; 445 break;
422 } 446 }
423 447
448 errno = req->errorno;
424 449
425 PUTBACK; 450 PUTBACK;
426 call_sv (req->callback, G_VOID | G_EVAL); 451 call_sv (req->callback, G_VOID | G_EVAL);
427 SPAGAIN; 452 SPAGAIN;
428 453
… …
490 req->flags |= FLAG_CANCELLED; 515 req->flags |= FLAG_CANCELLED;
491 516
492 req_cancel_subs (req); 517 req_cancel_subs (req);
493} 518}
494 519
520static void *aio_proc(void *arg);
521
522static void start_thread (void)
523{
524 sigset_t fullsigset, oldsigset;
525 pthread_attr_t attr;
526
527 worker *wrk = calloc (1, sizeof (worker));
528
529 if (!wrk)
530 croak ("unable to allocate worker thread data");
531
532 pthread_attr_init (&attr);
533 pthread_attr_setstacksize (&attr, STACKSIZE);
534 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
535#ifdef PTHREAD_SCOPE_PROCESS
536 pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
537#endif
538
539 sigfillset (&fullsigset);
540
541 LOCK (wrklock);
542 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
543
544 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
545 {
546 wrk->prev = &wrk_first;
547 wrk->next = wrk_first.next;
548 wrk_first.next->prev = wrk;
549 wrk_first.next = wrk;
550 ++started;
551 }
552 else
553 free (wrk);
554
555 sigprocmask (SIG_SETMASK, &oldsigset, 0);
556 UNLOCK (wrklock);
557}
558
559static void maybe_start_thread ()
560{
561#if 0
562 static struct timeval last;
563 struct timeval diff, now;
564#endif
565
566 if (started >= wanted)
567 return;
568
569 if (nready <= nreqs - get_nready () - get_npending ())
570 return;
571
572#if 0
573 gettimeofday (&now, 0);
574
575 diff.tv_sec = now.tv_sec - last.tv_sec;
576 diff.tv_usec = now.tv_usec - last.tv_usec;
577
578 if (diff.tv_usec < 0)
579 {
580 --diff.tv_sec;
581 diff.tv_usec += 1000000;
582 }
583
584 if (!diff.tv_sec && diff.tv_usec < 10000)
585 return;
586
587 last = now;
588#endif
589
590 start_thread ();
591}
592
593static void req_send (aio_req req)
594{
595 ++nreqs;
596
597 LOCK (reqlock);
598 ++nready;
599 reqq_push (&req_queue, req);
600 pthread_cond_signal (&reqwait);
601 UNLOCK (reqlock);
602
603 maybe_start_thread ();
604}
605
606static void end_thread (void)
607{
608 aio_req req;
609
610 Newz (0, req, 1, aio_cb);
611
612 req->type = REQ_QUIT;
613 req->pri = PRI_MAX + PRI_BIAS;
614
615 LOCK (reqlock);
616 reqq_push (&req_queue, req);
617 pthread_cond_signal (&reqwait);
618 UNLOCK (reqlock);
619
620 LOCK (wrklock);
621 --started;
622 UNLOCK (wrklock);
623}
624
625static void min_parallel (int nthreads)
626{
627 if (wanted < nthreads)
628 wanted = nthreads;
629}
630
631static void max_parallel (int nthreads)
632{
633 if (wanted > nthreads)
634 wanted = nthreads;
635
636 while (started > wanted)
637 end_thread ();
638}
639
640static void poll_wait ()
641{
642 fd_set rfd;
643
644 while (nreqs)
645 {
646 int size;
647 if (WORDREAD_UNSAFE) LOCK (reslock);
648 size = res_queue.size;
649 if (WORDREAD_UNSAFE) UNLOCK (reslock);
650
651 if (size)
652 return;
653
654 maybe_start_thread ();
655
656 FD_ZERO(&rfd);
657 FD_SET(respipe [0], &rfd);
658
659 select (respipe [0] + 1, &rfd, 0, 0, 0);
660 }
661}
662
495static int poll_cb () 663static int poll_cb (int max)
496{ 664{
497 dSP; 665 dSP;
498 int count = 0; 666 int count = 0;
499 int do_croak = 0; 667 int do_croak = 0;
500 aio_req req; 668 aio_req req;
501 669
502 for (;;) 670 for (;;)
503 { 671 {
504 LOCK (reslock); 672 while (max <= 0 || count < max)
505 req = reqq_shift (&res_queue);
506
507 if (req)
508 { 673 {
674 maybe_start_thread ();
675
676 LOCK (reslock);
677 req = reqq_shift (&res_queue);
678
509 if (!res_queue.size) 679 if (req)
510 { 680 {
681 --npending;
682
683 if (!res_queue.size)
684 {
511 /* read any signals sent by the worker threads */ 685 /* read any signals sent by the worker threads */
512 char buf [32]; 686 char buf [32];
513 while (read (respipe [0], buf, 32) == 32) 687 while (read (respipe [0], buf, 32) == 32)
688 ;
514 ; 689 }
515 } 690 }
691
692 UNLOCK (reslock);
693
694 if (!req)
695 break;
696
697 --nreqs;
698
699 if (req->type == REQ_GROUP && req->length)
700 {
701 req->fd = 1; /* mark request as delayed */
702 continue;
703 }
704 else
705 {
706 if (req->type == REQ_READ)
707 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
708
709 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
710 SvREADONLY_off (req->data);
711
712 if (req->statdata)
713 {
714 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
715 PL_laststatval = req->result;
716 PL_statcache = *(req->statdata);
717 }
718
719 req_invoke (req);
720
721 count++;
722 }
723
724 req_free (req);
516 } 725 }
517 726
518 UNLOCK (reslock); 727 if (nreqs <= max_outstanding)
519
520 if (!req)
521 break; 728 break;
522 729
523 --nreqs; 730 poll_wait ();
524 731
525 if (req->type == REQ_QUIT) 732 max = 0;
526 started--;
527 else if (req->type == REQ_GROUP && req->length)
528 {
529 req->fd = 1; /* mark request as delayed */
530 continue;
531 }
532 else
533 {
534 if (req->type == REQ_READ)
535 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
536
537 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
538 SvREADONLY_off (req->data);
539
540 if (req->statdata)
541 {
542 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
543 PL_laststatval = req->result;
544 PL_statcache = *(req->statdata);
545 }
546
547 req_invoke (req);
548
549 count++;
550 }
551
552 req_free (req);
553 } 733 }
554 734
555 return count; 735 return count;
556}
557
558static void *aio_proc(void *arg);
559
560static void start_thread (void)
561{
562 sigset_t fullsigset, oldsigset;
563 pthread_attr_t attr;
564
565 worker *wrk = calloc (1, sizeof (worker));
566
567 if (!wrk)
568 croak ("unable to allocate worker thread data");
569
570 pthread_attr_init (&attr);
571 pthread_attr_setstacksize (&attr, STACKSIZE);
572 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
573
574 sigfillset (&fullsigset);
575
576 LOCK (wrklock);
577 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
578
579 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
580 {
581 wrk->prev = &wrk_first;
582 wrk->next = wrk_first.next;
583 wrk_first.next->prev = wrk;
584 wrk_first.next = wrk;
585 started++;
586 }
587 else
588 free (wrk);
589
590 sigprocmask (SIG_SETMASK, &oldsigset, 0);
591 UNLOCK (wrklock);
592}
593
594static void req_send (aio_req req)
595{
596 while (started < wanted && nreqs >= started)
597 start_thread ();
598
599 ++nreqs;
600
601 LOCK (reqlock);
602 reqq_push (&req_queue, req);
603 pthread_cond_signal (&reqwait);
604 UNLOCK (reqlock);
605}
606
607static void end_thread (void)
608{
609 aio_req req;
610
611 Newz (0, req, 1, aio_cb);
612
613 req->type = REQ_QUIT;
614 req->pri = PRI_MAX + PRI_BIAS;
615
616 req_send (req);
617}
618
619static void min_parallel (int nthreads)
620{
621 if (wanted < nthreads)
622 wanted = nthreads;
623}
624
625static void max_parallel (int nthreads)
626{
627 int cur = started;
628
629 if (wanted > nthreads)
630 wanted = nthreads;
631
632 while (cur > wanted)
633 {
634 end_thread ();
635 cur--;
636 }
637
638 while (started > wanted)
639 {
640 poll_wait ();
641 poll_cb ();
642 }
643} 736}
644 737
645static void create_pipe () 738static void create_pipe ()
646{ 739{
647 if (pipe (respipe)) 740 if (pipe (respipe))
… …
702#if !HAVE_FDATASYNC 795#if !HAVE_FDATASYNC
703# define fdatasync fsync 796# define fdatasync fsync
704#endif 797#endif
705 798
706#if !HAVE_READAHEAD 799#if !HAVE_READAHEAD
707# define readahead aio_readahead 800# define readahead(fd,offset,count) aio_readahead (fd, offset, count, self)
708 801
709static ssize_t readahead (int fd, off_t offset, size_t count) 802static ssize_t aio_readahead (int fd, off_t offset, size_t count, worker *self)
710{ 803{
711 dBUF; 804 dBUF;
712 805
713 while (count > 0) 806 while (count > 0)
714 { 807 {
… …
719 count -= len; 812 count -= len;
720 } 813 }
721 814
722 errno = 0; 815 errno = 0;
723} 816}
817
724#endif 818#endif
725 819
726#if !HAVE_READDIR_R 820#if !HAVE_READDIR_R
727# define readdir_r aio_readdir_r 821# define readdir_r aio_readdir_r
728 822
… …
773 { 867 {
774 off_t sbytes; 868 off_t sbytes;
775 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0); 869 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
776 870
777 if (res < 0 && sbytes) 871 if (res < 0 && sbytes)
778 /* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */ 872 /* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
779 res = sbytes; 873 res = sbytes;
780 } 874 }
781 875
782# elif __hpux 876# elif __hpux
783 res = sendfile (ofd, ifd, offset, count, 0, 0); 877 res = sendfile (ofd, ifd, offset, count, 0, 0);
… …
863 int errorno; 957 int errorno;
864 958
865 LOCK (wrklock); 959 LOCK (wrklock);
866 self->dirp = dirp = opendir (req->dataptr); 960 self->dirp = dirp = opendir (req->dataptr);
867 self->dbuf = u = malloc (sizeof (*u)); 961 self->dbuf = u = malloc (sizeof (*u));
962 req->data2ptr = names = malloc (memlen);
868 UNLOCK (wrklock); 963 UNLOCK (wrklock);
869
870 req->data2ptr = names = malloc (memlen);
871 964
872 if (dirp && u && names) 965 if (dirp && u && names)
873 for (;;) 966 for (;;)
874 { 967 {
875 errno = 0; 968 errno = 0;
… …
911/*****************************************************************************/ 1004/*****************************************************************************/
912 1005
913static void *aio_proc (void *thr_arg) 1006static void *aio_proc (void *thr_arg)
914{ 1007{
915 aio_req req; 1008 aio_req req;
916 int type;
917 worker *self = (worker *)thr_arg; 1009 worker *self = (worker *)thr_arg;
918 1010
919 do 1011 for (;;)
920 { 1012 {
921 LOCK (reqlock); 1013 LOCK (reqlock);
922 1014
923 for (;;) 1015 for (;;)
924 { 1016 {
… …
928 break; 1020 break;
929 1021
930 pthread_cond_wait (&reqwait, &reqlock); 1022 pthread_cond_wait (&reqwait, &reqlock);
931 } 1023 }
932 1024
1025 --nready;
1026
933 UNLOCK (reqlock); 1027 UNLOCK (reqlock);
934 1028
935 errno = 0; /* strictly unnecessary */ 1029 errno = 0; /* strictly unnecessary */
936 type = req->type; /* remember type for QUIT check */
937 1030
938 if (!(req->flags & FLAG_CANCELLED)) 1031 if (!(req->flags & FLAG_CANCELLED))
939 switch (type) 1032 switch (req->type)
940 { 1033 {
941 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 1034 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
942 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 1035 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
943 1036
944 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 1037 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
953 case REQ_UNLINK: req->result = unlink (req->dataptr); break; 1046 case REQ_UNLINK: req->result = unlink (req->dataptr); break;
954 case REQ_RMDIR: req->result = rmdir (req->dataptr); break; 1047 case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
955 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break; 1048 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
956 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break; 1049 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
957 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break; 1050 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
1051 case REQ_MKNOD: req->result = mknod (req->data2ptr, req->mode, (dev_t)req->offset); break;
958 1052
959 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 1053 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
960 case REQ_FSYNC: req->result = fsync (req->fd); break; 1054 case REQ_FSYNC: req->result = fsync (req->fd); break;
961 case REQ_READDIR: scandir_ (req, self); break; 1055 case REQ_READDIR: scandir_ (req, self); break;
962 1056
… …
970 req->result = select (0, 0, 0, 0, &tv); 1064 req->result = select (0, 0, 0, 0, &tv);
971 } 1065 }
972 1066
973 case REQ_GROUP: 1067 case REQ_GROUP:
974 case REQ_NOP: 1068 case REQ_NOP:
1069 break;
1070
975 case REQ_QUIT: 1071 case REQ_QUIT:
1072 LOCK (wrklock);
1073 worker_free (self);
1074 --started;
1075 UNLOCK (wrklock);
976 break; 1076 return 0;
977 1077
978 default: 1078 default:
979 req->result = ENOSYS; 1079 req->result = ENOSYS;
980 break; 1080 break;
981 } 1081 }
982 1082
983 req->errorno = errno; 1083 req->errorno = errno;
984 1084
985 LOCK (reslock); 1085 LOCK (reslock);
986 1086
1087 ++npending;
1088
987 if (!reqq_push (&res_queue, req)) 1089 if (!reqq_push (&res_queue, req))
988 /* write a dummy byte to the pipe so fh becomes ready */ 1090 /* write a dummy byte to the pipe so fh becomes ready */
989 write (respipe [1], &respipe, 1); 1091 write (respipe [1], &respipe, 1);
990 1092
991 self->req = 0; 1093 self->req = 0;
992 worker_clear (self); 1094 worker_clear (self);
993 1095
994 UNLOCK (reslock); 1096 UNLOCK (reslock);
995 } 1097 }
996 while (type != REQ_QUIT);
997
998 LOCK (wrklock);
999 worker_free (self);
1000 UNLOCK (wrklock);
1001
1002 return 0;
1003} 1098}
1004 1099
1005/*****************************************************************************/ 1100/*****************************************************************************/
1006 1101
1007static void atfork_prepare (void) 1102static void atfork_prepare (void)
… …
1087PROTOTYPES: ENABLE 1182PROTOTYPES: ENABLE
1088 1183
1089BOOT: 1184BOOT:
1090{ 1185{
1091 HV *stash = gv_stashpv ("IO::AIO", 1); 1186 HV *stash = gv_stashpv ("IO::AIO", 1);
1187
1092 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); 1188 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1093 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1189 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1094 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1190 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1191 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
1192 newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
1193 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1095 1194
1096 create_pipe (); 1195 create_pipe ();
1097 pthread_atfork (atfork_prepare, atfork_parent, atfork_child); 1196 pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1098}
1099 1197
1198 start_thread ();
1199}
1200
1100void 1201void
1101min_parallel (nthreads) 1202min_parallel (int nthreads)
1102 int nthreads
1103 PROTOTYPE: $ 1203 PROTOTYPE: $
1104 1204
1105void 1205void
1106max_parallel (nthreads) 1206max_parallel (int nthreads)
1107 int nthreads
1108 PROTOTYPE: $ 1207 PROTOTYPE: $
1208
1209int
1210max_outstanding (int maxreqs)
1211 PROTOTYPE: $
1212 CODE:
1213 RETVAL = max_outstanding;
1214 max_outstanding = maxreqs;
1215 OUTPUT:
1216 RETVAL
1109 1217
1110void 1218void
1111aio_open (pathname,flags,mode,callback=&PL_sv_undef) 1219aio_open (pathname,flags,mode,callback=&PL_sv_undef)
1112 SV * pathname 1220 SV * pathname
1113 int flags 1221 int flags
… …
1328 1436
1329 REQ_SEND; 1437 REQ_SEND;
1330} 1438}
1331 1439
1332void 1440void
1441aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
1442 SV * pathname
1443 SV * callback
1444 UV mode
1445 UV dev
1446 PPCODE:
1447{
1448 dREQ;
1449
1450 req->type = REQ_MKNOD;
1451 req->data = newSVsv (pathname);
1452 req->dataptr = SvPVbyte_nolen (req->data);
1453 req->mode = (mode_t)mode;
1454 req->offset = dev;
1455
1456 REQ_SEND;
1457}
1458
1459void
1333aio_busy (delay,callback=&PL_sv_undef) 1460aio_busy (delay,callback=&PL_sv_undef)
1334 double delay 1461 double delay
1335 SV * callback 1462 SV * callback
1336 PPCODE: 1463 PPCODE:
1337{ 1464{
… …
1368 req->type = REQ_NOP; 1495 req->type = REQ_NOP;
1369 1496
1370 REQ_SEND; 1497 REQ_SEND;
1371} 1498}
1372 1499
1373void 1500int
1374aioreq_pri (int pri = DEFAULT_PRI) 1501aioreq_pri (int pri = 0)
1375 CODE: 1502 PROTOTYPE: ;$
1503 CODE:
1504 RETVAL = next_pri - PRI_BIAS;
1505 if (items > 0)
1506 {
1376 if (pri < PRI_MIN) pri = PRI_MIN; 1507 if (pri < PRI_MIN) pri = PRI_MIN;
1377 if (pri > PRI_MAX) pri = PRI_MAX; 1508 if (pri > PRI_MAX) pri = PRI_MAX;
1378 next_pri = pri + PRI_BIAS; 1509 next_pri = pri + PRI_BIAS;
1510 }
1511 OUTPUT:
1512 RETVAL
1379 1513
1380void 1514void
1381aioreq_nice (int nice = 0) 1515aioreq_nice (int nice = 0)
1382 CODE: 1516 CODE:
1383 nice = next_pri - nice; 1517 nice = next_pri - nice;
1384 if (nice < PRI_MIN) nice = PRI_MIN; 1518 if (nice < PRI_MIN) nice = PRI_MIN;
1385 if (nice > PRI_MAX) nice = PRI_MAX; 1519 if (nice > PRI_MAX) nice = PRI_MAX;
1386 next_pri = nice + PRI_BIAS; 1520 next_pri = nice + PRI_BIAS;
1387 1521
1388void 1522void
1389flush () 1523flush ()
1390 PROTOTYPE: 1524 PROTOTYPE:
1391 CODE: 1525 CODE:
1392 while (nreqs) 1526 while (nreqs)
1393 { 1527 {
1394 poll_wait (); 1528 poll_wait ();
1395 poll_cb (); 1529 poll_cb (0);
1396 } 1530 }
1397 1531
1398void 1532void
1399poll() 1533poll()
1400 PROTOTYPE: 1534 PROTOTYPE:
1401 CODE: 1535 CODE:
1402 if (nreqs) 1536 if (nreqs)
1403 { 1537 {
1404 poll_wait (); 1538 poll_wait ();
1405 poll_cb (); 1539 poll_cb (0);
1406 } 1540 }
1407 1541
1408int 1542int
1409poll_fileno() 1543poll_fileno()
1410 PROTOTYPE: 1544 PROTOTYPE:
… …
1415 1549
1416int 1550int
1417poll_cb(...) 1551poll_cb(...)
1418 PROTOTYPE: 1552 PROTOTYPE:
1419 CODE: 1553 CODE:
1420 RETVAL = poll_cb (); 1554 RETVAL = poll_cb (0);
1555 OUTPUT:
1556 RETVAL
1557
1558int
1559poll_some(int max = 0)
1560 PROTOTYPE: $
1561 CODE:
1562 RETVAL = poll_cb (max);
1421 OUTPUT: 1563 OUTPUT:
1422 RETVAL 1564 RETVAL
1423 1565
1424void 1566void
1425poll_wait() 1567poll_wait()
… …
1434 CODE: 1576 CODE:
1435 RETVAL = nreqs; 1577 RETVAL = nreqs;
1436 OUTPUT: 1578 OUTPUT:
1437 RETVAL 1579 RETVAL
1438 1580
1581int
1582nready()
1583 PROTOTYPE:
1584 CODE:
1585 RETVAL = get_nready ();
1586 OUTPUT:
1587 RETVAL
1588
1589int
1590npending()
1591 PROTOTYPE:
1592 CODE:
1593 RETVAL = get_npending ();
1594 OUTPUT:
1595 RETVAL
1596
1439PROTOTYPES: DISABLE 1597PROTOTYPES: DISABLE
1440 1598
1441MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1599MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1442 1600
1443void 1601void
… …
1494void 1652void
1495result (aio_req grp, ...) 1653result (aio_req grp, ...)
1496 CODE: 1654 CODE:
1497{ 1655{
1498 int i; 1656 int i;
1657 AV *av;
1658
1659 grp->errorno = errno;
1660
1499 AV *av = newAV (); 1661 av = newAV ();
1500 1662
1501 for (i = 1; i < items; ++i ) 1663 for (i = 1; i < items; ++i )
1502 av_push (av, newSVsv (ST (i))); 1664 av_push (av, newSVsv (ST (i)));
1503 1665
1504 SvREFCNT_dec (grp->data); 1666 SvREFCNT_dec (grp->data);
1505 grp->data = (SV *)av; 1667 grp->data = (SV *)av;
1506} 1668}
1669
1670void
1671errno (aio_req grp, int errorno = errno)
1672 CODE:
1673 grp->errorno = errorno;
1507 1674
1508void 1675void
1509limit (aio_req grp, int limit) 1676limit (aio_req grp, int limit)
1510 CODE: 1677 CODE:
1511 grp->fd2 = limit; 1678 grp->fd2 = limit;

Diff Legend

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