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.75 by root, Thu Oct 26 06:44:48 2006 UTC vs.
Revision 1.80 by root, Fri Oct 27 19:17:23 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"
… …
60# define STACKSIZE PTHREAD_STACK_MIN 61# define STACKSIZE PTHREAD_STACK_MIN
61#else 62#else
62# define STACKSIZE 16384 63# define STACKSIZE 16384
63#endif 64#endif
64 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
76#endif
77
65/* buffer size for various temporary buffers */ 78/* buffer size for various temporary buffers */
66#define AIO_BUFSIZE 65536 79#define AIO_BUFSIZE 65536
67 80
68#define dBUF \ 81#define dBUF \
69 char *aio_buf; \ 82 char *aio_buf; \
… …
131 NUM_PRI = PRI_MAX + PRI_BIAS + 1, 144 NUM_PRI = PRI_MAX + PRI_BIAS + 1,
132}; 145};
133 146
134static int next_pri = DEFAULT_PRI + PRI_BIAS; 147static int next_pri = DEFAULT_PRI + PRI_BIAS;
135 148
136static int started, wanted; 149static unsigned int started, wanted;
137static volatile int nreqs;
138static int respipe [2];
139 150
140#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP) 151#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
141# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP 152# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
142#else 153#else
143# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER 154# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
144#endif 155#endif
145 156
146#define LOCK(mutex) pthread_mutex_lock (&(mutex)) 157#define LOCK(mutex) pthread_mutex_lock (&(mutex))
147#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex)) 158#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
148 159
149/* worker threasd management */ 160/* worker threads management */
150static pthread_mutex_t wrklock = AIO_MUTEX_INIT; 161static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
151 162
152typedef struct worker { 163typedef struct worker {
153 /* locked by wrklock */ 164 /* locked by wrklock */
154 struct worker *prev, *next; 165 struct worker *prev, *next;
… …
184 wrk->prev->next = wrk->next; 195 wrk->prev->next = wrk->next;
185 196
186 free (wrk); 197 free (wrk);
187} 198}
188 199
200static volatile unsigned int nreqs, nready, npending;
201static volatile unsigned int max_outstanding = 0xffffffff;
202static int respipe [2];
203
189static pthread_mutex_t reslock = AIO_MUTEX_INIT; 204static pthread_mutex_t reslock = AIO_MUTEX_INIT;
190static pthread_mutex_t reqlock = AIO_MUTEX_INIT; 205static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
191static 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
192 238
193/* 239/*
194 * a somewhat faster data structure might be nice, but 240 * a somewhat faster data structure might be nice, but
195 * with 8 priorities this actually needs <20 insns 241 * with 8 priorities this actually needs <20 insns
196 * per shift, the most expensive operation. 242 * per shift, the most expensive operation.
… …
242 } 288 }
243 289
244 abort (); 290 abort ();
245} 291}
246 292
247static int poll_cb (); 293static int poll_cb (int max);
248static void req_invoke (aio_req req); 294static void req_invoke (aio_req req);
249static void req_free (aio_req req); 295static void req_free (aio_req req);
250static void req_cancel (aio_req req); 296static void req_cancel (aio_req req);
251 297
252/* must be called at most once */ 298/* must be called at most once */
… …
317 req_invoke (grp); 363 req_invoke (grp);
318 req_free (grp); 364 req_free (grp);
319 } 365 }
320} 366}
321 367
322static void poll_wait ()
323{
324 fd_set rfd;
325
326 while (nreqs)
327 {
328 int size;
329#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
330 LOCK (reslock);
331#endif
332 size = res_queue.size;
333#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
334 UNLOCK (reslock);
335#endif
336
337 if (size)
338 return;
339
340 FD_ZERO(&rfd);
341 FD_SET(respipe [0], &rfd);
342
343 select (respipe [0] + 1, &rfd, 0, 0, 0);
344 }
345}
346
347static void req_invoke (aio_req req) 368static void req_invoke (aio_req req)
348{ 369{
349 dSP; 370 dSP;
350 371
351 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback)) 372 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
352 { 373 {
353 errno = req->errorno;
354
355 ENTER; 374 ENTER;
356 SAVETMPS; 375 SAVETMPS;
357 PUSHMARK (SP); 376 PUSHMARK (SP);
358 EXTEND (SP, 1); 377 EXTEND (SP, 1);
359 378
… …
424 default: 443 default:
425 PUSHs (sv_2mortal (newSViv (req->result))); 444 PUSHs (sv_2mortal (newSViv (req->result)));
426 break; 445 break;
427 } 446 }
428 447
448 errno = req->errorno;
429 449
430 PUTBACK; 450 PUTBACK;
431 call_sv (req->callback, G_VOID | G_EVAL); 451 call_sv (req->callback, G_VOID | G_EVAL);
432 SPAGAIN; 452 SPAGAIN;
433 453
… …
495 req->flags |= FLAG_CANCELLED; 515 req->flags |= FLAG_CANCELLED;
496 516
497 req_cancel_subs (req); 517 req_cancel_subs (req);
498} 518}
499 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
500static int poll_cb () 660static int poll_cb (int max)
501{ 661{
502 dSP; 662 dSP;
503 int count = 0; 663 int count = 0;
504 int do_croak = 0; 664 int do_croak = 0;
505 aio_req req; 665 aio_req req;
506 666
507 for (;;) 667 for (;;)
508 { 668 {
509 LOCK (reslock); 669 while (max <= 0 || count < max)
510 req = reqq_shift (&res_queue);
511
512 if (req)
513 { 670 {
671 maybe_start_thread ();
672
673 LOCK (reslock);
674 req = reqq_shift (&res_queue);
675
514 if (!res_queue.size) 676 if (req)
515 { 677 {
678 --npending;
679
680 if (!res_queue.size)
681 {
516 /* read any signals sent by the worker threads */ 682 /* read any signals sent by the worker threads */
517 char buf [32]; 683 char buf [32];
518 while (read (respipe [0], buf, 32) == 32) 684 while (read (respipe [0], buf, 32) == 32)
685 ;
519 ; 686 }
520 } 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);
521 } 722 }
522 723
523 UNLOCK (reslock); 724 if (nreqs <= max_outstanding)
524
525 if (!req)
526 break; 725 break;
527 726
528 --nreqs; 727 poll_wait ();
529 728
530 if (req->type == REQ_QUIT) 729 max = 0;
531 started--;
532 else if (req->type == REQ_GROUP && req->length)
533 {
534 req->fd = 1; /* mark request as delayed */
535 continue;
536 }
537 else
538 {
539 if (req->type == REQ_READ)
540 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
541
542 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
543 SvREADONLY_off (req->data);
544
545 if (req->statdata)
546 {
547 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
548 PL_laststatval = req->result;
549 PL_statcache = *(req->statdata);
550 }
551
552 req_invoke (req);
553
554 count++;
555 }
556
557 req_free (req);
558 } 730 }
559 731
560 return count; 732 return count;
561}
562
563static void *aio_proc(void *arg);
564
565static void start_thread (void)
566{
567 sigset_t fullsigset, oldsigset;
568 pthread_attr_t attr;
569
570 worker *wrk = calloc (1, sizeof (worker));
571
572 if (!wrk)
573 croak ("unable to allocate worker thread data");
574
575 pthread_attr_init (&attr);
576 pthread_attr_setstacksize (&attr, STACKSIZE);
577 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
578
579 sigfillset (&fullsigset);
580
581 LOCK (wrklock);
582 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
583
584 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
585 {
586 wrk->prev = &wrk_first;
587 wrk->next = wrk_first.next;
588 wrk_first.next->prev = wrk;
589 wrk_first.next = wrk;
590 started++;
591 }
592 else
593 free (wrk);
594
595 sigprocmask (SIG_SETMASK, &oldsigset, 0);
596 UNLOCK (wrklock);
597}
598
599static void req_send (aio_req req)
600{
601 while (started < wanted && nreqs >= started)
602 start_thread ();
603
604 ++nreqs;
605
606 LOCK (reqlock);
607 reqq_push (&req_queue, req);
608 pthread_cond_signal (&reqwait);
609 UNLOCK (reqlock);
610}
611
612static void end_thread (void)
613{
614 aio_req req;
615
616 Newz (0, req, 1, aio_cb);
617
618 req->type = REQ_QUIT;
619 req->pri = PRI_MAX + PRI_BIAS;
620
621 req_send (req);
622}
623
624static void min_parallel (int nthreads)
625{
626 if (wanted < nthreads)
627 wanted = nthreads;
628}
629
630static void max_parallel (int nthreads)
631{
632 int cur = started;
633
634 if (wanted > nthreads)
635 wanted = nthreads;
636
637 while (cur > wanted)
638 {
639 end_thread ();
640 cur--;
641 }
642
643 while (started > wanted)
644 {
645 poll_wait ();
646 poll_cb ();
647 }
648} 733}
649 734
650static void create_pipe () 735static void create_pipe ()
651{ 736{
652 if (pipe (respipe)) 737 if (pipe (respipe))
… …
779 { 864 {
780 off_t sbytes; 865 off_t sbytes;
781 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0); 866 res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
782 867
783 if (res < 0 && sbytes) 868 if (res < 0 && sbytes)
784 /* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */ 869 /* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
785 res = sbytes; 870 res = sbytes;
786 } 871 }
787 872
788# elif __hpux 873# elif __hpux
789 res = sendfile (ofd, ifd, offset, count, 0, 0); 874 res = sendfile (ofd, ifd, offset, count, 0, 0);
… …
869 int errorno; 954 int errorno;
870 955
871 LOCK (wrklock); 956 LOCK (wrklock);
872 self->dirp = dirp = opendir (req->dataptr); 957 self->dirp = dirp = opendir (req->dataptr);
873 self->dbuf = u = malloc (sizeof (*u)); 958 self->dbuf = u = malloc (sizeof (*u));
959 req->data2ptr = names = malloc (memlen);
874 UNLOCK (wrklock); 960 UNLOCK (wrklock);
875
876 req->data2ptr = names = malloc (memlen);
877 961
878 if (dirp && u && names) 962 if (dirp && u && names)
879 for (;;) 963 for (;;)
880 { 964 {
881 errno = 0; 965 errno = 0;
… …
917/*****************************************************************************/ 1001/*****************************************************************************/
918 1002
919static void *aio_proc (void *thr_arg) 1003static void *aio_proc (void *thr_arg)
920{ 1004{
921 aio_req req; 1005 aio_req req;
922 int type;
923 worker *self = (worker *)thr_arg; 1006 worker *self = (worker *)thr_arg;
924 1007
925 do 1008 for (;;)
926 { 1009 {
927 LOCK (reqlock); 1010 LOCK (reqlock);
928 1011
929 for (;;) 1012 for (;;)
930 { 1013 {
… …
934 break; 1017 break;
935 1018
936 pthread_cond_wait (&reqwait, &reqlock); 1019 pthread_cond_wait (&reqwait, &reqlock);
937 } 1020 }
938 1021
1022 --nready;
1023
939 UNLOCK (reqlock); 1024 UNLOCK (reqlock);
940 1025
941 errno = 0; /* strictly unnecessary */ 1026 errno = 0; /* strictly unnecessary */
942 type = req->type; /* remember type for QUIT check */
943 1027
944 if (!(req->flags & FLAG_CANCELLED)) 1028 if (!(req->flags & FLAG_CANCELLED))
945 switch (type) 1029 switch (req->type)
946 { 1030 {
947 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;
948 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;
949 1033
950 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;
… …
976 req->result = select (0, 0, 0, 0, &tv); 1060 req->result = select (0, 0, 0, 0, &tv);
977 } 1061 }
978 1062
979 case REQ_GROUP: 1063 case REQ_GROUP:
980 case REQ_NOP: 1064 case REQ_NOP:
1065 break;
1066
981 case REQ_QUIT: 1067 case REQ_QUIT:
1068 LOCK (wrklock);
1069 worker_free (self);
1070 --started;
1071 UNLOCK (wrklock);
982 break; 1072 return 0;
983 1073
984 default: 1074 default:
985 req->result = ENOSYS; 1075 req->result = ENOSYS;
986 break; 1076 break;
987 } 1077 }
988 1078
989 req->errorno = errno; 1079 req->errorno = errno;
990 1080
991 LOCK (reslock); 1081 LOCK (reslock);
992 1082
1083 ++npending;
1084
993 if (!reqq_push (&res_queue, req)) 1085 if (!reqq_push (&res_queue, req))
994 /* write a dummy byte to the pipe so fh becomes ready */ 1086 /* write a dummy byte to the pipe so fh becomes ready */
995 write (respipe [1], &respipe, 1); 1087 write (respipe [1], &respipe, 1);
996 1088
997 self->req = 0; 1089 self->req = 0;
998 worker_clear (self); 1090 worker_clear (self);
999 1091
1000 UNLOCK (reslock); 1092 UNLOCK (reslock);
1001 } 1093 }
1002 while (type != REQ_QUIT);
1003
1004 LOCK (wrklock);
1005 worker_free (self);
1006 UNLOCK (wrklock);
1007
1008 return 0;
1009} 1094}
1010 1095
1011/*****************************************************************************/ 1096/*****************************************************************************/
1012 1097
1013static void atfork_prepare (void) 1098static void atfork_prepare (void)
… …
1099 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1184 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1100 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1185 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1101 1186
1102 create_pipe (); 1187 create_pipe ();
1103 pthread_atfork (atfork_prepare, atfork_parent, atfork_child); 1188 pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1104}
1105 1189
1190 start_thread ();
1191}
1192
1106void 1193void
1107min_parallel (nthreads) 1194min_parallel (int nthreads)
1108 int nthreads
1109 PROTOTYPE: $ 1195 PROTOTYPE: $
1110 1196
1111void 1197void
1112max_parallel (nthreads) 1198max_parallel (int nthreads)
1113 int nthreads
1114 PROTOTYPE: $ 1199 PROTOTYPE: $
1200
1201int
1202max_outstanding (int maxreqs)
1203 PROTOTYPE: $
1204 CODE:
1205 RETVAL = max_outstanding;
1206 max_outstanding = maxreqs;
1207 OUTPUT:
1208 RETVAL
1115 1209
1116void 1210void
1117aio_open (pathname,flags,mode,callback=&PL_sv_undef) 1211aio_open (pathname,flags,mode,callback=&PL_sv_undef)
1118 SV * pathname 1212 SV * pathname
1119 int flags 1213 int flags
… …
1374 req->type = REQ_NOP; 1468 req->type = REQ_NOP;
1375 1469
1376 REQ_SEND; 1470 REQ_SEND;
1377} 1471}
1378 1472
1379void 1473int
1380aioreq_pri (int pri = DEFAULT_PRI) 1474aioreq_pri (int pri = 0)
1381 CODE: 1475 PROTOTYPE: ;$
1476 CODE:
1477 RETVAL = next_pri - PRI_BIAS;
1478 if (items > 0)
1479 {
1382 if (pri < PRI_MIN) pri = PRI_MIN; 1480 if (pri < PRI_MIN) pri = PRI_MIN;
1383 if (pri > PRI_MAX) pri = PRI_MAX; 1481 if (pri > PRI_MAX) pri = PRI_MAX;
1384 next_pri = pri + PRI_BIAS; 1482 next_pri = pri + PRI_BIAS;
1483 }
1484 OUTPUT:
1485 RETVAL
1385 1486
1386void 1487void
1387aioreq_nice (int nice = 0) 1488aioreq_nice (int nice = 0)
1388 CODE: 1489 CODE:
1389 nice = next_pri - nice; 1490 nice = next_pri - nice;
1390 if (nice < PRI_MIN) nice = PRI_MIN; 1491 if (nice < PRI_MIN) nice = PRI_MIN;
1391 if (nice > PRI_MAX) nice = PRI_MAX; 1492 if (nice > PRI_MAX) nice = PRI_MAX;
1392 next_pri = nice + PRI_BIAS; 1493 next_pri = nice + PRI_BIAS;
1393 1494
1394void 1495void
1395flush () 1496flush ()
1396 PROTOTYPE: 1497 PROTOTYPE:
1397 CODE: 1498 CODE:
1398 while (nreqs) 1499 while (nreqs)
1399 { 1500 {
1400 poll_wait (); 1501 poll_wait ();
1401 poll_cb (); 1502 poll_cb (0);
1402 } 1503 }
1403 1504
1404void 1505void
1405poll() 1506poll()
1406 PROTOTYPE: 1507 PROTOTYPE:
1407 CODE: 1508 CODE:
1408 if (nreqs) 1509 if (nreqs)
1409 { 1510 {
1410 poll_wait (); 1511 poll_wait ();
1411 poll_cb (); 1512 poll_cb (0);
1412 } 1513 }
1413 1514
1414int 1515int
1415poll_fileno() 1516poll_fileno()
1416 PROTOTYPE: 1517 PROTOTYPE:
… …
1421 1522
1422int 1523int
1423poll_cb(...) 1524poll_cb(...)
1424 PROTOTYPE: 1525 PROTOTYPE:
1425 CODE: 1526 CODE:
1426 RETVAL = poll_cb (); 1527 RETVAL = poll_cb (0);
1528 OUTPUT:
1529 RETVAL
1530
1531int
1532poll_some(int max = 0)
1533 PROTOTYPE: $
1534 CODE:
1535 RETVAL = poll_cb (max);
1427 OUTPUT: 1536 OUTPUT:
1428 RETVAL 1537 RETVAL
1429 1538
1430void 1539void
1431poll_wait() 1540poll_wait()
… …
1440 CODE: 1549 CODE:
1441 RETVAL = nreqs; 1550 RETVAL = nreqs;
1442 OUTPUT: 1551 OUTPUT:
1443 RETVAL 1552 RETVAL
1444 1553
1554int
1555nready()
1556 PROTOTYPE:
1557 CODE:
1558 RETVAL = get_nready ();
1559 OUTPUT:
1560 RETVAL
1561
1562int
1563npending()
1564 PROTOTYPE:
1565 CODE:
1566 RETVAL = get_npending ();
1567 OUTPUT:
1568 RETVAL
1569
1445PROTOTYPES: DISABLE 1570PROTOTYPES: DISABLE
1446 1571
1447MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1572MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1448 1573
1449void 1574void
… …
1500void 1625void
1501result (aio_req grp, ...) 1626result (aio_req grp, ...)
1502 CODE: 1627 CODE:
1503{ 1628{
1504 int i; 1629 int i;
1630 AV *av;
1631
1632 grp->errorno = errno;
1633
1505 AV *av = newAV (); 1634 av = newAV ();
1506 1635
1507 for (i = 1; i < items; ++i ) 1636 for (i = 1; i < items; ++i )
1508 av_push (av, newSVsv (ST (i))); 1637 av_push (av, newSVsv (ST (i)));
1509 1638
1510 SvREFCNT_dec (grp->data); 1639 SvREFCNT_dec (grp->data);
1511 grp->data = (SV *)av; 1640 grp->data = (SV *)av;
1512} 1641}
1642
1643void
1644errno (aio_req grp, int errorno = errno)
1645 CODE:
1646 grp->errorno = errorno;
1513 1647
1514void 1648void
1515limit (aio_req grp, int limit) 1649limit (aio_req grp, int limit)
1516 CODE: 1650 CODE:
1517 grp->fd2 = limit; 1651 grp->fd2 = limit;

Diff Legend

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