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.287 by root, Mon Nov 17 07:03:12 2008 UTC vs.
Revision 1.301 by root, Wed Nov 19 02:41:31 2008 UTC

… …
95# define GV_NOTQUAL 0 95# define GV_NOTQUAL 0
96#endif 96#endif
97#ifndef newSV 97#ifndef newSV
98# define newSV(l) NEWSV(0,l) 98# define newSV(l) NEWSV(0,l)
99#endif 99#endif
100#ifndef CvISXSUB_on
101# define CvISXSUB_on(cv) (void)cv
102#endif
100 103
101/* 5.8.7 */ 104/* 5.8.7 */
102#ifndef SvRV_set 105#ifndef SvRV_set
103# define SvRV_set(s,v) SvRV(s) = (v) 106# define SvRV_set(s,v) SvRV(s) = (v)
104#endif 107#endif
… …
117#endif 120#endif
118 121
119/* The next macros try to return the current stack pointer, in an as 122/* The next macros try to return the current stack pointer, in an as
120 * portable way as possible. */ 123 * portable way as possible. */
121#if __GNUC__ >= 4 124#if __GNUC__ >= 4
125# define dSTACKLEVEL int stacklevel_dummy
122# define dSTACKLEVEL void *stacklevel = __builtin_frame_address (0) 126# define STACKLEVEL __builtin_frame_address (0)
123#else 127#else
124# define dSTACKLEVEL volatile void *stacklevel = (volatile void *)&stacklevel 128# define dSTACKLEVEL volatile void *stacklevel
129# define STACKLEVEL ((void *)&stacklevel)
125#endif 130#endif
126 131
127#define IN_DESTRUCT (PL_main_cv == Nullcv) 132#define IN_DESTRUCT (PL_main_cv == Nullcv)
128 133
129#if __GNUC__ >= 3 134#if __GNUC__ >= 3
… …
140#define expect_true(expr) expect ((expr) != 0, 1) 145#define expect_true(expr) expect ((expr) != 0, 1)
141 146
142#define NOINLINE attribute ((noinline)) 147#define NOINLINE attribute ((noinline))
143 148
144#include "CoroAPI.h" 149#include "CoroAPI.h"
150#define GCoroAPI (&coroapi) /* very sneaky */
145 151
146#ifdef USE_ITHREADS 152#ifdef USE_ITHREADS
147# if CORO_PTHREAD 153# if CORO_PTHREAD
148static void *coro_thx; 154static void *coro_thx;
149# endif 155# endif
150#endif 156#endif
151 157
152static double (*nvtime)(); /* so why doesn't it take void? */ 158static double (*nvtime)(); /* so why doesn't it take void? */
159
160/* we hijack an hopefully unused CV flag for our purposes */
161#define CVf_SLF 0x4000
162static OP *pp_slf (pTHX);
153 163
154static U32 cctx_gen; 164static U32 cctx_gen;
155static size_t cctx_stacksize = CORO_STACKSIZE; 165static size_t cctx_stacksize = CORO_STACKSIZE;
156static struct CoroAPI coroapi; 166static struct CoroAPI coroapi;
157static AV *main_mainstack; /* used to differentiate between $main and others */ 167static AV *main_mainstack; /* used to differentiate between $main and others */
… …
247 /* statistics */ 257 /* statistics */
248 int usecount; /* number of transfers to this coro */ 258 int usecount; /* number of transfers to this coro */
249 259
250 /* coro process data */ 260 /* coro process data */
251 int prio; 261 int prio;
252 SV *throw; /* exception to be thrown */ 262 SV *except; /* exception to be thrown */
253 263
254 /* async_pool */ 264 /* async_pool */
255 SV *saved_deffh; 265 SV *saved_deffh;
256 266
257 /* linked list */ 267 /* linked list */
… …
263 273
264/* the following variables are effectively part of the perl context */ 274/* the following variables are effectively part of the perl context */
265/* and get copied between struct coro and these variables */ 275/* and get copied between struct coro and these variables */
266/* the mainr easonw e don't support windows process emulation */ 276/* the mainr easonw e don't support windows process emulation */
267static struct CoroSLF slf_frame; /* the current slf frame */ 277static struct CoroSLF slf_frame; /* the current slf frame */
268static SV *coro_throw;
269 278
270/** Coro ********************************************************************/ 279/** Coro ********************************************************************/
271 280
272#define PRIO_MAX 3 281#define PRIO_MAX 3
273#define PRIO_HIGH 1 282#define PRIO_HIGH 1
… …
502 511
503 PUTBACK; 512 PUTBACK;
504 } 513 }
505 514
506 slf_frame = c->slf_frame; 515 slf_frame = c->slf_frame;
507 coro_throw = c->throw; 516 CORO_THROW = c->except;
508} 517}
509 518
510static void 519static void
511save_perl (pTHX_ Coro__State c) 520save_perl (pTHX_ Coro__State c)
512{ 521{
513 c->throw = coro_throw; 522 c->except = CORO_THROW;
514 c->slf_frame = slf_frame; 523 c->slf_frame = slf_frame;
515 524
516 { 525 {
517 dSP; 526 dSP;
518 I32 cxix = cxstack_ix; 527 I32 cxix = cxstack_ix;
… …
811slf_check_nop (pTHX_ struct CoroSLF *frame) 820slf_check_nop (pTHX_ struct CoroSLF *frame)
812{ 821{
813 return 0; 822 return 0;
814} 823}
815 824
825static UNOP coro_setup_op;
826
816static void NOINLINE /* noinline to keep it out of the transfer fast path */ 827static void NOINLINE /* noinline to keep it out of the transfer fast path */
817coro_setup (pTHX_ struct coro *coro) 828coro_setup (pTHX_ struct coro *coro)
818{ 829{
819 /* 830 /*
820 * emulate part of the perl startup here. 831 * emulate part of the perl startup here.
… …
848 { 859 {
849 dSP; 860 dSP;
850 UNOP myop; 861 UNOP myop;
851 862
852 Zero (&myop, 1, UNOP); 863 Zero (&myop, 1, UNOP);
853 myop.op_next = Nullop; 864 myop.op_next = Nullop;
854 myop.op_flags = OPf_WANT_VOID; 865 myop.op_flags = OPf_WANT_VOID;
855 866
856 PUSHMARK (SP); 867 PUSHMARK (SP);
857 XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv)))); 868 XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv))));
858 PUTBACK; 869 PUTBACK;
… …
865 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here. 876 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
866 */ 877 */
867 slf_frame.prepare = prepare_nop; /* provide a nop function for an eventual pp_slf */ 878 slf_frame.prepare = prepare_nop; /* provide a nop function for an eventual pp_slf */
868 slf_frame.check = slf_check_nop; /* signal pp_slf to not repeat */ 879 slf_frame.check = slf_check_nop; /* signal pp_slf to not repeat */
869 880
870 coro_throw = coro->throw; 881 /* 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;
883 coro_setup_op.op_type = OP_CUSTOM;
884 coro_setup_op.op_ppaddr = pp_slf;
885 /* no flags required, as an init function won't be called */
886
887 PL_op = (OP *)&coro_setup_op;
888
889 /* copy throw, in case it was set before coro_setup */
890 CORO_THROW = coro->except;
871} 891}
872 892
873static void 893static void
874coro_destruct (pTHX_ struct coro *coro) 894coro_destruct (pTHX_ struct coro *coro)
875{ 895{
… …
899 919
900 SvREFCNT_dec (PL_diehook); 920 SvREFCNT_dec (PL_diehook);
901 SvREFCNT_dec (PL_warnhook); 921 SvREFCNT_dec (PL_warnhook);
902 922
903 SvREFCNT_dec (coro->saved_deffh); 923 SvREFCNT_dec (coro->saved_deffh);
904 SvREFCNT_dec (coro_throw); 924 SvREFCNT_dec (CORO_THROW);
905 925
906 coro_destruct_stacks (aTHX); 926 coro_destruct_stacks (aTHX);
907} 927}
908 928
909INLINE void 929INLINE void
… …
1038 1058
1039 TAINT_NOT; 1059 TAINT_NOT;
1040 return 0; 1060 return 0;
1041} 1061}
1042 1062
1063static struct coro_cctx *cctx_ssl_cctx;
1064static struct CoroSLF cctx_ssl_frame;
1065
1043static void 1066static void
1044prepare_set_stacklevel (struct coro_transfer_args *ta, struct coro_cctx *cctx) 1067slf_prepare_set_stacklevel (pTHX_ struct coro_transfer_args *ta)
1045{ 1068{
1046 ta->prev = (struct coro *)cctx; 1069 ta->prev = (struct coro *)cctx_ssl_cctx;
1047 ta->next = 0; 1070 ta->next = 0;
1048} 1071}
1049 1072
1050/* inject a fake call to Coro::State::_cctx_init into the execution */ 1073static int
1051/* _cctx_init should be careful, as it could be called at almost any time */ 1074slf_check_set_stacklevel (pTHX_ struct CoroSLF *frame)
1052/* during execution of a perl program */ 1075{
1053/* also initialises PL_top_env */ 1076 *frame = cctx_ssl_frame;
1077
1078 return frame->check (aTHX_ frame); /* execute the restored frame - there must be one */
1079}
1080
1081/* initialises PL_top_env and injects a pseudo-slf-call to set the stacklevel */
1054static void NOINLINE 1082static void NOINLINE
1055cctx_prepare (pTHX_ coro_cctx *cctx) 1083cctx_prepare (pTHX_ coro_cctx *cctx)
1056{ 1084{
1057 dSP;
1058 UNOP myop;
1059
1060 PL_top_env = &PL_start_env; 1085 PL_top_env = &PL_start_env;
1061 1086
1062 if (cctx->flags & CC_TRACE) 1087 if (cctx->flags & CC_TRACE)
1063 PL_runops = runops_trace; 1088 PL_runops = runops_trace;
1064 1089
1065 Zero (&myop, 1, UNOP); 1090 /* we already must be executing an SLF op, there is no other valid way
1066 myop.op_next = PL_op; 1091 * that can lead to creation of a new cctx */
1067 myop.op_flags = OPf_WANT_VOID | OPf_STACKED; 1092 assert (("FATAL: can't prepare slf-less cctx in Coro module (please report)",
1093 slf_frame.prepare && PL_op->op_ppaddr == pp_slf));
1068 1094
1069 PUSHMARK (SP); 1095 /* we must emulate leaving pp_slf, which is done inside slf_check_set_stacklevel */
1070 EXTEND (SP, 2); 1096 cctx_ssl_cctx = cctx;
1071 PUSHs (sv_2mortal (newSViv ((IV)cctx))); 1097 cctx_ssl_frame = slf_frame;
1072 PUSHs ((SV *)get_cv ("Coro::State::_cctx_init", FALSE)); 1098
1073 PUTBACK; 1099 slf_frame.prepare = slf_prepare_set_stacklevel;
1074 PL_op = (OP *)&myop; 1100 slf_frame.check = slf_check_set_stacklevel;
1075 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
1076 SPAGAIN;
1077} 1101}
1078 1102
1079/* the tail of transfer: execute stuff we can only do after a transfer */ 1103/* the tail of transfer: execute stuff we can only do after a transfer */
1080INLINE void 1104INLINE void
1081transfer_tail (pTHX) 1105transfer_tail (pTHX)
… …
1297 dSTACKLEVEL; 1321 dSTACKLEVEL;
1298 1322
1299 /* sometimes transfer is only called to set idle_sp */ 1323 /* sometimes transfer is only called to set idle_sp */
1300 if (expect_false (!next)) 1324 if (expect_false (!next))
1301 { 1325 {
1302 ((coro_cctx *)prev)->idle_sp = (void *)stacklevel; 1326 ((coro_cctx *)prev)->idle_sp = STACKLEVEL;
1303 assert (((coro_cctx *)prev)->idle_te = PL_top_env); /* just for the side-effect when asserts are enabled */ 1327 assert (((coro_cctx *)prev)->idle_te = PL_top_env); /* just for the side-effect when asserts are enabled */
1304 } 1328 }
1305 else if (expect_true (prev != next)) 1329 else if (expect_true (prev != next))
1306 { 1330 {
1307 coro_cctx *prev__cctx; 1331 coro_cctx *prev__cctx;
… …
1332 1356
1333 prev__cctx = prev->cctx; 1357 prev__cctx = prev->cctx;
1334 1358
1335 /* possibly untie and reuse the cctx */ 1359 /* possibly untie and reuse the cctx */
1336 if (expect_true ( 1360 if (expect_true (
1337 prev__cctx->idle_sp == (void *)stacklevel 1361 prev__cctx->idle_sp == STACKLEVEL
1338 && !(prev__cctx->flags & CC_TRACE) 1362 && !(prev__cctx->flags & CC_TRACE)
1339 && !force_cctx 1363 && !force_cctx
1340 )) 1364 ))
1341 { 1365 {
1342 /* I assume that stacklevel is a stronger indicator than PL_top_env changes */ 1366 /* I assume that stacklevel is a stronger indicator than PL_top_env changes */
… …
1525 ENTER; 1549 ENTER;
1526 SAVETMPS; 1550 SAVETMPS;
1527 1551
1528 PUSHMARK (SP); 1552 PUSHMARK (SP);
1529 PUTBACK; 1553 PUTBACK;
1530 call_sv (sv_hook, G_DISCARD); 1554 call_sv (sv_hook, G_VOID | G_DISCARD);
1531 SPAGAIN;
1532 1555
1533 FREETMPS; 1556 FREETMPS;
1534 LEAVE; 1557 LEAVE;
1535 } 1558 }
1536 1559
… …
1563 ENTER; 1586 ENTER;
1564 SAVETMPS; 1587 SAVETMPS;
1565 1588
1566 PUSHMARK (SP); 1589 PUSHMARK (SP);
1567 PUTBACK; 1590 PUTBACK;
1568 call_sv (get_sv ("Coro::idle", FALSE), G_DISCARD); 1591 call_sv (get_sv ("Coro::idle", FALSE), G_VOID | G_DISCARD);
1569 SPAGAIN;
1570 1592
1571 FREETMPS; 1593 FREETMPS;
1572 LEAVE; 1594 LEAVE;
1573 continue; 1595 continue;
1574 } 1596 }
… …
1714 1736
1715 RETURNOP (slf_restore.op_first); 1737 RETURNOP (slf_restore.op_first);
1716} 1738}
1717 1739
1718static void 1740static void
1719slf_prepare_set_stacklevel (pTHX_ struct coro_transfer_args *ta)
1720{
1721 prepare_set_stacklevel (ta, (struct coro_cctx *)slf_frame.data);
1722}
1723
1724static void
1725slf_init_set_stacklevel (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1726{
1727 assert (("FATAL: set_stacklevel needs the coro cctx as sole argument", items == 1));
1728
1729 frame->prepare = slf_prepare_set_stacklevel;
1730 frame->check = slf_check_nop;
1731 frame->data = (void *)SvIV (arg [0]);
1732}
1733
1734static void
1735slf_prepare_transfer (pTHX_ struct coro_transfer_args *ta) 1741slf_prepare_transfer (pTHX_ struct coro_transfer_args *ta)
1736{ 1742{
1737 SV **arg = (SV **)slf_frame.data; 1743 SV **arg = (SV **)slf_frame.data;
1738 1744
1739 prepare_transfer (aTHX_ ta, arg [0], arg [1]); 1745 prepare_transfer (aTHX_ ta, arg [0], arg [1]);
… …
1768slf_init_cede_notself (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 1774slf_init_cede_notself (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1769{ 1775{
1770 frame->prepare = prepare_cede_notself; 1776 frame->prepare = prepare_cede_notself;
1771 frame->check = slf_check_nop; 1777 frame->check = slf_check_nop;
1772} 1778}
1773
1774/* we hijack an hopefully unused CV flag for our purposes */
1775#define CVf_SLF 0x4000
1776 1779
1777/* 1780/*
1778 * these not obviously related functions are all rolled into one 1781 * these not obviously related functions are all rolled into one
1779 * function to increase chances that they all will call transfer with the same 1782 * function to increase chances that they all will call transfer with the same
1780 * stack offset 1783 * stack offset
… …
1836 } 1839 }
1837 while (slf_frame.check (aTHX_ &slf_frame)); 1840 while (slf_frame.check (aTHX_ &slf_frame));
1838 1841
1839 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */ 1842 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
1840 1843
1841 /* return value handling - mostly like entersub */
1842 {
1843 dSP;
1844 SV **bot = PL_stack_base + checkmark;
1845 int gimme = GIMME_V;
1846
1847 /* make sure we put something on the stack in scalar context */
1848 if (gimme == G_SCALAR)
1849 {
1850 if (sp == bot)
1851 XPUSHs (&PL_sv_undef);
1852
1853 SP = bot + 1;
1854 }
1855
1856 PUTBACK;
1857 }
1858
1859 /* exception handling */ 1844 /* exception handling */
1860 if (expect_false (coro_throw)) 1845 if (expect_false (CORO_THROW))
1861 { 1846 {
1862 SV *exception = sv_2mortal (coro_throw); 1847 SV *exception = sv_2mortal (CORO_THROW);
1863 1848
1864 coro_throw = 0; 1849 CORO_THROW = 0;
1865 sv_setsv (ERRSV, exception); 1850 sv_setsv (ERRSV, exception);
1866 croak (0); 1851 croak (0);
1852 }
1853
1854 /* return value handling - mostly like entersub */
1855 /* make sure we put something on the stack in scalar context */
1856 if (GIMME_V == G_SCALAR)
1857 {
1858 dSP;
1859 SV **bot = PL_stack_base + checkmark;
1860
1861 if (sp == bot) /* too few, push undef */
1862 bot [1] = &PL_sv_undef;
1863 else if (sp != bot + 1) /* too many, take last one */
1864 bot [1] = *sp;
1865
1866 SP = bot + 1;
1867
1868 PUTBACK;
1867 } 1869 }
1868 1870
1869 return NORMAL; 1871 return NORMAL;
1870} 1872}
1871 1873
… …
1993 PerlIOBuf_get_cnt, 1995 PerlIOBuf_get_cnt,
1994 PerlIOBuf_set_ptrcnt, 1996 PerlIOBuf_set_ptrcnt,
1995}; 1997};
1996 1998
1997/*****************************************************************************/ 1999/*****************************************************************************/
2000/* Coro::Semaphore & Coro::Signal */
2001
2002static SV *
2003coro_waitarray_new (pTHX_ int count)
2004{
2005 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2006 AV *av = newAV ();
2007 SV **ary;
2008
2009 /* unfortunately, building manually saves memory */
2010 Newx (ary, 2, SV *);
2011 AvALLOC (av) = ary;
2012 AvARRAY (av) = ary;
2013 AvMAX (av) = 1;
2014 AvFILLp (av) = 0;
2015 ary [0] = newSViv (count);
2016
2017 return newRV_noinc ((SV *)av);
2018}
2019
1998/* Coro::Semaphore */ 2020/* semaphore */
1999 2021
2000static void 2022static void
2001coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2023coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2002{ 2024{
2003 SV *count_sv = AvARRAY (av)[0]; 2025 SV *count_sv = AvARRAY (av)[0];
… …
2033 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2055 /* call $sem->adjust (0) to possibly wake up some other waiters */
2034 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2056 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2035} 2057}
2036 2058
2037static int 2059static int
2038slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2060slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2039{ 2061{
2040 AV *av = (AV *)frame->data; 2062 AV *av = (AV *)frame->data;
2041 SV *count_sv = AvARRAY (av)[0]; 2063 SV *count_sv = AvARRAY (av)[0];
2042 2064
2065 /* if we are about to throw, don't actually acquire the lock, just throw */
2066 if (CORO_THROW)
2067 return 0;
2043 if (SvIVX (count_sv) > 0) 2068 else if (SvIVX (count_sv) > 0)
2044 { 2069 {
2045 SvSTATE_current->on_destroy = 0; 2070 SvSTATE_current->on_destroy = 0;
2071
2072 if (acquire)
2046 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2073 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2074 else
2075 coro_semaphore_adjust (aTHX_ av, 0);
2076
2047 return 0; 2077 return 0;
2048 } 2078 }
2049 else 2079 else
2050 { 2080 {
2051 int i; 2081 int i;
… …
2060 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2090 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2061 return 1; 2091 return 1;
2062 } 2092 }
2063} 2093}
2064 2094
2065static void 2095static int
2096slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2097{
2098 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2099}
2100
2101static int
2102slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2103{
2104 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2105}
2106
2107static void
2066slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2108slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2067{ 2109{
2068 AV *av = (AV *)SvRV (arg [0]); 2110 AV *av = (AV *)SvRV (arg [0]);
2069 2111
2070 if (SvIVX (AvARRAY (av)[0]) > 0) 2112 if (SvIVX (AvARRAY (av)[0]) > 0)
2071 { 2113 {
2072 frame->data = (void *)av; 2114 frame->data = (void *)av;
2073 frame->prepare = prepare_nop; 2115 frame->prepare = prepare_nop;
2074 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2075 } 2116 }
2076 else 2117 else
2077 { 2118 {
2078 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2119 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2079 2120
2080 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av)); 2121 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av));
2081 frame->prepare = prepare_schedule; 2122 frame->prepare = prepare_schedule;
2082 2123
2083 /* to avoid race conditions when a woken-up coro gets terminated */ 2124 /* to avoid race conditions when a woken-up coro gets terminated */
2084 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2125 /* we arrange for a temporary on_destroy that calls adjust (0) */
2085 assert (!SvSTATE_current->on_destroy);//D
2086 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2126 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2087 } 2127 }
2128}
2088 2129
2130static void
2131slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2132{
2133 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2089 frame->check = slf_check_semaphore_down; 2134 frame->check = slf_check_semaphore_down;
2135}
2090 2136
2137static void
2138slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2139{
2140 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2141 frame->check = slf_check_semaphore_wait;
2142}
2143
2144/* signal */
2145
2146static void
2147coro_signal_wake (pTHX_ AV *av, int count)
2148{
2149 SvIVX (AvARRAY (av)[0]) = 0;
2150
2151 /* now signal count waiters */
2152 while (count > 0 && AvFILLp (av) > 0)
2153 {
2154 SV *cb;
2155
2156 /* swap first two elements so we can shift a waiter */
2157 cb = AvARRAY (av)[0];
2158 AvARRAY (av)[0] = AvARRAY (av)[1];
2159 AvARRAY (av)[1] = cb;
2160
2161 cb = av_shift (av);
2162
2163 api_ready (aTHX_ cb);
2164 sv_setiv (cb, 0); /* signal waiter */
2165 SvREFCNT_dec (cb);
2166
2167 --count;
2168 }
2169}
2170
2171static int
2172slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2173{
2174 /* if we are about to throw, also stop waiting */
2175 return SvROK ((SV *)frame->data) && !CORO_THROW;
2176}
2177
2178static void
2179slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2180{
2181 AV *av = (AV *)SvRV (arg [0]);
2182
2183 if (SvIVX (AvARRAY (av)[0]))
2184 {
2185 SvIVX (AvARRAY (av)[0]) = 0;
2186 frame->prepare = prepare_nop;
2187 frame->check = slf_check_nop;
2188 }
2189 else
2190 {
2191 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2192
2193 av_push (av, waiter);
2194
2195 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2196 frame->prepare = prepare_schedule;
2197 frame->check = slf_check_signal_wait;
2198 }
2091} 2199}
2092 2200
2093/*****************************************************************************/ 2201/*****************************************************************************/
2094/* gensub: simple closure generation utility */ 2202/* gensub: simple closure generation utility */
2095 2203
… …
2168static int 2276static int
2169slf_check_aio_req (pTHX_ struct CoroSLF *frame) 2277slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2170{ 2278{
2171 AV *state = (AV *)frame->data; 2279 AV *state = (AV *)frame->data;
2172 2280
2281 /* if we are about to throw, return early */
2282 /* this does not cancel the aio request, but at least */
2283 /* it quickly returns */
2284 if (CORO_THROW)
2285 return 0;
2286
2173 /* one element that is an RV? repeat! */ 2287 /* one element that is an RV? repeat! */
2174 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2288 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2175 return 1; 2289 return 1;
2176 2290
2177 /* restore status */ 2291 /* restore status */
… …
2247 2361
2248 for (i = 0; i < items; ++i) 2362 for (i = 0; i < items; ++i)
2249 PUSHs (arg [i]); 2363 PUSHs (arg [i]);
2250 2364
2251 /* now push the callback closure */ 2365 /* now push the callback closure */
2252 PUSHs (sv_2mortal (gensub (coro_aio_callback, (void *)SvREFCNT_inc_NN ((SV *)state)))); 2366 PUSHs (sv_2mortal (gensub (aTHX_ coro_aio_callback, (void *)SvREFCNT_inc_NN ((SV *)state))));
2253 2367
2254 /* now call the AIO function - we assume our request is uncancelable */ 2368 /* now call the AIO function - we assume our request is uncancelable */
2255 PUTBACK; 2369 PUTBACK;
2256 call_sv ((SV *)req, G_VOID | G_DISCARD); 2370 call_sv ((SV *)req, G_VOID | G_DISCARD);
2257 } 2371 }
… …
2263} 2377}
2264 2378
2265static void 2379static void
2266coro_aio_req_xs (pTHX_ CV *cv) 2380coro_aio_req_xs (pTHX_ CV *cv)
2267{ 2381{
2268 dVAR;
2269 dXSARGS; 2382 dXSARGS;
2270 2383
2271 CORO_EXECUTE_SLF_XS (slf_init_aio_req); 2384 CORO_EXECUTE_SLF_XS (slf_init_aio_req);
2272 2385
2273 XSRETURN_EMPTY; 2386 XSRETURN_EMPTY;
… …
2374 for (i = 1; i < items; i++) 2487 for (i = 1; i < items; i++)
2375 av_push (coro->args, newSVsv (ST (i))); 2488 av_push (coro->args, newSVsv (ST (i)));
2376} 2489}
2377 OUTPUT: 2490 OUTPUT:
2378 RETVAL 2491 RETVAL
2379
2380void
2381_set_stacklevel (...)
2382 CODE:
2383 CORO_EXECUTE_SLF_XS (slf_init_set_stacklevel);
2384 2492
2385void 2493void
2386transfer (...) 2494transfer (...)
2387 PROTOTYPE: $$ 2495 PROTOTYPE: $$
2388 CODE: 2496 CODE:
… …
2514throw (Coro::State self, SV *throw = &PL_sv_undef) 2622throw (Coro::State self, SV *throw = &PL_sv_undef)
2515 PROTOTYPE: $;$ 2623 PROTOTYPE: $;$
2516 CODE: 2624 CODE:
2517{ 2625{
2518 struct coro *current = SvSTATE_current; 2626 struct coro *current = SvSTATE_current;
2519 SV **throwp = self == current ? &coro_throw : &self->throw; 2627 SV **throwp = self == current ? &CORO_THROW : &self->except;
2520 SvREFCNT_dec (*throwp); 2628 SvREFCNT_dec (*throwp);
2521 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2629 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2522} 2630}
2523 2631
2524void 2632void
… …
2613 coroapi.ready = api_ready; 2721 coroapi.ready = api_ready;
2614 coroapi.is_ready = api_is_ready; 2722 coroapi.is_ready = api_is_ready;
2615 coroapi.nready = coro_nready; 2723 coroapi.nready = coro_nready;
2616 coroapi.current = coro_current; 2724 coroapi.current = coro_current;
2617 2725
2618 GCoroAPI = &coroapi; 2726 /*GCoroAPI = &coroapi;*/
2619 sv_setiv (sv, (IV)&coroapi); 2727 sv_setiv (sv, (IV)&coroapi);
2620 SvREADONLY_on (sv); 2728 SvREADONLY_on (sv);
2621 } 2729 }
2622} 2730}
2623 2731
… …
2769 2877
2770 2878
2771MODULE = Coro::State PACKAGE = Coro::Semaphore 2879MODULE = Coro::State PACKAGE = Coro::Semaphore
2772 2880
2773SV * 2881SV *
2774new (SV *klass, SV *count_ = 0) 2882new (SV *klass, SV *count = 0)
2775 CODE: 2883 CODE:
2776{ 2884 RETVAL = sv_bless (
2777 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2885 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2778 AV *av = newAV (); 2886 GvSTASH (CvGV (cv))
2779 av_push (av, newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1)); 2887 );
2780 RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv))); 2888 OUTPUT:
2781} 2889 RETVAL
2890
2891# helper for Coro::Channel
2892SV *
2893_alloc (int count)
2894 CODE:
2895 RETVAL = coro_waitarray_new (aTHX_ count);
2782 OUTPUT: 2896 OUTPUT:
2783 RETVAL 2897 RETVAL
2784 2898
2785SV * 2899SV *
2786count (SV *self) 2900count (SV *self)
… …
2798 2912
2799void 2913void
2800down (SV *self) 2914down (SV *self)
2801 CODE: 2915 CODE:
2802 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 2916 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
2917
2918void
2919wait (SV *self)
2920 CODE:
2921 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2803 2922
2804void 2923void
2805try (SV *self) 2924try (SV *self)
2806 PPCODE: 2925 PPCODE:
2807{ 2926{
… …
2819 XSRETURN_NO; 2938 XSRETURN_NO;
2820} 2939}
2821 2940
2822void 2941void
2823waiters (SV *self) 2942waiters (SV *self)
2824 CODE: 2943 PPCODE:
2825{ 2944{
2826 AV *av = (AV *)SvRV (self); 2945 AV *av = (AV *)SvRV (self);
2946 int wcount = AvFILLp (av) + 1 - 1;
2827 2947
2828 if (GIMME_V == G_SCALAR) 2948 if (GIMME_V == G_SCALAR)
2829 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 2949 XPUSHs (sv_2mortal (newSViv (wcount)));
2830 else 2950 else
2831 { 2951 {
2832 int i; 2952 int i;
2833 EXTEND (SP, AvFILLp (av) + 1 - 1); 2953 EXTEND (SP, wcount);
2834 for (i = 1; i <= AvFILLp (av); ++i) 2954 for (i = 1; i <= wcount; ++i)
2835 PUSHs (newSVsv (AvARRAY (av)[i])); 2955 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2836 } 2956 }
2837} 2957}
2958
2959MODULE = Coro::State PACKAGE = Coro::Signal
2960
2961SV *
2962new (SV *klass)
2963 CODE:
2964 RETVAL = sv_bless (
2965 coro_waitarray_new (aTHX_ 0),
2966 GvSTASH (CvGV (cv))
2967 );
2968 OUTPUT:
2969 RETVAL
2970
2971void
2972wait (SV *self)
2973 CODE:
2974 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
2975
2976void
2977broadcast (SV *self)
2978 CODE:
2979{
2980 AV *av = (AV *)SvRV (self);
2981 coro_signal_wake (aTHX_ av, AvFILLp (av));
2982}
2983
2984void
2985send (SV *self)
2986 CODE:
2987{
2988 AV *av = (AV *)SvRV (self);
2989
2990 if (AvFILLp (av))
2991 coro_signal_wake (aTHX_ av, 1);
2992 else
2993 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
2994}
2995
2996IV
2997awaited (SV *self)
2998 CODE:
2999 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3000 OUTPUT:
3001 RETVAL
2838 3002
2839 3003
2840MODULE = Coro::State PACKAGE = Coro::AnyEvent 3004MODULE = Coro::State PACKAGE = Coro::AnyEvent
2841 3005
2842BOOT: 3006BOOT:
… …
2857 sv_setsv (sv_activity, &PL_sv_undef); 3021 sv_setsv (sv_activity, &PL_sv_undef);
2858 if (coro_nready >= incede) 3022 if (coro_nready >= incede)
2859 { 3023 {
2860 PUSHMARK (SP); 3024 PUSHMARK (SP);
2861 PUTBACK; 3025 PUTBACK;
2862 call_pv ("Coro::AnyEvent::_activity", G_DISCARD | G_EVAL); 3026 call_pv ("Coro::AnyEvent::_activity", G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
2863 SPAGAIN;
2864 } 3027 }
2865 3028
2866 --incede; 3029 --incede;
2867} 3030}
2868 3031
… …
2874 CODE: 3037 CODE:
2875{ 3038{
2876 HV *st; 3039 HV *st;
2877 GV *gvp; 3040 GV *gvp;
2878 CV *req_cv = sv_2cv (req, &st, &gvp, 0); 3041 CV *req_cv = sv_2cv (req, &st, &gvp, 0);
3042 /* newXSproto doesn't return the CV on 5.8 */
2879 CV *slf_cv = newXSproto (target, coro_aio_req_xs, __FILE__, proto); 3043 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3044 sv_setpv ((SV *)slf_cv, proto);
2880 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0); 3045 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
2881} 3046}
2882 3047

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines