--- IO-AIO/AIO.xs 2008/05/10 19:25:33 1.116 +++ IO-AIO/AIO.xs 2008/09/25 07:56:07 1.126 @@ -15,6 +15,9 @@ #include #include +/* perl namespace pollution */ +#undef VERSION + #ifdef _WIN32 # define EIO_STRUCT_DIRENT Direntry_t @@ -64,6 +67,12 @@ #endif +/* perl stupidly overrides readdir and maybe others */ +/* with thread-unsafe versions, imagine that :( */ +#undef readdir +#undef opendir +#undef closedir + #define EIO_STRUCT_STAT Stat_t /* use NV for 32 bit perls as it allows larger offsets */ @@ -79,11 +88,13 @@ #define AIO_REQ_KLASS "IO::AIO::REQ" #define AIO_GRP_KLASS "IO::AIO::GRP" -#define EIO_COMMON \ +#define EIO_REQ_MEMBERS \ SV *callback; \ SV *sv1, *sv2; \ STRLEN stroffset; \ - SV *self + SV *self; + +#define EIO_NO_WRAPPERS 1 #include "libeio/eio.h" @@ -103,9 +114,9 @@ typedef eio_req *aio_req; typedef eio_req *aio_req_ornot; -static int max_outstanding = 0; - -static int next_pri = EIO_DEFAULT_PRI + EIO_PRI_BIAS; +static SV *on_next_submit; +static int next_pri = EIO_PRI_DEFAULT; +static int max_outstanding; static int respipe_osf [2], respipe [2] = { -1, -1 }; @@ -168,6 +179,23 @@ } } +static void req_submit (eio_req *req) +{ + eio_submit (req); + + if (on_next_submit) + { + dSP; + SV *cb = sv_2mortal (on_next_submit); + + on_next_submit = 0; + + PUSHMARK (SP); + PUTBACK; + call_sv (cb, G_DISCARD | G_EVAL); + } +} + static int req_invoke (eio_req *req) { dSP; @@ -262,11 +290,7 @@ case EIO_READLINK: if (req->result > 0) - { - SvCUR_set (req->sv2, req->result); - *SvEND (req->sv2) = 0; - PUSHs (req->sv2); - } + PUSHs (sv_2mortal (newSVpvn (req->ptr2, req->result))); break; case EIO_STAT: @@ -396,7 +420,7 @@ if (size) return; - maybe_start_thread (); + etp_maybe_start_thread (); FD_ZERO (&rfd); FD_SET (respipe [0], &rfd); @@ -407,18 +431,18 @@ static int poll_cb (void) { - int res; - - do + for (;;) { - res = eio_poll (); + int res = eio_poll (); if (res > 0) croak (0); - } - while (max_outstanding && max_outstanding <= eio_nreqs ()); - return res; + if (!max_outstanding || max_outstanding > eio_nreqs ()) + return res; + + poll_wait (); + } } static void atfork_child (void) @@ -429,7 +453,7 @@ #define dREQ \ aio_req req; \ int req_pri = next_pri; \ - next_pri = EIO_DEFAULT_PRI + EIO_PRI_BIAS; \ + next_pri = EIO_PRI_DEFAULT; \ \ if (SvOK (callback) && !SvROK (callback)) \ croak ("callback must be undef or of reference type"); \ @@ -442,7 +466,9 @@ req->pri = req_pri #define REQ_SEND \ - eio_submit (req); \ + PUTBACK; \ + req_submit (req); \ + SPAGAIN; \ \ if (GIMME_V != G_VOID) \ XPUSHs (req_sv (req, AIO_REQ_KLASS)); @@ -460,16 +486,7 @@ newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT)); newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC)); -#ifdef _WIN32 - X_MUTEX_CHECK (wrklock); - X_MUTEX_CHECK (reslock); - X_MUTEX_CHECK (reqlock); - X_MUTEX_CHECK (reqwait); - X_MUTEX_CHECK (preadwritelock); - X_MUTEX_CHECK (readdirlock); - - X_COND_CHECK (reqwait); -#else +#ifndef _WIN32 newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO)); #endif @@ -646,14 +663,9 @@ SV *data; dREQ; - data = newSV (NAME_MAX); - SvPOK_on (data); - req->type = EIO_READLINK; req->sv1 = newSVsv (path); req->ptr1 = SvPVbyte_nolen (req->sv1); - req->sv2 = data; - req->ptr2 = SvPVbyte_nolen (data); REQ_SEND; } @@ -701,17 +713,9 @@ { dREQ; - req->ptr2 = malloc (sizeof (EIO_STRUCT_STAT)); - if (!req->ptr2) - { - req_destroy (req); - croak ("out of memory during aio_stat statdata allocation"); - } - - req->flags |= EIO_FLAG_PTR2_FREE; req->sv1 = newSVsv (fh_or_path); - if (SvPOK (fh_or_path)) + if (SvPOK (req->sv1)) { req->type = ix; req->ptr1 = SvPVbyte_nolen (req->sv1); @@ -735,7 +739,7 @@ req->nv2 = SvOK (mtime) ? SvNV (mtime) : -1.; req->sv1 = newSVsv (fh_or_path); - if (SvPOK (fh_or_path)) + if (SvPOK (req->sv1)) { req->type = EIO_UTIME; req->ptr1 = SvPVbyte_nolen (req->sv1); @@ -758,7 +762,7 @@ req->sv1 = newSVsv (fh_or_path); req->offs = SvOK (offset) ? SvVAL64 (offset) : -1; - if (SvPOK (fh_or_path)) + if (SvPOK (req->sv1)) { req->type = EIO_TRUNCATE; req->ptr1 = SvPVbyte_nolen (req->sv1); @@ -776,20 +780,24 @@ aio_chmod (SV8 *fh_or_path, int mode, SV *callback=&PL_sv_undef) ALIAS: aio_chmod = EIO_CHMOD - aio_fchmod = EIO_FCHMOD aio_mkdir = EIO_MKDIR PPCODE: { dREQ; - req->type = ix; req->int2 = mode; req->sv1 = newSVsv (fh_or_path); - if (ix == EIO_FCHMOD) - req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); + if (SvPOK (req->sv1)) + { + req->type = ix; + req->ptr1 = SvPVbyte_nolen (req->sv1); + } else - req->ptr1 = SvPVbyte_nolen (req->sv1); + { + req->type = EIO_FCHMOD; + req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); + } REQ_SEND; } @@ -804,7 +812,7 @@ req->int3 = SvOK (gid) ? SvIV (gid) : -1; req->sv1 = newSVsv (fh_or_path); - if (SvPOK (fh_or_path)) + if (SvPOK (req->sv1)) { req->type = EIO_CHOWN; req->ptr1 = SvPVbyte_nolen (req->sv1); @@ -912,12 +920,12 @@ aioreq_pri (int pri = 0) PROTOTYPE: ;$ CODE: - RETVAL = next_pri - EIO_PRI_BIAS; + RETVAL = next_pri; if (items > 0) { if (pri < EIO_PRI_MIN) pri = EIO_PRI_MIN; if (pri > EIO_PRI_MAX) pri = EIO_PRI_MAX; - next_pri = pri + EIO_PRI_BIAS; + next_pri = pri; } OUTPUT: RETVAL @@ -928,7 +936,7 @@ nice = next_pri - nice; if (nice < EIO_PRI_MIN) nice = EIO_PRI_MIN; if (nice > EIO_PRI_MAX) nice = EIO_PRI_MAX; - next_pri = nice + EIO_PRI_BIAS; + next_pri = nice; void flush () @@ -999,12 +1007,15 @@ nthreads() PROTOTYPE: CODE: - if (WORDACCESS_UNSAFE) X_LOCK (wrklock); - RETVAL = started; - if (WORDACCESS_UNSAFE) X_UNLOCK (wrklock); + RETVAL = eio_nthreads (); OUTPUT: RETVAL +void _on_next_submit (SV *cb) + CODE: + SvREFCNT_dec (on_next_submit); + on_next_submit = SvOK (cb) ? newSVsv (cb) : 0; + PROTOTYPES: DISABLE MODULE = IO::AIO PACKAGE = IO::AIO::REQ