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.44 by root, Sat Oct 21 23:20:29 2006 UTC vs.
Revision 1.61 by root, Mon Oct 23 22:44:21 2006 UTC

9 9
10#include <pthread.h> 10#include <pthread.h>
11 11
12#include <stddef.h> 12#include <stddef.h>
13#include <errno.h> 13#include <errno.h>
14#include <sys/time.h>
15#include <sys/select.h>
14#include <sys/types.h> 16#include <sys/types.h>
15#include <sys/stat.h> 17#include <sys/stat.h>
16#include <limits.h> 18#include <limits.h>
17#include <unistd.h> 19#include <unistd.h>
18#include <fcntl.h> 20#include <fcntl.h>
53 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 55 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
54 REQ_FSYNC, REQ_FDATASYNC, 56 REQ_FSYNC, REQ_FDATASYNC,
55 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 57 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
56 REQ_READDIR, 58 REQ_READDIR,
57 REQ_LINK, REQ_SYMLINK, 59 REQ_LINK, REQ_SYMLINK,
58 REQ_GROUP, 60 REQ_GROUP, REQ_NOP,
61 REQ_SLEEP,
59}; 62};
60 63
61#define AIO_REQ_KLASS "IO::AIO::REQ" 64#define AIO_REQ_KLASS "IO::AIO::REQ"
62#define AIO_GRP_KLASS "IO::AIO::GRP" 65#define AIO_GRP_KLASS "IO::AIO::GRP"
63 66
64typedef struct aio_cb 67typedef struct aio_cb
65{ 68{
66 struct aio_cb *grp, *grp_prev, *grp_next;
67
68 struct aio_cb *volatile next; 69 struct aio_cb *volatile next;
69
70 SV *self; /* the perl counterpart of this request, if any */
71 70
72 SV *data, *callback; 71 SV *data, *callback;
73 SV *fh, *fh2; 72 SV *fh, *fh2;
74 void *dataptr, *data2ptr; 73 void *dataptr, *data2ptr;
75 Stat_t *statdata; 74 Stat_t *statdata;
76 off_t offset; 75 off_t offset;
77 size_t length; 76 size_t length;
78 ssize_t result; 77 ssize_t result;
79 78
79 STRLEN dataoffset;
80 int type; 80 int type;
81 int fd, fd2; 81 int fd, fd2;
82 int errorno; 82 int errorno;
83 STRLEN dataoffset;
84 mode_t mode; /* open */ 83 mode_t mode; /* open */
84
85 unsigned char cancelled; 85 unsigned char flags;
86 unsigned char pri;
87
88 SV *self; /* the perl counterpart of this request, if any */
89 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
86} aio_cb; 90} aio_cb;
91
92enum {
93 FLAG_CANCELLED = 0x01,
94};
87 95
88typedef aio_cb *aio_req; 96typedef aio_cb *aio_req;
89typedef aio_cb *aio_req_ornot; 97typedef aio_cb *aio_req_ornot;
90typedef aio_cb *aio_group; 98
99enum {
100 PRI_MIN = -4,
101 PRI_MAX = 4,
102
103 DEFAULT_PRI = 0,
104 PRI_BIAS = -PRI_MIN,
105};
106
107static int next_pri = DEFAULT_PRI + PRI_BIAS;
91 108
92static int started, wanted; 109static int started, wanted;
93static volatile int nreqs; 110static volatile int nreqs;
94static int max_outstanding = 1<<30; 111static int max_outstanding = 1<<30;
95static int respipe [2]; 112static int respipe [2];
99static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 116static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
100 117
101static volatile aio_req reqs, reqe; /* queue start, queue end */ 118static volatile aio_req reqs, reqe; /* queue start, queue end */
102static volatile aio_req ress, rese; /* queue start, queue end */ 119static volatile aio_req ress, rese; /* queue start, queue end */
103 120
121static void req_invoke (aio_req req);
122static void req_free (aio_req req);
123
104/* must be called at most once */ 124/* must be called at most once */
105static SV *req_sv (aio_req req, const char *klass) 125static SV *req_sv (aio_req req, const char *klass)
106{ 126{
127 if (!req->self)
128 {
107 req->self = (SV *)newHV (); 129 req->self = (SV *)newHV ();
108 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0); 130 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
131 }
109 132
110 return sv_2mortal (sv_bless (newRV_noinc (req->self), gv_stashpv (klass, 1))); 133 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
111} 134}
112 135
113static aio_req SvAIO_REQ (SV *sv, const char *klass) 136static aio_req SvAIO_REQ (SV *sv)
114{ 137{
138 MAGIC *mg;
139
115 if (!sv_derived_from (sv, klass) || !SvROK (sv)) 140 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
116 croak ("object of class %s expected", klass); 141 croak ("object of class " AIO_REQ_KLASS " expected");
117 142
118 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 143 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
119 144
120 return mg ? (aio_req)mg->mg_ptr : 0; 145 return mg ? (aio_req)mg->mg_ptr : 0;
121} 146}
122 147
123static void req_free (aio_req req) 148static void aio_grp_feed (aio_req grp)
124{ 149{
125 if (req->self) 150 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
126 {
127 sv_unmagic (req->self, PERL_MAGIC_ext);
128 SvREFCNT_dec (req->self);
129 } 151 {
152 int old_len = grp->length;
130 153
131 if (req->data) 154 if (grp->fh2 && SvOK (grp->fh2))
132 SvREFCNT_dec (req->data); 155 {
156 dSP;
133 157
134 if (req->fh) 158 ENTER;
135 SvREFCNT_dec (req->fh); 159 SAVETMPS;
160 PUSHMARK (SP);
161 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
162 PUTBACK;
163 call_sv (grp->fh2, G_VOID | G_EVAL);
164 SPAGAIN;
165 FREETMPS;
166 LEAVE;
167 }
136 168
137 if (req->fh2) 169 /* stop if no progress has been made */
170 if (old_len == grp->length)
171 {
138 SvREFCNT_dec (req->fh2); 172 SvREFCNT_dec (grp->fh2);
139 173 grp->fh2 = 0;
140 if (req->statdata) 174 break;
141 Safefree (req->statdata); 175 }
142
143 if (req->callback)
144 SvREFCNT_dec (req->callback);
145
146 if (req->type == REQ_READDIR && req->result >= 0)
147 free (req->data2ptr);
148
149 Safefree (req);
150}
151
152static void
153poll_wait ()
154{
155 if (nreqs && !ress)
156 { 176 }
177}
178
179static void aio_grp_dec (aio_req grp)
180{
181 --grp->length;
182
183 /* call feeder, if applicable */
184 aio_grp_feed (grp);
185
186 /* finish, if done */
187 if (!grp->length && grp->fd)
188 {
189 req_invoke (grp);
190 req_free (grp);
191 }
192}
193
194static void poll_wait ()
195{
196 while (nreqs)
197 {
198 aio_req req;
199 pthread_mutex_lock (&reslock);
200 req = ress;
201 pthread_mutex_unlock (&reslock);
202
203 if (req)
204 return;
205
157 fd_set rfd; 206 fd_set rfd;
158 FD_ZERO(&rfd); 207 FD_ZERO(&rfd);
159 FD_SET(respipe [0], &rfd); 208 FD_SET(respipe [0], &rfd);
160 209
161 select (respipe [0] + 1, &rfd, 0, 0, 0); 210 select (respipe [0] + 1, &rfd, 0, 0, 0);
162 } 211 }
163} 212}
164 213
165static int 214static void req_invoke (aio_req req)
166poll_cb () 215{
216 dSP;
217 int errorno = errno;
218
219 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
220 return;
221
222 errno = req->errorno;
223
224 ENTER;
225 SAVETMPS;
226 PUSHMARK (SP);
227 EXTEND (SP, 1);
228
229 switch (req->type)
230 {
231 case REQ_READDIR:
232 {
233 SV *rv = &PL_sv_undef;
234
235 if (req->result >= 0)
236 {
237 char *buf = req->data2ptr;
238 AV *av = newAV ();
239
240 while (req->result)
241 {
242 SV *sv = newSVpv (buf, 0);
243
244 av_push (av, sv);
245 buf += SvCUR (sv) + 1;
246 req->result--;
247 }
248
249 rv = sv_2mortal (newRV_noinc ((SV *)av));
250 }
251
252 PUSHs (rv);
253 }
254 break;
255
256 case REQ_OPEN:
257 {
258 /* convert fd to fh */
259 SV *fh;
260
261 PUSHs (sv_2mortal (newSViv (req->result)));
262 PUTBACK;
263 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
264 SPAGAIN;
265
266 fh = SvREFCNT_inc (POPs);
267
268 PUSHMARK (SP);
269 XPUSHs (sv_2mortal (fh));
270 }
271 break;
272
273 case REQ_GROUP:
274 req->fd = 2; /* mark group as finished */
275
276 if (req->data)
277 {
278 int i;
279 AV *av = (AV *)req->data;
280
281 EXTEND (SP, AvFILL (av) + 1);
282 for (i = 0; i <= AvFILL (av); ++i)
283 PUSHs (*av_fetch (av, i, 0));
284 }
285 break;
286
287 case REQ_NOP:
288 case REQ_SLEEP:
289 break;
290
291 default:
292 PUSHs (sv_2mortal (newSViv (req->result)));
293 break;
294 }
295
296
297 PUTBACK;
298 call_sv (req->callback, G_VOID | G_EVAL);
299 SPAGAIN;
300
301 FREETMPS;
302 LEAVE;
303
304 errno = errorno;
305
306 if (SvTRUE (ERRSV))
307 {
308 req_free (req);
309 croak (0);
310 }
311}
312
313static void req_free (aio_req req)
314{
315 if (req->grp)
316 {
317 aio_req grp = req->grp;
318
319 /* unlink request */
320 if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
321 if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
322
323 if (grp->grp_first == req)
324 grp->grp_first = req->grp_next;
325
326 aio_grp_dec (grp);
327 }
328
329 if (req->self)
330 {
331 sv_unmagic (req->self, PERL_MAGIC_ext);
332 SvREFCNT_dec (req->self);
333 }
334
335 SvREFCNT_dec (req->data);
336 SvREFCNT_dec (req->fh);
337 SvREFCNT_dec (req->fh2);
338 SvREFCNT_dec (req->callback);
339 Safefree (req->statdata);
340
341 if (req->type == REQ_READDIR && req->result >= 0)
342 free (req->data2ptr);
343
344 Safefree (req);
345}
346
347static void req_cancel (aio_req req)
348{
349 req->flags |= FLAG_CANCELLED;
350
351 if (req->type == REQ_GROUP)
352 {
353 aio_req sub;
354
355 for (sub = req->grp_first; sub; sub = sub->grp_next)
356 req_cancel (sub);
357 }
358}
359
360static int poll_cb ()
167{ 361{
168 dSP; 362 dSP;
169 int count = 0; 363 int count = 0;
170 int do_croak = 0; 364 int do_croak = 0;
171 aio_req req; 365 aio_req req;
193 pthread_mutex_unlock (&reslock); 387 pthread_mutex_unlock (&reslock);
194 388
195 if (!req) 389 if (!req)
196 break; 390 break;
197 391
198 nreqs--; 392 --nreqs;
199 393
200 if (req->type == REQ_QUIT) 394 if (req->type == REQ_QUIT)
201 started--; 395 started--;
396 else if (req->type == REQ_GROUP && req->length)
397 {
398 req->fd = 1; /* mark request as delayed */
399 continue;
400 }
202 else 401 else
203 { 402 {
204 int errorno = errno;
205 errno = req->errorno;
206
207 if (req->type == REQ_READ) 403 if (req->type == REQ_READ)
208 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0)); 404 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
209 405
210 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE)) 406 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
211 SvREADONLY_off (req->data); 407 SvREADONLY_off (req->data);
215 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; 411 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
216 PL_laststatval = req->result; 412 PL_laststatval = req->result;
217 PL_statcache = *(req->statdata); 413 PL_statcache = *(req->statdata);
218 } 414 }
219 415
220 ENTER; 416 req_invoke (req);
221 PUSHMARK (SP);
222 417
223 if (req->type == REQ_READDIR)
224 {
225 SV *rv = &PL_sv_undef;
226
227 if (req->result >= 0)
228 {
229 char *buf = req->data2ptr;
230 AV *av = newAV ();
231
232 while (req->result)
233 {
234 SV *sv = newSVpv (buf, 0);
235
236 av_push (av, sv);
237 buf += SvCUR (sv) + 1;
238 req->result--;
239 }
240
241 rv = sv_2mortal (newRV_noinc ((SV *)av));
242 }
243
244 XPUSHs (rv);
245 }
246 else
247 {
248 XPUSHs (sv_2mortal (newSViv (req->result)));
249
250 if (req->type == REQ_OPEN)
251 {
252 /* convert fd to fh */
253 SV *fh;
254
255 PUTBACK;
256 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
257 SPAGAIN;
258
259 fh = SvREFCNT_inc (POPs);
260
261 PUSHMARK (SP);
262 XPUSHs (sv_2mortal (fh));
263 }
264 }
265
266 if (SvOK (req->callback) && !req->cancelled)
267 {
268 PUTBACK;
269 call_sv (req->callback, G_VOID | G_EVAL);
270 SPAGAIN;
271
272 if (SvTRUE (ERRSV))
273 {
274 req_free (req);
275 croak (0);
276 }
277 }
278
279 LEAVE;
280
281 errno = errorno;
282 count++; 418 count++;
283 } 419 }
284 420
285 req_free (req); 421 req_free (req);
286 } 422 }
288 return count; 424 return count;
289} 425}
290 426
291static void *aio_proc(void *arg); 427static void *aio_proc(void *arg);
292 428
293static void
294start_thread (void) 429static void start_thread (void)
295{ 430{
296 sigset_t fullsigset, oldsigset; 431 sigset_t fullsigset, oldsigset;
297 pthread_t tid; 432 pthread_t tid;
298 pthread_attr_t attr; 433 pthread_attr_t attr;
299 434
308 started++; 443 started++;
309 444
310 sigprocmask (SIG_SETMASK, &oldsigset, 0); 445 sigprocmask (SIG_SETMASK, &oldsigset, 0);
311} 446}
312 447
313static void
314req_send (aio_req req) 448static void req_send (aio_req req)
315{ 449{
316 while (started < wanted && nreqs >= started) 450 while (started < wanted && nreqs >= started)
317 start_thread (); 451 start_thread ();
318 452
319 nreqs++; 453 ++nreqs;
320 454
321 pthread_mutex_lock (&reqlock); 455 pthread_mutex_lock (&reqlock);
322 456
323 req->next = 0; 457 req->next = 0;
324 458
343 477
344 poll_wait (); 478 poll_wait ();
345 } 479 }
346} 480}
347 481
348static void 482static void end_thread (void)
349end_thread (void)
350{ 483{
351 aio_req req; 484 aio_req req;
352 Newz (0, req, 1, aio_cb); 485 Newz (0, req, 1, aio_cb);
353 req->type = REQ_QUIT; 486 req->type = REQ_QUIT;
354 487
405 * normal read/write by using a mutex. slows down execution a lot, 538 * normal read/write by using a mutex. slows down execution a lot,
406 * but that's your problem, not mine. 539 * but that's your problem, not mine.
407 */ 540 */
408static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER; 541static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER;
409 542
410static ssize_t 543static ssize_t pread (int fd, void *buf, size_t count, off_t offset)
411pread (int fd, void *buf, size_t count, off_t offset)
412{ 544{
413 ssize_t res; 545 ssize_t res;
414 off_t ooffset; 546 off_t ooffset;
415 547
416 pthread_mutex_lock (&preadwritelock); 548 pthread_mutex_lock (&preadwritelock);
421 pthread_mutex_unlock (&preadwritelock); 553 pthread_mutex_unlock (&preadwritelock);
422 554
423 return res; 555 return res;
424} 556}
425 557
426static ssize_t
427pwrite (int fd, void *buf, size_t count, off_t offset) 558static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset)
428{ 559{
429 ssize_t res; 560 ssize_t res;
430 off_t ooffset; 561 off_t ooffset;
431 562
432 pthread_mutex_lock (&preadwritelock); 563 pthread_mutex_lock (&preadwritelock);
445#endif 576#endif
446 577
447#if !HAVE_READAHEAD 578#if !HAVE_READAHEAD
448# define readahead aio_readahead 579# define readahead aio_readahead
449 580
450static ssize_t
451readahead (int fd, off_t offset, size_t count) 581static ssize_t readahead (int fd, off_t offset, size_t count)
452{ 582{
453 char readahead_buf[4096]; 583 char readahead_buf[4096];
454 584
455 while (count > 0) 585 while (count > 0)
456 { 586 {
468#if !HAVE_READDIR_R 598#if !HAVE_READDIR_R
469# define readdir_r aio_readdir_r 599# define readdir_r aio_readdir_r
470 600
471static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER; 601static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;
472 602
473static int
474readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res) 603static int readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res)
475{ 604{
476 struct dirent *e; 605 struct dirent *e;
477 int errorno; 606 int errorno;
478 607
479 pthread_mutex_lock (&readdirlock); 608 pthread_mutex_lock (&readdirlock);
495 return e ? 0 : -1; 624 return e ? 0 : -1;
496} 625}
497#endif 626#endif
498 627
499/* sendfile always needs emulation */ 628/* sendfile always needs emulation */
500static ssize_t
501sendfile_ (int ofd, int ifd, off_t offset, size_t count) 629static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count)
502{ 630{
503 ssize_t res; 631 ssize_t res;
504 632
505 if (!count) 633 if (!count)
506 return 0; 634 return 0;
588 716
589 return res; 717 return res;
590} 718}
591 719
592/* read a full directory */ 720/* read a full directory */
593static int
594scandir_ (const char *path, void **namesp) 721static int scandir_ (const char *path, void **namesp)
595{ 722{
596 DIR *dirp = opendir (path); 723 DIR *dirp = opendir (path);
597 union 724 union
598 { 725 {
599 struct dirent d; 726 struct dirent d;
653 return res; 780 return res;
654} 781}
655 782
656/*****************************************************************************/ 783/*****************************************************************************/
657 784
658static void *
659aio_proc (void *thr_arg) 785static void *aio_proc (void *thr_arg)
660{ 786{
661 aio_req req; 787 aio_req req;
662 int type; 788 int type;
663 789
664 do 790 do
682 } 808 }
683 809
684 pthread_mutex_unlock (&reqlock); 810 pthread_mutex_unlock (&reqlock);
685 811
686 errno = 0; /* strictly unnecessary */ 812 errno = 0; /* strictly unnecessary */
813 type = req->type; /* remember type for QUIT check */
687 814
688 if (!req->cancelled) 815 if (!(req->flags & FLAG_CANCELLED))
689 switch (req->type) 816 switch (type)
690 { 817 {
691 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 818 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
692 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 819 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
693 820
694 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 821 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
708 835
709 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 836 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
710 case REQ_FSYNC: req->result = fsync (req->fd); break; 837 case REQ_FSYNC: req->result = fsync (req->fd); break;
711 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break; 838 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;
712 839
840 case REQ_SLEEP:
841 {
842 struct timeval tv;
843
844 tv.tv_sec = req->fd;
845 tv.tv_usec = req->fd2;
846
847 req->result = select (0, 0, 0, 0, &tv);
848 }
849
850 case REQ_GROUP:
851 case REQ_NOP:
713 case REQ_QUIT: 852 case REQ_QUIT:
714 break; 853 break;
715 854
716 default: 855 default:
717 req->result = ENOSYS; 856 req->result = ENOSYS;
801 atfork_parent (); 940 atfork_parent ();
802} 941}
803 942
804#define dREQ \ 943#define dREQ \
805 aio_req req; \ 944 aio_req req; \
945 int req_pri = next_pri; \
946 next_pri = DEFAULT_PRI + PRI_BIAS; \
806 \ 947 \
807 if (SvOK (callback) && !SvROK (callback)) \ 948 if (SvOK (callback) && !SvROK (callback)) \
808 croak ("callback must be undef or of reference type"); \ 949 croak ("callback must be undef or of reference type"); \
809 \ 950 \
810 Newz (0, req, 1, aio_cb); \ 951 Newz (0, req, 1, aio_cb); \
811 if (!req) \ 952 if (!req) \
812 croak ("out of memory during aio_req allocation"); \ 953 croak ("out of memory during aio_req allocation"); \
813 \ 954 \
814 req->callback = newSVsv (callback) 955 req->callback = newSVsv (callback); \
956 req->pri = req_pri
815 957
816#define REQ_SEND \ 958#define REQ_SEND \
817 req_send (req); \ 959 req_send (req); \
818 \ 960 \
819 if (GIMME_V != G_VOID) \ 961 if (GIMME_V != G_VOID) \
1073 1215
1074 REQ_SEND; 1216 REQ_SEND;
1075} 1217}
1076 1218
1077void 1219void
1220aio_sleep (delay,callback=&PL_sv_undef)
1221 double delay
1222 SV * callback
1223 PPCODE:
1224{
1225 dREQ;
1226
1227 req->type = REQ_SLEEP;
1228 req->fd = delay < 0. ? 0 : delay;
1229 req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);
1230
1231 REQ_SEND;
1232}
1233
1234void
1078aio_group (callback=&PL_sv_undef) 1235aio_group (callback=&PL_sv_undef)
1079 SV * callback 1236 SV * callback
1080 PROTOTYPE: ;& 1237 PROTOTYPE: ;$
1081 PPCODE: 1238 PPCODE:
1082{ 1239{
1083 dREQ; 1240 dREQ;
1241
1084 req->type = REQ_GROUP; 1242 req->type = REQ_GROUP;
1243 req_send (req);
1244
1085 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1245 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1086} 1246}
1247
1248void
1249aio_nop (callback=&PL_sv_undef)
1250 SV * callback
1251 PPCODE:
1252{
1253 dREQ;
1254
1255 req->type = REQ_NOP;
1256
1257 REQ_SEND;
1258}
1259
1260#if 0
1261
1262void
1263aio_pri (int pri = DEFAULT_PRI)
1264 CODE:
1265 if (pri < PRI_MIN) pri = PRI_MIN;
1266 if (pri > PRI_MAX) pri = PRI_MAX;
1267 next_pri = pri + PRI_BIAS;
1268
1269#endif
1087 1270
1088void 1271void
1089flush () 1272flush ()
1090 PROTOTYPE: 1273 PROTOTYPE:
1091 CODE: 1274 CODE:
1134 CODE: 1317 CODE:
1135 RETVAL = nreqs; 1318 RETVAL = nreqs;
1136 OUTPUT: 1319 OUTPUT:
1137 RETVAL 1320 RETVAL
1138 1321
1322PROTOTYPES: DISABLE
1323
1139MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1324MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1140 1325
1141void 1326void
1142cancel (aio_req_ornot req) 1327cancel (aio_req_ornot req)
1143 PROTOTYPE: 1328 PROTOTYPE:
1144 CODE: 1329 CODE:
1145 req->cancelled = 1; 1330 req_cancel (req);
1146 1331
1332void
1333cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1334 CODE:
1335 SvREFCNT_dec (req->callback);
1336 req->callback = newSVsv (callback);
1337
1338MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1339
1340void
1341add (aio_req grp, ...)
1342 PPCODE:
1343{
1344 int i;
1345 aio_req req;
1346
1347 if (grp->fd == 2)
1348 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1349
1350 for (i = 1; i < items; ++i )
1351 {
1352 if (GIMME_V != G_VOID)
1353 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1354
1355 req = SvAIO_REQ (ST (i));
1356
1357 if (req)
1358 {
1359 ++grp->length;
1360 req->grp = grp;
1361
1362 req->grp_prev = 0;
1363 req->grp_next = grp->grp_first;
1364
1365 if (grp->grp_first)
1366 grp->grp_first->grp_prev = req;
1367
1368 grp->grp_first = req;
1369 }
1370 }
1371}
1372
1373void
1374result (aio_req grp, ...)
1375 CODE:
1376{
1377 int i;
1378 AV *av = newAV ();
1379
1380 for (i = 1; i < items; ++i )
1381 av_push (av, newSVsv (ST (i)));
1382
1383 SvREFCNT_dec (grp->data);
1384 grp->data = (SV *)av;
1385}
1386
1387void
1388feed_limit (aio_req grp, int limit)
1389 CODE:
1390 grp->fd2 = limit;
1391 aio_grp_feed (grp);
1392
1393void
1394feed (aio_req grp, SV *callback=&PL_sv_undef)
1395 CODE:
1396{
1397 SvREFCNT_dec (grp->fh2);
1398 grp->fh2 = newSVsv (callback);
1399
1400 if (grp->fd2 <= 0)
1401 grp->fd2 = 2;
1402
1403 aio_grp_feed (grp);
1404}
1405

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines