--- Coro/Event/Event.xs 2006/11/23 23:54:02 1.10 +++ Coro/Event/Event.xs 2007/09/22 20:11:53 1.20 @@ -2,70 +2,86 @@ #include "perl.h" #include "XSUB.h" +#include #include -/* this useful idiom is unfortunately missing... */ -static void -confess (const char *msg) -{ - dSP; - - PUSHMARK(SP); - XPUSHs (sv_2mortal(newSVpv("only one coroutine can wait for an event at any given time",0))); - PUTBACK; - call_pv ("Carp::confess", G_VOID); -} - #include "EventAPI.h" #include "../Coro/CoroAPI.h" -#ifndef PE_PERLCB -# define PE_PERLCB 0x020 /* not public, but we need it :( */ -#endif - -#define CD_CORO 0 +#define CD_WAIT 0 /* wait queue */ #define CD_TYPE 1 #define CD_OK 2 -#define CD_PRIO 3 /* hardcoded in Coro::Event */ -#define CD_HITS 4 /* hardcoded in Coro::Event */ -#define CD_GOT 5 /* hardcoded in Coro::Event, Coro::Handle */ -#define CD_MAX 5 +#define CD_HITS 3 /* hardcoded in Coro::Event */ +#define CD_GOT 4 /* hardcoded in Coro::Event, Coro::Handle */ +#define CD_MAX 4 -#define EV_CLASS "Coro::Event" +static HV *coro_event_event_stash; -static void -coro_std_cb(pe_event *pe) +static int +coro_ev_free (pTHX_ SV *sv, MAGIC *mg) { - AV *priv = (AV *)pe->ext_data; - IV type = SvIV (*av_fetch (priv, CD_TYPE, 1)); - SV **cd_coro = &AvARRAY(priv)[CD_CORO]; + *(void **)mg->mg_ptr = 0; - sv_setiv (AvARRAY(priv)[CD_PRIO], pe->prio); - sv_setiv (AvARRAY(priv)[CD_HITS], pe->hits); + return 0; +} - if (type == 1) - sv_setiv (AvARRAY(priv)[CD_GOT], ((pe_ioevent *)pe)->got); +#define PERL_MAGIC_coro_event 0x18 /* to avoid clashes with e.g. event */ - if (*cd_coro != &PL_sv_undef) - { - CORO_READY (*cd_coro); - SvREFCNT_dec (*cd_coro); - *cd_coro = &PL_sv_undef; - } - else +static MGVTBL vtbl_coro_event = {0, 0, 0, 0, coro_ev_free}; + +static void +coro_std_cb (pe_event *pe) +{ + if (pe->up->ext_data) { - AvARRAY(priv)[CD_OK] = &PL_sv_yes; - GEventAPI->stop (pe->up, 0); + AV *priv = (AV *)pe->ext_data; + IV type = SvIV (AvARRAY (priv)[CD_TYPE]); + AV *cd_wait; + SV *coro; + + SvIV_set (AvARRAY (priv)[CD_HITS], pe->hits); + SvIV_set (AvARRAY (priv)[CD_GOT], type ? ((pe_ioevent *)pe)->got : 0); + + AvARRAY (priv)[CD_OK] = &PL_sv_yes; + + cd_wait = (AV *)AvARRAY(priv)[CD_WAIT]; + + coro = av_shift (cd_wait); + if (coro != &PL_sv_undef) + { + CORO_READY (coro); + SvREFCNT_dec (coro); + } + + if (av_len (cd_wait) < 0) + GEventAPI->stop (pe->up, 0); } } +static void +asynccheck_hook (void *data) +{ + /* this loops as long as we have _other_ coros with the same or higher priority */ + while (CORO_NREADY && CORO_CEDE) + ; +} + static double prepare_hook (void *data) { - while (CORO_NREADY) - CORO_CEDE; - - return 1e20; + /* this yields once to another coro with any priority */ + if (CORO_NREADY) + { + CORO_CEDE_NOTSELF; + /* + * timers might have changed, and Event fails to notice this + * so we have to assume the worst. If Event didn't have that bug, + * we would only need to do this if CORO_NREADY is != 0 now. + */ + return 0.; + } + else + return 85197.73; /* this is as good as any value, but it factors badly with common values */ } MODULE = Coro::Event PACKAGE = Coro::Event @@ -74,65 +90,85 @@ BOOT: { + coro_event_event_stash = gv_stashpv ("Coro::Event::Event", TRUE); + I_EVENT_API ("Coro::Event"); I_CORO_API ("Coro::Event"); - GEventAPI->add_hook ("prepare", (void *)prepare_hook, 0); + GEventAPI->add_hook ("asynccheck", (void *)asynccheck_hook, 0); + GEventAPI->add_hook ("prepare", (void *)prepare_hook, 0); } void -_install_std_cb(self,type) - SV * self - int type +_install_std_cb (SV *self, int type) CODE: +{ pe_watcher *w = GEventAPI->sv_2watcher (self); - if (WaFLAGS (w) & PE_PERLCB) - croak ("Coro::Event watchers must not have a perl callback (see Coro::Event), caught"); + if (w->callback) + croak ("Coro::Event watchers must not have a callback (see Coro::Event), caught"); + { AV *priv = newAV (); - SV *rv = newRV_noinc ((SV *)priv); - av_extend (priv, CD_MAX); - av_store (priv, CD_CORO, &PL_sv_undef); - av_store (priv, CD_TYPE, newSViv (type)); - av_store (priv, CD_OK , &PL_sv_no); - av_store (priv, CD_PRIO, newSViv (0)); - av_store (priv, CD_HITS, newSViv (0)); - av_store (priv, CD_GOT , type ? newSViv (0) : &PL_sv_undef); + av_fill (priv, CD_MAX); + AvARRAY (priv)[CD_WAIT] = (SV *)newAV (); /* AV in AV _should_ not be exposed to perl */ + AvARRAY (priv)[CD_TYPE] = newSViv (type); + AvARRAY (priv)[CD_OK ] = &PL_sv_no; + AvARRAY (priv)[CD_HITS] = newSViv (0); + AvARRAY (priv)[CD_GOT ] = newSViv (0); SvREADONLY_on (priv); w->callback = coro_std_cb; w->ext_data = priv; - hv_store ((HV *)SvRV (self), - EV_CLASS, strlen (EV_CLASS), - rv, 0); - - GEventAPI->start (w, 0); + sv_magicext (SvRV (self), newRV_noinc ((SV *)priv), PERL_MAGIC_coro_event, &vtbl_coro_event, &w->ext_data, 0); } +} void -_next(self) - SV * self +_next (SV *self) CODE: +{ pe_watcher *w = GEventAPI->sv_2watcher (self); AV *priv = (AV *)w->ext_data; + if (AvARRAY (priv)[CD_OK] == &PL_sv_yes) + { + AvARRAY (priv)[CD_OK] = &PL_sv_no; + XSRETURN_NO; /* got an event */ + } + + av_push ((AV *)AvARRAY (priv)[CD_WAIT], SvREFCNT_inc (CORO_CURRENT)); + if (!w->running) GEventAPI->start (w, 1); - if (AvARRAY(priv)[CD_OK] == &PL_sv_yes) - { - AvARRAY(priv)[CD_OK] = &PL_sv_no; - XSRETURN_NO; - } - else - { - if (AvARRAY(priv)[CD_CORO] != &PL_sv_undef) - confess ("only one coroutine can wait for an event"); + XSRETURN_YES; /* schedule */ +} - AvARRAY(priv)[CD_CORO] = SvREFCNT_inc (CORO_CURRENT); - XSRETURN_YES; - } +SV * +_event (SV *self) + CODE: +{ + if (GIMME_V == G_VOID) + XSRETURN_EMPTY; + + { + pe_watcher *w = GEventAPI->sv_2watcher (self); + AV *priv = (AV *)w->ext_data; + + RETVAL = newRV_inc ((SV *)priv); + + /* may need to bless it now */ + if (!SvOBJECT (priv)) + { + SvREADONLY_off ((SV *)priv); + sv_bless (RETVAL, coro_event_event_stash); + SvREADONLY_on ((SV *)priv); + } + } +} + OUTPUT: + RETVAL