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.72 by root, Tue Oct 24 17:22:17 2006 UTC vs.
Revision 1.82 by root, Fri Oct 27 20:11:58 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 max_outstanding = 1<<30;
134static int respipe [2];
135 150
136#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP) 151#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
137# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP 152# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
138#else 153#else
139# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER 154# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
140#endif 155#endif
141 156
142#define LOCK(mutex) pthread_mutex_lock (&(mutex)) 157#define LOCK(mutex) pthread_mutex_lock (&(mutex))
143#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex)) 158#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
144 159
145/* worker threasd management */ 160/* worker threads management */
146static pthread_mutex_t wrklock = AIO_MUTEX_INIT; 161static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
147 162
148typedef struct worker { 163typedef struct worker {
149 /* locked by wrklock */ 164 /* locked by wrklock */
150 struct worker *prev, *next; 165 struct worker *prev, *next;
… …
180 wrk->prev->next = wrk->next; 195 wrk->prev->next = wrk->next;
181 196
182 free (wrk); 197 free (wrk);
183} 198}
184 199
200static volatile unsigned int nreqs, nready, npending;
201static volatile unsigned int max_outstanding = 0xffffffff;
202static int respipe [2];
203
185static pthread_mutex_t reslock = AIO_MUTEX_INIT; 204static pthread_mutex_t reslock = AIO_MUTEX_INIT;
186static pthread_mutex_t reqlock = AIO_MUTEX_INIT; 205static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
187static 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
188 238
189/* 239/*
190 * a somewhat faster data structure might be nice, but 240 * a somewhat faster data structure might be nice, but
191 * with 8 priorities this actually needs <20 insns 241 * with 8 priorities this actually needs <20 insns
192 * per shift, the most expensive operation. 242 * per shift, the most expensive operation.
… …
238 } 288 }
239 289
240 abort (); 290 abort ();
241} 291}
242 292
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 req_send (req);
616
617 LOCK (wrklock);
618 --started;
619 UNLOCK (wrklock);
620}
621
622static void min_parallel (int nthreads)
623{
624 if (wanted < nthreads)
625 wanted = nthreads;
626}
627
628static void max_parallel (int nthreads)
629{
630 if (wanted > nthreads)
631 wanted = nthreads;
632
633 while (started > wanted)
634 end_thread ();
635}
636
637static void poll_wait ()
638{
639 fd_set rfd;
640
641 while (nreqs)
642 {
643 int size;
644 if (WORDREAD_UNSAFE) LOCK (reslock);
645 size = res_queue.size;
646 if (WORDREAD_UNSAFE) UNLOCK (reslock);
647
648 if (size)
649 return;
650
651 maybe_start_thread ();
652
653 FD_ZERO(&rfd);
654 FD_SET(respipe [0], &rfd);
655
656 select (respipe [0] + 1, &rfd, 0, 0, 0);
657 }
658}
659
495static int poll_cb () 660static int poll_cb (int max)
496{ 661{
497 dSP; 662 dSP;
498 int count = 0; 663 int count = 0;
499 int do_croak = 0; 664 int do_croak = 0;
500 aio_req req; 665 aio_req req;
501 666
502 for (;;) 667 for (;;)
503 { 668 {
504 LOCK (reslock); 669 while (max <= 0 || count < max)
505 req = reqq_shift (&res_queue);
506
507 if (req)
508 { 670 {
671 maybe_start_thread ();
672
673 LOCK (reslock);
674 req = reqq_shift (&res_queue);
675
509 if (!res_queue.size) 676 if (req)
510 { 677 {
678 --npending;
679
680 if (!res_queue.size)
681 {
511 /* read any signals sent by the worker threads */ 682 /* read any signals sent by the worker threads */
512 char buf [32]; 683 char buf [32];
513 while (read (respipe [0], buf, 32) == 32) 684 while (read (respipe [0], buf, 32) == 32)
685 ;
514 ; 686 }
515 } 687 }
688
689 UNLOCK (reslock);
690
691 if (!req)
692 break;
693
694 --nreqs;
695
696 if (req->type == REQ_GROUP && req->length)
697 {
698 req->fd = 1; /* mark request as delayed */
699 continue;
700 }
701 else
702 {
703 if (req->type == REQ_READ)
704 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
705
706 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
707 SvREADONLY_off (req->data);
708
709 if (req->statdata)
710 {
711 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
712 PL_laststatval = req->result;
713 PL_statcache = *(req->statdata);
714 }
715
716 req_invoke (req);
717
718 count++;
719 }
720
721 req_free (req);
516 } 722 }
517 723
518 UNLOCK (reslock); 724 if (nreqs <= max_outstanding)
519
520 if (!req)
521 break; 725 break;
522 726
523 --nreqs; 727 poll_wait ();
524 728
525 if (req->type == REQ_QUIT) 729 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 } 730 }
554 731
555 return count; 732 return count;
556}
557
558static void *aio_proc(void *arg);
559
560static void start_thread (void)
561{
562 worker *wrk = calloc (1, sizeof (worker));
563
564 if (!wrk)
565 croak ("unable to allocate worker thread data");
566
567 sigset_t fullsigset, oldsigset;
568 pthread_attr_t attr;
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 if (nreqs > max_outstanding)
607 for (;;)
608 {
609 poll_cb ();
610
611 if (nreqs <= max_outstanding)
612 break;
613
614 poll_wait ();
615 }
616}
617
618static void end_thread (void)
619{
620 aio_req req;
621
622 Newz (0, req, 1, aio_cb);
623
624 req->type = REQ_QUIT;
625 req->pri = PRI_MAX + PRI_BIAS;
626
627 req_send (req);
628}
629
630static void min_parallel (int nthreads)
631{
632 if (wanted < nthreads)
633 wanted = nthreads;
634}
635
636static void max_parallel (int nthreads)
637{
638 int cur = started;
639
640 if (wanted > nthreads)
641 wanted = nthreads;
642
643 while (cur > wanted)
644 {
645 end_thread ();
646 cur--;
647 }
648
649 while (started > wanted)
650 {
651 poll_wait ();
652 poll_cb ();
653 }
654} 733}
655 734
656static void create_pipe () 735static void create_pipe ()
657{ 736{
658 if (pipe (respipe)) 737 if (pipe (respipe))
… …
713#if !HAVE_FDATASYNC 792#if !HAVE_FDATASYNC
714# define fdatasync fsync 793# define fdatasync fsync
715#endif 794#endif
716 795
717#if !HAVE_READAHEAD 796#if !HAVE_READAHEAD
718# define readahead aio_readahead 797# define readahead(fd,offset,count) aio_readahead (fd, offset, count, self)
719 798
720static ssize_t readahead (int fd, off_t offset, size_t count) 799static ssize_t aio_readahead (int fd, off_t offset, size_t count, worker *self)
721{ 800{
722 dBUF; 801 dBUF;
723 802
724 while (count > 0) 803 while (count > 0)
725 { 804 {
… …
730 count -= len; 809 count -= len;
731 } 810 }
732 811
733 errno = 0; 812 errno = 0;
734} 813}
814
735#endif 815#endif
736 816
737#if !HAVE_READDIR_R 817#if !HAVE_READDIR_R
738# define readdir_r aio_readdir_r 818# define readdir_r aio_readdir_r
739 819
… …
784 { 864 {
785 off_t sbytes; 865 off_t sbytes;
786 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0); 866 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
787 867
788 if (res < 0 && sbytes) 868 if (res < 0 && sbytes)
789 /* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */ 869 /* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
790 res = sbytes; 870 res = sbytes;
791 } 871 }
792 872
793# elif __hpux 873# elif __hpux
794 res = sendfile (ofd, ifd, offset, count, 0, 0); 874 res = sendfile (ofd, ifd, offset, count, 0, 0);
… …
874 int errorno; 954 int errorno;
875 955
876 LOCK (wrklock); 956 LOCK (wrklock);
877 self->dirp = dirp = opendir (req->dataptr); 957 self->dirp = dirp = opendir (req->dataptr);
878 self->dbuf = u = malloc (sizeof (*u)); 958 self->dbuf = u = malloc (sizeof (*u));
959 req->data2ptr = names = malloc (memlen);
879 UNLOCK (wrklock); 960 UNLOCK (wrklock);
880
881 req->data2ptr = names = malloc (memlen);
882 961
883 if (dirp && u && names) 962 if (dirp && u && names)
884 for (;;) 963 for (;;)
885 { 964 {
886 errno = 0; 965 errno = 0;
… …
922/*****************************************************************************/ 1001/*****************************************************************************/
923 1002
924static void *aio_proc (void *thr_arg) 1003static void *aio_proc (void *thr_arg)
925{ 1004{
926 aio_req req; 1005 aio_req req;
927 int type;
928 worker *self = (worker *)thr_arg; 1006 worker *self = (worker *)thr_arg;
929 1007
930 do 1008 for (;;)
931 { 1009 {
932 LOCK (reqlock); 1010 LOCK (reqlock);
933 1011
934 for (;;) 1012 for (;;)
935 { 1013 {
… …
939 break; 1017 break;
940 1018
941 pthread_cond_wait (&reqwait, &reqlock); 1019 pthread_cond_wait (&reqwait, &reqlock);
942 } 1020 }
943 1021
1022 --nready;
1023
944 UNLOCK (reqlock); 1024 UNLOCK (reqlock);
945 1025
946 errno = 0; /* strictly unnecessary */ 1026 errno = 0; /* strictly unnecessary */
947 type = req->type; /* remember type for QUIT check */
948 1027
949 if (!(req->flags & FLAG_CANCELLED)) 1028 if (!(req->flags & FLAG_CANCELLED))
950 switch (type) 1029 switch (req->type)
951 { 1030 {
952 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 1031 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
953 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 1032 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
954 1033
955 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 1034 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
964 case REQ_UNLINK: req->result = unlink (req->dataptr); break; 1043 case REQ_UNLINK: req->result = unlink (req->dataptr); break;
965 case REQ_RMDIR: req->result = rmdir (req->dataptr); break; 1044 case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
966 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break; 1045 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
967 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break; 1046 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
968 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break; 1047 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
1048 case REQ_MKNOD: req->result = mknod (req->data2ptr, req->mode, (dev_t)req->offset); break;
969 1049
970 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 1050 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
971 case REQ_FSYNC: req->result = fsync (req->fd); break; 1051 case REQ_FSYNC: req->result = fsync (req->fd); break;
972 case REQ_READDIR: scandir_ (req, self); break; 1052 case REQ_READDIR: scandir_ (req, self); break;
973 1053
… …
981 req->result = select (0, 0, 0, 0, &tv); 1061 req->result = select (0, 0, 0, 0, &tv);
982 } 1062 }
983 1063
984 case REQ_GROUP: 1064 case REQ_GROUP:
985 case REQ_NOP: 1065 case REQ_NOP:
1066 break;
1067
986 case REQ_QUIT: 1068 case REQ_QUIT:
1069 LOCK (wrklock);
1070 worker_free (self);
1071 --started;
1072 UNLOCK (wrklock);
987 break; 1073 return 0;
988 1074
989 default: 1075 default:
990 req->result = ENOSYS; 1076 req->result = ENOSYS;
991 break; 1077 break;
992 } 1078 }
993 1079
994 req->errorno = errno; 1080 req->errorno = errno;
995 1081
996 LOCK (reslock); 1082 LOCK (reslock);
997 1083
1084 ++npending;
1085
998 if (!reqq_push (&res_queue, req)) 1086 if (!reqq_push (&res_queue, req))
999 /* write a dummy byte to the pipe so fh becomes ready */ 1087 /* write a dummy byte to the pipe so fh becomes ready */
1000 write (respipe [1], &respipe, 1); 1088 write (respipe [1], &respipe, 1);
1001 1089
1002 self->req = 0; 1090 self->req = 0;
1003 worker_clear (self); 1091 worker_clear (self);
1004 1092
1005 UNLOCK (reslock); 1093 UNLOCK (reslock);
1006 } 1094 }
1007 while (type != REQ_QUIT);
1008
1009 LOCK (wrklock);
1010 worker_free (self);
1011 UNLOCK (wrklock);
1012
1013 return 0;
1014} 1095}
1015 1096
1016/*****************************************************************************/ 1097/*****************************************************************************/
1017 1098
1018static void atfork_prepare (void) 1099static void atfork_prepare (void)
… …
1098PROTOTYPES: ENABLE 1179PROTOTYPES: ENABLE
1099 1180
1100BOOT: 1181BOOT:
1101{ 1182{
1102 HV *stash = gv_stashpv ("IO::AIO", 1); 1183 HV *stash = gv_stashpv ("IO::AIO", 1);
1184
1103 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); 1185 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1104 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1186 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1105 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1187 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1188 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
1189 newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
1190 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1106 1191
1107 create_pipe (); 1192 create_pipe ();
1108 pthread_atfork (atfork_prepare, atfork_parent, atfork_child); 1193 pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1109}
1110 1194
1195 start_thread ();
1196}
1197
1111void 1198void
1112min_parallel (nthreads) 1199min_parallel (int nthreads)
1113 int nthreads
1114 PROTOTYPE: $ 1200 PROTOTYPE: $
1115 1201
1116void 1202void
1117max_parallel (nthreads) 1203max_parallel (int nthreads)
1118 int nthreads
1119 PROTOTYPE: $ 1204 PROTOTYPE: $
1120 1205
1121int 1206int
1122max_outstanding (nreqs) 1207max_outstanding (int maxreqs)
1123 int nreqs 1208 PROTOTYPE: $
1124 PROTOTYPE: $
1125 CODE: 1209 CODE:
1126 RETVAL = max_outstanding; 1210 RETVAL = max_outstanding;
1127 max_outstanding = nreqs; 1211 max_outstanding = maxreqs;
1212 OUTPUT:
1213 RETVAL
1128 1214
1129void 1215void
1130aio_open (pathname,flags,mode,callback=&PL_sv_undef) 1216aio_open (pathname,flags,mode,callback=&PL_sv_undef)
1131 SV * pathname 1217 SV * pathname
1132 int flags 1218 int flags
… …
1347 1433
1348 REQ_SEND; 1434 REQ_SEND;
1349} 1435}
1350 1436
1351void 1437void
1438aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
1439 SV * pathname
1440 SV * callback
1441 UV mode
1442 UV dev
1443 PPCODE:
1444{
1445 dREQ;
1446
1447 req->type = REQ_MKNOD;
1448 req->data = newSVsv (pathname);
1449 req->dataptr = SvPVbyte_nolen (req->data);
1450 req->mode = (mode_t)mode;
1451 req->offset = dev;
1452
1453 REQ_SEND;
1454}
1455
1456void
1352aio_busy (delay,callback=&PL_sv_undef) 1457aio_busy (delay,callback=&PL_sv_undef)
1353 double delay 1458 double delay
1354 SV * callback 1459 SV * callback
1355 PPCODE: 1460 PPCODE:
1356{ 1461{
… …
1387 req->type = REQ_NOP; 1492 req->type = REQ_NOP;
1388 1493
1389 REQ_SEND; 1494 REQ_SEND;
1390} 1495}
1391 1496
1392void 1497int
1393aioreq_pri (int pri = DEFAULT_PRI) 1498aioreq_pri (int pri = 0)
1394 CODE: 1499 PROTOTYPE: ;$
1500 CODE:
1501 RETVAL = next_pri - PRI_BIAS;
1502 if (items > 0)
1503 {
1395 if (pri < PRI_MIN) pri = PRI_MIN; 1504 if (pri < PRI_MIN) pri = PRI_MIN;
1396 if (pri > PRI_MAX) pri = PRI_MAX; 1505 if (pri > PRI_MAX) pri = PRI_MAX;
1397 next_pri = pri + PRI_BIAS; 1506 next_pri = pri + PRI_BIAS;
1507 }
1508 OUTPUT:
1509 RETVAL
1398 1510
1399void 1511void
1400aioreq_nice (int nice = 0) 1512aioreq_nice (int nice = 0)
1401 CODE: 1513 CODE:
1402 nice = next_pri - nice; 1514 nice = next_pri - nice;
1403 if (nice < PRI_MIN) nice = PRI_MIN; 1515 if (nice < PRI_MIN) nice = PRI_MIN;
1404 if (nice > PRI_MAX) nice = PRI_MAX; 1516 if (nice > PRI_MAX) nice = PRI_MAX;
1405 next_pri = nice + PRI_BIAS; 1517 next_pri = nice + PRI_BIAS;
1406 1518
1407void 1519void
1408flush () 1520flush ()
1409 PROTOTYPE: 1521 PROTOTYPE:
1410 CODE: 1522 CODE:
1411 while (nreqs) 1523 while (nreqs)
1412 { 1524 {
1413 poll_wait (); 1525 poll_wait ();
1414 poll_cb (); 1526 poll_cb (0);
1415 } 1527 }
1416 1528
1417void 1529void
1418poll() 1530poll()
1419 PROTOTYPE: 1531 PROTOTYPE:
1420 CODE: 1532 CODE:
1421 if (nreqs) 1533 if (nreqs)
1422 { 1534 {
1423 poll_wait (); 1535 poll_wait ();
1424 poll_cb (); 1536 poll_cb (0);
1425 } 1537 }
1426 1538
1427int 1539int
1428poll_fileno() 1540poll_fileno()
1429 PROTOTYPE: 1541 PROTOTYPE:
… …
1434 1546
1435int 1547int
1436poll_cb(...) 1548poll_cb(...)
1437 PROTOTYPE: 1549 PROTOTYPE:
1438 CODE: 1550 CODE:
1439 RETVAL = poll_cb (); 1551 RETVAL = poll_cb (0);
1552 OUTPUT:
1553 RETVAL
1554
1555int
1556poll_some(int max = 0)
1557 PROTOTYPE: $
1558 CODE:
1559 RETVAL = poll_cb (max);
1440 OUTPUT: 1560 OUTPUT:
1441 RETVAL 1561 RETVAL
1442 1562
1443void 1563void
1444poll_wait() 1564poll_wait()
… …
1453 CODE: 1573 CODE:
1454 RETVAL = nreqs; 1574 RETVAL = nreqs;
1455 OUTPUT: 1575 OUTPUT:
1456 RETVAL 1576 RETVAL
1457 1577
1578int
1579nready()
1580 PROTOTYPE:
1581 CODE:
1582 RETVAL = get_nready ();
1583 OUTPUT:
1584 RETVAL
1585
1586int
1587npending()
1588 PROTOTYPE:
1589 CODE:
1590 RETVAL = get_npending ();
1591 OUTPUT:
1592 RETVAL
1593
1458PROTOTYPES: DISABLE 1594PROTOTYPES: DISABLE
1459 1595
1460MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1596MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1461 1597
1462void 1598void
… …
1513void 1649void
1514result (aio_req grp, ...) 1650result (aio_req grp, ...)
1515 CODE: 1651 CODE:
1516{ 1652{
1517 int i; 1653 int i;
1654 AV *av;
1655
1656 grp->errorno = errno;
1657
1518 AV *av = newAV (); 1658 av = newAV ();
1519 1659
1520 for (i = 1; i < items; ++i ) 1660 for (i = 1; i < items; ++i )
1521 av_push (av, newSVsv (ST (i))); 1661 av_push (av, newSVsv (ST (i)));
1522 1662
1523 SvREFCNT_dec (grp->data); 1663 SvREFCNT_dec (grp->data);
1524 grp->data = (SV *)av; 1664 grp->data = (SV *)av;
1525} 1665}
1666
1667void
1668errno (aio_req grp, int errorno = errno)
1669 CODE:
1670 grp->errorno = errorno;
1526 1671
1527void 1672void
1528limit (aio_req grp, int limit) 1673limit (aio_req grp, int limit)
1529 CODE: 1674 CODE:
1530 grp->fd2 = limit; 1675 grp->fd2 = limit;

Diff Legend

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