--- IO-AIO/AIO.xs 2005/08/17 03:52:20 1.26 +++ IO-AIO/AIO.xs 2005/08/17 05:06:59 1.28 @@ -64,7 +64,6 @@ static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; -static pthread_mutex_t frklock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; static volatile aio_req reqs, reqe; /* queue start, queue end */ @@ -78,7 +77,7 @@ if (req->fh) SvREFCNT_dec (req->fh); - if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT) + if (req->statdata) Safefree (req->statdata); if (req->callback) @@ -106,24 +105,33 @@ dSP; int count = 0; int do_croak = 0; - aio_req req, prv; + aio_req req; - pthread_mutex_lock (&reslock); + for (;;) + { + pthread_mutex_lock (&reslock); + req = ress; - { - /* read any signals sent by the worker threads */ - char buf [32]; - while (read (respipe [0], buf, 32) == 32) - ; - } + if (req) + { + ress = req->next; - req = ress; - ress = rese = 0; + if (!ress) + { + rese = 0; - pthread_mutex_unlock (&reslock); + /* read any signals sent by the worker threads */ + char buf [32]; + while (read (respipe [0], buf, 32) == 32) + ; + } + } + + pthread_mutex_unlock (&reslock); + + if (!req) + break; - while (req) - { nreqs--; if (req->type == REQ_QUIT) @@ -134,10 +142,12 @@ errno = req->errorno; if (req->type == REQ_READ) - SvCUR_set (req->data, req->dataoffset - + req->result > 0 ? req->result : 0); + SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0)); + + if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE)) + SvREADONLY_off (req->data); - if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT) + if (req->statdata) { PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; PL_laststatval = req->result; @@ -168,22 +178,21 @@ PUTBACK; call_sv (req->callback, G_VOID | G_EVAL); SPAGAIN; + + if (SvTRUE (ERRSV)) + { + free_req (req); + croak (0); + } } LEAVE; - do_croak = SvTRUE (ERRSV); - errno = errorno; count++; } - prv = req; - req = req->next; - free_req (prv); - - if (do_croak) - croak (0); + free_req (req); } return count; @@ -231,11 +240,16 @@ pthread_cond_signal (&reqwait); pthread_mutex_unlock (&reqlock); - while (nreqs > max_outstanding) - { - poll_wait (); - poll_cb (); - } + if (nreqs > max_outstanding) + for (;;) + { + poll_cb (); + + if (nreqs <= max_outstanding) + break; + + poll_wait (); + } } static void @@ -248,7 +262,6 @@ send_req (req); } - static void min_parallel (int nthreads) { while (nthreads > started) @@ -258,6 +271,7 @@ static void max_parallel (int nthreads) { int cur = started; + while (cur > nthreads) { end_thread (); @@ -285,7 +299,6 @@ static void atfork_prepare (void) { - pthread_mutex_lock (&frklock); pthread_mutex_lock (&reqlock); pthread_mutex_lock (&reslock); } @@ -294,35 +307,33 @@ { pthread_mutex_unlock (&reslock); pthread_mutex_unlock (&reqlock); - pthread_mutex_unlock (&frklock); } static void atfork_child (void) { + aio_req prv; + int restart = started; started = 0; while (reqs) { - free_req (reqs); - reqs = reqs->next; + prv = reqs; + reqs = prv->next; + free_req (prv); } reqs = reqe = 0; - + while (ress) { - free_req (ress); - ress = ress->next; + prv = ress; + ress = prv->next; + free_req (prv); } - + ress = rese = 0; - close (respipe [0]); - close (respipe [1]); - - create_pipe (); - atfork_parent (); min_parallel (restart); @@ -498,7 +509,7 @@ if (!req) \ croak ("out of memory during aio_req allocation"); \ \ - req->callback = SvREFCNT_inc (callback); + req->callback = newSVsv (callback); MODULE = IO::AIO PACKAGE = IO::AIO @@ -621,7 +632,12 @@ req->length = length; req->data = SvREFCNT_inc (data); req->dataptr = (char *)svptr + dataoffset; - req->callback = SvREFCNT_inc (callback); + + if (!SvREADONLY (data)) + { + SvREADONLY_on (data); + req->data2ptr = (void *)data; + } send_req (req); } @@ -660,7 +676,10 @@ New (0, req->statdata, 1, Stat_t); if (!req->statdata) - croak ("out of memory during aio_req->statdata allocation (sorry, i just leaked memory, too)"); + { + free_req (req); + croak ("out of memory during aio_req->statdata allocation"); + } if (SvPOK (fh_or_path)) {