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.57 by root, Mon Oct 23 18:38:15 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;
70
71 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
69 72
70 SV *self; /* the perl counterpart of this request, if any */ 73 SV *self; /* the perl counterpart of this request, if any */
71 74
72 SV *data, *callback; 75 SV *data, *callback;
73 SV *fh, *fh2; 76 SV *fh, *fh2;
85 unsigned char cancelled; 88 unsigned char cancelled;
86} aio_cb; 89} aio_cb;
87 90
88typedef aio_cb *aio_req; 91typedef aio_cb *aio_req;
89typedef aio_cb *aio_req_ornot; 92typedef aio_cb *aio_req_ornot;
90typedef aio_cb *aio_group;
91 93
92static int started, wanted; 94static int started, wanted;
93static volatile int nreqs; 95static volatile int nreqs;
94static int max_outstanding = 1<<30; 96static int max_outstanding = 1<<30;
95static int respipe [2]; 97static int respipe [2];
99static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
100 102
101static volatile aio_req reqs, reqe; /* queue start, queue end */ 103static volatile aio_req reqs, reqe; /* queue start, queue end */
102static volatile aio_req ress, rese; /* queue start, queue end */ 104static volatile aio_req ress, rese; /* queue start, queue end */
103 105
106static void req_invoke (aio_req req);
107static void req_free (aio_req req);
108
104/* must be called at most once */ 109/* must be called at most once */
105static SV *req_sv (aio_req req, const char *klass) 110static SV *req_sv (aio_req req, const char *klass)
106{ 111{
112 if (!req->self)
113 {
107 req->self = (SV *)newHV (); 114 req->self = (SV *)newHV ();
108 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0); 115 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
116 }
109 117
110 return sv_2mortal (sv_bless (newRV_noinc (req->self), gv_stashpv (klass, 1))); 118 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
111} 119}
112 120
113static aio_req SvAIO_REQ (SV *sv, const char *klass) 121static aio_req SvAIO_REQ (SV *sv)
114{ 122{
123 MAGIC *mg;
124
115 if (!sv_derived_from (sv, klass) || !SvROK (sv)) 125 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
116 croak ("object of class %s expected", klass); 126 croak ("object of class " AIO_REQ_KLASS " expected");
117 127
118 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 128 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
119 129
120 return mg ? (aio_req)mg->mg_ptr : 0; 130 return mg ? (aio_req)mg->mg_ptr : 0;
121} 131}
122 132
123static void req_free (aio_req req) 133static void aio_grp_feed (aio_req grp)
124{ 134{
125 if (req->self) 135 while (grp->length < grp->fd2 && !grp->cancelled)
126 {
127 sv_unmagic (req->self, PERL_MAGIC_ext);
128 SvREFCNT_dec (req->self);
129 } 136 {
137 int old_len = grp->length;
130 138
131 if (req->data) 139 if (grp->fh2 && SvOK (grp->fh2))
132 SvREFCNT_dec (req->data); 140 {
141 dSP;
133 142
134 if (req->fh) 143 ENTER;
135 SvREFCNT_dec (req->fh); 144 SAVETMPS;
145 PUSHMARK (SP);
146 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
147 PUTBACK;
148 call_sv (grp->fh2, G_VOID | G_EVAL);
149 SPAGAIN;
150 FREETMPS;
151 LEAVE;
152 }
136 153
137 if (req->fh2) 154 /* stop if no progress has been made */
155 if (old_len == grp->length)
156 {
138 SvREFCNT_dec (req->fh2); 157 SvREFCNT_dec (grp->fh2);
139 158 grp->fh2 = 0;
140 if (req->statdata) 159 break;
141 Safefree (req->statdata); 160 }
142 161 }
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} 162}
151 163
152static void 164static void aio_grp_dec (aio_req grp)
153poll_wait () 165{
166 --grp->length;
167
168 /* call feeder, if applicable */
169 aio_grp_feed (grp);
170
171 /* finish, if done */
172 if (!grp->length && grp->fd)
173 {
174 req_invoke (grp);
175 req_free (grp);
176 }
177}
178
179static void poll_wait ()
154{ 180{
155 if (nreqs && !ress) 181 if (nreqs && !ress)
156 { 182 {
157 fd_set rfd; 183 fd_set rfd;
158 FD_ZERO(&rfd); 184 FD_ZERO(&rfd);
160 186
161 select (respipe [0] + 1, &rfd, 0, 0, 0); 187 select (respipe [0] + 1, &rfd, 0, 0, 0);
162 } 188 }
163} 189}
164 190
165static int 191static void req_invoke (aio_req req)
166poll_cb () 192{
193 dSP;
194 int errorno = errno;
195
196 if (req->cancelled || !SvOK (req->callback))
197 return;
198
199 errno = req->errorno;
200
201 ENTER;
202 SAVETMPS;
203 PUSHMARK (SP);
204 EXTEND (SP, 1);
205
206 switch (req->type)
207 {
208 case REQ_READDIR:
209 {
210 SV *rv = &PL_sv_undef;
211
212 if (req->result >= 0)
213 {
214 char *buf = req->data2ptr;
215 AV *av = newAV ();
216
217 while (req->result)
218 {
219 SV *sv = newSVpv (buf, 0);
220
221 av_push (av, sv);
222 buf += SvCUR (sv) + 1;
223 req->result--;
224 }
225
226 rv = sv_2mortal (newRV_noinc ((SV *)av));
227 }
228
229 PUSHs (rv);
230 }
231 break;
232
233 case REQ_OPEN:
234 {
235 /* convert fd to fh */
236 SV *fh;
237
238 PUSHs (sv_2mortal (newSViv (req->result)));
239 PUTBACK;
240 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
241 SPAGAIN;
242
243 fh = SvREFCNT_inc (POPs);
244
245 PUSHMARK (SP);
246 XPUSHs (sv_2mortal (fh));
247 }
248 break;
249
250 case REQ_GROUP:
251 req->fd = 2; /* mark group as finished */
252
253 if (req->data)
254 {
255 int i;
256 AV *av = (AV *)req->data;
257
258 EXTEND (SP, AvFILL (av) + 1);
259 for (i = 0; i <= AvFILL (av); ++i)
260 PUSHs (*av_fetch (av, i, 0));
261 }
262 break;
263
264 case REQ_NOP:
265 case REQ_SLEEP:
266 break;
267
268 default:
269 PUSHs (sv_2mortal (newSViv (req->result)));
270 break;
271 }
272
273
274 PUTBACK;
275 call_sv (req->callback, G_VOID | G_EVAL);
276 SPAGAIN;
277
278 FREETMPS;
279 LEAVE;
280
281 errno = errorno;
282
283 if (SvTRUE (ERRSV))
284 {
285 req_free (req);
286 croak (0);
287 }
288}
289
290static void req_free (aio_req req)
291{
292 if (req->grp)
293 {
294 aio_req grp = req->grp;
295
296 /* unlink request */
297 if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
298 if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
299
300 if (grp->grp_first == req)
301 grp->grp_first = req->grp_next;
302
303 aio_grp_dec (grp);
304 }
305
306 if (req->self)
307 {
308 sv_unmagic (req->self, PERL_MAGIC_ext);
309 SvREFCNT_dec (req->self);
310 }
311
312 SvREFCNT_dec (req->data);
313 SvREFCNT_dec (req->fh);
314 SvREFCNT_dec (req->fh2);
315 SvREFCNT_dec (req->callback);
316 Safefree (req->statdata);
317
318 if (req->type == REQ_READDIR && req->result >= 0)
319 free (req->data2ptr);
320
321 Safefree (req);
322}
323
324static void req_cancel (aio_req req)
325{
326 req->cancelled = 1;
327
328 if (req->type == REQ_GROUP)
329 {
330 aio_req sub;
331
332 for (sub = req->grp_first; sub; sub = sub->grp_next)
333 req_cancel (sub);
334 }
335}
336
337static int poll_cb ()
167{ 338{
168 dSP; 339 dSP;
169 int count = 0; 340 int count = 0;
170 int do_croak = 0; 341 int do_croak = 0;
171 aio_req req; 342 aio_req req;
193 pthread_mutex_unlock (&reslock); 364 pthread_mutex_unlock (&reslock);
194 365
195 if (!req) 366 if (!req)
196 break; 367 break;
197 368
198 nreqs--; 369 --nreqs;
199 370
200 if (req->type == REQ_QUIT) 371 if (req->type == REQ_QUIT)
201 started--; 372 started--;
373 else if (req->type == REQ_GROUP && req->length)
374 {
375 req->fd = 1; /* mark request as delayed */
376 continue;
377 }
202 else 378 else
203 { 379 {
204 int errorno = errno;
205 errno = req->errorno;
206
207 if (req->type == REQ_READ) 380 if (req->type == REQ_READ)
208 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0)); 381 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
209 382
210 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE)) 383 if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
211 SvREADONLY_off (req->data); 384 SvREADONLY_off (req->data);
215 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; 388 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
216 PL_laststatval = req->result; 389 PL_laststatval = req->result;
217 PL_statcache = *(req->statdata); 390 PL_statcache = *(req->statdata);
218 } 391 }
219 392
220 ENTER; 393 req_invoke (req);
221 PUSHMARK (SP);
222 394
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++; 395 count++;
283 } 396 }
284 397
285 req_free (req); 398 req_free (req);
286 } 399 }
288 return count; 401 return count;
289} 402}
290 403
291static void *aio_proc(void *arg); 404static void *aio_proc(void *arg);
292 405
293static void
294start_thread (void) 406static void start_thread (void)
295{ 407{
296 sigset_t fullsigset, oldsigset; 408 sigset_t fullsigset, oldsigset;
297 pthread_t tid; 409 pthread_t tid;
298 pthread_attr_t attr; 410 pthread_attr_t attr;
299 411
308 started++; 420 started++;
309 421
310 sigprocmask (SIG_SETMASK, &oldsigset, 0); 422 sigprocmask (SIG_SETMASK, &oldsigset, 0);
311} 423}
312 424
313static void
314req_send (aio_req req) 425static void req_send (aio_req req)
315{ 426{
316 while (started < wanted && nreqs >= started) 427 while (started < wanted && nreqs >= started)
317 start_thread (); 428 start_thread ();
318 429
319 nreqs++; 430 ++nreqs;
320 431
321 pthread_mutex_lock (&reqlock); 432 pthread_mutex_lock (&reqlock);
322 433
323 req->next = 0; 434 req->next = 0;
324 435
343 454
344 poll_wait (); 455 poll_wait ();
345 } 456 }
346} 457}
347 458
348static void 459static void end_thread (void)
349end_thread (void)
350{ 460{
351 aio_req req; 461 aio_req req;
352 Newz (0, req, 1, aio_cb); 462 Newz (0, req, 1, aio_cb);
353 req->type = REQ_QUIT; 463 req->type = REQ_QUIT;
354 464
405 * normal read/write by using a mutex. slows down execution a lot, 515 * normal read/write by using a mutex. slows down execution a lot,
406 * but that's your problem, not mine. 516 * but that's your problem, not mine.
407 */ 517 */
408static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER; 518static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER;
409 519
410static ssize_t 520static 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{ 521{
413 ssize_t res; 522 ssize_t res;
414 off_t ooffset; 523 off_t ooffset;
415 524
416 pthread_mutex_lock (&preadwritelock); 525 pthread_mutex_lock (&preadwritelock);
421 pthread_mutex_unlock (&preadwritelock); 530 pthread_mutex_unlock (&preadwritelock);
422 531
423 return res; 532 return res;
424} 533}
425 534
426static ssize_t
427pwrite (int fd, void *buf, size_t count, off_t offset) 535static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset)
428{ 536{
429 ssize_t res; 537 ssize_t res;
430 off_t ooffset; 538 off_t ooffset;
431 539
432 pthread_mutex_lock (&preadwritelock); 540 pthread_mutex_lock (&preadwritelock);
445#endif 553#endif
446 554
447#if !HAVE_READAHEAD 555#if !HAVE_READAHEAD
448# define readahead aio_readahead 556# define readahead aio_readahead
449 557
450static ssize_t
451readahead (int fd, off_t offset, size_t count) 558static ssize_t readahead (int fd, off_t offset, size_t count)
452{ 559{
453 char readahead_buf[4096]; 560 char readahead_buf[4096];
454 561
455 while (count > 0) 562 while (count > 0)
456 { 563 {
468#if !HAVE_READDIR_R 575#if !HAVE_READDIR_R
469# define readdir_r aio_readdir_r 576# define readdir_r aio_readdir_r
470 577
471static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER; 578static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;
472 579
473static int
474readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res) 580static int readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res)
475{ 581{
476 struct dirent *e; 582 struct dirent *e;
477 int errorno; 583 int errorno;
478 584
479 pthread_mutex_lock (&readdirlock); 585 pthread_mutex_lock (&readdirlock);
495 return e ? 0 : -1; 601 return e ? 0 : -1;
496} 602}
497#endif 603#endif
498 604
499/* sendfile always needs emulation */ 605/* sendfile always needs emulation */
500static ssize_t
501sendfile_ (int ofd, int ifd, off_t offset, size_t count) 606static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count)
502{ 607{
503 ssize_t res; 608 ssize_t res;
504 609
505 if (!count) 610 if (!count)
506 return 0; 611 return 0;
588 693
589 return res; 694 return res;
590} 695}
591 696
592/* read a full directory */ 697/* read a full directory */
593static int
594scandir_ (const char *path, void **namesp) 698static int scandir_ (const char *path, void **namesp)
595{ 699{
596 DIR *dirp = opendir (path); 700 DIR *dirp = opendir (path);
597 union 701 union
598 { 702 {
599 struct dirent d; 703 struct dirent d;
653 return res; 757 return res;
654} 758}
655 759
656/*****************************************************************************/ 760/*****************************************************************************/
657 761
658static void *
659aio_proc (void *thr_arg) 762static void *aio_proc (void *thr_arg)
660{ 763{
661 aio_req req; 764 aio_req req;
662 int type; 765 int type;
663 766
664 do 767 do
684 pthread_mutex_unlock (&reqlock); 787 pthread_mutex_unlock (&reqlock);
685 788
686 errno = 0; /* strictly unnecessary */ 789 errno = 0; /* strictly unnecessary */
687 790
688 if (!req->cancelled) 791 if (!req->cancelled)
689 switch (req->type) 792 switch (type = req->type) /* remember type for QUIT check */
690 { 793 {
691 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 794 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; 795 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
693 796
694 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 797 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
708 811
709 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; 812 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
710 case REQ_FSYNC: req->result = fsync (req->fd); break; 813 case REQ_FSYNC: req->result = fsync (req->fd); break;
711 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break; 814 case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;
712 815
816 case REQ_SLEEP:
817 {
818 struct timeval tv;
819
820 tv.tv_sec = req->fd;
821 tv.tv_usec = req->fd2;
822
823 req->result = select (0, 0, 0, 0, &tv);
824 }
825
826 case REQ_GROUP:
827 case REQ_NOP:
713 case REQ_QUIT: 828 case REQ_QUIT:
714 break; 829 break;
715 830
716 default: 831 default:
717 req->result = ENOSYS; 832 req->result = ENOSYS;
1073 1188
1074 REQ_SEND; 1189 REQ_SEND;
1075} 1190}
1076 1191
1077void 1192void
1193aio_sleep (delay,callback=&PL_sv_undef)
1194 double delay
1195 SV * callback
1196 PPCODE:
1197{
1198 dREQ;
1199
1200 req->type = REQ_SLEEP;
1201 req->fd = delay < 0. ? 0 : delay;
1202 req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);
1203
1204 REQ_SEND;
1205}
1206
1207void
1078aio_group (callback=&PL_sv_undef) 1208aio_group (callback=&PL_sv_undef)
1079 SV * callback 1209 SV * callback
1080 PROTOTYPE: ;& 1210 PROTOTYPE: ;$
1081 PPCODE: 1211 PPCODE:
1082{ 1212{
1083 dREQ; 1213 dREQ;
1084 req->type = REQ_GROUP; 1214 req->type = REQ_GROUP;
1215 req_send (req);
1085 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1216 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1217}
1218
1219void
1220aio_nop (callback=&PL_sv_undef)
1221 SV * callback
1222 PPCODE:
1223{
1224 dREQ;
1225
1226 req->type = REQ_NOP;
1227
1228 REQ_SEND;
1086} 1229}
1087 1230
1088void 1231void
1089flush () 1232flush ()
1090 PROTOTYPE: 1233 PROTOTYPE:
1134 CODE: 1277 CODE:
1135 RETVAL = nreqs; 1278 RETVAL = nreqs;
1136 OUTPUT: 1279 OUTPUT:
1137 RETVAL 1280 RETVAL
1138 1281
1282PROTOTYPES: DISABLE
1283
1139MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1284MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1140 1285
1141void 1286void
1142cancel (aio_req_ornot req) 1287cancel (aio_req_ornot req)
1143 PROTOTYPE: 1288 PROTOTYPE:
1144 CODE: 1289 CODE:
1145 req->cancelled = 1; 1290 req_cancel (req);
1146 1291
1292void
1293cb (aio_req req, SV *callback=&PL_sv_undef)
1294 CODE:
1295 SvREFCNT_dec (req->callback);
1296 req->callback = newSVsv (callback);
1297
1298MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1299
1300void
1301add (aio_req grp, ...)
1302 PPCODE:
1303{
1304 int i;
1305 aio_req req;
1306
1307 if (grp->fd == 2)
1308 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1309
1310 for (i = 1; i < items; ++i )
1311 {
1312 if (GIMME_V != G_VOID)
1313 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1314
1315 req = SvAIO_REQ (ST (i));
1316
1317 if (req)
1318 {
1319 ++grp->length;
1320 req->grp = grp;
1321
1322 req->grp_prev = 0;
1323 req->grp_next = grp->grp_first;
1324
1325 if (grp->grp_first)
1326 grp->grp_first->grp_prev = req;
1327
1328 grp->grp_first = req;
1329 }
1330 }
1331}
1332
1333void
1334result (aio_req grp, ...)
1335 CODE:
1336{
1337 int i;
1338 AV *av = newAV ();
1339
1340 for (i = 1; i < items; ++i )
1341 av_push (av, newSVsv (ST (i)));
1342
1343 SvREFCNT_dec (grp->data);
1344 grp->data = (SV *)av;
1345}
1346
1347void
1348feed_limit (aio_req grp, int limit)
1349 CODE:
1350 grp->fd2 = limit;
1351 aio_grp_feed (grp);
1352
1353void
1354feed (aio_req grp, SV *callback=&PL_sv_undef)
1355 CODE:
1356{
1357 SvREFCNT_dec (grp->fh2);
1358 grp->fh2 = newSVsv (callback);
1359
1360 if (grp->fd2 <= 0)
1361 grp->fd2 = 2;
1362
1363 aio_grp_feed (grp);
1364}
1365

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines