--- IO-AIO/AIO.xs 2006/10/22 00:49:29 1.46 +++ IO-AIO/AIO.xs 2006/10/23 00:47:29 1.55 @@ -57,8 +57,8 @@ REQ_UNLINK, REQ_RMDIR, REQ_RENAME, REQ_READDIR, REQ_LINK, REQ_SYMLINK, + REQ_GROUP, REQ_NOP, REQ_SLEEP, - REQ_GROUP, }; #define AIO_REQ_KLASS "IO::AIO::REQ" @@ -66,10 +66,10 @@ typedef struct aio_cb { - struct aio_cb *grp, *grp_prev, *grp_next; - struct aio_cb *volatile next; + struct aio_cb *grp, *grp_prev, *grp_next, *grp_first; + SV *self; /* the perl counterpart of this request, if any */ SV *data, *callback; @@ -103,27 +103,79 @@ static volatile aio_req reqs, reqe; /* queue start, queue end */ static volatile aio_req ress, rese; /* queue start, queue end */ +static void req_invoke (aio_req req); static void req_free (aio_req req); /* must be called at most once */ static SV *req_sv (aio_req req, const char *klass) { - req->self = (SV *)newHV (); - sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0); + if (!req->self) + { + req->self = (SV *)newHV (); + sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0); + } return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1))); } static aio_req SvAIO_REQ (SV *sv) { + MAGIC *mg; + if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) croak ("object of class " AIO_REQ_KLASS " expected"); - MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); + mg = mg_find (SvRV (sv), PERL_MAGIC_ext); return mg ? (aio_req)mg->mg_ptr : 0; } +static void aio_grp_feed (aio_req grp) +{ + while (grp->length < grp->fd2) + { + int old_len = grp->length; + + if (grp->fh2 && SvOK (grp->fh2)) + { + dSP; + + ENTER; + SAVETMPS; + PUSHMARK (SP); + XPUSHs (req_sv (grp, AIO_GRP_KLASS)); + PUTBACK; + call_sv (grp->fh2, G_VOID | G_EVAL); + SPAGAIN; + FREETMPS; + LEAVE; + } + + /* stop if no progress has been made */ + if (old_len == grp->length) + { + SvREFCNT_dec (grp->fh2); + grp->fh2 = 0; + break; + } + } +} + +static void aio_grp_dec (aio_req grp) +{ + --grp->length; + + /* call feeder, if applicable */ + aio_grp_feed (grp); + + /* finish, if done */ + if (!grp->length && grp->fd) + { + req_invoke (grp); + req_free (grp); + } +} + static void poll_wait () { if (nreqs && !ress) @@ -147,7 +199,9 @@ errno = req->errorno; ENTER; + SAVETMPS; PUSHMARK (SP); + EXTEND (SP, 1); switch (req->type) { @@ -172,7 +226,7 @@ rv = sv_2mortal (newRV_noinc ((SV *)av)); } - XPUSHs (rv); + PUSHs (rv); } break; @@ -181,7 +235,7 @@ /* convert fd to fh */ SV *fh; - XPUSHs (sv_2mortal (newSViv (req->result))); + PUSHs (sv_2mortal (newSViv (req->result))); PUTBACK; call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); SPAGAIN; @@ -193,12 +247,26 @@ } break; - case REQ_SLEEP: case REQ_GROUP: + req->fd = 2; /* mark group as finished */ + + if (req->data) + { + int i; + AV *av = (AV *)req->data; + + EXTEND (SP, AvFILL (av) + 1); + for (i = 0; i <= AvFILL (av); ++i) + PUSHs (*av_fetch (av, i, 0)); + } + break; + + case REQ_NOP: + case REQ_SLEEP: break; default: - XPUSHs (sv_2mortal (newSViv (req->result))); + PUSHs (sv_2mortal (newSViv (req->result))); break; } @@ -207,15 +275,16 @@ call_sv (req->callback, G_VOID | G_EVAL); SPAGAIN; + FREETMPS; + LEAVE; + + errno = errorno; + if (SvTRUE (ERRSV)) { req_free (req); croak (0); } - - LEAVE; - - errno = errorno; } static void req_free (aio_req req) @@ -225,14 +294,13 @@ aio_req grp = req->grp; /* unlink request */ - req->grp_next->grp_prev = req->grp_prev; - req->grp_prev->grp_next = req->grp_next; + if (req->grp_next) req->grp_next->grp_prev = req->grp_prev; + if (req->grp_prev) req->grp_prev->grp_next = req->grp_next; - if (grp->grp_next == grp && grp->fd) - { - req_invoke (grp); - req_free (grp); - } + if (grp->grp_first == req) + grp->grp_first = req->grp_next; + + aio_grp_dec (grp); } if (req->self) @@ -241,20 +309,11 @@ SvREFCNT_dec (req->self); } - if (req->data) - SvREFCNT_dec (req->data); - - if (req->fh) - SvREFCNT_dec (req->fh); - - if (req->fh2) - SvREFCNT_dec (req->fh2); - - if (req->statdata) - Safefree (req->statdata); - - if (req->callback) - SvREFCNT_dec (req->callback); + SvREFCNT_dec (req->data); + SvREFCNT_dec (req->fh); + SvREFCNT_dec (req->fh2); + SvREFCNT_dec (req->callback); + Safefree (req->statdata); if (req->type == REQ_READDIR && req->result >= 0) free (req->data2ptr); @@ -270,7 +329,7 @@ { aio_req sub; - for (sub = req->grp_next; sub != req; sub = sub->grp_next) + for (sub = req->grp_first; sub; sub = sub->grp_next) req_cancel (sub); } } @@ -307,11 +366,11 @@ if (!req) break; - nreqs--; + --nreqs; if (req->type == REQ_QUIT) started--; - else if (req->type == REQ_GROUP && req->grp_next != req) + else if (req->type == REQ_GROUP && req->length) { req->fd = 1; /* mark request as delayed */ continue; @@ -368,7 +427,7 @@ while (started < wanted && nreqs >= started) start_thread (); - nreqs++; + ++nreqs; pthread_mutex_lock (&reqlock); @@ -730,7 +789,7 @@ errno = 0; /* strictly unnecessary */ if (!req->cancelled) - switch (req->type) + switch (type = req->type) /* remember type for QUIT check */ { case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; @@ -764,6 +823,8 @@ req->result = select (0, 0, 0, 0, &tv); } + case REQ_GROUP: + case REQ_NOP: case REQ_QUIT: break; @@ -1151,14 +1212,23 @@ { dREQ; req->type = REQ_GROUP; - req->grp_next = req; - req->grp_prev = req; - req_send (req); XPUSHs (req_sv (req, AIO_GRP_KLASS)); } void +aio_nop (callback=&PL_sv_undef) + SV * callback + PPCODE: +{ + dREQ; + + req->type = REQ_NOP; + + REQ_SEND; +} + +void flush () PROTOTYPE: CODE: @@ -1209,6 +1279,8 @@ OUTPUT: RETVAL +PROTOTYPES: DISABLE + MODULE = IO::AIO PACKAGE = IO::AIO::REQ void @@ -1221,28 +1293,77 @@ void add (aio_req grp, ...) - PROTOTYPE: $;@ PPCODE: { int i; + aio_req req; + + if (grp->fd == 2) + croak ("cannot add requests to IO::AIO::GRP after the group finished"); for (i = 1; i < items; ++i ) { if (GIMME_V != G_VOID) XPUSHs (sv_2mortal (newSVsv (ST (i)))); - aio_req req = SvAIO_REQ (ST (i)); - printf ("req = %p\n", req);//D + req = SvAIO_REQ (ST (i)); if (req) { - req->grp_prev = grp; - req->grp_next = grp->grp_next; - grp->grp_next->grp_prev = req; - grp->grp_next = req; - + ++grp->length; req->grp = grp; + + req->grp_prev = 0; + req->grp_next = grp->grp_first; + + if (grp->grp_first) + grp->grp_first->grp_prev = req; + + grp->grp_first = req; } } } +void +result (aio_req grp, ...) + CODE: +{ + int i; + AV *av = newAV (); + + for (i = 1; i < items; ++i ) + av_push (av, newSVsv (ST (i))); + + SvREFCNT_dec (grp->data); + grp->data = (SV *)av; +} + +void +lock (aio_req grp) + CODE: + ++grp->length; + +void +unlock (aio_req grp) + CODE: + aio_grp_dec (grp); + +void +feeder_limit (aio_req grp, int limit) + CODE: + grp->fd2 = limit; + aio_grp_feed (grp); + +void +set_feeder (aio_req grp, SV *callback=&PL_sv_undef) + CODE: +{ + SvREFCNT_dec (grp->fh2); + grp->fh2 = newSVsv (callback); + + if (grp->fd2 <= 0) + grp->fd2 = 2; + + aio_grp_feed (grp); +} +