--- IO-AIO/AIO.xs 2008/05/10 18:06:41 1.115 +++ IO-AIO/AIO.xs 2009/06/03 12:24:49 1.140 @@ -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 */ @@ -73,24 +82,37 @@ # define SvVAL64 SvNV #endif +/*****************************************************************************/ + +#if __GNUC__ >= 3 +# define expect(expr,value) __builtin_expect ((expr),(value)) +#else +# define expect(expr,value) (expr) +#endif + +#define expect_false(expr) expect ((expr) != 0, 0) +#define expect_true(expr) expect ((expr) != 0, 1) + +/*****************************************************************************/ + static HV *stash; typedef SV SV8; /* byte-sv, used for argument-checking */ #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" static int req_invoke (eio_req *req); #define EIO_FINISH(req) req_invoke (req) -static void aio_grp_feed (eio_req *grp); -#define EIO_FEED(req) aio_grp_feed (req) static void req_destroy (eio_req *grp); #define EIO_DESTROY(req) req_destroy (req) @@ -103,9 +125,9 @@ typedef eio_req *aio_req; typedef eio_req *aio_req_ornot; -static unsigned int max_outstanding = 0x7fffffff; - -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 +190,23 @@ } } +static void req_submit (eio_req *req) +{ + eio_submit (req); + + if (expect_false (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; @@ -175,7 +214,7 @@ if (req->flags & FLAG_SV2_RO_OFF) SvREADONLY_off (req->sv2); - if (!EIO_CANCELLED (req) && SvOK (req->callback)) + if (!EIO_CANCELLED (req) && req->callback) { ENTER; SAVETMPS; @@ -262,11 +301,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: @@ -279,9 +314,13 @@ break; case EIO_READ: - SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0)); - *SvEND (req->sv2) = 0; - PUSHs (sv_2mortal (newSViv (req->result))); + { + SvCUR_set (req->sv2, req->stroffset + (req->result > 0 ? req->result : 0)); + *SvEND (req->sv2) = 0; + SvPOK_only (req->sv2); + SvSETMAGIC (req->sv2); + PUSHs (sv_2mortal (newSViv (req->result))); + } break; case EIO_DUP2: @@ -396,7 +435,7 @@ if (size) return; - maybe_start_thread (); + etp_maybe_start_thread (); FD_ZERO (&rfd); FD_SET (respipe [0], &rfd); @@ -407,35 +446,77 @@ static int poll_cb (void) { - int res = eio_poll (); + for (;;) + { + int res = eio_poll (); + + if (res > 0) + croak (0); + + if (!max_outstanding || max_outstanding > eio_nreqs ()) + return res; + + poll_wait (); + } +} + +static void atfork_child (void) +{ + create_respipe (); +} + +static SV * +get_cb (SV *cb_sv) +{ + HV *st; + GV *gvp; + CV *cv; + + if (!SvOK (cb_sv)) + return 0; + + cv = sv_2cv (cb_sv, &st, &gvp, 0); - if (res > 0) - croak (0); + if (!cv) + croak ("IO::AIO callback must be undef or a CODE reference"); - return res; + return (SV *)cv; } #define dREQ \ + SV *cb_cv; \ 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"); \ + cb_cv = get_cb (callback); \ \ Newz (0, req, 1, eio_req); \ if (!req) \ croak ("out of memory during eio_req allocation"); \ \ - req->callback = newSVsv (callback); \ + req->callback = SvREFCNT_inc (cb_cv); \ 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)); - + +static int +extract_fd (SV *fh, int wr) +{ + int fd = PerlIO_fileno (wr ? IoOFP (sv_2io (fh)) : IoIFP (sv_2io (fh))); + + if (fd < 0) + croak ("illegal fh argument, either not an OS file or read/write mode mismatch"); + + return fd; +} + MODULE = IO::AIO PACKAGE = IO::AIO PROTOTYPES: ENABLE @@ -445,27 +526,26 @@ stash = gv_stashpv ("IO::AIO", 1); newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); + newCONSTSUB (stash, "ENOSYS", newSViv (ENOSYS)); newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 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 + newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO)); + newCONSTSUB (stash, "SYNC_FILE_RANGE_WAIT_BEFORE", newSViv (EIO_SYNC_FILE_RANGE_WAIT_BEFORE)); + newCONSTSUB (stash, "SYNC_FILE_RANGE_WRITE" , newSViv (EIO_SYNC_FILE_RANGE_WRITE)); + newCONSTSUB (stash, "SYNC_FILE_RANGE_WAIT_AFTER" , newSViv (EIO_SYNC_FILE_RANGE_WAIT_AFTER)); create_respipe (); if (eio_init (want_poll, done_poll) < 0) croak ("IO::AIO: unable to initialise eio library"); + + /* atfork child called in fifo order, so before eio's handler */ + X_THREAD_ATFORK (0, 0, atfork_child); } void @@ -528,11 +608,30 @@ aio_fdatasync = EIO_FDATASYNC PPCODE: { + int fd = extract_fd (fh, 0); dREQ; req->type = ix; req->sv1 = newSVsv (fh); - req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh))); + req->int1 = fd; + + REQ_SEND (req); +} + +void +aio_sync_file_range (SV *fh, SV *offset, SV *nbytes, IV flags, SV *callback=&PL_sv_undef) + PROTOTYPE: $$$$;$ + PPCODE: +{ + int fd = extract_fd (fh, 0); + dREQ; + + req->type = EIO_SYNC_FILE_RANGE; + req->sv1 = newSVsv (fh); + req->int1 = fd; + req->offs = SvVAL64 (offset); + req->size = SvVAL64 (nbytes); + req->int2 = flags; REQ_SEND (req); } @@ -543,6 +642,7 @@ PPCODE: { static int close_pipe = -1; /* dummy fd to close fds via dup2 */ + int fd = extract_fd (fh, 0); dREQ; if (close_pipe < 0) @@ -560,7 +660,7 @@ req->type = EIO_DUP2; req->int1 = close_pipe; req->sv2 = newSVsv (fh); - req->int2 = PerlIO_fileno (IoIFP (sv_2io (fh))); + req->int2 = fd; REQ_SEND (req); } @@ -574,12 +674,10 @@ PPCODE: { STRLEN svlen; + int fd = extract_fd (fh, ix == EIO_WRITE); char *svptr = SvPVbyte (data, svlen); UV len = SvUV (length); - SvUPGRADE (data, SVt_PV); - SvPOK_on (data); - if (dataoffset < 0) dataoffset += svlen; @@ -594,20 +692,17 @@ } else { - /* read: grow scalar as necessary */ + /* read: check type and grow scalar as necessary */ + SvUPGRADE (data, SVt_PV); svptr = SvGROW (data, len + dataoffset + 1); } - if (len < 0) - croak ("length must not be negative"); - { dREQ; req->type = ix; req->sv1 = newSVsv (fh); - req->int1 = PerlIO_fileno (ix == EIO_READ ? IoIFP (sv_2io (fh)) - : IoOFP (sv_2io (fh))); + req->int1 = fd; req->offs = SvOK (offset) ? SvVAL64 (offset) : -1; req->size = len; req->sv2 = SvREFCNT_inc (data); @@ -632,14 +727,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; } @@ -649,13 +739,15 @@ PROTOTYPE: $$$$;$ PPCODE: { + int ifd = extract_fd (in_fh , 0); + int ofd = extract_fd (out_fh, 0); dREQ; req->type = EIO_SENDFILE; req->sv1 = newSVsv (out_fh); - req->int1 = PerlIO_fileno (IoIFP (sv_2io (out_fh))); + req->int1 = ofd; req->sv2 = newSVsv (in_fh); - req->int2 = PerlIO_fileno (IoIFP (sv_2io (in_fh))); + req->int2 = ifd; req->offs = SvVAL64 (in_offset); req->size = length; @@ -667,11 +759,12 @@ PROTOTYPE: $$$;$ PPCODE: { + int fd = extract_fd (fh, 0); dREQ; req->type = EIO_READAHEAD; req->sv1 = newSVsv (fh); - req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh))); + req->int1 = fd; req->offs = SvVAL64 (offset); req->size = length; @@ -687,17 +780,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); @@ -721,7 +806,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); @@ -744,7 +829,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); @@ -762,20 +847,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; } @@ -790,7 +879,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); @@ -876,7 +965,7 @@ req->type = EIO_GROUP; - eio_submit (req); + req_submit (req); XPUSHs (req_sv (req, AIO_GRP_KLASS)); } @@ -898,12 +987,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 @@ -914,13 +1003,13 @@ 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 () PROTOTYPE: CODE: - while (nreqs) + while (eio_nreqs ()) { poll_wait (); poll_cb (); @@ -985,12 +1074,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 @@ -1002,9 +1094,19 @@ void cb (aio_req_ornot req, SV *callback=&PL_sv_undef) - CODE: - SvREFCNT_dec (req->callback); - req->callback = newSVsv (callback); + PPCODE: +{ + if (GIMME_V != G_VOID) + XPUSHs (req->callback ? sv_2mortal (newRV_inc (req->callback)) : &PL_sv_undef); + + if (items > 1) + { + SV *cb_cv = get_cb (callback); + + SvREFCNT_dec (req->callback); + req->callback = SvREFCNT_inc (cb_cv); + } +} MODULE = IO::AIO PACKAGE = IO::AIO::GRP @@ -1069,7 +1171,8 @@ CODE: { SvREFCNT_dec (grp->sv2); - grp->sv2 = newSVsv (callback); + grp->sv2 = newSVsv (callback); + grp->feed = aio_grp_feed; if (grp->int2 <= 0) grp->int2 = 2;