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.77 by root, Thu Oct 26 13:25:40 2006 UTC vs.
Revision 1.81 by root, Fri Oct 27 20:10:06 2006 UTC

… …
61# define STACKSIZE PTHREAD_STACK_MIN 61# define STACKSIZE PTHREAD_STACK_MIN
62#else 62#else
63# define STACKSIZE 16384 63# define STACKSIZE 16384
64#endif 64#endif
65 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
66/* buffer size for various temporary buffers */ 78/* buffer size for various temporary buffers */
67#define AIO_BUFSIZE 65536 79#define AIO_BUFSIZE 65536
68 80
69#define dBUF \ 81#define dBUF \
70 char *aio_buf; \ 82 char *aio_buf; \
… …
80 REQ_READ, REQ_WRITE, REQ_READAHEAD, 92 REQ_READ, REQ_WRITE, REQ_READAHEAD,
81 REQ_SENDFILE, 93 REQ_SENDFILE,
82 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 94 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
83 REQ_FSYNC, REQ_FDATASYNC, 95 REQ_FSYNC, REQ_FDATASYNC,
84 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 96 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
85 REQ_READDIR, 97 REQ_MKNOD, REQ_READDIR,
86 REQ_LINK, REQ_SYMLINK, 98 REQ_LINK, REQ_SYMLINK,
87 REQ_GROUP, REQ_NOP, 99 REQ_GROUP, REQ_NOP,
88 REQ_BUSY, 100 REQ_BUSY,
89}; 101};
90 102
… …
132 NUM_PRI = PRI_MAX + PRI_BIAS + 1, 144 NUM_PRI = PRI_MAX + PRI_BIAS + 1,
133}; 145};
134 146
135static int next_pri = DEFAULT_PRI + PRI_BIAS; 147static int next_pri = DEFAULT_PRI + PRI_BIAS;
136 148
137static int started, wanted; 149static unsigned int started, wanted;
138static volatile int nreqs;
139static int respipe [2];
140 150
141#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP) 151#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
142# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP 152# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
143#else 153#else
144# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER 154# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
… …
185 wrk->prev->next = wrk->next; 195 wrk->prev->next = wrk->next;
186 196
187 free (wrk); 197 free (wrk);
188} 198}
189 199
200static volatile unsigned int nreqs, nready, npending;
201static volatile unsigned int max_outstanding = 0xffffffff;
202static int respipe [2];
203
190static pthread_mutex_t reslock = AIO_MUTEX_INIT; 204static pthread_mutex_t reslock = AIO_MUTEX_INIT;
191static pthread_mutex_t reqlock = AIO_MUTEX_INIT; 205static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
192static 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
193 238
194/* 239/*
195 * a somewhat faster data structure might be nice, but 240 * a somewhat faster data structure might be nice, but
196 * with 8 priorities this actually needs <20 insns 241 * with 8 priorities this actually needs <20 insns
197 * per shift, the most expensive operation. 242 * per shift, the most expensive operation.
… …
318 req_invoke (grp); 363 req_invoke (grp);
319 req_free (grp); 364 req_free (grp);
320 } 365 }
321} 366}
322 367
323static void poll_wait ()
324{
325 fd_set rfd;
326
327 while (nreqs)
328 {
329 int size;
330#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
331 LOCK (reslock);
332#endif
333 size = res_queue.size;
334#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
335 UNLOCK (reslock);
336#endif
337
338 if (size)
339 return;
340
341 FD_ZERO(&rfd);
342 FD_SET(respipe [0], &rfd);
343
344 select (respipe [0] + 1, &rfd, 0, 0, 0);
345 }
346}
347
348static void req_invoke (aio_req req) 368static void req_invoke (aio_req req)
349{ 369{
350 dSP; 370 dSP;
351 371
352 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback)) 372 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
353 { 373 {
354 errno = req->errorno;
355
356 ENTER; 374 ENTER;
357 SAVETMPS; 375 SAVETMPS;
358 PUSHMARK (SP); 376 PUSHMARK (SP);
359 EXTEND (SP, 1); 377 EXTEND (SP, 1);
360 378
… …
425 default: 443 default:
426 PUSHs (sv_2mortal (newSViv (req->result))); 444 PUSHs (sv_2mortal (newSViv (req->result)));
427 break; 445 break;
428 } 446 }
429 447
448 errno = req->errorno;
430 449
431 PUTBACK; 450 PUTBACK;
432 call_sv (req->callback, G_VOID | G_EVAL); 451 call_sv (req->callback, G_VOID | G_EVAL);
433 SPAGAIN; 452 SPAGAIN;
434 453
… …
496 req->flags |= FLAG_CANCELLED; 515 req->flags |= FLAG_CANCELLED;
497 516
498 req_cancel_subs (req); 517 req_cancel_subs (req);
499} 518}
500 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
501static int poll_cb (int max) 660static int poll_cb (int max)
502{ 661{
503 dSP; 662 dSP;
504 int count = 0; 663 int count = 0;
505 int do_croak = 0; 664 int do_croak = 0;
506 aio_req req; 665 aio_req req;
507 666
667 for (;;)
668 {
508 while (max <= 0 || count < max) 669 while (max <= 0 || count < max)
509 {
510 LOCK (reslock);
511 req = reqq_shift (&res_queue);
512
513 if (req)
514 { 670 {
671 maybe_start_thread ();
672
673 LOCK (reslock);
674 req = reqq_shift (&res_queue);
675
515 if (!res_queue.size) 676 if (req)
516 { 677 {
678 --npending;
679
680 if (!res_queue.size)
681 {
517 /* read any signals sent by the worker threads */ 682 /* read any signals sent by the worker threads */
518 char buf [32]; 683 char buf [32];
519 while (read (respipe [0], buf, 32) == 32) 684 while (read (respipe [0], buf, 32) == 32)
685 ;
520 ; 686 }
521 } 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);
522 } 722 }
523 723
524 UNLOCK (reslock); 724 if (nreqs <= max_outstanding)
525
526 if (!req)
527 break; 725 break;
528 726
529 --nreqs; 727 poll_wait ();
530 728
531 if (req->type == REQ_QUIT) 729 max = 0;
532 started--;
533 else if (req->type == REQ_GROUP && req->length)
534 {
535 req->fd = 1; /* mark request as delayed */
536 continue;
537 }
538 else
539 {
540 if (req->type == REQ_READ)
541 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
542
543 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
544 SvREADONLY_off (req->data);
545
546 if (req->statdata)
547 {
548 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
549 PL_laststatval = req->result;
550 PL_statcache = *(req->statdata);
551 }
552
553 req_invoke (req);
554
555 count++;
556 }
557
558 req_free (req);
559 } 730 }
560 731
561 return count; 732 return count;
562}
563
564static void *aio_proc(void *arg);
565
566static void start_thread (void)
567{
568 sigset_t fullsigset, oldsigset;
569 pthread_attr_t attr;
570
571 worker *wrk = calloc (1, sizeof (worker));
572
573 if (!wrk)
574 croak ("unable to allocate worker thread data");
575
576 pthread_attr_init (&attr);
577 pthread_attr_setstacksize (&attr, STACKSIZE);
578 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
579
580 sigfillset (&fullsigset);
581
582 LOCK (wrklock);
583 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
584
585 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
586 {
587 wrk->prev = &wrk_first;
588 wrk->next = wrk_first.next;
589 wrk_first.next->prev = wrk;
590 wrk_first.next = wrk;
591 started++;
592 }
593 else
594 free (wrk);
595
596 sigprocmask (SIG_SETMASK, &oldsigset, 0);
597 UNLOCK (wrklock);
598}
599
600static void req_send (aio_req req)
601{
602 while (started < wanted && nreqs >= started)
603 start_thread ();
604
605 ++nreqs;
606
607 LOCK (reqlock);
608 reqq_push (&req_queue, req);
609 pthread_cond_signal (&reqwait);
610 UNLOCK (reqlock);
611}
612
613static void end_thread (void)
614{
615 aio_req req;
616
617 Newz (0, req, 1, aio_cb);
618
619 req->type = REQ_QUIT;
620 req->pri = PRI_MAX + PRI_BIAS;
621
622 req_send (req);
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 int cur = started;
634
635 if (wanted > nthreads)
636 wanted = nthreads;
637
638 while (cur > wanted)
639 {
640 end_thread ();
641 cur--;
642 }
643
644 while (started > wanted)
645 {
646 poll_wait ();
647 poll_cb (0);
648 }
649} 733}
650 734
651static void create_pipe () 735static void create_pipe ()
652{ 736{
653 if (pipe (respipe)) 737 if (pipe (respipe))
… …
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;
… …
959 case REQ_UNLINK: req->result = unlink (req->dataptr); break; 1043 case REQ_UNLINK: req->result = unlink (req->dataptr); break;
960 case REQ_RMDIR: req->result = rmdir (req->dataptr); break; 1044 case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
961 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break; 1045 case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
962 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break; 1046 case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
963 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;
964 1049
965 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 1050 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
966 case REQ_FSYNC: req->result = fsync (req->fd); break; 1051 case REQ_FSYNC: req->result = fsync (req->fd); break;
967 case REQ_READDIR: scandir_ (req, self); break; 1052 case REQ_READDIR: scandir_ (req, self); break;
968 1053
… …
976 req->result = select (0, 0, 0, 0, &tv); 1061 req->result = select (0, 0, 0, 0, &tv);
977 } 1062 }
978 1063
979 case REQ_GROUP: 1064 case REQ_GROUP:
980 case REQ_NOP: 1065 case REQ_NOP:
1066 break;
1067
981 case REQ_QUIT: 1068 case REQ_QUIT:
1069 LOCK (wrklock);
1070 worker_free (self);
1071 --started;
1072 UNLOCK (wrklock);
982 break; 1073 return 0;
983 1074
984 default: 1075 default:
985 req->result = ENOSYS; 1076 req->result = ENOSYS;
986 break; 1077 break;
987 } 1078 }
988 1079
989 req->errorno = errno; 1080 req->errorno = errno;
990 1081
991 LOCK (reslock); 1082 LOCK (reslock);
992 1083
1084 ++npending;
1085
993 if (!reqq_push (&res_queue, req)) 1086 if (!reqq_push (&res_queue, req))
994 /* write a dummy byte to the pipe so fh becomes ready */ 1087 /* write a dummy byte to the pipe so fh becomes ready */
995 write (respipe [1], &respipe, 1); 1088 write (respipe [1], &respipe, 1);
996 1089
997 self->req = 0; 1090 self->req = 0;
998 worker_clear (self); 1091 worker_clear (self);
999 1092
1000 UNLOCK (reslock); 1093 UNLOCK (reslock);
1001 } 1094 }
1002 while (type != REQ_QUIT);
1003
1004 LOCK (wrklock);
1005 worker_free (self);
1006 UNLOCK (wrklock);
1007
1008 return 0;
1009} 1095}
1010 1096
1011/*****************************************************************************/ 1097/*****************************************************************************/
1012 1098
1013static void atfork_prepare (void) 1099static void atfork_prepare (void)
… …
1096{ 1182{
1097 HV *stash = gv_stashpv ("IO::AIO", 1); 1183 HV *stash = gv_stashpv ("IO::AIO", 1);
1098 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); 1184 newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1099 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1185 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1100 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1186 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1187 newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
1188 newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
1101 1189
1102 create_pipe (); 1190 create_pipe ();
1103 pthread_atfork (atfork_prepare, atfork_parent, atfork_child); 1191 pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1104}
1105 1192
1193 start_thread ();
1194}
1195
1106void 1196void
1107min_parallel (nthreads) 1197min_parallel (int nthreads)
1108 int nthreads
1109 PROTOTYPE: $ 1198 PROTOTYPE: $
1110 1199
1111void 1200void
1112max_parallel (nthreads) 1201max_parallel (int nthreads)
1113 int nthreads
1114 PROTOTYPE: $ 1202 PROTOTYPE: $
1203
1204int
1205max_outstanding (int maxreqs)
1206 PROTOTYPE: $
1207 CODE:
1208 RETVAL = max_outstanding;
1209 max_outstanding = maxreqs;
1210 OUTPUT:
1211 RETVAL
1115 1212
1116void 1213void
1117aio_open (pathname,flags,mode,callback=&PL_sv_undef) 1214aio_open (pathname,flags,mode,callback=&PL_sv_undef)
1118 SV * pathname 1215 SV * pathname
1119 int flags 1216 int flags
… …
1334 1431
1335 REQ_SEND; 1432 REQ_SEND;
1336} 1433}
1337 1434
1338void 1435void
1436aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
1437 SV * pathname
1438 SV * callback
1439 UV mode
1440 UV dev
1441 PPCODE:
1442{
1443 dREQ;
1444
1445 req->type = REQ_MKNOD;
1446 req->data = newSVsv (pathname);
1447 req->dataptr = SvPVbyte_nolen (req->data);
1448 req->mode = (mode_t)mode;
1449 req->offset = dev;
1450
1451 REQ_SEND;
1452}
1453
1454void
1339aio_busy (delay,callback=&PL_sv_undef) 1455aio_busy (delay,callback=&PL_sv_undef)
1340 double delay 1456 double delay
1341 SV * callback 1457 SV * callback
1342 PPCODE: 1458 PPCODE:
1343{ 1459{
… …
1374 req->type = REQ_NOP; 1490 req->type = REQ_NOP;
1375 1491
1376 REQ_SEND; 1492 REQ_SEND;
1377} 1493}
1378 1494
1379void 1495int
1380aioreq_pri (int pri = DEFAULT_PRI) 1496aioreq_pri (int pri = 0)
1381 CODE: 1497 PROTOTYPE: ;$
1498 CODE:
1499 RETVAL = next_pri - PRI_BIAS;
1500 if (items > 0)
1501 {
1382 if (pri < PRI_MIN) pri = PRI_MIN; 1502 if (pri < PRI_MIN) pri = PRI_MIN;
1383 if (pri > PRI_MAX) pri = PRI_MAX; 1503 if (pri > PRI_MAX) pri = PRI_MAX;
1384 next_pri = pri + PRI_BIAS; 1504 next_pri = pri + PRI_BIAS;
1505 }
1506 OUTPUT:
1507 RETVAL
1385 1508
1386void 1509void
1387aioreq_nice (int nice = 0) 1510aioreq_nice (int nice = 0)
1388 CODE: 1511 CODE:
1389 nice = next_pri - nice; 1512 nice = next_pri - nice;
1390 if (nice < PRI_MIN) nice = PRI_MIN; 1513 if (nice < PRI_MIN) nice = PRI_MIN;
1391 if (nice > PRI_MAX) nice = PRI_MAX; 1514 if (nice > PRI_MAX) nice = PRI_MAX;
1392 next_pri = nice + PRI_BIAS; 1515 next_pri = nice + PRI_BIAS;
1393 1516
1394void 1517void
1395flush () 1518flush ()
1396 PROTOTYPE: 1519 PROTOTYPE:
1397 CODE: 1520 CODE:
… …
1448 CODE: 1571 CODE:
1449 RETVAL = nreqs; 1572 RETVAL = nreqs;
1450 OUTPUT: 1573 OUTPUT:
1451 RETVAL 1574 RETVAL
1452 1575
1576int
1577nready()
1578 PROTOTYPE:
1579 CODE:
1580 RETVAL = get_nready ();
1581 OUTPUT:
1582 RETVAL
1583
1584int
1585npending()
1586 PROTOTYPE:
1587 CODE:
1588 RETVAL = get_npending ();
1589 OUTPUT:
1590 RETVAL
1591
1453PROTOTYPES: DISABLE 1592PROTOTYPES: DISABLE
1454 1593
1455MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1594MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1456 1595
1457void 1596void
… …
1508void 1647void
1509result (aio_req grp, ...) 1648result (aio_req grp, ...)
1510 CODE: 1649 CODE:
1511{ 1650{
1512 int i; 1651 int i;
1652 AV *av;
1653
1654 grp->errorno = errno;
1655
1513 AV *av = newAV (); 1656 av = newAV ();
1514 1657
1515 for (i = 1; i < items; ++i ) 1658 for (i = 1; i < items; ++i )
1516 av_push (av, newSVsv (ST (i))); 1659 av_push (av, newSVsv (ST (i)));
1517 1660
1518 SvREFCNT_dec (grp->data); 1661 SvREFCNT_dec (grp->data);
1519 grp->data = (SV *)av; 1662 grp->data = (SV *)av;
1520} 1663}
1664
1665void
1666errno (aio_req grp, int errorno = errno)
1667 CODE:
1668 grp->errorno = errorno;
1521 1669
1522void 1670void
1523limit (aio_req grp, int limit) 1671limit (aio_req grp, int limit)
1524 CODE: 1672 CODE:
1525 grp->fd2 = limit; 1673 grp->fd2 = limit;

Diff Legend

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