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.283 by root, Sun Nov 16 11:12:57 2008 UTC vs.
Revision 1.304 by root, Wed Nov 19 08:22:48 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
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
152/* helper storage struct for Coro::AIO */
153struct io_state
154{
155 AV *res;
156 int errorno;
157 I32 laststype; /* U16 in 5.10.0 */
158 int laststatval;
159 Stat_t statcache;
160};
161
162static 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);
163 163
164static U32 cctx_gen; 164static U32 cctx_gen;
165static size_t cctx_stacksize = CORO_STACKSIZE; 165static size_t cctx_stacksize = CORO_STACKSIZE;
166static struct CoroAPI coroapi; 166static struct CoroAPI coroapi;
167static AV *main_mainstack; /* used to differentiate between $main and others */ 167static AV *main_mainstack; /* used to differentiate between $main and others */
241/* this is a structure representing a perl-level coroutine */ 241/* this is a structure representing a perl-level coroutine */
242struct coro { 242struct coro {
243 /* the C coroutine allocated to this perl coroutine, if any */ 243 /* the C coroutine allocated to this perl coroutine, if any */
244 coro_cctx *cctx; 244 coro_cctx *cctx;
245 245
246 /* process data */ 246 /* state data */
247 struct CoroSLF slf_frame; /* saved slf frame */ 247 struct CoroSLF slf_frame; /* saved slf frame */
248 AV *mainstack; 248 AV *mainstack;
249 perl_slots *slot; /* basically the saved sp */ 249 perl_slots *slot; /* basically the saved sp */
250 250
251 AV *args; /* data associated with this coroutine (initial args) */ 251 AV *args; /* data associated with this coroutine (initial args) */
257 /* statistics */ 257 /* statistics */
258 int usecount; /* number of transfers to this coro */ 258 int usecount; /* number of transfers to this coro */
259 259
260 /* coro process data */ 260 /* coro process data */
261 int prio; 261 int prio;
262 SV *throw; /* exception to be thrown */ 262 SV *except; /* exception to be thrown */
263 SV *rouse_cb;
263 264
264 /* async_pool */ 265 /* async_pool */
265 SV *saved_deffh; 266 SV *saved_deffh;
266 267
267 /* linked list */ 268 /* linked list */
269}; 270};
270 271
271typedef struct coro *Coro__State; 272typedef struct coro *Coro__State;
272typedef struct coro *Coro__State_or_hashref; 273typedef struct coro *Coro__State_or_hashref;
273 274
275/* the following variables are effectively part of the perl context */
276/* and get copied between struct coro and these variables */
277/* the mainr easonw e don't support windows process emulation */
274static struct CoroSLF slf_frame; /* the current slf frame */ 278static struct CoroSLF slf_frame; /* the current slf frame */
275 279
276/** Coro ********************************************************************/ 280/** Coro ********************************************************************/
277 281
278#define PRIO_MAX 3 282#define PRIO_MAX 3
380 SvREFCNT_dec (av); /* sv_magicext increased the refcount */ 384 SvREFCNT_dec (av); /* sv_magicext increased the refcount */
381 385
382 return 0; 386 return 0;
383} 387}
384 388
385#define CORO_MAGIC_type_cv PERL_MAGIC_ext 389#define CORO_MAGIC_type_cv 26
386#define CORO_MAGIC_type_state PERL_MAGIC_ext 390#define CORO_MAGIC_type_state PERL_MAGIC_ext
387 391
388static MGVTBL coro_cv_vtbl = { 392static MGVTBL coro_cv_vtbl = {
389 0, 0, 0, 0, 393 0, 0, 0, 0,
390 coro_cv_free 394 coro_cv_free
391}; 395};
392 396
397#define CORO_MAGIC_NN(sv, type) \
398 (expect_true (SvMAGIC (sv)->mg_type == type) \
399 ? SvMAGIC (sv) \
400 : mg_find (sv, type))
401
393#define CORO_MAGIC(sv, type) \ 402#define CORO_MAGIC(sv, type) \
394 expect_true (SvMAGIC (sv)) \ 403 (expect_true (SvMAGIC (sv)) \
395 ? expect_true (SvMAGIC (sv)->mg_type == type) \ 404 ? CORO_MAGIC_NN (sv, type) \
396 ? SvMAGIC (sv) \
397 : mg_find (sv, type) \
398 : 0 405 : 0)
399 406
400#define CORO_MAGIC_cv(cv) CORO_MAGIC (((SV *)(cv)), CORO_MAGIC_type_cv) 407#define CORO_MAGIC_cv(cv) CORO_MAGIC (((SV *)(cv)), CORO_MAGIC_type_cv)
401#define CORO_MAGIC_state(sv) CORO_MAGIC (((SV *)(sv)), CORO_MAGIC_type_state) 408#define CORO_MAGIC_state(sv) CORO_MAGIC_NN (((SV *)(sv)), CORO_MAGIC_type_state)
402 409
403INLINE struct coro * 410INLINE struct coro *
404SvSTATE_ (pTHX_ SV *coro) 411SvSTATE_ (pTHX_ SV *coro)
405{ 412{
406 HV *stash; 413 HV *stash;
424 return (struct coro *)mg->mg_ptr; 431 return (struct coro *)mg->mg_ptr;
425} 432}
426 433
427#define SvSTATE(sv) SvSTATE_ (aTHX_ (sv)) 434#define SvSTATE(sv) SvSTATE_ (aTHX_ (sv))
428 435
429/* fastert than SvSTATE, but expects a coroutine hv */ 436/* faster than SvSTATE, but expects a coroutine hv */
430INLINE struct coro * 437#define SvSTATE_hv(hv) ((struct coro *)CORO_MAGIC_NN ((SV *)hv, CORO_MAGIC_type_state)->mg_ptr)
431SvSTATE_hv (SV *sv)
432{
433 MAGIC *mg = expect_true (SvMAGIC (sv)->mg_type == CORO_MAGIC_type_state)
434 ? SvMAGIC (sv)
435 : mg_find (sv, CORO_MAGIC_type_state);
436
437 return (struct coro *)mg->mg_ptr;
438}
439
440#define SvSTATE_current SvSTATE_hv (SvRV (coro_current)) 438#define SvSTATE_current SvSTATE_hv (SvRV (coro_current))
441 439
442/* the next two functions merely cache the padlists */ 440/* the next two functions merely cache the padlists */
443static void 441static void
444get_padlist (pTHX_ CV *cv) 442get_padlist (pTHX_ CV *cv)
513 } 511 }
514 512
515 PUTBACK; 513 PUTBACK;
516 } 514 }
517 515
518 slf_frame = c->slf_frame; 516 slf_frame = c->slf_frame;
517 CORO_THROW = c->except;
519} 518}
520 519
521static void 520static void
522save_perl (pTHX_ Coro__State c) 521save_perl (pTHX_ Coro__State c)
523{ 522{
523 c->except = CORO_THROW;
524 c->slf_frame = slf_frame; 524 c->slf_frame = slf_frame;
525 525
526 { 526 {
527 dSP; 527 dSP;
528 I32 cxix = cxstack_ix; 528 I32 cxix = cxstack_ix;
821slf_check_nop (pTHX_ struct CoroSLF *frame) 821slf_check_nop (pTHX_ struct CoroSLF *frame)
822{ 822{
823 return 0; 823 return 0;
824} 824}
825 825
826static void 826static UNOP coro_setup_op;
827
828static void NOINLINE /* noinline to keep it out of the transfer fast path */
827coro_setup (pTHX_ struct coro *coro) 829coro_setup (pTHX_ struct coro *coro)
828{ 830{
829 /* 831 /*
830 * emulate part of the perl startup here. 832 * emulate part of the perl startup here.
831 */ 833 */
858 { 860 {
859 dSP; 861 dSP;
860 UNOP myop; 862 UNOP myop;
861 863
862 Zero (&myop, 1, UNOP); 864 Zero (&myop, 1, UNOP);
863 myop.op_next = Nullop; 865 myop.op_next = Nullop;
864 myop.op_flags = OPf_WANT_VOID; 866 myop.op_flags = OPf_WANT_VOID;
865 867
866 PUSHMARK (SP); 868 PUSHMARK (SP);
867 XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv)))); 869 XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv))));
868 PUTBACK; 870 PUTBACK;
874 /* this newly created coroutine might be run on an existing cctx which most 876 /* this newly created coroutine might be run on an existing cctx which most
875 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here. 877 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
876 */ 878 */
877 slf_frame.prepare = prepare_nop; /* provide a nop function for an eventual pp_slf */ 879 slf_frame.prepare = prepare_nop; /* provide a nop function for an eventual pp_slf */
878 slf_frame.check = slf_check_nop; /* signal pp_slf to not repeat */ 880 slf_frame.check = slf_check_nop; /* signal pp_slf to not repeat */
881
882 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
883 coro_setup_op.op_next = PL_op;
884 coro_setup_op.op_type = OP_CUSTOM;
885 coro_setup_op.op_ppaddr = pp_slf;
886 /* no flags required, as an init function won't be called */
887
888 PL_op = (OP *)&coro_setup_op;
889
890 /* copy throw, in case it was set before coro_setup */
891 CORO_THROW = coro->except;
879} 892}
880 893
881static void 894static void
882coro_destruct (pTHX_ struct coro *coro) 895coro_destruct (pTHX_ struct coro *coro)
883{ 896{
906 SvREFCNT_dec (GvSV (irsgv)); 919 SvREFCNT_dec (GvSV (irsgv));
907 920
908 SvREFCNT_dec (PL_diehook); 921 SvREFCNT_dec (PL_diehook);
909 SvREFCNT_dec (PL_warnhook); 922 SvREFCNT_dec (PL_warnhook);
910 923
924 SvREFCNT_dec (CORO_THROW);
911 SvREFCNT_dec (coro->saved_deffh); 925 SvREFCNT_dec (coro->saved_deffh);
912 SvREFCNT_dec (coro->throw); 926 SvREFCNT_dec (coro->rouse_cb);
913 927
914 coro_destruct_stacks (aTHX); 928 coro_destruct_stacks (aTHX);
915} 929}
916 930
917INLINE void 931INLINE void
1046 1060
1047 TAINT_NOT; 1061 TAINT_NOT;
1048 return 0; 1062 return 0;
1049} 1063}
1050 1064
1065static struct coro_cctx *cctx_ssl_cctx;
1066static struct CoroSLF cctx_ssl_frame;
1067
1051static void 1068static void
1052prepare_set_stacklevel (struct coro_transfer_args *ta, struct coro_cctx *cctx) 1069slf_prepare_set_stacklevel (pTHX_ struct coro_transfer_args *ta)
1053{ 1070{
1054 ta->prev = (struct coro *)cctx; 1071 ta->prev = (struct coro *)cctx_ssl_cctx;
1055 ta->next = 0; 1072 ta->next = 0;
1056} 1073}
1057 1074
1058/* inject a fake call to Coro::State::_cctx_init into the execution */ 1075static int
1059/* _cctx_init should be careful, as it could be called at almost any time */ 1076slf_check_set_stacklevel (pTHX_ struct CoroSLF *frame)
1060/* during execution of a perl program */ 1077{
1061/* also initialises PL_top_env */ 1078 *frame = cctx_ssl_frame;
1079
1080 return frame->check (aTHX_ frame); /* execute the restored frame - there must be one */
1081}
1082
1083/* initialises PL_top_env and injects a pseudo-slf-call to set the stacklevel */
1062static void NOINLINE 1084static void NOINLINE
1063cctx_prepare (pTHX_ coro_cctx *cctx) 1085cctx_prepare (pTHX_ coro_cctx *cctx)
1064{ 1086{
1065 dSP;
1066 UNOP myop;
1067
1068 PL_top_env = &PL_start_env; 1087 PL_top_env = &PL_start_env;
1069 1088
1070 if (cctx->flags & CC_TRACE) 1089 if (cctx->flags & CC_TRACE)
1071 PL_runops = runops_trace; 1090 PL_runops = runops_trace;
1072 1091
1073 Zero (&myop, 1, UNOP); 1092 /* we already must be executing an SLF op, there is no other valid way
1074 myop.op_next = PL_op; 1093 * that can lead to creation of a new cctx */
1075 myop.op_flags = OPf_WANT_VOID | OPf_STACKED; 1094 assert (("FATAL: can't prepare slf-less cctx in Coro module (please report)",
1095 slf_frame.prepare && PL_op->op_ppaddr == pp_slf));
1076 1096
1077 PUSHMARK (SP); 1097 /* we must emulate leaving pp_slf, which is done inside slf_check_set_stacklevel */
1078 EXTEND (SP, 2); 1098 cctx_ssl_cctx = cctx;
1079 PUSHs (sv_2mortal (newSViv ((IV)cctx))); 1099 cctx_ssl_frame = slf_frame;
1080 PUSHs ((SV *)get_cv ("Coro::State::_cctx_init", FALSE)); 1100
1081 PUTBACK; 1101 slf_frame.prepare = slf_prepare_set_stacklevel;
1082 PL_op = (OP *)&myop; 1102 slf_frame.check = slf_check_set_stacklevel;
1083 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
1084 SPAGAIN;
1085} 1103}
1086 1104
1087/* the tail of transfer: execute stuff we can only do after a transfer */ 1105/* the tail of transfer: execute stuff we can only do after a transfer */
1088INLINE void 1106INLINE void
1089transfer_tail (pTHX) 1107transfer_tail (pTHX)
1276/** coroutine switching *****************************************************/ 1294/** coroutine switching *****************************************************/
1277 1295
1278static void 1296static void
1279transfer_check (pTHX_ struct coro *prev, struct coro *next) 1297transfer_check (pTHX_ struct coro *prev, struct coro *next)
1280{ 1298{
1299 /* TODO: throwing up here is considered harmful */
1300
1281 if (expect_true (prev != next)) 1301 if (expect_true (prev != next))
1282 { 1302 {
1283 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW)))) 1303 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW))))
1284 croak ("Coro::State::transfer called with non-running/new prev Coro::State, but can only transfer from running or new states,"); 1304 croak ("Coro::State::transfer called with non-running/new prev Coro::State, but can only transfer from running or new states,");
1285 1305
1295#endif 1315#endif
1296 } 1316 }
1297} 1317}
1298 1318
1299/* always use the TRANSFER macro */ 1319/* always use the TRANSFER macro */
1300static void NOINLINE 1320static void NOINLINE /* noinline so we have a fixed stackframe */
1301transfer (pTHX_ struct coro *prev, struct coro *next, int force_cctx) 1321transfer (pTHX_ struct coro *prev, struct coro *next, int force_cctx)
1302{ 1322{
1303 dSTACKLEVEL; 1323 dSTACKLEVEL;
1304 1324
1305 /* sometimes transfer is only called to set idle_sp */ 1325 /* sometimes transfer is only called to set idle_sp */
1306 if (expect_false (!next)) 1326 if (expect_false (!next))
1307 { 1327 {
1308 ((coro_cctx *)prev)->idle_sp = (void *)stacklevel; 1328 ((coro_cctx *)prev)->idle_sp = STACKLEVEL;
1309 assert (((coro_cctx *)prev)->idle_te = PL_top_env); /* just for the side-effect when asserts are enabled */ 1329 assert (((coro_cctx *)prev)->idle_te = PL_top_env); /* just for the side-effect when asserts are enabled */
1310 } 1330 }
1311 else if (expect_true (prev != next)) 1331 else if (expect_true (prev != next))
1312 { 1332 {
1313 coro_cctx *prev__cctx; 1333 coro_cctx *prev__cctx;
1338 1358
1339 prev__cctx = prev->cctx; 1359 prev__cctx = prev->cctx;
1340 1360
1341 /* possibly untie and reuse the cctx */ 1361 /* possibly untie and reuse the cctx */
1342 if (expect_true ( 1362 if (expect_true (
1343 prev__cctx->idle_sp == (void *)stacklevel 1363 prev__cctx->idle_sp == STACKLEVEL
1344 && !(prev__cctx->flags & CC_TRACE) 1364 && !(prev__cctx->flags & CC_TRACE)
1345 && !force_cctx 1365 && !force_cctx
1346 )) 1366 ))
1347 { 1367 {
1348 /* I assume that stacklevel is a stronger indicator than PL_top_env changes */ 1368 /* I assume that stacklevel is a stronger indicator than PL_top_env changes */
1479 1499
1480 prepare_transfer (aTHX_ &ta, prev_sv, next_sv); 1500 prepare_transfer (aTHX_ &ta, prev_sv, next_sv);
1481 TRANSFER (ta, 1); 1501 TRANSFER (ta, 1);
1482} 1502}
1483 1503
1504/*****************************************************************************/
1505/* gensub: simple closure generation utility */
1506
1507#define GENSUB_ARG CvXSUBANY (cv).any_ptr
1508
1509/* create a closure from XS, returns a code reference */
1510/* the arg can be accessed via GENSUB_ARG from the callback */
1511/* the callback must use dXSARGS/XSRETURN */
1512static SV *
1513gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
1514{
1515 CV *cv = (CV *)newSV (0);
1516
1517 sv_upgrade ((SV *)cv, SVt_PVCV);
1518
1519 CvANON_on (cv);
1520 CvISXSUB_on (cv);
1521 CvXSUB (cv) = xsub;
1522 GENSUB_ARG = arg;
1523
1524 return newRV_noinc ((SV *)cv);
1525}
1526
1484/** Coro ********************************************************************/ 1527/** Coro ********************************************************************/
1485 1528
1486INLINE void 1529INLINE void
1487coro_enq (pTHX_ struct coro *coro) 1530coro_enq (pTHX_ struct coro *coro)
1488{ 1531{
1531 ENTER; 1574 ENTER;
1532 SAVETMPS; 1575 SAVETMPS;
1533 1576
1534 PUSHMARK (SP); 1577 PUSHMARK (SP);
1535 PUTBACK; 1578 PUTBACK;
1536 call_sv (sv_hook, G_DISCARD); 1579 call_sv (sv_hook, G_VOID | G_DISCARD);
1537 SPAGAIN;
1538 1580
1539 FREETMPS; 1581 FREETMPS;
1540 LEAVE; 1582 LEAVE;
1541 } 1583 }
1542 1584
1569 ENTER; 1611 ENTER;
1570 SAVETMPS; 1612 SAVETMPS;
1571 1613
1572 PUSHMARK (SP); 1614 PUSHMARK (SP);
1573 PUTBACK; 1615 PUTBACK;
1574 call_sv (get_sv ("Coro::idle", FALSE), G_DISCARD); 1616 call_sv (get_sv ("Coro::idle", FALSE), G_VOID | G_DISCARD);
1575 SPAGAIN;
1576 1617
1577 FREETMPS; 1618 FREETMPS;
1578 LEAVE; 1619 LEAVE;
1579 continue; 1620 continue;
1580 } 1621 }
1687 if (coro->flags & CF_RUNNING) 1728 if (coro->flags & CF_RUNNING)
1688 PL_runops = RUNOPS_DEFAULT; 1729 PL_runops = RUNOPS_DEFAULT;
1689 else 1730 else
1690 coro->slot->runops = RUNOPS_DEFAULT; 1731 coro->slot->runops = RUNOPS_DEFAULT;
1691 } 1732 }
1733}
1734
1735/*****************************************************************************/
1736/* rouse callback */
1737
1738#define CORO_MAGIC_type_rouse PERL_MAGIC_ext
1739
1740static void
1741coro_rouse_callback (pTHX_ CV *cv)
1742{
1743 dXSARGS;
1744 SV *data = (SV *)GENSUB_ARG;
1745
1746 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1747 {
1748 /* first call, set args */
1749 int i;
1750 AV *av = newAV ();
1751 SV *coro = SvRV (data);
1752
1753 SvRV_set (data, (SV *)av);
1754 api_ready (aTHX_ coro);
1755 SvREFCNT_dec (coro);
1756
1757 /* better take a full copy of the arguments */
1758 while (items--)
1759 av_store (av, items, newSVsv (ST (items)));
1760 }
1761
1762 XSRETURN_EMPTY;
1763}
1764
1765static int
1766slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
1767{
1768 SV *data = (SV *)frame->data;
1769
1770 if (CORO_THROW)
1771 return 0;
1772
1773 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1774 return 1;
1775
1776 /* now push all results on the stack */
1777 {
1778 dSP;
1779 AV *av = (AV *)SvRV (data);
1780 int i;
1781
1782 EXTEND (SP, AvFILLp (av) + 1);
1783 for (i = 0; i <= AvFILLp (av); ++i)
1784 PUSHs (sv_2mortal (AvARRAY (av)[i]));
1785
1786 /* we have stolen the elements, so ste length to zero and free */
1787 AvFILLp (av) = -1;
1788 av_undef (av);
1789
1790 PUTBACK;
1791 }
1792
1793 return 0;
1794}
1795
1796static void
1797slf_init_rouse_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1798{
1799 SV *cb;
1800
1801 if (items)
1802 cb = arg [0];
1803 else
1804 {
1805 struct coro *coro = SvSTATE_current;
1806
1807 if (!coro->rouse_cb)
1808 croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
1809
1810 cb = sv_2mortal (coro->rouse_cb);
1811 coro->rouse_cb = 0;
1812 }
1813
1814 if (!SvROK (cb)
1815 || SvTYPE (SvRV (cb)) != SVt_PVCV
1816 || CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
1817 croak ("Coro::rouse_wait called with illegal callback argument,");
1818
1819 {
1820 CV *cv = (CV *)SvRV (cb); /* for GENSUB_ARG */
1821 SV *data = (SV *)GENSUB_ARG;
1822
1823 frame->data = (void *)data;
1824 frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
1825 frame->check = slf_check_rouse_wait;
1826 }
1827}
1828
1829static SV *
1830coro_new_rouse_cb (pTHX)
1831{
1832 HV *hv = (HV *)SvRV (coro_current);
1833 struct coro *coro = SvSTATE_hv (hv);
1834 SV *data = newRV_inc ((SV *)hv);
1835 SV *cb = gensub (aTHX_ coro_rouse_callback, (void *)data);
1836
1837 sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
1838 SvREFCNT_dec (data); /* magicext increases the refcount */
1839
1840 SvREFCNT_dec (coro->rouse_cb);
1841 coro->rouse_cb = SvREFCNT_inc_NN (cb);
1842
1843 return cb;
1844}
1845
1846/*****************************************************************************/
1847/* schedule-like-function opcode (SLF) */
1848
1849static UNOP slf_restore; /* restore stack as entersub did, for first-re-run */
1850static const CV *slf_cv;
1851static SV **slf_argv;
1852static int slf_argc, slf_arga; /* count, allocated */
1853static I32 slf_ax; /* top of stack, for restore */
1854
1855/* this restores the stack in the case we patched the entersub, to */
1856/* recreate the stack frame as perl will on following calls */
1857/* since entersub cleared the stack */
1858static OP *
1859pp_restore (pTHX)
1860{
1861 int i;
1862 SV **SP = PL_stack_base + slf_ax;
1863
1864 PUSHMARK (SP);
1865
1866 EXTEND (SP, slf_argc + 1);
1867
1868 for (i = 0; i < slf_argc; ++i)
1869 PUSHs (sv_2mortal (slf_argv [i]));
1870
1871 PUSHs ((SV *)CvGV (slf_cv));
1872
1873 RETURNOP (slf_restore.op_first);
1874}
1875
1876static void
1877slf_prepare_transfer (pTHX_ struct coro_transfer_args *ta)
1878{
1879 SV **arg = (SV **)slf_frame.data;
1880
1881 prepare_transfer (aTHX_ ta, arg [0], arg [1]);
1882}
1883
1884static void
1885slf_init_transfer (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1886{
1887 if (items != 2)
1888 croak ("Coro::State::transfer (prev, next) expects two arguments, not %d,", items);
1889
1890 frame->prepare = slf_prepare_transfer;
1891 frame->check = slf_check_nop;
1892 frame->data = (void *)arg; /* let's hope it will stay valid */
1893}
1894
1895static void
1896slf_init_schedule (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1897{
1898 frame->prepare = prepare_schedule;
1899 frame->check = slf_check_nop;
1900}
1901
1902static void
1903slf_init_cede (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1904{
1905 frame->prepare = prepare_cede;
1906 frame->check = slf_check_nop;
1907}
1908
1909static void
1910slf_init_cede_notself (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1911{
1912 frame->prepare = prepare_cede_notself;
1913 frame->check = slf_check_nop;
1914}
1915
1916/*
1917 * these not obviously related functions are all rolled into one
1918 * function to increase chances that they all will call transfer with the same
1919 * stack offset
1920 * SLF stands for "schedule-like-function".
1921 */
1922static OP *
1923pp_slf (pTHX)
1924{
1925 I32 checkmark; /* mark SP to see how many elements check has pushed */
1926
1927 /* set up the slf frame, unless it has already been set-up */
1928 /* the latter happens when a new coro has been started */
1929 /* or when a new cctx was attached to an existing coroutine */
1930 if (expect_true (!slf_frame.prepare))
1931 {
1932 /* first iteration */
1933 dSP;
1934 SV **arg = PL_stack_base + TOPMARK + 1;
1935 int items = SP - arg; /* args without function object */
1936 SV *gv = *sp;
1937
1938 /* do a quick consistency check on the "function" object, and if it isn't */
1939 /* for us, divert to the real entersub */
1940 if (SvTYPE (gv) != SVt_PVGV
1941 || !GvCV (gv)
1942 || !(CvFLAGS (GvCV (gv)) & CVf_SLF))
1943 return PL_ppaddr[OP_ENTERSUB](aTHX);
1944
1945 if (!(PL_op->op_flags & OPf_STACKED))
1946 {
1947 /* ampersand-form of call, use @_ instead of stack */
1948 AV *av = GvAV (PL_defgv);
1949 arg = AvARRAY (av);
1950 items = AvFILLp (av) + 1;
1951 }
1952
1953 /* now call the init function, which needs to set up slf_frame */
1954 ((coro_slf_cb)CvXSUBANY (GvCV (gv)).any_ptr)
1955 (aTHX_ &slf_frame, GvCV (gv), arg, items);
1956
1957 /* pop args */
1958 SP = PL_stack_base + POPMARK;
1959
1960 PUTBACK;
1961 }
1962
1963 /* now that we have a slf_frame, interpret it! */
1964 /* we use a callback system not to make the code needlessly */
1965 /* complicated, but so we can run multiple perl coros from one cctx */
1966
1967 do
1968 {
1969 struct coro_transfer_args ta;
1970
1971 slf_frame.prepare (aTHX_ &ta);
1972 TRANSFER (ta, 0);
1973
1974 checkmark = PL_stack_sp - PL_stack_base;
1975 }
1976 while (slf_frame.check (aTHX_ &slf_frame));
1977
1978 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
1979
1980 /* exception handling */
1981 if (expect_false (CORO_THROW))
1982 {
1983 SV *exception = sv_2mortal (CORO_THROW);
1984
1985 CORO_THROW = 0;
1986 sv_setsv (ERRSV, exception);
1987 croak (0);
1988 }
1989
1990 /* return value handling - mostly like entersub */
1991 /* make sure we put something on the stack in scalar context */
1992 if (GIMME_V == G_SCALAR)
1993 {
1994 dSP;
1995 SV **bot = PL_stack_base + checkmark;
1996
1997 if (sp == bot) /* too few, push undef */
1998 bot [1] = &PL_sv_undef;
1999 else if (sp != bot + 1) /* too many, take last one */
2000 bot [1] = *sp;
2001
2002 SP = bot + 1;
2003
2004 PUTBACK;
2005 }
2006
2007 return NORMAL;
2008}
2009
2010static void
2011api_execute_slf (pTHX_ CV *cv, coro_slf_cb init_cb, I32 ax)
2012{
2013 int i;
2014 SV **arg = PL_stack_base + ax;
2015 int items = PL_stack_sp - arg + 1;
2016
2017 assert (("FATAL: SLF call with illegal CV value", !CvANON (cv)));
2018
2019 if (PL_op->op_ppaddr != PL_ppaddr [OP_ENTERSUB]
2020 && PL_op->op_ppaddr != pp_slf)
2021 croak ("FATAL: Coro SLF calls can only be made normally, not via goto or any other means, caught");
2022
2023 CvFLAGS (cv) |= CVf_SLF;
2024 CvXSUBANY (cv).any_ptr = (void *)init_cb;
2025 slf_cv = cv;
2026
2027 /* we patch the op, and then re-run the whole call */
2028 /* we have to put the same argument on the stack for this to work */
2029 /* and this will be done by pp_restore */
2030 slf_restore.op_next = (OP *)&slf_restore;
2031 slf_restore.op_type = OP_CUSTOM;
2032 slf_restore.op_ppaddr = pp_restore;
2033 slf_restore.op_first = PL_op;
2034
2035 slf_ax = ax - 1; /* undo the ax++ inside dAXMARK */
2036
2037 if (PL_op->op_flags & OPf_STACKED)
2038 {
2039 if (items > slf_arga)
2040 {
2041 slf_arga = items;
2042 free (slf_argv);
2043 slf_argv = malloc (slf_arga * sizeof (SV *));
2044 }
2045
2046 slf_argc = items;
2047
2048 for (i = 0; i < items; ++i)
2049 slf_argv [i] = SvREFCNT_inc (arg [i]);
2050 }
2051 else
2052 slf_argc = 0;
2053
2054 PL_op->op_ppaddr = pp_slf;
2055 PL_op->op_type = OP_CUSTOM; /* maybe we should leave it at entersub? */
2056
2057 PL_op = (OP *)&slf_restore;
1692} 2058}
1693 2059
1694/*****************************************************************************/ 2060/*****************************************************************************/
1695/* PerlIO::cede */ 2061/* PerlIO::cede */
1696 2062
1765 PerlIOBuf_get_cnt, 2131 PerlIOBuf_get_cnt,
1766 PerlIOBuf_set_ptrcnt, 2132 PerlIOBuf_set_ptrcnt,
1767}; 2133};
1768 2134
1769/*****************************************************************************/ 2135/*****************************************************************************/
2136/* Coro::Semaphore & Coro::Signal */
1770 2137
1771static const CV *slf_cv; /* for quick consistency check */
1772
1773static UNOP slf_restore; /* restore stack as entersub did, for first-re-run */
1774static SV *slf_arg0;
1775static SV *slf_arg1;
1776static SV *slf_arg2;
1777
1778/* this restores the stack in the case we patched the entersub, to */
1779/* recreate the stack frame as perl will on following calls */
1780/* since entersub cleared the stack */
1781static OP * 2138static SV *
1782pp_restore (pTHX) 2139coro_waitarray_new (pTHX_ int count)
1783{ 2140{
1784 dSP; 2141 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2142 AV *av = newAV ();
2143 SV **ary;
1785 2144
1786 PUSHMARK (SP); 2145 /* unfortunately, building manually saves memory */
2146 Newx (ary, 2, SV *);
2147 AvALLOC (av) = ary;
2148 SvPV_set ((SV *)av, (char *)ary); /* 5.8.8 needs this syntax instead of AvARRAY = ary */
2149 AvMAX (av) = 1;
2150 AvFILLp (av) = 0;
2151 ary [0] = newSViv (count);
1787 2152
1788 EXTEND (SP, 3); 2153 return newRV_noinc ((SV *)av);
1789 if (slf_arg0) PUSHs (sv_2mortal (slf_arg0));
1790 if (slf_arg1) PUSHs (sv_2mortal (slf_arg1));
1791 if (slf_arg2) PUSHs (sv_2mortal (slf_arg2));
1792 PUSHs ((SV *)CvGV (slf_cv));
1793
1794 RETURNOP (slf_restore.op_first);
1795} 2154}
1796 2155
1797static void 2156/* semaphore */
1798slf_prepare_set_stacklevel (pTHX_ struct coro_transfer_args *ta)
1799{
1800 prepare_set_stacklevel (ta, (struct coro_cctx *)slf_frame.data);
1801}
1802
1803static void
1804slf_init_set_stacklevel (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1805{
1806 assert (("FATAL: set_stacklevel needs the coro cctx as sole argument", items == 1));
1807
1808 frame->prepare = slf_prepare_set_stacklevel;
1809 frame->check = slf_check_nop;
1810 frame->data = (void *)SvIV (arg [0]);
1811}
1812
1813static void
1814slf_prepare_transfer (pTHX_ struct coro_transfer_args *ta)
1815{
1816 SV **arg = (SV **)slf_frame.data;
1817
1818 prepare_transfer (aTHX_ ta, arg [0], arg [1]);
1819
1820 /* if the destination has ->throw set, then copy it */
1821 /* into the current coro's throw slot, so it will be raised */
1822 /* after the schedule */
1823 if (expect_false (ta->next->throw))
1824 {
1825 struct coro *coro = SvSTATE_current;
1826 SvREFCNT_dec (coro->throw);
1827 coro->throw = ta->next->throw;
1828 ta->next->throw = 0;
1829 }
1830}
1831
1832static void
1833slf_init_transfer (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1834{
1835 if (items != 2)
1836 croak ("Coro::State::transfer (prev, next) expects two arguments, not %d,", items);
1837
1838 frame->prepare = slf_prepare_transfer;
1839 frame->check = slf_check_nop;
1840 frame->data = (void *)arg; /* let's hope it will stay valid */
1841}
1842
1843static void
1844slf_init_schedule (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1845{
1846 frame->prepare = prepare_schedule;
1847 frame->check = slf_check_nop;
1848}
1849
1850static void
1851slf_init_cede (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1852{
1853 frame->prepare = prepare_cede;
1854 frame->check = slf_check_nop;
1855}
1856
1857static void
1858slf_init_cede_notself (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1859{
1860 frame->prepare = prepare_cede_notself;
1861 frame->check = slf_check_nop;
1862}
1863
1864/* we hijack an hopefully unused CV flag for our purposes */
1865#define CVf_SLF 0x4000
1866
1867/*
1868 * these not obviously related functions are all rolled into one
1869 * function to increase chances that they all will call transfer with the same
1870 * stack offset
1871 * SLF stands for "schedule-like-function".
1872 */
1873static OP *
1874pp_slf (pTHX)
1875{
1876 I32 checkmark; /* mark SP to see how many elements check has pushed */
1877
1878 /* set up the slf frame, unless it has already been set-up */
1879 /* the latter happens when a new coro has been started */
1880 /* or when a new cctx was attached to an existing coroutine */
1881 if (expect_true (!slf_frame.prepare))
1882 {
1883 /* first iteration */
1884 dSP;
1885 SV **arg = PL_stack_base + TOPMARK + 1;
1886 int items = SP - arg; /* args without function object */
1887 SV *gv = *sp;
1888
1889 /* do a quick consistency check on the "function" object, and if it isn't */
1890 /* for us, divert to the real entersub */
1891 if (SvTYPE (gv) != SVt_PVGV || !(CvFLAGS (GvCV (gv)) & CVf_SLF))
1892 return PL_ppaddr[OP_ENTERSUB](aTHX);
1893
1894 if (!(PL_op->op_flags & OPf_STACKED))
1895 {
1896 /* ampersand-form of call, use @_ instead of stack */
1897 AV *av = GvAV (PL_defgv);
1898 arg = AvARRAY (av);
1899 items = AvFILLp (av) + 1;
1900 }
1901
1902 PUTBACK;
1903
1904 /* now call the init function, which needs to set up slf_frame */
1905 ((coro_slf_cb)CvXSUBANY (GvCV (gv)).any_ptr)
1906 (aTHX_ &slf_frame, GvCV (gv), arg, items);
1907
1908 /* pop args */
1909 SP = PL_stack_base + POPMARK;
1910
1911 PUTBACK;
1912 }
1913
1914 /* now that we have a slf_frame, interpret it! */
1915 /* we use a callback system not to make the code needlessly */
1916 /* complicated, but so we can run multiple perl coros from one cctx */
1917
1918 do
1919 {
1920 struct coro_transfer_args ta;
1921
1922 slf_frame.prepare (aTHX_ &ta);
1923 TRANSFER (ta, 0);
1924
1925 checkmark = PL_stack_sp - PL_stack_base;
1926 }
1927 while (slf_frame.check (aTHX_ &slf_frame));
1928
1929 {
1930 dSP;
1931 SV **bot = PL_stack_base + checkmark;
1932 int gimme = GIMME_V;
1933
1934 slf_frame.prepare = 0; /* invalidate the frame, so it gets initialised again next time */
1935
1936 /* make sure we put something on the stack in scalar context */
1937 if (gimme == G_SCALAR)
1938 {
1939 if (sp == bot)
1940 XPUSHs (&PL_sv_undef);
1941
1942 SP = bot + 1;
1943 }
1944
1945 PUTBACK;
1946 }
1947
1948 {
1949 struct coro *coro = SvSTATE_current;
1950
1951 if (expect_false (coro->throw))
1952 {
1953 SV *exception = sv_2mortal (coro->throw);
1954
1955 coro->throw = 0;
1956 sv_setsv (ERRSV, exception);
1957 croak (0);
1958 }
1959 }
1960
1961 return NORMAL;
1962}
1963
1964static void
1965api_execute_slf (pTHX_ CV *cv, coro_slf_cb init_cb, SV **arg, int items)
1966{
1967 assert (("FATAL: SLF call with illegal CV value", !CvANON (cv)));
1968
1969 if (PL_op->op_ppaddr != PL_ppaddr [OP_ENTERSUB]
1970 && PL_op->op_ppaddr != pp_slf)
1971 croak ("FATAL: Coro SLF calls can only be made normally, not via goto or any other means, caught");
1972
1973 if (items > 3)
1974 croak ("Coro only supports up to three arguments to SLF functions currently (not %d), caught", items);
1975
1976 CvFLAGS (cv) |= CVf_SLF;
1977 CvXSUBANY (cv).any_ptr = (void *)init_cb;
1978 slf_cv = cv;
1979
1980 /* we patch the op, and then re-run the whole call */
1981 /* we have to put the same argument on the stack for this to work */
1982 /* and this will be done by pp_restore */
1983 slf_restore.op_next = (OP *)&slf_restore;
1984 slf_restore.op_type = OP_CUSTOM;
1985 slf_restore.op_ppaddr = pp_restore;
1986 slf_restore.op_first = PL_op;
1987
1988 slf_arg0 = items > 0 ? SvREFCNT_inc (arg [0]) : 0;
1989 slf_arg1 = items > 1 ? SvREFCNT_inc (arg [1]) : 0;
1990 slf_arg2 = items > 2 ? SvREFCNT_inc (arg [2]) : 0;
1991
1992 PL_op->op_ppaddr = pp_slf;
1993
1994 PL_op = (OP *)&slf_restore;
1995}
1996
1997/*****************************************************************************/
1998 2157
1999static void 2158static void
2000coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2159coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2001{ 2160{
2002 SV *count_sv = AvARRAY (av)[0]; 2161 SV *count_sv = AvARRAY (av)[0];
2032 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2191 /* call $sem->adjust (0) to possibly wake up some other waiters */
2033 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2192 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2034} 2193}
2035 2194
2036static int 2195static int
2037slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2196slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2038{ 2197{
2039 AV *av = (AV *)frame->data; 2198 AV *av = (AV *)frame->data;
2040 SV *count_sv = AvARRAY (av)[0]; 2199 SV *count_sv = AvARRAY (av)[0];
2041 2200
2201 /* if we are about to throw, don't actually acquire the lock, just throw */
2202 if (CORO_THROW)
2203 return 0;
2042 if (SvIVX (count_sv) > 0) 2204 else if (SvIVX (count_sv) > 0)
2043 { 2205 {
2044 SvSTATE_current->on_destroy = 0; 2206 SvSTATE_current->on_destroy = 0;
2207
2208 if (acquire)
2045 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2209 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2210 else
2211 coro_semaphore_adjust (aTHX_ av, 0);
2212
2046 return 0; 2213 return 0;
2047 } 2214 }
2048 else 2215 else
2049 { 2216 {
2050 int i; 2217 int i;
2059 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2226 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2060 return 1; 2227 return 1;
2061 } 2228 }
2062} 2229}
2063 2230
2064static void 2231static int
2232slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2233{
2234 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2235}
2236
2237static int
2238slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2239{
2240 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2241}
2242
2243static void
2065slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2244slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2066{ 2245{
2067 AV *av = (AV *)SvRV (arg [0]); 2246 AV *av = (AV *)SvRV (arg [0]);
2068 2247
2069 if (SvIVX (AvARRAY (av)[0]) > 0) 2248 if (SvIVX (AvARRAY (av)[0]) > 0)
2070 { 2249 {
2071 frame->data = (void *)av; 2250 frame->data = (void *)av;
2072 frame->prepare = prepare_nop; 2251 frame->prepare = prepare_nop;
2073 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2074 } 2252 }
2075 else 2253 else
2076 { 2254 {
2077 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2255 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2078 2256
2079 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av)); 2257 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av));
2080 frame->prepare = prepare_schedule; 2258 frame->prepare = prepare_schedule;
2081 2259
2082 /* to avoid race conditions when a woken-up coro gets terminated */ 2260 /* to avoid race conditions when a woken-up coro gets terminated */
2083 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2261 /* we arrange for a temporary on_destroy that calls adjust (0) */
2084 assert (!SvSTATE_current->on_destroy);//D
2085 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2262 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2086 } 2263 }
2264}
2087 2265
2266static void
2267slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2268{
2269 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2088 frame->check = slf_check_semaphore_down; 2270 frame->check = slf_check_semaphore_down;
2271}
2089 2272
2273static void
2274slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2275{
2276 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2277 frame->check = slf_check_semaphore_wait;
2278}
2279
2280/* signal */
2281
2282static void
2283coro_signal_wake (pTHX_ AV *av, int count)
2284{
2285 SvIVX (AvARRAY (av)[0]) = 0;
2286
2287 /* now signal count waiters */
2288 while (count > 0 && AvFILLp (av) > 0)
2289 {
2290 SV *cb;
2291
2292 /* swap first two elements so we can shift a waiter */
2293 cb = AvARRAY (av)[0];
2294 AvARRAY (av)[0] = AvARRAY (av)[1];
2295 AvARRAY (av)[1] = cb;
2296
2297 cb = av_shift (av);
2298
2299 api_ready (aTHX_ cb);
2300 sv_setiv (cb, 0); /* signal waiter */
2301 SvREFCNT_dec (cb);
2302
2303 --count;
2304 }
2305}
2306
2307static int
2308slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2309{
2310 /* if we are about to throw, also stop waiting */
2311 return SvROK ((SV *)frame->data) && !CORO_THROW;
2312}
2313
2314static void
2315slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2316{
2317 AV *av = (AV *)SvRV (arg [0]);
2318
2319 if (SvIVX (AvARRAY (av)[0]))
2320 {
2321 SvIVX (AvARRAY (av)[0]) = 0;
2322 frame->prepare = prepare_nop;
2323 frame->check = slf_check_nop;
2324 }
2325 else
2326 {
2327 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2328
2329 av_push (av, waiter);
2330
2331 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2332 frame->prepare = prepare_schedule;
2333 frame->check = slf_check_signal_wait;
2334 }
2090} 2335}
2091 2336
2092/*****************************************************************************/ 2337/*****************************************************************************/
2338/* Coro::AIO */
2093 2339
2094#define GENSUB_ARG CvXSUBANY (cv).any_ptr 2340#define CORO_MAGIC_type_aio PERL_MAGIC_ext
2095 2341
2096/* create a closure from XS, returns a code reference */ 2342/* helper storage struct */
2097/* the arg can be accessed via GENSUB_ARG from the callback */ 2343struct io_state
2098/* the callback must use dXSARGS/XSRETURN */
2099static SV *
2100gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
2101{ 2344{
2102 CV *cv = (CV *)NEWSV (0, 0); 2345 int errorno;
2346 I32 laststype; /* U16 in 5.10.0 */
2347 int laststatval;
2348 Stat_t statcache;
2349};
2103 2350
2351static void
2352coro_aio_callback (pTHX_ CV *cv)
2353{
2354 dXSARGS;
2355 AV *state = (AV *)GENSUB_ARG;
2356 SV *coro = av_pop (state);
2357 SV *data_sv = newSV (sizeof (struct io_state));
2358
2359 av_extend (state, items);
2360
2104 sv_upgrade ((SV *)cv, SVt_PVCV); 2361 sv_upgrade (data_sv, SVt_PV);
2362 SvCUR_set (data_sv, sizeof (struct io_state));
2363 SvPOK_only (data_sv);
2105 2364
2106 CvANON_on (cv); 2365 {
2107 CvISXSUB_on (cv); 2366 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2108 CvXSUB (cv) = xsub;
2109 GENSUB_ARG = arg;
2110 2367
2111 return newRV_noinc ((SV *)cv); 2368 data->errorno = errno;
2369 data->laststype = PL_laststype;
2370 data->laststatval = PL_laststatval;
2371 data->statcache = PL_statcache;
2372 }
2373
2374 /* now build the result vector out of all the parameters and the data_sv */
2375 {
2376 int i;
2377
2378 for (i = 0; i < items; ++i)
2379 av_push (state, SvREFCNT_inc_NN (ST (i)));
2380 }
2381
2382 av_push (state, data_sv);
2383
2384 api_ready (aTHX_ coro);
2385 SvREFCNT_dec (coro);
2386 SvREFCNT_dec ((AV *)state);
2387}
2388
2389static int
2390slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2391{
2392 AV *state = (AV *)frame->data;
2393
2394 /* if we are about to throw, return early */
2395 /* this does not cancel the aio request, but at least */
2396 /* it quickly returns */
2397 if (CORO_THROW)
2398 return 0;
2399
2400 /* one element that is an RV? repeat! */
2401 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2402 return 1;
2403
2404 /* restore status */
2405 {
2406 SV *data_sv = av_pop (state);
2407 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2408
2409 errno = data->errorno;
2410 PL_laststype = data->laststype;
2411 PL_laststatval = data->laststatval;
2412 PL_statcache = data->statcache;
2413
2414 SvREFCNT_dec (data_sv);
2415 }
2416
2417 /* push result values */
2418 {
2419 dSP;
2420 int i;
2421
2422 EXTEND (SP, AvFILLp (state) + 1);
2423 for (i = 0; i <= AvFILLp (state); ++i)
2424 PUSHs (sv_2mortal (SvREFCNT_inc_NN (AvARRAY (state)[i])));
2425
2426 PUTBACK;
2427 }
2428
2429 return 0;
2430}
2431
2432static void
2433slf_init_aio_req (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2434{
2435 AV *state = (AV *)sv_2mortal ((SV *)newAV ());
2436 SV *coro_hv = SvRV (coro_current);
2437 struct coro *coro = SvSTATE_hv (coro_hv);
2438
2439 /* put our coroutine id on the state arg */
2440 av_push (state, SvREFCNT_inc_NN (coro_hv));
2441
2442 /* first see whether we have a non-zero priority and set it as AIO prio */
2443 if (coro->prio)
2444 {
2445 dSP;
2446
2447 static SV *prio_cv;
2448 static SV *prio_sv;
2449
2450 if (expect_false (!prio_cv))
2451 {
2452 prio_cv = (SV *)get_cv ("IO::AIO::aioreq_pri", 0);
2453 prio_sv = newSViv (0);
2454 }
2455
2456 PUSHMARK (SP);
2457 sv_setiv (prio_sv, coro->prio);
2458 XPUSHs (prio_sv);
2459
2460 PUTBACK;
2461 call_sv (prio_cv, G_VOID | G_DISCARD);
2462 }
2463
2464 /* now call the original request */
2465 {
2466 dSP;
2467 CV *req = (CV *)CORO_MAGIC_NN ((SV *)cv, CORO_MAGIC_type_aio)->mg_obj;
2468 int i;
2469
2470 PUSHMARK (SP);
2471
2472 /* first push all args to the stack */
2473 EXTEND (SP, items + 1);
2474
2475 for (i = 0; i < items; ++i)
2476 PUSHs (arg [i]);
2477
2478 /* now push the callback closure */
2479 PUSHs (sv_2mortal (gensub (aTHX_ coro_aio_callback, (void *)SvREFCNT_inc_NN ((SV *)state))));
2480
2481 /* now call the AIO function - we assume our request is uncancelable */
2482 PUTBACK;
2483 call_sv ((SV *)req, G_VOID | G_DISCARD);
2484 }
2485
2486 /* now that the requets is going, we loop toll we have a result */
2487 frame->data = (void *)state;
2488 frame->prepare = prepare_schedule;
2489 frame->check = slf_check_aio_req;
2490}
2491
2492static void
2493coro_aio_req_xs (pTHX_ CV *cv)
2494{
2495 dXSARGS;
2496
2497 CORO_EXECUTE_SLF_XS (slf_init_aio_req);
2498
2499 XSRETURN_EMPTY;
2112} 2500}
2113 2501
2114/*****************************************************************************/ 2502/*****************************************************************************/
2115 2503
2116MODULE = Coro::State PACKAGE = Coro::State PREFIX = api_ 2504MODULE = Coro::State PACKAGE = Coro::State PREFIX = api_
2214} 2602}
2215 OUTPUT: 2603 OUTPUT:
2216 RETVAL 2604 RETVAL
2217 2605
2218void 2606void
2219_set_stacklevel (...)
2220 CODE:
2221 api_execute_slf (aTHX_ cv, slf_init_set_stacklevel, &ST (0), items);
2222
2223void
2224transfer (...) 2607transfer (...)
2225 PROTOTYPE: $$ 2608 PROTOTYPE: $$
2226 CODE: 2609 CODE:
2227 api_execute_slf (aTHX_ cv, slf_init_transfer, &ST (0), items); 2610 CORO_EXECUTE_SLF_XS (slf_init_transfer);
2228 2611
2229bool 2612bool
2230_destroy (SV *coro_sv) 2613_destroy (SV *coro_sv)
2231 CODE: 2614 CODE:
2232 RETVAL = coro_state_destroy (aTHX_ SvSTATE (coro_sv)); 2615 RETVAL = coro_state_destroy (aTHX_ SvSTATE (coro_sv));
2350 2733
2351void 2734void
2352throw (Coro::State self, SV *throw = &PL_sv_undef) 2735throw (Coro::State self, SV *throw = &PL_sv_undef)
2353 PROTOTYPE: $;$ 2736 PROTOTYPE: $;$
2354 CODE: 2737 CODE:
2738{
2739 struct coro *current = SvSTATE_current;
2740 SV **throwp = self == current ? &CORO_THROW : &self->except;
2355 SvREFCNT_dec (self->throw); 2741 SvREFCNT_dec (*throwp);
2356 self->throw = SvOK (throw) ? newSVsv (throw) : 0; 2742 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2743}
2357 2744
2358void 2745void
2359api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB) 2746api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
2360 PROTOTYPE: $;$ 2747 PROTOTYPE: $;$
2361 C_ARGS: aTHX_ coro, flags 2748 C_ARGS: aTHX_ coro, flags
2410 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv; 2797 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv;
2411 2798
2412 SV *tmp = *src; *src = *dst; *dst = tmp; 2799 SV *tmp = *src; *src = *dst; *dst = tmp;
2413 } 2800 }
2414 2801
2802
2415MODULE = Coro::State PACKAGE = Coro 2803MODULE = Coro::State PACKAGE = Coro
2416 2804
2417BOOT: 2805BOOT:
2418{ 2806{
2419 int i; 2807 int i;
2446 coroapi.ready = api_ready; 2834 coroapi.ready = api_ready;
2447 coroapi.is_ready = api_is_ready; 2835 coroapi.is_ready = api_is_ready;
2448 coroapi.nready = coro_nready; 2836 coroapi.nready = coro_nready;
2449 coroapi.current = coro_current; 2837 coroapi.current = coro_current;
2450 2838
2451 GCoroAPI = &coroapi; 2839 /*GCoroAPI = &coroapi;*/
2452 sv_setiv (sv, (IV)&coroapi); 2840 sv_setiv (sv, (IV)&coroapi);
2453 SvREADONLY_on (sv); 2841 SvREADONLY_on (sv);
2454 } 2842 }
2455} 2843}
2456 2844
2457void 2845void
2458schedule (...) 2846schedule (...)
2459 CODE: 2847 CODE:
2460 api_execute_slf (aTHX_ cv, slf_init_schedule, &ST (0), 0); 2848 CORO_EXECUTE_SLF_XS (slf_init_schedule);
2461 2849
2462void 2850void
2463cede (...) 2851cede (...)
2464 CODE: 2852 CODE:
2465 api_execute_slf (aTHX_ cv, slf_init_cede, &ST (0), 0); 2853 CORO_EXECUTE_SLF_XS (slf_init_cede);
2466 2854
2467void 2855void
2468cede_notself (...) 2856cede_notself (...)
2469 CODE: 2857 CODE:
2470 api_execute_slf (aTHX_ cv, slf_init_cede_notself, &ST (0), 0); 2858 CORO_EXECUTE_SLF_XS (slf_init_cede_notself);
2471 2859
2472void 2860void
2473_set_current (SV *current) 2861_set_current (SV *current)
2474 PROTOTYPE: $ 2862 PROTOTYPE: $
2475 CODE: 2863 CODE:
2563 2951
2564void 2952void
2565_pool_2 (SV *cb) 2953_pool_2 (SV *cb)
2566 CODE: 2954 CODE:
2567{ 2955{
2568 struct coro *coro = SvSTATE_current; 2956 HV *hv = (HV *)SvRV (coro_current);
2957 struct coro *coro = SvSTATE_hv ((SV *)hv);
2569 2958
2570 sv_setsv (cb, &PL_sv_undef); 2959 sv_setsv (cb, &PL_sv_undef);
2571 2960
2572 SvREFCNT_dec ((SV *)PL_defoutgv); PL_defoutgv = (GV *)coro->saved_deffh; 2961 SvREFCNT_dec ((SV *)PL_defoutgv); PL_defoutgv = (GV *)coro->saved_deffh;
2573 coro->saved_deffh = 0; 2962 coro->saved_deffh = 0;
2580 SvREFCNT_dec (old); 2969 SvREFCNT_dec (old);
2581 croak ("\3async_pool terminate\2\n"); 2970 croak ("\3async_pool terminate\2\n");
2582 } 2971 }
2583 2972
2584 av_clear (GvAV (PL_defgv)); 2973 av_clear (GvAV (PL_defgv));
2585 hv_store ((HV *)SvRV (coro_current), "desc", sizeof ("desc") - 1, 2974 hv_store (hv, "desc", sizeof ("desc") - 1,
2586 newSVpvn ("[async_pool idle]", sizeof ("[async_pool idle]") - 1), 0); 2975 newSVpvn ("[async_pool idle]", sizeof ("[async_pool idle]") - 1), 0);
2587 2976
2588 coro->prio = 0; 2977 coro->prio = 0;
2589 2978
2590 if (coro->cctx && (coro->cctx->flags & CC_TRACE)) 2979 if (coro->cctx && (coro->cctx->flags & CC_TRACE))
2591 api_trace (aTHX_ coro_current, 0); 2980 api_trace (aTHX_ coro_current, 0);
2592 2981
2593 av_push (av_async_pool, newSVsv (coro_current)); 2982 av_push (av_async_pool, newSVsv (coro_current));
2594} 2983}
2595 2984
2596 2985SV *
2597MODULE = Coro::State PACKAGE = Coro::AIO 2986rouse_cb ()
2987 PROTOTYPE:
2988 CODE:
2989 RETVAL = coro_new_rouse_cb (aTHX);
2990 OUTPUT:
2991 RETVAL
2598 2992
2599void 2993void
2600_get_state (SV *self) 2994rouse_wait (SV *cb = 0)
2601 PROTOTYPE: $ 2995 PROTOTYPE: ;$
2602 PPCODE: 2996 PPCODE:
2603{ 2997 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
2604 AV *defav = GvAV (PL_defgv);
2605 AV *av = newAV ();
2606 int i;
2607 SV *data_sv = newSV (sizeof (struct io_state));
2608 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2609 SvCUR_set (data_sv, sizeof (struct io_state));
2610 SvPOK_only (data_sv);
2611
2612 data->errorno = errno;
2613 data->laststype = PL_laststype;
2614 data->laststatval = PL_laststatval;
2615 data->statcache = PL_statcache;
2616
2617 av_extend (av, AvFILLp (defav) + 1 + 1);
2618
2619 for (i = 0; i <= AvFILLp (defav); ++i)
2620 av_push (av, SvREFCNT_inc_NN (AvARRAY (defav)[i]));
2621
2622 av_push (av, data_sv);
2623
2624 XPUSHs (sv_2mortal (newRV_noinc ((SV *)av)));
2625
2626 api_ready (aTHX_ self);
2627}
2628
2629void
2630_set_state (SV *state)
2631 PROTOTYPE: $
2632 PPCODE:
2633{
2634 AV *av = (AV *)SvRV (state);
2635 struct io_state *data = (struct io_state *)SvPVX (AvARRAY (av)[AvFILLp (av)]);
2636 int i;
2637
2638 errno = data->errorno;
2639 PL_laststype = data->laststype;
2640 PL_laststatval = data->laststatval;
2641 PL_statcache = data->statcache;
2642
2643 EXTEND (SP, AvFILLp (av));
2644 for (i = 0; i < AvFILLp (av); ++i)
2645 PUSHs (sv_2mortal (SvREFCNT_inc_NN (AvARRAY (av)[i])));
2646}
2647
2648
2649MODULE = Coro::State PACKAGE = Coro::AnyEvent
2650
2651BOOT:
2652 sv_activity = coro_get_sv (aTHX_ "Coro::AnyEvent::ACTIVITY", TRUE);
2653
2654void
2655_schedule (...)
2656 CODE:
2657{
2658 static int incede;
2659
2660 api_cede_notself (aTHX);
2661
2662 ++incede;
2663 while (coro_nready >= incede && api_cede (aTHX))
2664 ;
2665
2666 sv_setsv (sv_activity, &PL_sv_undef);
2667 if (coro_nready >= incede)
2668 {
2669 PUSHMARK (SP);
2670 PUTBACK;
2671 call_pv ("Coro::AnyEvent::_activity", G_DISCARD | G_EVAL);
2672 SPAGAIN;
2673 }
2674
2675 --incede;
2676}
2677 2998
2678 2999
2679MODULE = Coro::State PACKAGE = PerlIO::cede 3000MODULE = Coro::State PACKAGE = PerlIO::cede
2680 3001
2681BOOT: 3002BOOT:
2682 PerlIO_define_layer (aTHX_ &PerlIO_cede); 3003 PerlIO_define_layer (aTHX_ &PerlIO_cede);
2683 3004
3005
2684MODULE = Coro::State PACKAGE = Coro::Semaphore 3006MODULE = Coro::State PACKAGE = Coro::Semaphore
2685 3007
2686SV * 3008SV *
2687new (SV *klass, SV *count_ = 0) 3009new (SV *klass, SV *count = 0)
2688 CODE: 3010 CODE:
2689{ 3011 RETVAL = sv_bless (
2690 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 3012 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2691 AV *av = newAV (); 3013 GvSTASH (CvGV (cv))
2692 av_push (av, newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1)); 3014 );
2693 RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv))); 3015 OUTPUT:
2694} 3016 RETVAL
3017
3018# helper for Coro::Channel
3019SV *
3020_alloc (int count)
3021 CODE:
3022 RETVAL = coro_waitarray_new (aTHX_ count);
2695 OUTPUT: 3023 OUTPUT:
2696 RETVAL 3024 RETVAL
2697 3025
2698SV * 3026SV *
2699count (SV *self) 3027count (SV *self)
2710 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1); 3038 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
2711 3039
2712void 3040void
2713down (SV *self) 3041down (SV *self)
2714 CODE: 3042 CODE:
2715 api_execute_slf (aTHX_ cv, slf_init_semaphore_down, &ST (0), 1); 3043 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3044
3045void
3046wait (SV *self)
3047 CODE:
3048 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2716 3049
2717void 3050void
2718try (SV *self) 3051try (SV *self)
2719 PPCODE: 3052 PPCODE:
2720{ 3053{
2732 XSRETURN_NO; 3065 XSRETURN_NO;
2733} 3066}
2734 3067
2735void 3068void
2736waiters (SV *self) 3069waiters (SV *self)
2737 CODE: 3070 PPCODE:
2738{ 3071{
2739 AV *av = (AV *)SvRV (self); 3072 AV *av = (AV *)SvRV (self);
3073 int wcount = AvFILLp (av) + 1 - 1;
2740 3074
2741 if (GIMME_V == G_SCALAR) 3075 if (GIMME_V == G_SCALAR)
2742 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 3076 XPUSHs (sv_2mortal (newSViv (wcount)));
2743 else 3077 else
2744 { 3078 {
2745 int i; 3079 int i;
2746 EXTEND (SP, AvFILLp (av) + 1 - 1); 3080 EXTEND (SP, wcount);
2747 for (i = 1; i <= AvFILLp (av); ++i) 3081 for (i = 1; i <= wcount; ++i)
2748 PUSHs (newSVsv (AvARRAY (av)[i])); 3082 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2749 } 3083 }
2750} 3084}
2751 3085
3086MODULE = Coro::State PACKAGE = Coro::Signal
3087
3088SV *
3089new (SV *klass)
3090 CODE:
3091 RETVAL = sv_bless (
3092 coro_waitarray_new (aTHX_ 0),
3093 GvSTASH (CvGV (cv))
3094 );
3095 OUTPUT:
3096 RETVAL
3097
3098void
3099wait (SV *self)
3100 CODE:
3101 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3102
3103void
3104broadcast (SV *self)
3105 CODE:
3106{
3107 AV *av = (AV *)SvRV (self);
3108 coro_signal_wake (aTHX_ av, AvFILLp (av));
3109}
3110
3111void
3112send (SV *self)
3113 CODE:
3114{
3115 AV *av = (AV *)SvRV (self);
3116
3117 if (AvFILLp (av))
3118 coro_signal_wake (aTHX_ av, 1);
3119 else
3120 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
3121}
3122
3123IV
3124awaited (SV *self)
3125 CODE:
3126 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3127 OUTPUT:
3128 RETVAL
3129
3130
3131MODULE = Coro::State PACKAGE = Coro::AnyEvent
3132
3133BOOT:
3134 sv_activity = coro_get_sv (aTHX_ "Coro::AnyEvent::ACTIVITY", TRUE);
3135
3136void
3137_schedule (...)
3138 CODE:
3139{
3140 static int incede;
3141
3142 api_cede_notself (aTHX);
3143
3144 ++incede;
3145 while (coro_nready >= incede && api_cede (aTHX))
3146 ;
3147
3148 sv_setsv (sv_activity, &PL_sv_undef);
3149 if (coro_nready >= incede)
3150 {
3151 PUSHMARK (SP);
3152 PUTBACK;
3153 call_pv ("Coro::AnyEvent::_activity", G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3154 }
3155
3156 --incede;
3157}
3158
3159
3160MODULE = Coro::State PACKAGE = Coro::AIO
3161
3162void
3163_register (char *target, char *proto, SV *req)
3164 CODE:
3165{
3166 HV *st;
3167 GV *gvp;
3168 CV *req_cv = sv_2cv (req, &st, &gvp, 0);
3169 /* newXSproto doesn't return the CV on 5.8 */
3170 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3171 sv_setpv ((SV *)slf_cv, proto);
3172 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
3173}
3174

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines