ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Coro/Coro/State.xs
(Generate patch)

Comparing Coro/Coro/State.xs (file contents):
Revision 1.295 by root, Tue Nov 18 10:35:01 2008 UTC vs.
Revision 1.309 by root, Wed Nov 19 15:29:57 2008 UTC

16 16
17#ifdef WIN32 17#ifdef WIN32
18# undef setjmp 18# undef setjmp
19# undef longjmp 19# undef longjmp
20# undef _exit 20# undef _exit
21# define setjmp _setjmp // deep magic, don't ask 21# define setjmp _setjmp /* deep magic */
22#else 22#else
23# include <inttypes.h> /* most portable stdint.h */ 23# include <inttypes.h> /* most portable stdint.h */
24#endif 24#endif
25 25
26#ifdef HAVE_MMAP 26#ifdef HAVE_MMAP
145#define expect_true(expr) expect ((expr) != 0, 1) 145#define expect_true(expr) expect ((expr) != 0, 1)
146 146
147#define NOINLINE attribute ((noinline)) 147#define NOINLINE attribute ((noinline))
148 148
149#include "CoroAPI.h" 149#include "CoroAPI.h"
150#define GCoroAPI (&coroapi) /* very sneaky */
150 151
151#ifdef USE_ITHREADS 152#ifdef USE_ITHREADS
152# if CORO_PTHREAD 153# if CORO_PTHREAD
153static void *coro_thx; 154static void *coro_thx;
154# endif 155# endif
164static size_t cctx_stacksize = CORO_STACKSIZE; 165static size_t cctx_stacksize = CORO_STACKSIZE;
165static struct CoroAPI coroapi; 166static struct CoroAPI coroapi;
166static AV *main_mainstack; /* used to differentiate between $main and others */ 167static AV *main_mainstack; /* used to differentiate between $main and others */
167static JMPENV *main_top_env; 168static JMPENV *main_top_env;
168static HV *coro_state_stash, *coro_stash; 169static HV *coro_state_stash, *coro_stash;
170static CV *coro_run_cv;
169static volatile SV *coro_mortal; /* will be freed/thrown after next transfer */ 171static volatile SV *coro_mortal; /* will be freed/thrown after next transfer */
170 172
171static GV *irsgv; /* $/ */ 173static GV *irsgv; /* $/ */
172static GV *stdoutgv; /* *STDOUT */ 174static GV *stdoutgv; /* *STDOUT */
173static SV *rv_diehook; 175static SV *rv_diehook;
245 /* state data */ 247 /* state data */
246 struct CoroSLF slf_frame; /* saved slf frame */ 248 struct CoroSLF slf_frame; /* saved slf frame */
247 AV *mainstack; 249 AV *mainstack;
248 perl_slots *slot; /* basically the saved sp */ 250 perl_slots *slot; /* basically the saved sp */
249 251
252 CV *startcv; /* the CV to execute */
250 AV *args; /* data associated with this coroutine (initial args) */ 253 AV *args; /* data associated with this coroutine (initial args) */
251 int refcnt; /* coroutines are refcounted, yes */ 254 int refcnt; /* coroutines are refcounted, yes */
252 int flags; /* CF_ flags */ 255 int flags; /* CF_ flags */
253 HV *hv; /* the perl hash associated with this coro, if any */ 256 HV *hv; /* the perl hash associated with this coro, if any */
254 void (*on_destroy)(pTHX_ struct coro *coro); 257 void (*on_destroy)(pTHX_ struct coro *coro);
255 258
256 /* statistics */ 259 /* statistics */
257 int usecount; /* number of transfers to this coro */ 260 int usecount; /* number of transfers to this coro */
258 261
259 /* coro process data */ 262 /* coro process data */
260 int prio; 263 int prio;
261 SV *throw; /* exception to be thrown */ 264 SV *except; /* exception to be thrown */
265 SV *rouse_cb;
262 266
263 /* async_pool */ 267 /* async_pool */
264 SV *saved_deffh; 268 SV *saved_deffh;
265 269
266 /* linked list */ 270 /* linked list */
272 276
273/* the following variables are effectively part of the perl context */ 277/* the following variables are effectively part of the perl context */
274/* and get copied between struct coro and these variables */ 278/* and get copied between struct coro and these variables */
275/* the mainr easonw e don't support windows process emulation */ 279/* the mainr easonw e don't support windows process emulation */
276static struct CoroSLF slf_frame; /* the current slf frame */ 280static struct CoroSLF slf_frame; /* the current slf frame */
277static SV *coro_throw;
278 281
279/** Coro ********************************************************************/ 282/** Coro ********************************************************************/
280 283
281#define PRIO_MAX 3 284#define PRIO_MAX 3
282#define PRIO_HIGH 1 285#define PRIO_HIGH 1
320#if PERL_VERSION_ATLEAST (5,10,0) 323#if PERL_VERSION_ATLEAST (5,10,0)
321 /* silence stupid and wrong 5.10 warning that I am unable to switch off */ 324 /* silence stupid and wrong 5.10 warning that I am unable to switch off */
322 get_hv (name, create); 325 get_hv (name, create);
323#endif 326#endif
324 return get_hv (name, create); 327 return get_hv (name, create);
328}
329
330/* may croak */
331INLINE CV *
332coro_sv_2cv (SV *sv)
333{
334 HV *st;
335 GV *gvp;
336 return sv_2cv (sv, &st, &gvp, 0);
325} 337}
326 338
327static AV * 339static AV *
328coro_clone_padlist (pTHX_ CV *cv) 340coro_clone_padlist (pTHX_ CV *cv)
329{ 341{
448 else 460 else
449 { 461 {
450#if CORO_PREFER_PERL_FUNCTIONS 462#if CORO_PREFER_PERL_FUNCTIONS
451 /* this is probably cleaner? but also slower! */ 463 /* this is probably cleaner? but also slower! */
452 /* in practise, it seems to be less stable */ 464 /* in practise, it seems to be less stable */
453 CV *cp = Perl_cv_clone (cv); 465 CV *cp = Perl_cv_clone (aTHX_ cv);
454 CvPADLIST (cv) = CvPADLIST (cp); 466 CvPADLIST (cv) = CvPADLIST (cp);
455 CvPADLIST (cp) = 0; 467 CvPADLIST (cp) = 0;
456 SvREFCNT_dec (cp); 468 SvREFCNT_dec (cp);
457#else 469#else
458 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv); 470 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv);
511 523
512 PUTBACK; 524 PUTBACK;
513 } 525 }
514 526
515 slf_frame = c->slf_frame; 527 slf_frame = c->slf_frame;
516 coro_throw = c->throw; 528 CORO_THROW = c->except;
517} 529}
518 530
519static void 531static void
520save_perl (pTHX_ Coro__State c) 532save_perl (pTHX_ Coro__State c)
521{ 533{
522 c->throw = coro_throw; 534 c->except = CORO_THROW;
523 c->slf_frame = slf_frame; 535 c->slf_frame = slf_frame;
524 536
525 { 537 {
526 dSP; 538 dSP;
527 I32 cxix = cxstack_ix; 539 I32 cxix = cxstack_ix;
602 * of perl.c:init_stacks, except that it uses less memory 614 * of perl.c:init_stacks, except that it uses less memory
603 * on the (sometimes correct) assumption that coroutines do 615 * on the (sometimes correct) assumption that coroutines do
604 * not usually need a lot of stackspace. 616 * not usually need a lot of stackspace.
605 */ 617 */
606#if CORO_PREFER_PERL_FUNCTIONS 618#if CORO_PREFER_PERL_FUNCTIONS
607# define coro_init_stacks init_stacks 619# define coro_init_stacks(thx) init_stacks ()
608#else 620#else
609static void 621static void
610coro_init_stacks (pTHX) 622coro_init_stacks (pTHX)
611{ 623{
612 PL_curstackinfo = new_stackinfo(32, 8); 624 PL_curstackinfo = new_stackinfo(32, 8);
860 dSP; 872 dSP;
861 UNOP myop; 873 UNOP myop;
862 874
863 Zero (&myop, 1, UNOP); 875 Zero (&myop, 1, UNOP);
864 myop.op_next = Nullop; 876 myop.op_next = Nullop;
877 myop.op_type = OP_ENTERSUB;
865 myop.op_flags = OPf_WANT_VOID; 878 myop.op_flags = OPf_WANT_VOID;
866 879
867 PUSHMARK (SP); 880 PUSHMARK (SP);
868 XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv)))); 881 PUSHs ((SV *)coro->startcv);
869 PUTBACK; 882 PUTBACK;
870 PL_op = (OP *)&myop; 883 PL_op = (OP *)&myop;
871 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX); 884 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
872 SPAGAIN;
873 } 885 }
874 886
875 /* this newly created coroutine might be run on an existing cctx which most 887 /* this newly created coroutine might be run on an existing cctx which most
876 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here. 888 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
877 */ 889 */
880 892
881 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */ 893 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
882 coro_setup_op.op_next = PL_op; 894 coro_setup_op.op_next = PL_op;
883 coro_setup_op.op_type = OP_CUSTOM; 895 coro_setup_op.op_type = OP_CUSTOM;
884 coro_setup_op.op_ppaddr = pp_slf; 896 coro_setup_op.op_ppaddr = pp_slf;
885 /* no flags required, as an init function won't be called */ 897 /* no flags etc. required, as an init function won't be called */
886 898
887 PL_op = (OP *)&coro_setup_op; 899 PL_op = (OP *)&coro_setup_op;
888 900
889 /* copy throw, in case it was set before coro_setup */ 901 /* copy throw, in case it was set before coro_setup */
890 coro_throw = coro->throw; 902 CORO_THROW = coro->except;
891} 903}
892 904
893static void 905static void
894coro_destruct (pTHX_ struct coro *coro) 906coro_destruct (pTHX_ struct coro *coro)
895{ 907{
919 931
920 SvREFCNT_dec (PL_diehook); 932 SvREFCNT_dec (PL_diehook);
921 SvREFCNT_dec (PL_warnhook); 933 SvREFCNT_dec (PL_warnhook);
922 934
923 SvREFCNT_dec (coro->saved_deffh); 935 SvREFCNT_dec (coro->saved_deffh);
924 SvREFCNT_dec (coro_throw); 936 SvREFCNT_dec (coro->rouse_cb);
925 937
926 coro_destruct_stacks (aTHX); 938 coro_destruct_stacks (aTHX);
927} 939}
928 940
929INLINE void 941INLINE void
1016 SAVETMPS; 1028 SAVETMPS;
1017 EXTEND (SP, 3); 1029 EXTEND (SP, 3);
1018 PUSHMARK (SP); 1030 PUSHMARK (SP);
1019 PUSHs (&PL_sv_yes); 1031 PUSHs (&PL_sv_yes);
1020 PUSHs (fullname); 1032 PUSHs (fullname);
1021 PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef); 1033 PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef);
1022 PUTBACK; 1034 PUTBACK;
1023 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0); 1035 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1024 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD); 1036 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1025 SPAGAIN; 1037 SPAGAIN;
1026 FREETMPS; 1038 FREETMPS;
1132 transfer_tail (aTHX); 1144 transfer_tail (aTHX);
1133 1145
1134 /* somebody or something will hit me for both perl_run and PL_restartop */ 1146 /* somebody or something will hit me for both perl_run and PL_restartop */
1135 PL_restartop = PL_op; 1147 PL_restartop = PL_op;
1136 perl_run (PL_curinterp); 1148 perl_run (PL_curinterp);
1149 /*
1150 * Unfortunately, there is no way to get at the return values of the
1151 * coro body here, as perl_run destroys these
1152 */
1137 1153
1138 /* 1154 /*
1139 * If perl-run returns we assume exit() was being called or the coro 1155 * If perl-run returns we assume exit() was being called or the coro
1140 * fell off the end, which seems to be the only valid (non-bug) 1156 * fell off the end, which seems to be the only valid (non-bug)
1141 * reason for perl_run to return. We try to exit by jumping to the 1157 * reason for perl_run to return. We try to exit by jumping to the
1433 1449
1434 coro->slot = 0; 1450 coro->slot = 0;
1435 } 1451 }
1436 1452
1437 cctx_destroy (coro->cctx); 1453 cctx_destroy (coro->cctx);
1454 SvREFCNT_dec (coro->startcv);
1438 SvREFCNT_dec (coro->args); 1455 SvREFCNT_dec (coro->args);
1456 SvREFCNT_dec (CORO_THROW);
1439 1457
1440 if (coro->next) coro->next->prev = coro->prev; 1458 if (coro->next) coro->next->prev = coro->prev;
1441 if (coro->prev) coro->prev->next = coro->next; 1459 if (coro->prev) coro->prev->next = coro->next;
1442 if (coro == coro_first) coro_first = coro->next; 1460 if (coro == coro_first) coro_first = coro->next;
1443 1461
1497 1515
1498 prepare_transfer (aTHX_ &ta, prev_sv, next_sv); 1516 prepare_transfer (aTHX_ &ta, prev_sv, next_sv);
1499 TRANSFER (ta, 1); 1517 TRANSFER (ta, 1);
1500} 1518}
1501 1519
1520/*****************************************************************************/
1521/* gensub: simple closure generation utility */
1522
1523#define GENSUB_ARG CvXSUBANY (cv).any_ptr
1524
1525/* create a closure from XS, returns a code reference */
1526/* the arg can be accessed via GENSUB_ARG from the callback */
1527/* the callback must use dXSARGS/XSRETURN */
1528static SV *
1529gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
1530{
1531 CV *cv = (CV *)newSV (0);
1532
1533 sv_upgrade ((SV *)cv, SVt_PVCV);
1534
1535 CvANON_on (cv);
1536 CvISXSUB_on (cv);
1537 CvXSUB (cv) = xsub;
1538 GENSUB_ARG = arg;
1539
1540 return newRV_noinc ((SV *)cv);
1541}
1542
1502/** Coro ********************************************************************/ 1543/** Coro ********************************************************************/
1503 1544
1504INLINE void 1545INLINE void
1505coro_enq (pTHX_ struct coro *coro) 1546coro_enq (pTHX_ struct coro *coro)
1506{ 1547{
1703 if (coro->flags & CF_RUNNING) 1744 if (coro->flags & CF_RUNNING)
1704 PL_runops = RUNOPS_DEFAULT; 1745 PL_runops = RUNOPS_DEFAULT;
1705 else 1746 else
1706 coro->slot->runops = RUNOPS_DEFAULT; 1747 coro->slot->runops = RUNOPS_DEFAULT;
1707 } 1748 }
1749}
1750
1751/*****************************************************************************/
1752/* rouse callback */
1753
1754#define CORO_MAGIC_type_rouse PERL_MAGIC_ext
1755
1756static void
1757coro_rouse_callback (pTHX_ CV *cv)
1758{
1759 dXSARGS;
1760 SV *data = (SV *)GENSUB_ARG;
1761
1762 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1763 {
1764 /* first call, set args */
1765 int i;
1766 AV *av = newAV ();
1767 SV *coro = SvRV (data);
1768
1769 SvRV_set (data, (SV *)av);
1770 api_ready (aTHX_ coro);
1771 SvREFCNT_dec (coro);
1772
1773 /* better take a full copy of the arguments */
1774 while (items--)
1775 av_store (av, items, newSVsv (ST (items)));
1776 }
1777
1778 XSRETURN_EMPTY;
1779}
1780
1781static int
1782slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
1783{
1784 SV *data = (SV *)frame->data;
1785
1786 if (CORO_THROW)
1787 return 0;
1788
1789 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1790 return 1;
1791
1792 /* now push all results on the stack */
1793 {
1794 dSP;
1795 AV *av = (AV *)SvRV (data);
1796 int i;
1797
1798 EXTEND (SP, AvFILLp (av) + 1);
1799 for (i = 0; i <= AvFILLp (av); ++i)
1800 PUSHs (sv_2mortal (AvARRAY (av)[i]));
1801
1802 /* we have stolen the elements, so ste length to zero and free */
1803 AvFILLp (av) = -1;
1804 av_undef (av);
1805
1806 PUTBACK;
1807 }
1808
1809 return 0;
1810}
1811
1812static void
1813slf_init_rouse_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1814{
1815 SV *cb;
1816
1817 if (items)
1818 cb = arg [0];
1819 else
1820 {
1821 struct coro *coro = SvSTATE_current;
1822
1823 if (!coro->rouse_cb)
1824 croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
1825
1826 cb = sv_2mortal (coro->rouse_cb);
1827 coro->rouse_cb = 0;
1828 }
1829
1830 if (!SvROK (cb)
1831 || SvTYPE (SvRV (cb)) != SVt_PVCV
1832 || CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
1833 croak ("Coro::rouse_wait called with illegal callback argument,");
1834
1835 {
1836 CV *cv = (CV *)SvRV (cb); /* for GENSUB_ARG */
1837 SV *data = (SV *)GENSUB_ARG;
1838
1839 frame->data = (void *)data;
1840 frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
1841 frame->check = slf_check_rouse_wait;
1842 }
1843}
1844
1845static SV *
1846coro_new_rouse_cb (pTHX)
1847{
1848 HV *hv = (HV *)SvRV (coro_current);
1849 struct coro *coro = SvSTATE_hv (hv);
1850 SV *data = newRV_inc ((SV *)hv);
1851 SV *cb = gensub (aTHX_ coro_rouse_callback, (void *)data);
1852
1853 sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
1854 SvREFCNT_dec (data); /* magicext increases the refcount */
1855
1856 SvREFCNT_dec (coro->rouse_cb);
1857 coro->rouse_cb = SvREFCNT_inc_NN (cb);
1858
1859 return cb;
1708} 1860}
1709 1861
1710/*****************************************************************************/ 1862/*****************************************************************************/
1711/* schedule-like-function opcode (SLF) */ 1863/* schedule-like-function opcode (SLF) */
1712 1864
1839 } 1991 }
1840 while (slf_frame.check (aTHX_ &slf_frame)); 1992 while (slf_frame.check (aTHX_ &slf_frame));
1841 1993
1842 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */ 1994 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
1843 1995
1996 /* exception handling */
1997 if (expect_false (CORO_THROW))
1998 {
1999 SV *exception = sv_2mortal (CORO_THROW);
2000
2001 CORO_THROW = 0;
2002 sv_setsv (ERRSV, exception);
2003 croak (0);
2004 }
2005
1844 /* return value handling - mostly like entersub */ 2006 /* return value handling - mostly like entersub */
1845 /* make sure we put something on the stack in scalar context */ 2007 /* make sure we put something on the stack in scalar context */
1846 if (GIMME_V == G_SCALAR) 2008 if (GIMME_V == G_SCALAR)
1847 { 2009 {
1848 dSP; 2010 dSP;
1854 bot [1] = *sp; 2016 bot [1] = *sp;
1855 2017
1856 SP = bot + 1; 2018 SP = bot + 1;
1857 2019
1858 PUTBACK; 2020 PUTBACK;
1859 }
1860
1861 /* exception handling */
1862 if (expect_false (coro_throw))
1863 {
1864 SV *exception = sv_2mortal (coro_throw);
1865
1866 coro_throw = 0;
1867 sv_setsv (ERRSV, exception);
1868 croak (0);
1869 } 2021 }
1870 2022
1871 return NORMAL; 2023 return NORMAL;
1872} 2024}
1873 2025
1995 PerlIOBuf_get_cnt, 2147 PerlIOBuf_get_cnt,
1996 PerlIOBuf_set_ptrcnt, 2148 PerlIOBuf_set_ptrcnt,
1997}; 2149};
1998 2150
1999/*****************************************************************************/ 2151/*****************************************************************************/
2152/* Coro::Semaphore & Coro::Signal */
2153
2154static SV *
2155coro_waitarray_new (pTHX_ int count)
2156{
2157 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2158 AV *av = newAV ();
2159 SV **ary;
2160
2161 /* unfortunately, building manually saves memory */
2162 Newx (ary, 2, SV *);
2163 AvALLOC (av) = ary;
2164 /*AvARRAY (av) = ary;*/
2165 SvPVX ((SV *)av) = (char *)ary; /* 5.8.8 needs this syntax instead of AvARRAY = ary */
2166 AvMAX (av) = 1;
2167 AvFILLp (av) = 0;
2168 ary [0] = newSViv (count);
2169
2170 return newRV_noinc ((SV *)av);
2171}
2172
2000/* Coro::Semaphore */ 2173/* semaphore */
2001 2174
2002static void 2175static void
2003coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2176coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2004{ 2177{
2005 SV *count_sv = AvARRAY (av)[0]; 2178 SV *count_sv = AvARRAY (av)[0];
2017 AvARRAY (av)[0] = AvARRAY (av)[1]; 2190 AvARRAY (av)[0] = AvARRAY (av)[1];
2018 AvARRAY (av)[1] = count_sv; 2191 AvARRAY (av)[1] = count_sv;
2019 cb = av_shift (av); 2192 cb = av_shift (av);
2020 2193
2021 if (SvOBJECT (cb)) 2194 if (SvOBJECT (cb))
2195 {
2022 api_ready (aTHX_ cb); 2196 api_ready (aTHX_ cb);
2023 else 2197 --count;
2024 croak ("callbacks not yet supported"); 2198 }
2199 else if (SvTYPE (cb) == SVt_PVCV)
2200 {
2201 dSP;
2202 PUSHMARK (SP);
2203 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2204 PUTBACK;
2205 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2206 }
2025 2207
2026 SvREFCNT_dec (cb); 2208 SvREFCNT_dec (cb);
2027
2028 --count;
2029 } 2209 }
2030} 2210}
2031 2211
2032static void 2212static void
2033coro_semaphore_on_destroy (pTHX_ struct coro *coro) 2213coro_semaphore_on_destroy (pTHX_ struct coro *coro)
2035 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2215 /* call $sem->adjust (0) to possibly wake up some other waiters */
2036 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2216 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2037} 2217}
2038 2218
2039static int 2219static int
2040slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2220slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2041{ 2221{
2042 AV *av = (AV *)frame->data; 2222 AV *av = (AV *)frame->data;
2043 SV *count_sv = AvARRAY (av)[0]; 2223 SV *count_sv = AvARRAY (av)[0];
2044 2224
2045 /* if we are about to throw, don't actually acquire the lock, just throw */ 2225 /* if we are about to throw, don't actually acquire the lock, just throw */
2046 if (coro_throw && 0)//D 2226 if (CORO_THROW)
2047 return 0; 2227 return 0;
2048 else if (SvIVX (count_sv) > 0) 2228 else if (SvIVX (count_sv) > 0)
2049 { 2229 {
2050 SvSTATE_current->on_destroy = 0; 2230 SvSTATE_current->on_destroy = 0;
2231
2232 if (acquire)
2051 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2233 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2234 else
2235 coro_semaphore_adjust (aTHX_ av, 0);
2236
2052 return 0; 2237 return 0;
2053 } 2238 }
2054 else 2239 else
2055 { 2240 {
2056 int i; 2241 int i;
2065 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2250 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2066 return 1; 2251 return 1;
2067 } 2252 }
2068} 2253}
2069 2254
2070static void 2255static int
2256slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2257{
2258 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2259}
2260
2261static int
2262slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2263{
2264 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2265}
2266
2267static void
2071slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2268slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2072{ 2269{
2073 AV *av = (AV *)SvRV (arg [0]); 2270 AV *av = (AV *)SvRV (arg [0]);
2074 2271
2075 if (SvIVX (AvARRAY (av)[0]) > 0) 2272 if (SvIVX (AvARRAY (av)[0]) > 0)
2076 { 2273 {
2086 2283
2087 /* to avoid race conditions when a woken-up coro gets terminated */ 2284 /* to avoid race conditions when a woken-up coro gets terminated */
2088 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2285 /* we arrange for a temporary on_destroy that calls adjust (0) */
2089 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2286 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2090 } 2287 }
2288}
2091 2289
2290static void
2291slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2292{
2293 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2092 frame->check = slf_check_semaphore_down; 2294 frame->check = slf_check_semaphore_down;
2093
2094} 2295}
2095 2296
2096/*****************************************************************************/ 2297static void
2097/* gensub: simple closure generation utility */ 2298slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2098
2099#define GENSUB_ARG CvXSUBANY (cv).any_ptr
2100
2101/* create a closure from XS, returns a code reference */
2102/* the arg can be accessed via GENSUB_ARG from the callback */
2103/* the callback must use dXSARGS/XSRETURN */
2104static SV *
2105gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
2106{ 2299{
2107 CV *cv = (CV *)newSV (0); 2300 if (items >= 2)
2301 {
2302 /* callback form */
2303 AV *av = (AV *)SvRV (arg [0]);
2304 CV *cb_cv = coro_sv_2cv (arg [1]);
2108 2305
2109 sv_upgrade ((SV *)cv, SVt_PVCV); 2306 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
2110 2307
2111 CvANON_on (cv); 2308 if (SvIVX (AvARRAY (av)[0]) > 0)
2112 CvISXSUB_on (cv); 2309 coro_semaphore_adjust (aTHX_ av, 0);
2113 CvXSUB (cv) = xsub;
2114 GENSUB_ARG = arg;
2115 2310
2116 return newRV_noinc ((SV *)cv); 2311 frame->prepare = prepare_nop;
2312 frame->check = slf_check_nop;
2313 }
2314 else
2315 {
2316 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2317 frame->check = slf_check_semaphore_wait;
2318 }
2319}
2320
2321/* signal */
2322
2323static void
2324coro_signal_wake (pTHX_ AV *av, int count)
2325{
2326 SvIVX (AvARRAY (av)[0]) = 0;
2327
2328 /* now signal count waiters */
2329 while (count > 0 && AvFILLp (av) > 0)
2330 {
2331 SV *cb;
2332
2333 /* swap first two elements so we can shift a waiter */
2334 cb = AvARRAY (av)[0];
2335 AvARRAY (av)[0] = AvARRAY (av)[1];
2336 AvARRAY (av)[1] = cb;
2337
2338 cb = av_shift (av);
2339
2340 api_ready (aTHX_ cb);
2341 sv_setiv (cb, 0); /* signal waiter */
2342 SvREFCNT_dec (cb);
2343
2344 --count;
2345 }
2346}
2347
2348static int
2349slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2350{
2351 /* if we are about to throw, also stop waiting */
2352 return SvROK ((SV *)frame->data) && !CORO_THROW;
2353}
2354
2355static void
2356slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2357{
2358 AV *av = (AV *)SvRV (arg [0]);
2359
2360 if (SvIVX (AvARRAY (av)[0]))
2361 {
2362 SvIVX (AvARRAY (av)[0]) = 0;
2363 frame->prepare = prepare_nop;
2364 frame->check = slf_check_nop;
2365 }
2366 else
2367 {
2368 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2369
2370 av_push (av, waiter);
2371
2372 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2373 frame->prepare = prepare_schedule;
2374 frame->check = slf_check_signal_wait;
2375 }
2117} 2376}
2118 2377
2119/*****************************************************************************/ 2378/*****************************************************************************/
2120/* Coro::AIO */ 2379/* Coro::AIO */
2121 2380
2170 2429
2171static int 2430static int
2172slf_check_aio_req (pTHX_ struct CoroSLF *frame) 2431slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2173{ 2432{
2174 AV *state = (AV *)frame->data; 2433 AV *state = (AV *)frame->data;
2434
2435 /* if we are about to throw, return early */
2436 /* this does not cancel the aio request, but at least */
2437 /* it quickly returns */
2438 if (CORO_THROW)
2439 return 0;
2175 2440
2176 /* one element that is an RV? repeat! */ 2441 /* one element that is an RV? repeat! */
2177 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2442 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2178 return 1; 2443 return 1;
2179 2444
2350 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL)); 2615 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL));
2351} 2616}
2352 2617
2353SV * 2618SV *
2354new (char *klass, ...) 2619new (char *klass, ...)
2620 ALIAS:
2621 Coro::new = 1
2355 CODE: 2622 CODE:
2356{ 2623{
2357 struct coro *coro; 2624 struct coro *coro;
2358 MAGIC *mg; 2625 MAGIC *mg;
2359 HV *hv; 2626 HV *hv;
2627 CV *cb;
2360 int i; 2628 int i;
2629
2630 if (items > 1)
2631 {
2632 cb = coro_sv_2cv (ST (1));
2633
2634 if (CvISXSUB (cb))
2635 croak ("Coro::State doesn't support XS functions as coroutine start, caught");
2636
2637 if (!CvROOT (cb))
2638 croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught");
2639 }
2361 2640
2362 Newz (0, coro, 1, struct coro); 2641 Newz (0, coro, 1, struct coro);
2363 coro->args = newAV (); 2642 coro->args = newAV ();
2364 coro->flags = CF_NEW; 2643 coro->flags = CF_NEW;
2365 2644
2370 coro->hv = hv = newHV (); 2649 coro->hv = hv = newHV ();
2371 mg = sv_magicext ((SV *)hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)coro, 0); 2650 mg = sv_magicext ((SV *)hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)coro, 0);
2372 mg->mg_flags |= MGf_DUP; 2651 mg->mg_flags |= MGf_DUP;
2373 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1)); 2652 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));
2374 2653
2654 if (items > 1)
2655 {
2375 av_extend (coro->args, items - 1); 2656 av_extend (coro->args, items - 1 + ix);
2657
2658 if (ix)
2659 {
2660 av_push (coro->args, SvREFCNT_inc_NN ((SV *)cb));
2661 cb = coro_run_cv;
2662 }
2663
2664 coro->startcv = (CV *)SvREFCNT_inc_NN ((SV *)cb);
2665
2376 for (i = 1; i < items; i++) 2666 for (i = 2; i < items; i++)
2377 av_push (coro->args, newSVsv (ST (i))); 2667 av_push (coro->args, newSVsv (ST (i)));
2668 }
2378} 2669}
2379 OUTPUT: 2670 OUTPUT:
2380 RETVAL 2671 RETVAL
2381 2672
2382void 2673void
2511throw (Coro::State self, SV *throw = &PL_sv_undef) 2802throw (Coro::State self, SV *throw = &PL_sv_undef)
2512 PROTOTYPE: $;$ 2803 PROTOTYPE: $;$
2513 CODE: 2804 CODE:
2514{ 2805{
2515 struct coro *current = SvSTATE_current; 2806 struct coro *current = SvSTATE_current;
2516 SV **throwp = self == current ? &coro_throw : &self->throw; 2807 SV **throwp = self == current ? &CORO_THROW : &self->except;
2517 SvREFCNT_dec (*throwp); 2808 SvREFCNT_dec (*throwp);
2518 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2809 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2519} 2810}
2520 2811
2521void 2812void
2583 int i; 2874 int i;
2584 2875
2585 av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE); 2876 av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE);
2586 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE); 2877 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
2587 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE); 2878 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
2588 2879 coro_run_cv = get_cv (aTHX_ "Coro::_terminate", GV_ADD);
2589 coro_current = coro_get_sv (aTHX_ "Coro::current", FALSE); 2880 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE);
2590 SvREADONLY_on (coro_current); 2881 SvREADONLY_on (coro_current);
2591 2882
2592 coro_stash = gv_stashpv ("Coro", TRUE); 2883 coro_stash = gv_stashpv ("Coro", TRUE);
2593 2884
2594 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX)); 2885 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX));
2610 coroapi.ready = api_ready; 2901 coroapi.ready = api_ready;
2611 coroapi.is_ready = api_is_ready; 2902 coroapi.is_ready = api_is_ready;
2612 coroapi.nready = coro_nready; 2903 coroapi.nready = coro_nready;
2613 coroapi.current = coro_current; 2904 coroapi.current = coro_current;
2614 2905
2615 GCoroAPI = &coroapi; 2906 /*GCoroAPI = &coroapi;*/
2616 sv_setiv (sv, (IV)&coroapi); 2907 sv_setiv (sv, (IV)&coroapi);
2617 SvREADONLY_on (sv); 2908 SvREADONLY_on (sv);
2618 } 2909 }
2619} 2910}
2620 2911
2756 api_trace (aTHX_ coro_current, 0); 3047 api_trace (aTHX_ coro_current, 0);
2757 3048
2758 av_push (av_async_pool, newSVsv (coro_current)); 3049 av_push (av_async_pool, newSVsv (coro_current));
2759} 3050}
2760 3051
3052SV *
3053rouse_cb ()
3054 PROTOTYPE:
3055 CODE:
3056 RETVAL = coro_new_rouse_cb (aTHX);
3057 OUTPUT:
3058 RETVAL
3059
3060void
3061rouse_wait (...)
3062 PROTOTYPE: ;$
3063 PPCODE:
3064 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
3065
2761 3066
2762MODULE = Coro::State PACKAGE = PerlIO::cede 3067MODULE = Coro::State PACKAGE = PerlIO::cede
2763 3068
2764BOOT: 3069BOOT:
2765 PerlIO_define_layer (aTHX_ &PerlIO_cede); 3070 PerlIO_define_layer (aTHX_ &PerlIO_cede);
2766 3071
2767 3072
2768MODULE = Coro::State PACKAGE = Coro::Semaphore 3073MODULE = Coro::State PACKAGE = Coro::Semaphore
2769 3074
2770SV * 3075SV *
2771new (SV *klass, SV *count_ = 0) 3076new (SV *klass, SV *count = 0)
2772 CODE: 3077 CODE:
2773{ 3078 RETVAL = sv_bless (
2774 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 3079 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2775 AV *av = newAV (); 3080 GvSTASH (CvGV (cv))
2776 SV **ary; 3081 );
3082 OUTPUT:
3083 RETVAL
2777 3084
2778 /* unfortunately, building manually saves memory */ 3085# helper for Coro::Channel
2779 Newx (ary, 2, SV *); 3086SV *
2780 AvALLOC (av) = ary; 3087_alloc (int count)
2781 AvARRAY (av) = ary; 3088 CODE:
2782 AvMAX (av) = 1; 3089 RETVAL = coro_waitarray_new (aTHX_ count);
2783 AvFILLp (av) = 0;
2784 ary [0] = newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1);
2785
2786 RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv)));
2787}
2788 OUTPUT: 3090 OUTPUT:
2789 RETVAL 3091 RETVAL
2790 3092
2791SV * 3093SV *
2792count (SV *self) 3094count (SV *self)
2801 adjust = 1 3103 adjust = 1
2802 CODE: 3104 CODE:
2803 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1); 3105 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
2804 3106
2805void 3107void
2806down (SV *self) 3108down (...)
2807 CODE: 3109 CODE:
2808 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 3110 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3111
3112void
3113wait (...)
3114 CODE:
3115 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2809 3116
2810void 3117void
2811try (SV *self) 3118try (SV *self)
2812 PPCODE: 3119 PPCODE:
2813{ 3120{
2825 XSRETURN_NO; 3132 XSRETURN_NO;
2826} 3133}
2827 3134
2828void 3135void
2829waiters (SV *self) 3136waiters (SV *self)
2830 CODE: 3137 PPCODE:
2831{ 3138{
2832 AV *av = (AV *)SvRV (self); 3139 AV *av = (AV *)SvRV (self);
3140 int wcount = AvFILLp (av) + 1 - 1;
2833 3141
2834 if (GIMME_V == G_SCALAR) 3142 if (GIMME_V == G_SCALAR)
2835 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 3143 XPUSHs (sv_2mortal (newSViv (wcount)));
2836 else 3144 else
2837 { 3145 {
2838 int i; 3146 int i;
2839 EXTEND (SP, AvFILLp (av) + 1 - 1); 3147 EXTEND (SP, wcount);
2840 for (i = 1; i <= AvFILLp (av); ++i) 3148 for (i = 1; i <= wcount; ++i)
2841 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 3149 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2842 } 3150 }
2843} 3151}
3152
3153MODULE = Coro::State PACKAGE = Coro::Signal
3154
3155SV *
3156new (SV *klass)
3157 CODE:
3158 RETVAL = sv_bless (
3159 coro_waitarray_new (aTHX_ 0),
3160 GvSTASH (CvGV (cv))
3161 );
3162 OUTPUT:
3163 RETVAL
3164
3165void
3166wait (...)
3167 CODE:
3168 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3169
3170void
3171broadcast (SV *self)
3172 CODE:
3173{
3174 AV *av = (AV *)SvRV (self);
3175 coro_signal_wake (aTHX_ av, AvFILLp (av));
3176}
3177
3178void
3179send (SV *self)
3180 CODE:
3181{
3182 AV *av = (AV *)SvRV (self);
3183
3184 if (AvFILLp (av))
3185 coro_signal_wake (aTHX_ av, 1);
3186 else
3187 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
3188}
3189
3190IV
3191awaited (SV *self)
3192 CODE:
3193 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3194 OUTPUT:
3195 RETVAL
2844 3196
2845 3197
2846MODULE = Coro::State PACKAGE = Coro::AnyEvent 3198MODULE = Coro::State PACKAGE = Coro::AnyEvent
2847 3199
2848BOOT: 3200BOOT:
2876 3228
2877void 3229void
2878_register (char *target, char *proto, SV *req) 3230_register (char *target, char *proto, SV *req)
2879 CODE: 3231 CODE:
2880{ 3232{
2881 HV *st;
2882 GV *gvp;
2883 CV *req_cv = sv_2cv (req, &st, &gvp, 0); 3233 CV *req_cv = coro_sv_2cv (req);
2884 /* newXSproto doesn't return the CV on 5.8 */ 3234 /* newXSproto doesn't return the CV on 5.8 */
2885 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__); 3235 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
2886 sv_setpv ((SV *)slf_cv, proto); 3236 sv_setpv ((SV *)slf_cv, proto);
2887 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0); 3237 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
2888} 3238}

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines