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.6 by root, Tue Jul 17 15:42:28 2001 UTC vs.
Revision 1.318 by root, Thu Nov 20 07:02:43 2008 UTC

1#include "libcoro/coro.c"
2
3#define PERL_NO_GET_CONTEXT
4#define PERL_EXT
5
1#include "EXTERN.h" 6#include "EXTERN.h"
2#include "perl.h" 7#include "perl.h"
3#include "XSUB.h" 8#include "XSUB.h"
9#include "perliol.h"
4 10
5#if 0 11#include "patchlevel.h"
6# define CHK(x) (void *)0 12
13#include <stdio.h>
14#include <errno.h>
15#include <assert.h>
16
17#ifdef WIN32
18# undef setjmp
19# undef longjmp
20# undef _exit
21# define setjmp _setjmp /* deep magic */
7#else 22#else
8# define CHK(x) if (!(x)) croak("FATAL, CHK: " #x) 23# include <inttypes.h> /* most portable stdint.h */
9#endif 24#endif
10 25
26#ifdef HAVE_MMAP
27# include <unistd.h>
28# include <sys/mman.h>
29# ifndef MAP_ANONYMOUS
30# ifdef MAP_ANON
31# define MAP_ANONYMOUS MAP_ANON
32# else
33# undef HAVE_MMAP
34# endif
35# endif
36# include <limits.h>
37# ifndef PAGESIZE
38# define PAGESIZE pagesize
39# define BOOT_PAGESIZE pagesize = sysconf (_SC_PAGESIZE)
40static long pagesize;
41# else
42# define BOOT_PAGESIZE (void)0
43# endif
44#else
45# define PAGESIZE 0
46# define BOOT_PAGESIZE (void)0
47#endif
48
49#if CORO_USE_VALGRIND
50# include <valgrind/valgrind.h>
51#endif
52
53/* the maximum number of idle cctx that will be pooled */
54static int cctx_max_idle = 4;
55
56#define PERL_VERSION_ATLEAST(a,b,c) \
57 (PERL_REVISION > (a) \
58 || (PERL_REVISION == (a) \
59 && (PERL_VERSION > (b) \
60 || (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))
61
62#if !PERL_VERSION_ATLEAST (5,6,0)
63# ifndef PL_ppaddr
64# define PL_ppaddr ppaddr
65# endif
66# ifndef call_sv
67# define call_sv perl_call_sv
68# endif
69# ifndef get_sv
70# define get_sv perl_get_sv
71# endif
72# ifndef get_cv
73# define get_cv perl_get_cv
74# endif
75# ifndef IS_PADGV
76# define IS_PADGV(v) 0
77# endif
78# ifndef IS_PADCONST
79# define IS_PADCONST(v) 0
80# endif
81#endif
82
83/* 5.11 */
84#ifndef CxHASARGS
85# define CxHASARGS(cx) (cx)->blk_sub.hasargs
86#endif
87
88/* 5.10.0 */
89#ifndef SvREFCNT_inc_NN
90# define SvREFCNT_inc_NN(sv) SvREFCNT_inc (sv)
91#endif
92
93/* 5.8.8 */
94#ifndef GV_NOTQUAL
95# define GV_NOTQUAL 0
96#endif
97#ifndef newSV
98# define newSV(l) NEWSV(0,l)
99#endif
100#ifndef CvISXSUB_on
101# define CvISXSUB_on(cv) (void)cv
102#endif
103#ifndef CvISXSUB
104# define CvISXSUB(cv) (CvXSUB (cv) ? TRUE : FALSE)
105#endif
106
107/* 5.8.7 */
108#ifndef SvRV_set
109# define SvRV_set(s,v) SvRV(s) = (v)
110#endif
111
112#if !__i386 && !__x86_64 && !__powerpc && !__m68k && !__alpha && !__mips && !__sparc64
113# undef CORO_STACKGUARD
114#endif
115
116#ifndef CORO_STACKGUARD
117# define CORO_STACKGUARD 0
118#endif
119
120/* prefer perl internal functions over our own? */
121#ifndef CORO_PREFER_PERL_FUNCTIONS
122# define CORO_PREFER_PERL_FUNCTIONS 0
123#endif
124
125/* The next macros try to return the current stack pointer, in an as
126 * portable way as possible. */
127#if __GNUC__ >= 4
128# define dSTACKLEVEL int stacklevel_dummy
129# define STACKLEVEL __builtin_frame_address (0)
130#else
131# define dSTACKLEVEL volatile void *stacklevel
132# define STACKLEVEL ((void *)&stacklevel)
133#endif
134
135#define IN_DESTRUCT (PL_main_cv == Nullcv)
136
137#if __GNUC__ >= 3
138# define attribute(x) __attribute__(x)
139# define expect(expr,value) __builtin_expect ((expr),(value))
140# define INLINE static inline
141#else
142# define attribute(x)
143# define expect(expr,value) (expr)
144# define INLINE static
145#endif
146
147#define expect_false(expr) expect ((expr) != 0, 0)
148#define expect_true(expr) expect ((expr) != 0, 1)
149
150#define NOINLINE attribute ((noinline))
151
152#include "CoroAPI.h"
153#define GCoroAPI (&coroapi) /* very sneaky */
154
155#ifdef USE_ITHREADS
156# if CORO_PTHREAD
157static void *coro_thx;
158# endif
159#endif
160
161static double (*nvtime)(); /* so why doesn't it take void? */
162
163/* we hijack an hopefully unused CV flag for our purposes */
164#define CVf_SLF 0x4000
165static OP *pp_slf (pTHX);
166
167static U32 cctx_gen;
168static size_t cctx_stacksize = CORO_STACKSIZE;
169static struct CoroAPI coroapi;
170static AV *main_mainstack; /* used to differentiate between $main and others */
171static JMPENV *main_top_env;
172static HV *coro_state_stash, *coro_stash;
173static volatile SV *coro_mortal; /* will be freed/thrown after next transfer */
174
175static AV *av_destroy; /* destruction queue */
176static SV *sv_manager; /* the manager coro */
177
178static GV *irsgv; /* $/ */
179static GV *stdoutgv; /* *STDOUT */
180static SV *rv_diehook;
181static SV *rv_warnhook;
182static HV *hv_sig; /* %SIG */
183
184/* async_pool helper stuff */
185static SV *sv_pool_rss;
186static SV *sv_pool_size;
187static SV *sv_async_pool_idle; /* description string */
188static AV *av_async_pool; /* idle pool */
189static SV *sv_Coro; /* class string */
190static CV *cv_pool_handler;
191static CV *cv_coro_state_new;
192
193/* Coro::AnyEvent */
194static SV *sv_activity;
195
196static struct coro_cctx *cctx_first;
197static int cctx_count, cctx_idle;
198
199enum {
200 CC_MAPPED = 0x01,
201 CC_NOREUSE = 0x02, /* throw this away after tracing */
202 CC_TRACE = 0x04,
203 CC_TRACE_SUB = 0x08, /* trace sub calls */
204 CC_TRACE_LINE = 0x10, /* trace each statement */
205 CC_TRACE_ALL = CC_TRACE_SUB | CC_TRACE_LINE,
206};
207
208/* this is a structure representing a c-level coroutine */
209typedef struct coro_cctx
210{
211 struct coro_cctx *next;
212
213 /* the stack */
214 void *sptr;
215 size_t ssize;
216
217 /* cpu state */
218 void *idle_sp; /* sp of top-level transfer/schedule/cede call */
219 JMPENV *idle_te; /* same as idle_sp, but for top_env, TODO: remove once stable */
220 JMPENV *top_env;
221 coro_context cctx;
222
223 U32 gen;
224#if CORO_USE_VALGRIND
225 int valgrind_id;
226#endif
227 unsigned char flags;
228} coro_cctx;
229
230enum {
231 CF_RUNNING = 0x0001, /* coroutine is running */
232 CF_READY = 0x0002, /* coroutine is ready */
233 CF_NEW = 0x0004, /* has never been switched to */
234 CF_DESTROYED = 0x0008, /* coroutine data has been freed */
235};
236
237/* the structure where most of the perl state is stored, overlaid on the cxstack */
238typedef struct
239{
240 SV *defsv;
241 AV *defav;
242 SV *errsv;
243 SV *irsgv;
244#define VAR(name,type) type name;
245# include "state.h"
246#undef VAR
247} perl_slots;
248
249#define SLOT_COUNT ((sizeof (perl_slots) + sizeof (PERL_CONTEXT) - 1) / sizeof (PERL_CONTEXT))
250
251/* this is a structure representing a perl-level coroutine */
11struct coro { 252struct coro {
12 U8 dowarn; 253 /* the C coroutine allocated to this perl coroutine, if any */
13 AV *defav; 254 coro_cctx *cctx;
14 255
15 PERL_SI *curstackinfo; 256 /* state data */
16 AV *curstack; 257 struct CoroSLF slf_frame; /* saved slf frame */
17 AV *mainstack; 258 AV *mainstack;
18 SV **stack_sp; 259 perl_slots *slot; /* basically the saved sp */
19 OP *op;
20 SV **curpad;
21 SV **stack_base;
22 SV **stack_max;
23 SV **tmps_stack;
24 I32 tmps_floor;
25 I32 tmps_ix;
26 I32 tmps_max;
27 I32 *markstack;
28 I32 *markstack_ptr;
29 I32 *markstack_max;
30 I32 *scopestack;
31 I32 scopestack_ix;
32 I32 scopestack_max;
33 ANY *savestack;
34 I32 savestack_ix;
35 I32 savestack_max;
36 OP **retstack;
37 I32 retstack_ix;
38 I32 retstack_max;
39 COP *curcop;
40 260
41 AV *args; 261 CV *startcv; /* the CV to execute */
262 AV *args; /* data associated with this coroutine (initial args) */
263 int refcnt; /* coroutines are refcounted, yes */
264 int flags; /* CF_ flags */
265 HV *hv; /* the perl hash associated with this coro, if any */
266 void (*on_destroy)(pTHX_ struct coro *coro);
267
268 /* statistics */
269 int usecount; /* number of transfers to this coro */
270
271 /* coro process data */
272 int prio;
273 SV *except; /* exception to be thrown */
274 SV *rouse_cb;
275
276 /* async_pool */
277 SV *saved_deffh;
278 SV *invoke_cb;
279 AV *invoke_av;
280
281 /* linked list */
282 struct coro *next, *prev;
42}; 283};
43 284
44typedef struct coro *Coro__State; 285typedef struct coro *Coro__State;
45typedef struct coro *Coro__State_or_hashref; 286typedef struct coro *Coro__State_or_hashref;
46 287
47static HV *padlist_cache; 288/* the following variables are effectively part of the perl context */
289/* and get copied between struct coro and these variables */
290/* the mainr easonw e don't support windows process emulation */
291static struct CoroSLF slf_frame; /* the current slf frame */
48 292
49/* mostly copied from op.c:cv_clone2 */ 293/** Coro ********************************************************************/
50STATIC AV * 294
51clone_padlist (AV *protopadlist) 295#define PRIO_MAX 3
296#define PRIO_HIGH 1
297#define PRIO_NORMAL 0
298#define PRIO_LOW -1
299#define PRIO_IDLE -3
300#define PRIO_MIN -4
301
302/* for Coro.pm */
303static SV *coro_current;
304static SV *coro_readyhook;
305static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1];
306static CV *cv_coro_run, *cv_coro_terminate;
307static struct coro *coro_first;
308#define coro_nready coroapi.nready
309
310/** lowlevel stuff **********************************************************/
311
312static SV *
313coro_get_sv (pTHX_ const char *name, int create)
52{ 314{
53 AV *av; 315#if PERL_VERSION_ATLEAST (5,10,0)
54 I32 ix; 316 /* silence stupid and wrong 5.10 warning that I am unable to switch off */
55 AV *protopad_name = (AV *) * av_fetch (protopadlist, 0, FALSE); 317 get_sv (name, create);
56 AV *protopad = (AV *) * av_fetch (protopadlist, 1, FALSE); 318#endif
57 SV **pname = AvARRAY (protopad_name); 319 return get_sv (name, create);
58 SV **ppad = AvARRAY (protopad); 320}
59 I32 fname = AvFILLp (protopad_name); 321
60 I32 fpad = AvFILLp (protopad); 322static AV *
323coro_get_av (pTHX_ const char *name, int create)
324{
325#if PERL_VERSION_ATLEAST (5,10,0)
326 /* silence stupid and wrong 5.10 warning that I am unable to switch off */
327 get_av (name, create);
328#endif
329 return get_av (name, create);
330}
331
332static HV *
333coro_get_hv (pTHX_ const char *name, int create)
334{
335#if PERL_VERSION_ATLEAST (5,10,0)
336 /* silence stupid and wrong 5.10 warning that I am unable to switch off */
337 get_hv (name, create);
338#endif
339 return get_hv (name, create);
340}
341
342/* may croak */
343INLINE CV *
344coro_sv_2cv (pTHX_ SV *sv)
345{
346 HV *st;
347 GV *gvp;
348 return sv_2cv (sv, &st, &gvp, 0);
349}
350
351static AV *
352coro_clone_padlist (pTHX_ CV *cv)
353{
354 AV *padlist = CvPADLIST (cv);
61 AV *newpadlist, *newpad_name, *newpad; 355 AV *newpadlist, *newpad;
62 SV **npad;
63
64 newpad_name = newAV ();
65 for (ix = fname; ix >= 0; ix--)
66 av_store (newpad_name, ix, SvREFCNT_inc (pname[ix]));
67
68 newpad = newAV ();
69 av_fill (newpad, AvFILLp (protopad));
70 npad = AvARRAY (newpad);
71 356
72 newpadlist = newAV (); 357 newpadlist = newAV ();
73 AvREAL_off (newpadlist); 358 AvREAL_off (newpadlist);
74 av_store (newpadlist, 0, (SV *) newpad_name); 359#if PERL_VERSION_ATLEAST (5,10,0)
360 Perl_pad_push (aTHX_ padlist, AvFILLp (padlist) + 1);
361#else
362 Perl_pad_push (aTHX_ padlist, AvFILLp (padlist) + 1, 1);
363#endif
364 newpad = (AV *)AvARRAY (padlist)[AvFILLp (padlist)];
365 --AvFILLp (padlist);
366
367 av_store (newpadlist, 0, SvREFCNT_inc_NN (*av_fetch (padlist, 0, FALSE)));
75 av_store (newpadlist, 1, (SV *) newpad); 368 av_store (newpadlist, 1, (SV *)newpad);
76 369
77 av = newAV (); /* will be @_ */ 370 return newpadlist;
78 av_extend (av, 0); 371}
79 av_store (newpad, 0, (SV *) av);
80 AvFLAGS (av) = AVf_REIFY;
81 372
82 for (ix = fpad; ix > 0; ix--) 373static void
374free_padlist (pTHX_ AV *padlist)
375{
376 /* may be during global destruction */
377 if (SvREFCNT (padlist))
83 { 378 {
84 SV *namesv = (ix <= fname) ? pname[ix] : Nullsv; 379 I32 i = AvFILLp (padlist);
85 if (namesv && namesv != &PL_sv_undef) 380 while (i >= 0)
86 { 381 {
87 char *name = SvPVX (namesv); /* XXX */ 382 SV **svp = av_fetch (padlist, i--, FALSE);
88 if (SvFLAGS (namesv) & SVf_FAKE || *name == '&') 383 if (svp)
89 { /* lexical from outside? */
90 npad[ix] = SvREFCNT_inc (ppad[ix]);
91 } 384 {
92 else
93 { /* our own lexical */
94 SV *sv; 385 SV *sv;
95 if (*name == '&') 386 while (&PL_sv_undef != (sv = av_pop ((AV *)*svp)))
96 sv = SvREFCNT_inc (ppad[ix]); 387 SvREFCNT_dec (sv);
97 else if (*name == '@') 388
98 sv = (SV *) newAV (); 389 SvREFCNT_dec (*svp);
99 else if (*name == '%')
100 sv = (SV *) newHV ();
101 else
102 sv = NEWSV (0, 0);
103 if (!SvPADBUSY (sv))
104 SvPADMY_on (sv);
105 npad[ix] = sv;
106 } 390 }
107 } 391 }
108 else if (IS_PADGV (ppad[ix]) || IS_PADCONST (ppad[ix])) 392
109 { 393 SvREFCNT_dec ((SV*)padlist);
110 npad[ix] = SvREFCNT_inc (ppad[ix]); 394 }
111 } 395}
112 else 396
113 { 397static int
114 SV *sv = NEWSV (0, 0); 398coro_cv_free (pTHX_ SV *sv, MAGIC *mg)
115 SvPADTMP_on (sv); 399{
116 npad[ix] = sv; 400 AV *padlist;
117 } 401 AV *av = (AV *)mg->mg_obj;
402
403 /* casting is fun. */
404 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av)))
405 free_padlist (aTHX_ padlist);
406
407 SvREFCNT_dec (av); /* sv_magicext increased the refcount */
408
409 return 0;
410}
411
412#define CORO_MAGIC_type_cv 26
413#define CORO_MAGIC_type_state PERL_MAGIC_ext
414
415static MGVTBL coro_cv_vtbl = {
416 0, 0, 0, 0,
417 coro_cv_free
418};
419
420#define CORO_MAGIC_NN(sv, type) \
421 (expect_true (SvMAGIC (sv)->mg_type == type) \
422 ? SvMAGIC (sv) \
423 : mg_find (sv, type))
424
425#define CORO_MAGIC(sv, type) \
426 (expect_true (SvMAGIC (sv)) \
427 ? CORO_MAGIC_NN (sv, type) \
428 : 0)
429
430#define CORO_MAGIC_cv(cv) CORO_MAGIC (((SV *)(cv)), CORO_MAGIC_type_cv)
431#define CORO_MAGIC_state(sv) CORO_MAGIC_NN (((SV *)(sv)), CORO_MAGIC_type_state)
432
433INLINE struct coro *
434SvSTATE_ (pTHX_ SV *coro)
435{
436 HV *stash;
437 MAGIC *mg;
438
439 if (SvROK (coro))
440 coro = SvRV (coro);
441
442 if (expect_false (SvTYPE (coro) != SVt_PVHV))
443 croak ("Coro::State object required");
444
445 stash = SvSTASH (coro);
446 if (expect_false (stash != coro_stash && stash != coro_state_stash))
118 } 447 {
119 448 /* very slow, but rare, check */
120#if 0 /* NONOTUNDERSTOOD */ 449 if (!sv_derived_from (sv_2mortal (newRV_inc (coro)), "Coro::State"))
121 /* Now that vars are all in place, clone nested closures. */ 450 croak ("Coro::State object required");
122
123 for (ix = fpad; ix > 0; ix--) {
124 SV* namesv = (ix <= fname) ? pname[ix] : Nullsv;
125 if (namesv
126 && namesv != &PL_sv_undef
127 && !(SvFLAGS(namesv) & SVf_FAKE)
128 && *SvPVX(namesv) == '&'
129 && CvCLONE(ppad[ix]))
130 {
131 CV *kid = cv_clone((CV*)ppad[ix]);
132 SvREFCNT_dec(ppad[ix]);
133 CvCLONE_on(kid);
134 SvPADMY_on(kid);
135 npad[ix] = (SV*)kid;
136 }
137 } 451 }
452
453 mg = CORO_MAGIC_state (coro);
454 return (struct coro *)mg->mg_ptr;
455}
456
457#define SvSTATE(sv) SvSTATE_ (aTHX_ (sv))
458
459/* faster than SvSTATE, but expects a coroutine hv */
460#define SvSTATE_hv(hv) ((struct coro *)CORO_MAGIC_NN ((SV *)hv, CORO_MAGIC_type_state)->mg_ptr)
461#define SvSTATE_current SvSTATE_hv (SvRV (coro_current))
462
463/* the next two functions merely cache the padlists */
464static void
465get_padlist (pTHX_ CV *cv)
466{
467 MAGIC *mg = CORO_MAGIC_cv (cv);
468 AV *av;
469
470 if (expect_true (mg && AvFILLp ((av = (AV *)mg->mg_obj)) >= 0))
471 CvPADLIST (cv) = (AV *)AvARRAY (av)[AvFILLp (av)--];
472 else
473 {
474#if CORO_PREFER_PERL_FUNCTIONS
475 /* this is probably cleaner? but also slower! */
476 /* in practise, it seems to be less stable */
477 CV *cp = Perl_cv_clone (aTHX_ cv);
478 CvPADLIST (cv) = CvPADLIST (cp);
479 CvPADLIST (cp) = 0;
480 SvREFCNT_dec (cp);
481#else
482 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv);
138#endif 483#endif
139 484 }
140 return newpadlist;
141} 485}
142 486
143STATIC AV * 487static void
144free_padlist (AV *padlist) 488put_padlist (pTHX_ CV *cv)
145{ 489{
146 /* may be during global destruction */ 490 MAGIC *mg = CORO_MAGIC_cv (cv);
147 if (SvREFCNT(padlist)) 491 AV *av;
492
493 if (expect_false (!mg))
494 mg = sv_magicext ((SV *)cv, (SV *)newAV (), CORO_MAGIC_type_cv, &coro_cv_vtbl, 0, 0);
495
496 av = (AV *)mg->mg_obj;
497
498 if (expect_false (AvFILLp (av) >= AvMAX (av)))
499 av_extend (av, AvMAX (av) + 1);
500
501 AvARRAY (av)[++AvFILLp (av)] = (SV *)CvPADLIST (cv);
502}
503
504/** load & save, init *******************************************************/
505
506static void
507load_perl (pTHX_ Coro__State c)
508{
509 perl_slots *slot = c->slot;
510 c->slot = 0;
511
512 PL_mainstack = c->mainstack;
513
514 GvSV (PL_defgv) = slot->defsv;
515 GvAV (PL_defgv) = slot->defav;
516 GvSV (PL_errgv) = slot->errsv;
517 GvSV (irsgv) = slot->irsgv;
518
519 #define VAR(name,type) PL_ ## name = slot->name;
520 # include "state.h"
521 #undef VAR
522
148 { 523 {
149 I32 i = AvFILLp(padlist); 524 dSP;
150 while (i >= 0) 525
526 CV *cv;
527
528 /* now do the ugly restore mess */
529 while (expect_true (cv = (CV *)POPs))
151 { 530 {
152 SV **svp = av_fetch(padlist, i--, FALSE); 531 put_padlist (aTHX_ cv); /* mark this padlist as available */
153 SV *sv = svp ? *svp : Nullsv; 532 CvDEPTH (cv) = PTR2IV (POPs);
154 if (sv) 533 CvPADLIST (cv) = (AV *)POPs;
155 SvREFCNT_dec(sv);
156 } 534 }
157 535
158 SvREFCNT_dec((SV*)padlist); 536 PUTBACK;
159 } 537 }
160}
161 538
162/* the next tow functions merely cache the padlists */ 539 slf_frame = c->slf_frame;
163STATIC void 540 CORO_THROW = c->except;
164get_padlist (CV *cv)
165{
166 SV **he = hv_fetch (padlist_cache, (void *)&cv, sizeof (CV *), 0);
167
168 if (he && AvFILLp ((AV *)*he) >= 0)
169 CvPADLIST (cv) = (AV *)av_pop ((AV *)*he);
170 else
171 CvPADLIST (cv) = clone_padlist (CvPADLIST (cv));
172} 541}
173 542
174STATIC void
175put_padlist (CV *cv)
176{
177 SV **he = hv_fetch (padlist_cache, (void *)&cv, sizeof (CV *), 1);
178
179 if (SvTYPE (*he) != SVt_PVAV)
180 {
181 SvREFCNT_dec (*he);
182 *he = (SV *)newAV ();
183 }
184
185 av_push ((AV *)*he, (SV *)CvPADLIST (cv));
186}
187
188static void 543static void
189save_state(pTHX_ Coro__State c) 544save_perl (pTHX_ Coro__State c)
190{ 545{
546 c->except = CORO_THROW;
547 c->slf_frame = slf_frame;
548
191 { 549 {
192 dSP; 550 dSP;
193 I32 cxix = cxstack_ix; 551 I32 cxix = cxstack_ix;
552 PERL_CONTEXT *ccstk = cxstack;
194 PERL_SI *top_si = PL_curstackinfo; 553 PERL_SI *top_si = PL_curstackinfo;
195 PERL_CONTEXT *ccstk = cxstack;
196 554
197 /* 555 /*
198 * the worst thing you can imagine happens first - we have to save 556 * the worst thing you can imagine happens first - we have to save
199 * (and reinitialize) all cv's in the whole callchain :( 557 * (and reinitialize) all cv's in the whole callchain :(
200 */ 558 */
201 559
202 PUSHs (Nullsv); 560 XPUSHs (Nullsv);
203 /* this loop was inspired by pp_caller */ 561 /* this loop was inspired by pp_caller */
204 for (;;) 562 for (;;)
205 { 563 {
206 while (cxix >= 0) 564 while (expect_true (cxix >= 0))
207 { 565 {
208 PERL_CONTEXT *cx = &ccstk[cxix--]; 566 PERL_CONTEXT *cx = &ccstk[cxix--];
209 567
210 if (CxTYPE(cx) == CXt_SUB) 568 if (expect_true (CxTYPE (cx) == CXt_SUB || CxTYPE (cx) == CXt_FORMAT))
211 { 569 {
212 CV *cv = cx->blk_sub.cv; 570 CV *cv = cx->blk_sub.cv;
571
213 if (CvDEPTH(cv)) 572 if (expect_true (CvDEPTH (cv)))
214 { 573 {
215#ifdef USE_THREADS
216 XPUSHs ((SV *)CvOWNER(cv));
217#endif
218 EXTEND (SP, 3); 574 EXTEND (SP, 3);
219 PUSHs ((SV *)CvDEPTH(cv));
220 PUSHs ((SV *)CvPADLIST(cv)); 575 PUSHs ((SV *)CvPADLIST (cv));
576 PUSHs (INT2PTR (SV *, (IV)CvDEPTH (cv)));
221 PUSHs ((SV *)cv); 577 PUSHs ((SV *)cv);
222 578
223 get_padlist (cv);
224
225 CvDEPTH(cv) = 0; 579 CvDEPTH (cv) = 0;
226#ifdef USE_THREADS 580 get_padlist (aTHX_ cv);
227 CvOWNER(cv) = 0;
228 error must unlock this cv etc.. etc...
229 if you are here wondering about this error message then
230 the reason is that it will not work as advertised yet
231#endif
232 } 581 }
233 } 582 }
234 else if (CxTYPE(cx) == CXt_FORMAT) 583 }
584
585 if (expect_true (top_si->si_type == PERLSI_MAIN))
586 break;
587
588 top_si = top_si->si_prev;
589 ccstk = top_si->si_cxstack;
590 cxix = top_si->si_cxix;
591 }
592
593 PUTBACK;
594 }
595
596 /* allocate some space on the context stack for our purposes */
597 /* we manually unroll here, as usually 2 slots is enough */
598 if (SLOT_COUNT >= 1) CXINC;
599 if (SLOT_COUNT >= 2) CXINC;
600 if (SLOT_COUNT >= 3) CXINC;
601 {
602 int i;
603 for (i = 3; i < SLOT_COUNT; ++i)
604 CXINC;
605 }
606 cxstack_ix -= SLOT_COUNT; /* undo allocation */
607
608 c->mainstack = PL_mainstack;
609
610 {
611 perl_slots *slot = c->slot = (perl_slots *)(cxstack + cxstack_ix + 1);
612
613 slot->defav = GvAV (PL_defgv);
614 slot->defsv = DEFSV;
615 slot->errsv = ERRSV;
616 slot->irsgv = GvSV (irsgv);
617
618 #define VAR(name,type) slot->name = PL_ ## name;
619 # include "state.h"
620 #undef VAR
621 }
622}
623
624/*
625 * allocate various perl stacks. This is almost an exact copy
626 * of perl.c:init_stacks, except that it uses less memory
627 * on the (sometimes correct) assumption that coroutines do
628 * not usually need a lot of stackspace.
629 */
630#if CORO_PREFER_PERL_FUNCTIONS
631# define coro_init_stacks(thx) init_stacks ()
632#else
633static void
634coro_init_stacks (pTHX)
635{
636 PL_curstackinfo = new_stackinfo(32, 8);
637 PL_curstackinfo->si_type = PERLSI_MAIN;
638 PL_curstack = PL_curstackinfo->si_stack;
639 PL_mainstack = PL_curstack; /* remember in case we switch stacks */
640
641 PL_stack_base = AvARRAY(PL_curstack);
642 PL_stack_sp = PL_stack_base;
643 PL_stack_max = PL_stack_base + AvMAX(PL_curstack);
644
645 New(50,PL_tmps_stack,32,SV*);
646 PL_tmps_floor = -1;
647 PL_tmps_ix = -1;
648 PL_tmps_max = 32;
649
650 New(54,PL_markstack,16,I32);
651 PL_markstack_ptr = PL_markstack;
652 PL_markstack_max = PL_markstack + 16;
653
654#ifdef SET_MARK_OFFSET
655 SET_MARK_OFFSET;
656#endif
657
658 New(54,PL_scopestack,8,I32);
659 PL_scopestack_ix = 0;
660 PL_scopestack_max = 8;
661
662 New(54,PL_savestack,24,ANY);
663 PL_savestack_ix = 0;
664 PL_savestack_max = 24;
665
666#if !PERL_VERSION_ATLEAST (5,10,0)
667 New(54,PL_retstack,4,OP*);
668 PL_retstack_ix = 0;
669 PL_retstack_max = 4;
670#endif
671}
672#endif
673
674/*
675 * destroy the stacks, the callchain etc...
676 */
677static void
678coro_destruct_stacks (pTHX)
679{
680 while (PL_curstackinfo->si_next)
681 PL_curstackinfo = PL_curstackinfo->si_next;
682
683 while (PL_curstackinfo)
684 {
685 PERL_SI *p = PL_curstackinfo->si_prev;
686
687 if (!IN_DESTRUCT)
688 SvREFCNT_dec (PL_curstackinfo->si_stack);
689
690 Safefree (PL_curstackinfo->si_cxstack);
691 Safefree (PL_curstackinfo);
692 PL_curstackinfo = p;
693 }
694
695 Safefree (PL_tmps_stack);
696 Safefree (PL_markstack);
697 Safefree (PL_scopestack);
698 Safefree (PL_savestack);
699#if !PERL_VERSION_ATLEAST (5,10,0)
700 Safefree (PL_retstack);
701#endif
702}
703
704#define CORO_RSS \
705 rss += sizeof (SYM (curstackinfo)); \
706 rss += (SYM (curstackinfo->si_cxmax) + 1) * sizeof (PERL_CONTEXT); \
707 rss += sizeof (SV) + sizeof (struct xpvav) + (1 + AvMAX (SYM (curstack))) * sizeof (SV *); \
708 rss += SYM (tmps_max) * sizeof (SV *); \
709 rss += (SYM (markstack_max) - SYM (markstack_ptr)) * sizeof (I32); \
710 rss += SYM (scopestack_max) * sizeof (I32); \
711 rss += SYM (savestack_max) * sizeof (ANY);
712
713static size_t
714coro_rss (pTHX_ struct coro *coro)
715{
716 size_t rss = sizeof (*coro);
717
718 if (coro->mainstack)
719 {
720 if (coro->flags & CF_RUNNING)
721 {
722 #define SYM(sym) PL_ ## sym
723 CORO_RSS;
724 #undef SYM
725 }
726 else
727 {
728 #define SYM(sym) coro->slot->sym
729 CORO_RSS;
730 #undef SYM
731 }
732 }
733
734 return rss;
735}
736
737/** coroutine stack handling ************************************************/
738
739static int (*orig_sigelem_get) (pTHX_ SV *sv, MAGIC *mg);
740static int (*orig_sigelem_set) (pTHX_ SV *sv, MAGIC *mg);
741static int (*orig_sigelem_clr) (pTHX_ SV *sv, MAGIC *mg);
742
743/* apparently < 5.8.8 */
744#ifndef MgPV_nolen_const
745#define MgPV_nolen_const(mg) (((((int)(mg)->mg_len)) == HEf_SVKEY) ? \
746 SvPV_nolen((SV*)((mg)->mg_ptr)) : \
747 (const char*)(mg)->mg_ptr)
748#endif
749
750/*
751 * This overrides the default magic get method of %SIG elements.
752 * The original one doesn't provide for reading back of PL_diehook/PL_warnhook
753 * and instead of tryign to save and restore the hash elements, we just provide
754 * readback here.
755 * We only do this when the hook is != 0, as they are often set to 0 temporarily,
756 * not expecting this to actually change the hook. This is a potential problem
757 * when a schedule happens then, but we ignore this.
758 */
759static int
760coro_sigelem_get (pTHX_ SV *sv, MAGIC *mg)
761{
762 const char *s = MgPV_nolen_const (mg);
763
764 if (*s == '_')
765 {
766 SV **svp = 0;
767
768 if (strEQ (s, "__DIE__" )) svp = &PL_diehook;
769 if (strEQ (s, "__WARN__")) svp = &PL_warnhook;
770
771 if (svp)
772 {
773 sv_setsv (sv, *svp ? *svp : &PL_sv_undef);
774 return 0;
775 }
776 }
777
778 return orig_sigelem_get ? orig_sigelem_get (aTHX_ sv, mg) : 0;
779}
780
781static int
782coro_sigelem_clr (pTHX_ SV *sv, MAGIC *mg)
783{
784 const char *s = MgPV_nolen_const (mg);
785
786 if (*s == '_')
787 {
788 SV **svp = 0;
789
790 if (strEQ (s, "__DIE__" )) svp = &PL_diehook;
791 if (strEQ (s, "__WARN__")) svp = &PL_warnhook;
792
793 if (svp)
794 {
795 SV *old = *svp;
796 *svp = 0;
797 SvREFCNT_dec (old);
798 return 0;
799 }
800 }
801
802 return orig_sigelem_clr ? orig_sigelem_clr (aTHX_ sv, mg) : 0;
803}
804
805static int
806coro_sigelem_set (pTHX_ SV *sv, MAGIC *mg)
807{
808 const char *s = MgPV_nolen_const (mg);
809
810 if (*s == '_')
811 {
812 SV **svp = 0;
813
814 if (strEQ (s, "__DIE__" )) svp = &PL_diehook;
815 if (strEQ (s, "__WARN__")) svp = &PL_warnhook;
816
817 if (svp)
818 {
819 SV *old = *svp;
820 *svp = newSVsv (sv);
821 SvREFCNT_dec (old);
822 return 0;
823 }
824 }
825
826 return orig_sigelem_set ? orig_sigelem_set (aTHX_ sv, mg) : 0;
827}
828
829static void
830prepare_nop (pTHX_ struct coro_transfer_args *ta)
831{
832 /* kind of mega-hacky, but works */
833 ta->next = ta->prev = (struct coro *)ta;
834}
835
836static int
837slf_check_nop (pTHX_ struct CoroSLF *frame)
838{
839 return 0;
840}
841
842static int
843slf_check_repeat (pTHX_ struct CoroSLF *frame)
844{
845 return 1;
846}
847
848static UNOP coro_setup_op;
849
850static void NOINLINE /* noinline to keep it out of the transfer fast path */
851coro_setup (pTHX_ struct coro *coro)
852{
853 /*
854 * emulate part of the perl startup here.
855 */
856 coro_init_stacks (aTHX);
857
858 PL_runops = RUNOPS_DEFAULT;
859 PL_curcop = &PL_compiling;
860 PL_in_eval = EVAL_NULL;
861 PL_comppad = 0;
862 PL_curpm = 0;
863 PL_curpad = 0;
864 PL_localizing = 0;
865 PL_dirty = 0;
866 PL_restartop = 0;
867#if PERL_VERSION_ATLEAST (5,10,0)
868 PL_parser = 0;
869#endif
870
871 /* recreate the die/warn hooks */
872 PL_diehook = 0; SvSetMagicSV (*hv_fetch (hv_sig, "__DIE__" , sizeof ("__DIE__" ) - 1, 1), rv_diehook );
873 PL_warnhook = 0; SvSetMagicSV (*hv_fetch (hv_sig, "__WARN__", sizeof ("__WARN__") - 1, 1), rv_warnhook);
874
875 GvSV (PL_defgv) = newSV (0);
876 GvAV (PL_defgv) = coro->args; coro->args = 0;
877 GvSV (PL_errgv) = newSV (0);
878 GvSV (irsgv) = newSVpvn ("\n", 1); sv_magic (GvSV (irsgv), (SV *)irsgv, PERL_MAGIC_sv, "/", 0);
879 PL_rs = newSVsv (GvSV (irsgv));
880 PL_defoutgv = (GV *)SvREFCNT_inc_NN (stdoutgv);
881
882 {
883 dSP;
884 UNOP myop;
885
886 Zero (&myop, 1, UNOP);
887 myop.op_next = Nullop;
888 myop.op_type = OP_ENTERSUB;
889 myop.op_flags = OPf_WANT_VOID;
890
891 PUSHMARK (SP);
892 PUSHs ((SV *)coro->startcv);
893 PUTBACK;
894 PL_op = (OP *)&myop;
895 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
896 }
897
898 /* this newly created coroutine might be run on an existing cctx which most
899 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
900 */
901 slf_frame.prepare = prepare_nop; /* provide a nop function for an eventual pp_slf */
902 slf_frame.check = slf_check_nop; /* signal pp_slf to not repeat */
903
904 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
905 coro_setup_op.op_next = PL_op;
906 coro_setup_op.op_type = OP_CUSTOM;
907 coro_setup_op.op_ppaddr = pp_slf;
908 /* no flags etc. required, as an init function won't be called */
909
910 PL_op = (OP *)&coro_setup_op;
911
912 /* copy throw, in case it was set before coro_setup */
913 CORO_THROW = coro->except;
914}
915
916static void
917coro_destruct (pTHX_ struct coro *coro)
918{
919 if (!IN_DESTRUCT)
920 {
921 /* restore all saved variables and stuff */
922 LEAVE_SCOPE (0);
923 assert (PL_tmps_floor == -1);
924
925 /* free all temporaries */
926 FREETMPS;
927 assert (PL_tmps_ix == -1);
928
929 /* unwind all extra stacks */
930 POPSTACK_TO (PL_mainstack);
931
932 /* unwind main stack */
933 dounwind (-1);
934 }
935
936 SvREFCNT_dec (GvSV (PL_defgv));
937 SvREFCNT_dec (GvAV (PL_defgv));
938 SvREFCNT_dec (GvSV (PL_errgv));
939 SvREFCNT_dec (PL_defoutgv);
940 SvREFCNT_dec (PL_rs);
941 SvREFCNT_dec (GvSV (irsgv));
942
943 SvREFCNT_dec (PL_diehook);
944 SvREFCNT_dec (PL_warnhook);
945
946 SvREFCNT_dec (coro->saved_deffh);
947 SvREFCNT_dec (coro->rouse_cb);
948 SvREFCNT_dec (coro->invoke_cb);
949 SvREFCNT_dec (coro->invoke_av);
950
951 coro_destruct_stacks (aTHX);
952}
953
954INLINE void
955free_coro_mortal (pTHX)
956{
957 if (expect_true (coro_mortal))
958 {
959 SvREFCNT_dec (coro_mortal);
960 coro_mortal = 0;
961 }
962}
963
964static int
965runops_trace (pTHX)
966{
967 COP *oldcop = 0;
968 int oldcxix = -2;
969 struct coro *coro = SvSTATE_current; /* trace cctx is tied to specific coro */
970 coro_cctx *cctx = coro->cctx;
971
972 while ((PL_op = CALL_FPTR (PL_op->op_ppaddr) (aTHX)))
973 {
974 PERL_ASYNC_CHECK ();
975
976 if (cctx->flags & CC_TRACE_ALL)
977 {
978 if (PL_op->op_type == OP_LEAVESUB && cctx->flags & CC_TRACE_SUB)
979 {
980 PERL_CONTEXT *cx = &cxstack[cxstack_ix];
981 SV **bot, **top;
982 AV *av = newAV (); /* return values */
983 SV **cb;
984 dSP;
985
986 GV *gv = CvGV (cx->blk_sub.cv);
987 SV *fullname = sv_2mortal (newSV (0));
988 if (isGV (gv))
989 gv_efullname3 (fullname, gv, 0);
990
991 bot = PL_stack_base + cx->blk_oldsp + 1;
992 top = cx->blk_gimme == G_ARRAY ? SP + 1
993 : cx->blk_gimme == G_SCALAR ? bot + 1
994 : bot;
995
996 av_extend (av, top - bot);
997 while (bot < top)
998 av_push (av, SvREFCNT_inc_NN (*bot++));
999
1000 PL_runops = RUNOPS_DEFAULT;
1001 ENTER;
1002 SAVETMPS;
1003 EXTEND (SP, 3);
1004 PUSHMARK (SP);
1005 PUSHs (&PL_sv_no);
1006 PUSHs (fullname);
1007 PUSHs (sv_2mortal (newRV_noinc ((SV *)av)));
1008 PUTBACK;
1009 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1010 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1011 SPAGAIN;
1012 FREETMPS;
1013 LEAVE;
1014 PL_runops = runops_trace;
1015 }
1016
1017 if (oldcop != PL_curcop)
1018 {
1019 oldcop = PL_curcop;
1020
1021 if (PL_curcop != &PL_compiling)
1022 {
1023 SV **cb;
1024
1025 if (oldcxix != cxstack_ix && cctx->flags & CC_TRACE_SUB)
1026 {
1027 PERL_CONTEXT *cx = &cxstack[cxstack_ix];
1028
1029 if (CxTYPE (cx) == CXt_SUB && oldcxix < cxstack_ix)
1030 {
1031 runops_proc_t old_runops = PL_runops;
1032 dSP;
1033 GV *gv = CvGV (cx->blk_sub.cv);
1034 SV *fullname = sv_2mortal (newSV (0));
1035
1036 if (isGV (gv))
1037 gv_efullname3 (fullname, gv, 0);
1038
1039 PL_runops = RUNOPS_DEFAULT;
1040 ENTER;
1041 SAVETMPS;
1042 EXTEND (SP, 3);
1043 PUSHMARK (SP);
1044 PUSHs (&PL_sv_yes);
1045 PUSHs (fullname);
1046 PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef);
1047 PUTBACK;
1048 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1049 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1050 SPAGAIN;
1051 FREETMPS;
1052 LEAVE;
1053 PL_runops = runops_trace;
1054 }
1055
1056 oldcxix = cxstack_ix;
1057 }
1058
1059 if (cctx->flags & CC_TRACE_LINE)
1060 {
1061 dSP;
1062
1063 PL_runops = RUNOPS_DEFAULT;
1064 ENTER;
1065 SAVETMPS;
1066 EXTEND (SP, 3);
1067 PL_runops = RUNOPS_DEFAULT;
1068 PUSHMARK (SP);
1069 PUSHs (sv_2mortal (newSVpv (OutCopFILE (oldcop), 0)));
1070 PUSHs (sv_2mortal (newSViv (CopLINE (oldcop))));
1071 PUTBACK;
1072 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_line_cb", sizeof ("_trace_line_cb") - 1, 0);
1073 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1074 SPAGAIN;
1075 FREETMPS;
1076 LEAVE;
1077 PL_runops = runops_trace;
1078 }
1079 }
1080 }
1081 }
1082 }
1083
1084 TAINT_NOT;
1085 return 0;
1086}
1087
1088static struct coro_cctx *cctx_ssl_cctx;
1089static struct CoroSLF cctx_ssl_frame;
1090
1091static void
1092slf_prepare_set_stacklevel (pTHX_ struct coro_transfer_args *ta)
1093{
1094 ta->prev = (struct coro *)cctx_ssl_cctx;
1095 ta->next = 0;
1096}
1097
1098static int
1099slf_check_set_stacklevel (pTHX_ struct CoroSLF *frame)
1100{
1101 *frame = cctx_ssl_frame;
1102
1103 return frame->check (aTHX_ frame); /* execute the restored frame - there must be one */
1104}
1105
1106/* initialises PL_top_env and injects a pseudo-slf-call to set the stacklevel */
1107static void NOINLINE
1108cctx_prepare (pTHX_ coro_cctx *cctx)
1109{
1110 PL_top_env = &PL_start_env;
1111
1112 if (cctx->flags & CC_TRACE)
1113 PL_runops = runops_trace;
1114
1115 /* we already must be executing an SLF op, there is no other valid way
1116 * that can lead to creation of a new cctx */
1117 assert (("FATAL: can't prepare slf-less cctx in Coro module (please report)",
1118 slf_frame.prepare && PL_op->op_ppaddr == pp_slf));
1119
1120 /* we must emulate leaving pp_slf, which is done inside slf_check_set_stacklevel */
1121 cctx_ssl_cctx = cctx;
1122 cctx_ssl_frame = slf_frame;
1123
1124 slf_frame.prepare = slf_prepare_set_stacklevel;
1125 slf_frame.check = slf_check_set_stacklevel;
1126}
1127
1128/* the tail of transfer: execute stuff we can only do after a transfer */
1129INLINE void
1130transfer_tail (pTHX)
1131{
1132 free_coro_mortal (aTHX);
1133}
1134
1135/*
1136 * this is a _very_ stripped down perl interpreter ;)
1137 */
1138static void
1139cctx_run (void *arg)
1140{
1141#ifdef USE_ITHREADS
1142# if CORO_PTHREAD
1143 PERL_SET_CONTEXT (coro_thx);
1144# endif
1145#endif
1146 {
1147 dTHX;
1148
1149 /* normally we would need to skip the entersub here */
1150 /* not doing so will re-execute it, which is exactly what we want */
1151 /* PL_nop = PL_nop->op_next */
1152
1153 /* inject a fake subroutine call to cctx_init */
1154 cctx_prepare (aTHX_ (coro_cctx *)arg);
1155
1156 /* cctx_run is the alternative tail of transfer() */
1157 transfer_tail (aTHX);
1158
1159 /* somebody or something will hit me for both perl_run and PL_restartop */
1160 PL_restartop = PL_op;
1161 perl_run (PL_curinterp);
1162 /*
1163 * Unfortunately, there is no way to get at the return values of the
1164 * coro body here, as perl_run destroys these
1165 */
1166
1167 /*
1168 * If perl-run returns we assume exit() was being called or the coro
1169 * fell off the end, which seems to be the only valid (non-bug)
1170 * reason for perl_run to return. We try to exit by jumping to the
1171 * bootstrap-time "top" top_env, as we cannot restore the "main"
1172 * coroutine as Coro has no such concept
1173 */
1174 PL_top_env = main_top_env;
1175 JMPENV_JUMP (2); /* I do not feel well about the hardcoded 2 at all */
1176 }
1177}
1178
1179static coro_cctx *
1180cctx_new ()
1181{
1182 coro_cctx *cctx;
1183
1184 ++cctx_count;
1185 New (0, cctx, 1, coro_cctx);
1186
1187 cctx->gen = cctx_gen;
1188 cctx->flags = 0;
1189 cctx->idle_sp = 0; /* can be accessed by transfer between cctx_run and set_stacklevel, on throw */
1190
1191 return cctx;
1192}
1193
1194/* create a new cctx only suitable as source */
1195static coro_cctx *
1196cctx_new_empty ()
1197{
1198 coro_cctx *cctx = cctx_new ();
1199
1200 cctx->sptr = 0;
1201 coro_create (&cctx->cctx, 0, 0, 0, 0);
1202
1203 return cctx;
1204}
1205
1206/* create a new cctx suitable as destination/running a perl interpreter */
1207static coro_cctx *
1208cctx_new_run ()
1209{
1210 coro_cctx *cctx = cctx_new ();
1211 void *stack_start;
1212 size_t stack_size;
1213
1214#if HAVE_MMAP
1215 cctx->ssize = ((cctx_stacksize * sizeof (long) + PAGESIZE - 1) / PAGESIZE + CORO_STACKGUARD) * PAGESIZE;
1216 /* mmap supposedly does allocate-on-write for us */
1217 cctx->sptr = mmap (0, cctx->ssize, PROT_EXEC|PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, 0, 0);
1218
1219 if (cctx->sptr != (void *)-1)
1220 {
1221 #if CORO_STACKGUARD
1222 mprotect (cctx->sptr, CORO_STACKGUARD * PAGESIZE, PROT_NONE);
1223 #endif
1224 stack_start = (char *)cctx->sptr + CORO_STACKGUARD * PAGESIZE;
1225 stack_size = cctx->ssize - CORO_STACKGUARD * PAGESIZE;
1226 cctx->flags |= CC_MAPPED;
1227 }
1228 else
1229#endif
1230 {
1231 cctx->ssize = cctx_stacksize * (long)sizeof (long);
1232 New (0, cctx->sptr, cctx_stacksize, long);
1233
1234 if (!cctx->sptr)
1235 {
1236 perror ("FATAL: unable to allocate stack for coroutine, exiting.");
1237 _exit (EXIT_FAILURE);
1238 }
1239
1240 stack_start = cctx->sptr;
1241 stack_size = cctx->ssize;
1242 }
1243
1244 #if CORO_USE_VALGRIND
1245 cctx->valgrind_id = VALGRIND_STACK_REGISTER ((char *)stack_start, (char *)stack_start + stack_size);
1246 #endif
1247
1248 coro_create (&cctx->cctx, cctx_run, (void *)cctx, stack_start, stack_size);
1249
1250 return cctx;
1251}
1252
1253static void
1254cctx_destroy (coro_cctx *cctx)
1255{
1256 if (!cctx)
1257 return;
1258
1259 --cctx_count;
1260 coro_destroy (&cctx->cctx);
1261
1262 /* coro_transfer creates new, empty cctx's */
1263 if (cctx->sptr)
1264 {
1265 #if CORO_USE_VALGRIND
1266 VALGRIND_STACK_DEREGISTER (cctx->valgrind_id);
1267 #endif
1268
1269#if HAVE_MMAP
1270 if (cctx->flags & CC_MAPPED)
1271 munmap (cctx->sptr, cctx->ssize);
1272 else
1273#endif
1274 Safefree (cctx->sptr);
1275 }
1276
1277 Safefree (cctx);
1278}
1279
1280/* wether this cctx should be destructed */
1281#define CCTX_EXPIRED(cctx) ((cctx)->gen != cctx_gen || ((cctx)->flags & CC_NOREUSE))
1282
1283static coro_cctx *
1284cctx_get (pTHX)
1285{
1286 while (expect_true (cctx_first))
1287 {
1288 coro_cctx *cctx = cctx_first;
1289 cctx_first = cctx->next;
1290 --cctx_idle;
1291
1292 if (expect_true (!CCTX_EXPIRED (cctx)))
1293 return cctx;
1294
1295 cctx_destroy (cctx);
1296 }
1297
1298 return cctx_new_run ();
1299}
1300
1301static void
1302cctx_put (coro_cctx *cctx)
1303{
1304 assert (("FATAL: cctx_put called on non-initialised cctx in Coro (please report)", cctx->sptr));
1305
1306 /* free another cctx if overlimit */
1307 if (expect_false (cctx_idle >= cctx_max_idle))
1308 {
1309 coro_cctx *first = cctx_first;
1310 cctx_first = first->next;
1311 --cctx_idle;
1312
1313 cctx_destroy (first);
1314 }
1315
1316 ++cctx_idle;
1317 cctx->next = cctx_first;
1318 cctx_first = cctx;
1319}
1320
1321/** coroutine switching *****************************************************/
1322
1323static void
1324transfer_check (pTHX_ struct coro *prev, struct coro *next)
1325{
1326 /* TODO: throwing up here is considered harmful */
1327
1328 if (expect_true (prev != next))
1329 {
1330 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW))))
1331 croak ("Coro::State::transfer called with non-running/new prev Coro::State, but can only transfer from running or new states,");
1332
1333 if (expect_false (next->flags & CF_RUNNING))
1334 croak ("Coro::State::transfer called with running next Coro::State, but can only transfer to inactive states,");
1335
1336 if (expect_false (next->flags & CF_DESTROYED))
1337 croak ("Coro::State::transfer called with destroyed next Coro::State, but can only transfer to inactive states,");
1338
1339#if !PERL_VERSION_ATLEAST (5,10,0)
1340 if (expect_false (PL_lex_state != LEX_NOTPARSING))
1341 croak ("Coro::State::transfer called while parsing, but this is not supported in your perl version,");
1342#endif
1343 }
1344}
1345
1346/* always use the TRANSFER macro */
1347static void NOINLINE /* noinline so we have a fixed stackframe */
1348transfer (pTHX_ struct coro *prev, struct coro *next, int force_cctx)
1349{
1350 dSTACKLEVEL;
1351
1352 /* sometimes transfer is only called to set idle_sp */
1353 if (expect_false (!next))
1354 {
1355 ((coro_cctx *)prev)->idle_sp = STACKLEVEL;
1356 assert (((coro_cctx *)prev)->idle_te = PL_top_env); /* just for the side-effect when asserts are enabled */
1357 }
1358 else if (expect_true (prev != next))
1359 {
1360 coro_cctx *prev__cctx;
1361
1362 if (expect_false (prev->flags & CF_NEW))
1363 {
1364 /* create a new empty/source context */
1365 prev->cctx = cctx_new_empty ();
1366 prev->flags &= ~CF_NEW;
1367 prev->flags |= CF_RUNNING;
1368 }
1369
1370 prev->flags &= ~CF_RUNNING;
1371 next->flags |= CF_RUNNING;
1372
1373 /* first get rid of the old state */
1374 save_perl (aTHX_ prev);
1375
1376 if (expect_false (next->flags & CF_NEW))
1377 {
1378 /* need to start coroutine */
1379 next->flags &= ~CF_NEW;
1380 /* setup coroutine call */
1381 coro_setup (aTHX_ next);
1382 }
1383 else
1384 load_perl (aTHX_ next);
1385
1386 prev__cctx = prev->cctx;
1387
1388 /* possibly untie and reuse the cctx */
1389 if (expect_true (
1390 prev__cctx->idle_sp == STACKLEVEL
1391 && !(prev__cctx->flags & CC_TRACE)
1392 && !force_cctx
1393 ))
1394 {
1395 /* I assume that stacklevel is a stronger indicator than PL_top_env changes */
1396 assert (("FATAL: current top_env must equal previous top_env in Coro (please report)", PL_top_env == prev__cctx->idle_te));
1397
1398 prev->cctx = 0;
1399
1400 /* if the cctx is about to be destroyed we need to make sure we won't see it in cctx_get */
1401 /* without this the next cctx_get might destroy the prev__cctx while still in use */
1402 if (expect_false (CCTX_EXPIRED (prev__cctx)))
1403 if (!next->cctx)
1404 next->cctx = cctx_get (aTHX);
1405
1406 cctx_put (prev__cctx);
1407 }
1408
1409 ++next->usecount;
1410
1411 if (expect_true (!next->cctx))
1412 next->cctx = cctx_get (aTHX);
1413
1414 if (expect_false (prev__cctx != next->cctx))
1415 {
1416 prev__cctx->top_env = PL_top_env;
1417 PL_top_env = next->cctx->top_env;
1418 coro_transfer (&prev__cctx->cctx, &next->cctx->cctx);
1419 }
1420
1421 transfer_tail (aTHX);
1422 }
1423}
1424
1425#define TRANSFER(ta, force_cctx) transfer (aTHX_ (ta).prev, (ta).next, (force_cctx))
1426#define TRANSFER_CHECK(ta) transfer_check (aTHX_ (ta).prev, (ta).next)
1427
1428/** high level stuff ********************************************************/
1429
1430static int
1431coro_state_destroy (pTHX_ struct coro *coro)
1432{
1433 if (coro->flags & CF_DESTROYED)
1434 return 0;
1435
1436 if (coro->on_destroy)
1437 coro->on_destroy (aTHX_ coro);
1438
1439 coro->flags |= CF_DESTROYED;
1440
1441 if (coro->flags & CF_READY)
1442 {
1443 /* reduce nready, as destroying a ready coro effectively unreadies it */
1444 /* alternative: look through all ready queues and remove the coro */
1445 --coro_nready;
1446 }
1447 else
1448 coro->flags |= CF_READY; /* make sure it is NOT put into the readyqueue */
1449
1450 if (coro->mainstack && coro->mainstack != main_mainstack)
1451 {
1452 struct coro temp;
1453
1454 assert (("FATAL: tried to destroy currently running coroutine (please report)", !(coro->flags & CF_RUNNING)));
1455
1456 save_perl (aTHX_ &temp);
1457 load_perl (aTHX_ coro);
1458
1459 coro_destruct (aTHX_ coro);
1460
1461 load_perl (aTHX_ &temp);
1462
1463 coro->slot = 0;
1464 }
1465
1466 cctx_destroy (coro->cctx);
1467 SvREFCNT_dec (coro->startcv);
1468 SvREFCNT_dec (coro->args);
1469 SvREFCNT_dec (CORO_THROW);
1470
1471 if (coro->next) coro->next->prev = coro->prev;
1472 if (coro->prev) coro->prev->next = coro->next;
1473 if (coro == coro_first) coro_first = coro->next;
1474
1475 return 1;
1476}
1477
1478static int
1479coro_state_free (pTHX_ SV *sv, MAGIC *mg)
1480{
1481 struct coro *coro = (struct coro *)mg->mg_ptr;
1482 mg->mg_ptr = 0;
1483
1484 coro->hv = 0;
1485
1486 if (--coro->refcnt < 0)
1487 {
1488 coro_state_destroy (aTHX_ coro);
1489 Safefree (coro);
1490 }
1491
1492 return 0;
1493}
1494
1495static int
1496coro_state_dup (pTHX_ MAGIC *mg, CLONE_PARAMS *params)
1497{
1498 struct coro *coro = (struct coro *)mg->mg_ptr;
1499
1500 ++coro->refcnt;
1501
1502 return 0;
1503}
1504
1505static MGVTBL coro_state_vtbl = {
1506 0, 0, 0, 0,
1507 coro_state_free,
1508 0,
1509#ifdef MGf_DUP
1510 coro_state_dup,
1511#else
1512# define MGf_DUP 0
1513#endif
1514};
1515
1516static void
1517prepare_transfer (pTHX_ struct coro_transfer_args *ta, SV *prev_sv, SV *next_sv)
1518{
1519 ta->prev = SvSTATE (prev_sv);
1520 ta->next = SvSTATE (next_sv);
1521 TRANSFER_CHECK (*ta);
1522}
1523
1524static void
1525api_transfer (pTHX_ SV *prev_sv, SV *next_sv)
1526{
1527 struct coro_transfer_args ta;
1528
1529 prepare_transfer (aTHX_ &ta, prev_sv, next_sv);
1530 TRANSFER (ta, 1);
1531}
1532
1533/*****************************************************************************/
1534/* gensub: simple closure generation utility */
1535
1536#define GENSUB_ARG CvXSUBANY (cv).any_ptr
1537
1538/* create a closure from XS, returns a code reference */
1539/* the arg can be accessed via GENSUB_ARG from the callback */
1540/* the callback must use dXSARGS/XSRETURN */
1541static SV *
1542gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
1543{
1544 CV *cv = (CV *)newSV (0);
1545
1546 sv_upgrade ((SV *)cv, SVt_PVCV);
1547
1548 CvANON_on (cv);
1549 CvISXSUB_on (cv);
1550 CvXSUB (cv) = xsub;
1551 GENSUB_ARG = arg;
1552
1553 return newRV_noinc ((SV *)cv);
1554}
1555
1556/** Coro ********************************************************************/
1557
1558INLINE void
1559coro_enq (pTHX_ struct coro *coro)
1560{
1561 av_push (coro_ready [coro->prio - PRIO_MIN], SvREFCNT_inc_NN (coro->hv));
1562}
1563
1564INLINE SV *
1565coro_deq (pTHX)
1566{
1567 int prio;
1568
1569 for (prio = PRIO_MAX - PRIO_MIN + 1; --prio >= 0; )
1570 if (AvFILLp (coro_ready [prio]) >= 0)
1571 return av_shift (coro_ready [prio]);
1572
1573 return 0;
1574}
1575
1576static int
1577api_ready (pTHX_ SV *coro_sv)
1578{
1579 struct coro *coro;
1580 SV *sv_hook;
1581 void (*xs_hook)(void);
1582
1583 if (SvROK (coro_sv))
1584 coro_sv = SvRV (coro_sv);
1585
1586 coro = SvSTATE (coro_sv);
1587
1588 if (coro->flags & CF_READY)
1589 return 0;
1590
1591 coro->flags |= CF_READY;
1592
1593 sv_hook = coro_nready ? 0 : coro_readyhook;
1594 xs_hook = coro_nready ? 0 : coroapi.readyhook;
1595
1596 coro_enq (aTHX_ coro);
1597 ++coro_nready;
1598
1599 if (sv_hook)
1600 {
1601 dSP;
1602
1603 ENTER;
1604 SAVETMPS;
1605
1606 PUSHMARK (SP);
1607 PUTBACK;
1608 call_sv (sv_hook, G_VOID | G_DISCARD);
1609
1610 FREETMPS;
1611 LEAVE;
1612 }
1613
1614 if (xs_hook)
1615 xs_hook ();
1616
1617 return 1;
1618}
1619
1620static int
1621api_is_ready (pTHX_ SV *coro_sv)
1622{
1623 return !!(SvSTATE (coro_sv)->flags & CF_READY);
1624}
1625
1626/* expects to own a reference to next->hv */
1627INLINE void
1628prepare_schedule_to (pTHX_ struct coro_transfer_args *ta, struct coro *next)
1629{
1630 SV *prev_sv = SvRV (coro_current);
1631
1632 ta->prev = SvSTATE_hv (prev_sv);
1633 ta->next = next;
1634
1635 TRANSFER_CHECK (*ta);
1636
1637 SvRV_set (coro_current, (SV *)next->hv);
1638
1639 free_coro_mortal (aTHX);
1640 coro_mortal = prev_sv;
1641}
1642
1643static void
1644prepare_schedule (pTHX_ struct coro_transfer_args *ta)
1645{
1646 for (;;)
1647 {
1648 SV *next_sv = coro_deq (aTHX);
1649
1650 if (expect_true (next_sv))
1651 {
1652 struct coro *next = SvSTATE_hv (next_sv);
1653
1654 /* cannot transfer to destroyed coros, skip and look for next */
1655 if (expect_false (next->flags & CF_DESTROYED))
1656 SvREFCNT_dec (next_sv); /* coro_nready has already been taken care of by destroy */
1657 else
1658 {
1659 next->flags &= ~CF_READY;
1660 --coro_nready;
1661
1662 return prepare_schedule_to (aTHX_ ta, next);
1663 }
1664 }
1665 else
1666 {
1667 /* nothing to schedule: call the idle handler */
1668 dSP;
1669
1670 ENTER;
1671 SAVETMPS;
1672
1673 PUSHMARK (SP);
1674 PUTBACK;
1675 call_sv (get_sv ("Coro::idle", FALSE), G_VOID | G_DISCARD);
1676
1677 FREETMPS;
1678 LEAVE;
1679 }
1680 }
1681}
1682
1683INLINE void
1684prepare_cede (pTHX_ struct coro_transfer_args *ta)
1685{
1686 api_ready (aTHX_ coro_current);
1687 prepare_schedule (aTHX_ ta);
1688}
1689
1690INLINE void
1691prepare_cede_notself (pTHX_ struct coro_transfer_args *ta)
1692{
1693 SV *prev = SvRV (coro_current);
1694
1695 if (coro_nready)
1696 {
1697 prepare_schedule (aTHX_ ta);
1698 api_ready (aTHX_ prev);
1699 }
1700 else
1701 prepare_nop (aTHX_ ta);
1702}
1703
1704static void
1705api_schedule (pTHX)
1706{
1707 struct coro_transfer_args ta;
1708
1709 prepare_schedule (aTHX_ &ta);
1710 TRANSFER (ta, 1);
1711}
1712
1713static void
1714api_schedule_to (pTHX_ SV *coro_sv)
1715{
1716 struct coro_transfer_args ta;
1717 struct coro *next = SvSTATE (coro_sv);
1718
1719 SvREFCNT_inc_NN (coro_sv);
1720 prepare_schedule_to (aTHX_ &ta, next);
1721}
1722
1723static int
1724api_cede (pTHX)
1725{
1726 struct coro_transfer_args ta;
1727
1728 prepare_cede (aTHX_ &ta);
1729
1730 if (expect_true (ta.prev != ta.next))
1731 {
1732 TRANSFER (ta, 1);
1733 return 1;
1734 }
1735 else
1736 return 0;
1737}
1738
1739static int
1740api_cede_notself (pTHX)
1741{
1742 if (coro_nready)
1743 {
1744 struct coro_transfer_args ta;
1745
1746 prepare_cede_notself (aTHX_ &ta);
1747 TRANSFER (ta, 1);
1748 return 1;
1749 }
1750 else
1751 return 0;
1752}
1753
1754static void
1755api_trace (pTHX_ SV *coro_sv, int flags)
1756{
1757 struct coro *coro = SvSTATE (coro_sv);
1758
1759 if (flags & CC_TRACE)
1760 {
1761 if (!coro->cctx)
1762 coro->cctx = cctx_new_run ();
1763 else if (!(coro->cctx->flags & CC_TRACE))
1764 croak ("cannot enable tracing on coroutine with custom stack,");
1765
1766 coro->cctx->flags |= CC_NOREUSE | (flags & (CC_TRACE | CC_TRACE_ALL));
1767 }
1768 else if (coro->cctx && coro->cctx->flags & CC_TRACE)
1769 {
1770 coro->cctx->flags &= ~(CC_TRACE | CC_TRACE_ALL);
1771
1772 if (coro->flags & CF_RUNNING)
1773 PL_runops = RUNOPS_DEFAULT;
1774 else
1775 coro->slot->runops = RUNOPS_DEFAULT;
1776 }
1777}
1778
1779static void
1780coro_call_on_destroy (pTHX_ struct coro *coro)
1781{
1782 SV **on_destroyp = hv_fetch (coro->hv, "_on_destroy", sizeof ("_on_destroy") - 1, 0);
1783 SV **statusp = hv_fetch (coro->hv, "_status", sizeof ("_status") - 1, 0);
1784
1785 if (on_destroyp)
1786 {
1787 AV *on_destroy = (AV *)SvRV (*on_destroyp);
1788
1789 while (AvFILLp (on_destroy) >= 0)
1790 {
1791 dSP; /* don't disturb outer sp */
1792 SV *cb = av_pop (on_destroy);
1793
1794 PUSHMARK (SP);
1795
1796 if (statusp)
1797 {
1798 int i;
1799 AV *status = (AV *)SvRV (*statusp);
1800 EXTEND (SP, AvFILLp (status) + 1);
1801
1802 for (i = 0; i <= AvFILLp (status); ++i)
1803 PUSHs (AvARRAY (status)[i]);
1804 }
1805
1806 PUTBACK;
1807 call_sv (sv_2mortal (cb), G_VOID | G_DISCARD);
1808 }
1809 }
1810}
1811
1812static void
1813slf_init_terminate (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1814{
1815 int i;
1816 HV *hv = (HV *)SvRV (coro_current);
1817 AV *av = newAV ();
1818
1819 av_extend (av, items - 1);
1820 for (i = 0; i < items; ++i)
1821 av_push (av, SvREFCNT_inc_NN (arg [i]));
1822
1823 hv_store (hv, "_status", sizeof ("_status") - 1, newRV_noinc ((SV *)av), 0);
1824
1825 av_push (av_destroy, (SV *)newRV_inc ((SV *)hv)); /* RVinc for perl */
1826 api_ready (aTHX_ sv_manager);
1827
1828 frame->prepare = prepare_schedule;
1829 frame->check = slf_check_repeat;
1830}
1831
1832/*****************************************************************************/
1833/* async pool handler */
1834
1835static int
1836slf_check_pool_handler (pTHX_ struct CoroSLF *frame)
1837{
1838 HV *hv = (HV *)SvRV (coro_current);
1839 struct coro *coro = (struct coro *)frame->data;
1840
1841 if (!coro->invoke_cb)
1842 return 1; /* loop till we have invoke */
1843 else
1844 {
1845 hv_store (hv, "desc", sizeof ("desc") - 1,
1846 newSVpvn ("[async_pool]", sizeof ("[async_pool]") - 1), 0);
1847
1848 coro->saved_deffh = SvREFCNT_inc_NN ((SV *)PL_defoutgv);
1849
1850 {
1851 dSP;
1852 XPUSHs (sv_2mortal (coro->invoke_cb)); coro->invoke_cb = 0;
1853 PUTBACK;
1854 }
1855
1856 SvREFCNT_dec (GvAV (PL_defgv));
1857 GvAV (PL_defgv) = coro->invoke_av;
1858 coro->invoke_av = 0;
1859
1860 return 0;
1861 }
1862}
1863
1864static void
1865slf_init_pool_handler (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1866{
1867 HV *hv = (HV *)SvRV (coro_current);
1868 struct coro *coro = SvSTATE_hv ((SV *)hv);
1869
1870 if (expect_true (coro->saved_deffh))
1871 {
1872 /* subsequent iteration */
1873 SvREFCNT_dec ((SV *)PL_defoutgv); PL_defoutgv = (GV *)coro->saved_deffh;
1874 coro->saved_deffh = 0;
1875
1876 if (coro_rss (aTHX_ coro) > SvUV (sv_pool_rss)
1877 || av_len (av_async_pool) + 1 >= SvIV (sv_pool_size))
1878 {
1879 coro->invoke_cb = SvREFCNT_inc_NN ((SV *)cv_coro_terminate);
1880 coro->invoke_av = newAV ();
1881
1882 frame->prepare = prepare_nop;
1883 }
1884 else
1885 {
1886 av_clear (GvAV (PL_defgv));
1887 hv_store (hv, "desc", sizeof ("desc") - 1, SvREFCNT_inc_NN (sv_async_pool_idle), 0);
1888
1889 coro->prio = 0;
1890
1891 if (coro->cctx && (coro->cctx->flags & CC_TRACE))
1892 api_trace (aTHX_ coro_current, 0);
1893
1894 frame->prepare = prepare_schedule;
1895 av_push (av_async_pool, SvREFCNT_inc (hv));
1896 }
1897 }
1898 else
1899 {
1900 /* first iteration, simply fall through */
1901 frame->prepare = prepare_nop;
1902 }
1903
1904 frame->check = slf_check_pool_handler;
1905 frame->data = (void *)coro;
1906}
1907
1908/*****************************************************************************/
1909/* rouse callback */
1910
1911#define CORO_MAGIC_type_rouse PERL_MAGIC_ext
1912
1913static void
1914coro_rouse_callback (pTHX_ CV *cv)
1915{
1916 dXSARGS;
1917 SV *data = (SV *)GENSUB_ARG;
1918
1919 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1920 {
1921 /* first call, set args */
1922 AV *av = newAV ();
1923 SV *coro = SvRV (data);
1924
1925 SvRV_set (data, (SV *)av);
1926 api_ready (aTHX_ coro);
1927 SvREFCNT_dec (coro);
1928
1929 /* better take a full copy of the arguments */
1930 while (items--)
1931 av_store (av, items, newSVsv (ST (items)));
1932 }
1933
1934 XSRETURN_EMPTY;
1935}
1936
1937static int
1938slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
1939{
1940 SV *data = (SV *)frame->data;
1941
1942 if (CORO_THROW)
1943 return 0;
1944
1945 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1946 return 1;
1947
1948 /* now push all results on the stack */
1949 {
1950 dSP;
1951 AV *av = (AV *)SvRV (data);
1952 int i;
1953
1954 EXTEND (SP, AvFILLp (av) + 1);
1955 for (i = 0; i <= AvFILLp (av); ++i)
1956 PUSHs (sv_2mortal (AvARRAY (av)[i]));
1957
1958 /* we have stolen the elements, so ste length to zero and free */
1959 AvFILLp (av) = -1;
1960 av_undef (av);
1961
1962 PUTBACK;
1963 }
1964
1965 return 0;
1966}
1967
1968static void
1969slf_init_rouse_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1970{
1971 SV *cb;
1972
1973 if (items)
1974 cb = arg [0];
1975 else
1976 {
1977 struct coro *coro = SvSTATE_current;
1978
1979 if (!coro->rouse_cb)
1980 croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
1981
1982 cb = sv_2mortal (coro->rouse_cb);
1983 coro->rouse_cb = 0;
1984 }
1985
1986 if (!SvROK (cb)
1987 || SvTYPE (SvRV (cb)) != SVt_PVCV
1988 || CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
1989 croak ("Coro::rouse_wait called with illegal callback argument,");
1990
1991 {
1992 CV *cv = (CV *)SvRV (cb); /* for GENSUB_ARG */
1993 SV *data = (SV *)GENSUB_ARG;
1994
1995 frame->data = (void *)data;
1996 frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
1997 frame->check = slf_check_rouse_wait;
1998 }
1999}
2000
2001static SV *
2002coro_new_rouse_cb (pTHX)
2003{
2004 HV *hv = (HV *)SvRV (coro_current);
2005 struct coro *coro = SvSTATE_hv (hv);
2006 SV *data = newRV_inc ((SV *)hv);
2007 SV *cb = gensub (aTHX_ coro_rouse_callback, (void *)data);
2008
2009 sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
2010 SvREFCNT_dec (data); /* magicext increases the refcount */
2011
2012 SvREFCNT_dec (coro->rouse_cb);
2013 coro->rouse_cb = SvREFCNT_inc_NN (cb);
2014
2015 return cb;
2016}
2017
2018/*****************************************************************************/
2019/* schedule-like-function opcode (SLF) */
2020
2021static UNOP slf_restore; /* restore stack as entersub did, for first-re-run */
2022static const CV *slf_cv;
2023static SV **slf_argv;
2024static int slf_argc, slf_arga; /* count, allocated */
2025static I32 slf_ax; /* top of stack, for restore */
2026
2027/* this restores the stack in the case we patched the entersub, to */
2028/* recreate the stack frame as perl will on following calls */
2029/* since entersub cleared the stack */
2030static OP *
2031pp_restore (pTHX)
2032{
2033 int i;
2034 SV **SP = PL_stack_base + slf_ax;
2035
2036 PUSHMARK (SP);
2037
2038 EXTEND (SP, slf_argc + 1);
2039
2040 for (i = 0; i < slf_argc; ++i)
2041 PUSHs (sv_2mortal (slf_argv [i]));
2042
2043 PUSHs ((SV *)CvGV (slf_cv));
2044
2045 RETURNOP (slf_restore.op_first);
2046}
2047
2048static void
2049slf_prepare_transfer (pTHX_ struct coro_transfer_args *ta)
2050{
2051 SV **arg = (SV **)slf_frame.data;
2052
2053 prepare_transfer (aTHX_ ta, arg [0], arg [1]);
2054}
2055
2056static void
2057slf_init_transfer (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2058{
2059 if (items != 2)
2060 croak ("Coro::State::transfer (prev, next) expects two arguments, not %d,", items);
2061
2062 frame->prepare = slf_prepare_transfer;
2063 frame->check = slf_check_nop;
2064 frame->data = (void *)arg; /* let's hope it will stay valid */
2065}
2066
2067static void
2068slf_init_schedule (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2069{
2070 frame->prepare = prepare_schedule;
2071 frame->check = slf_check_nop;
2072}
2073
2074static void
2075slf_prepare_schedule_to (pTHX_ struct coro_transfer_args *ta)
2076{
2077 struct coro *next = (struct coro *)slf_frame.data;
2078
2079 SvREFCNT_inc_NN (next->hv);
2080 prepare_schedule_to (aTHX_ ta, next);
2081}
2082
2083static void
2084slf_init_schedule_to (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2085{
2086 if (!items)
2087 croak ("Coro::schedule_to expects a coroutine argument, caught");
2088
2089 frame->data = (void *)SvSTATE (arg [0]);
2090 frame->prepare = slf_prepare_schedule_to;
2091 frame->check = slf_check_nop;
2092}
2093
2094static void
2095slf_init_cede_to (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2096{
2097 api_ready (aTHX_ SvRV (coro_current));
2098
2099 slf_init_schedule_to (aTHX_ frame, cv, arg, items);
2100}
2101
2102static void
2103slf_init_cede (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2104{
2105 frame->prepare = prepare_cede;
2106 frame->check = slf_check_nop;
2107}
2108
2109static void
2110slf_init_cede_notself (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2111{
2112 frame->prepare = prepare_cede_notself;
2113 frame->check = slf_check_nop;
2114}
2115
2116/*
2117 * these not obviously related functions are all rolled into one
2118 * function to increase chances that they all will call transfer with the same
2119 * stack offset
2120 * SLF stands for "schedule-like-function".
2121 */
2122static OP *
2123pp_slf (pTHX)
2124{
2125 I32 checkmark; /* mark SP to see how many elements check has pushed */
2126
2127 /* set up the slf frame, unless it has already been set-up */
2128 /* the latter happens when a new coro has been started */
2129 /* or when a new cctx was attached to an existing coroutine */
2130 if (expect_true (!slf_frame.prepare))
2131 {
2132 /* first iteration */
2133 dSP;
2134 SV **arg = PL_stack_base + TOPMARK + 1;
2135 int items = SP - arg; /* args without function object */
2136 SV *gv = *sp;
2137
2138 /* do a quick consistency check on the "function" object, and if it isn't */
2139 /* for us, divert to the real entersub */
2140 if (SvTYPE (gv) != SVt_PVGV
2141 || !GvCV (gv)
2142 || !(CvFLAGS (GvCV (gv)) & CVf_SLF))
2143 return PL_ppaddr[OP_ENTERSUB](aTHX);
2144
2145 if (!(PL_op->op_flags & OPf_STACKED))
2146 {
2147 /* ampersand-form of call, use @_ instead of stack */
2148 AV *av = GvAV (PL_defgv);
2149 arg = AvARRAY (av);
2150 items = AvFILLp (av) + 1;
2151 }
2152
2153 /* now call the init function, which needs to set up slf_frame */
2154 ((coro_slf_cb)CvXSUBANY (GvCV (gv)).any_ptr)
2155 (aTHX_ &slf_frame, GvCV (gv), arg, items);
2156
2157 /* pop args */
2158 SP = PL_stack_base + POPMARK;
2159
2160 PUTBACK;
2161 }
2162
2163 /* now that we have a slf_frame, interpret it! */
2164 /* we use a callback system not to make the code needlessly */
2165 /* complicated, but so we can run multiple perl coros from one cctx */
2166
2167 do
2168 {
2169 struct coro_transfer_args ta;
2170
2171 slf_frame.prepare (aTHX_ &ta);
2172 TRANSFER (ta, 0);
2173
2174 checkmark = PL_stack_sp - PL_stack_base;
2175 }
2176 while (slf_frame.check (aTHX_ &slf_frame));
2177
2178 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
2179
2180 /* exception handling */
2181 if (expect_false (CORO_THROW))
2182 {
2183 SV *exception = sv_2mortal (CORO_THROW);
2184
2185 CORO_THROW = 0;
2186 sv_setsv (ERRSV, exception);
2187 croak (0);
2188 }
2189
2190 /* return value handling - mostly like entersub */
2191 /* make sure we put something on the stack in scalar context */
2192 if (GIMME_V == G_SCALAR)
2193 {
2194 dSP;
2195 SV **bot = PL_stack_base + checkmark;
2196
2197 if (sp == bot) /* too few, push undef */
2198 bot [1] = &PL_sv_undef;
2199 else if (sp != bot + 1) /* too many, take last one */
2200 bot [1] = *sp;
2201
2202 SP = bot + 1;
2203
2204 PUTBACK;
2205 }
2206
2207 return NORMAL;
2208}
2209
2210static void
2211api_execute_slf (pTHX_ CV *cv, coro_slf_cb init_cb, I32 ax)
2212{
2213 int i;
2214 SV **arg = PL_stack_base + ax;
2215 int items = PL_stack_sp - arg + 1;
2216
2217 assert (("FATAL: SLF call with illegal CV value", !CvANON (cv)));
2218
2219 if (PL_op->op_ppaddr != PL_ppaddr [OP_ENTERSUB]
2220 && PL_op->op_ppaddr != pp_slf)
2221 croak ("FATAL: Coro SLF calls can only be made normally, not via goto or any other means, caught");
2222
2223 CvFLAGS (cv) |= CVf_SLF;
2224 CvXSUBANY (cv).any_ptr = (void *)init_cb;
2225 slf_cv = cv;
2226
2227 /* we patch the op, and then re-run the whole call */
2228 /* we have to put the same argument on the stack for this to work */
2229 /* and this will be done by pp_restore */
2230 slf_restore.op_next = (OP *)&slf_restore;
2231 slf_restore.op_type = OP_CUSTOM;
2232 slf_restore.op_ppaddr = pp_restore;
2233 slf_restore.op_first = PL_op;
2234
2235 slf_ax = ax - 1; /* undo the ax++ inside dAXMARK */
2236
2237 if (PL_op->op_flags & OPf_STACKED)
2238 {
2239 if (items > slf_arga)
2240 {
2241 slf_arga = items;
2242 free (slf_argv);
2243 slf_argv = malloc (slf_arga * sizeof (SV *));
2244 }
2245
2246 slf_argc = items;
2247
2248 for (i = 0; i < items; ++i)
2249 slf_argv [i] = SvREFCNT_inc (arg [i]);
2250 }
2251 else
2252 slf_argc = 0;
2253
2254 PL_op->op_ppaddr = pp_slf;
2255 PL_op->op_type = OP_CUSTOM; /* maybe we should leave it at entersub? */
2256
2257 PL_op = (OP *)&slf_restore;
2258}
2259
2260/*****************************************************************************/
2261/* PerlIO::cede */
2262
2263typedef struct
2264{
2265 PerlIOBuf base;
2266 NV next, every;
2267} PerlIOCede;
2268
2269static IV
2270PerlIOCede_pushed (pTHX_ PerlIO *f, const char *mode, SV *arg, PerlIO_funcs *tab)
2271{
2272 PerlIOCede *self = PerlIOSelf (f, PerlIOCede);
2273
2274 self->every = SvCUR (arg) ? SvNV (arg) : 0.01;
2275 self->next = nvtime () + self->every;
2276
2277 return PerlIOBuf_pushed (aTHX_ f, mode, Nullsv, tab);
2278}
2279
2280static SV *
2281PerlIOCede_getarg (pTHX_ PerlIO *f, CLONE_PARAMS *param, int flags)
2282{
2283 PerlIOCede *self = PerlIOSelf (f, PerlIOCede);
2284
2285 return newSVnv (self->every);
2286}
2287
2288static IV
2289PerlIOCede_flush (pTHX_ PerlIO *f)
2290{
2291 PerlIOCede *self = PerlIOSelf (f, PerlIOCede);
2292 double now = nvtime ();
2293
2294 if (now >= self->next)
2295 {
2296 api_cede (aTHX);
2297 self->next = now + self->every;
2298 }
2299
2300 return PerlIOBuf_flush (aTHX_ f);
2301}
2302
2303static PerlIO_funcs PerlIO_cede =
2304{
2305 sizeof(PerlIO_funcs),
2306 "cede",
2307 sizeof(PerlIOCede),
2308 PERLIO_K_DESTRUCT | PERLIO_K_RAW,
2309 PerlIOCede_pushed,
2310 PerlIOBuf_popped,
2311 PerlIOBuf_open,
2312 PerlIOBase_binmode,
2313 PerlIOCede_getarg,
2314 PerlIOBase_fileno,
2315 PerlIOBuf_dup,
2316 PerlIOBuf_read,
2317 PerlIOBuf_unread,
2318 PerlIOBuf_write,
2319 PerlIOBuf_seek,
2320 PerlIOBuf_tell,
2321 PerlIOBuf_close,
2322 PerlIOCede_flush,
2323 PerlIOBuf_fill,
2324 PerlIOBase_eof,
2325 PerlIOBase_error,
2326 PerlIOBase_clearerr,
2327 PerlIOBase_setlinebuf,
2328 PerlIOBuf_get_base,
2329 PerlIOBuf_bufsiz,
2330 PerlIOBuf_get_ptr,
2331 PerlIOBuf_get_cnt,
2332 PerlIOBuf_set_ptrcnt,
2333};
2334
2335/*****************************************************************************/
2336/* Coro::Semaphore & Coro::Signal */
2337
2338static SV *
2339coro_waitarray_new (pTHX_ int count)
2340{
2341 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2342 AV *av = newAV ();
2343 SV **ary;
2344
2345 /* unfortunately, building manually saves memory */
2346 Newx (ary, 2, SV *);
2347 AvALLOC (av) = ary;
2348 /*AvARRAY (av) = ary;*/
2349 SvPVX ((SV *)av) = (char *)ary; /* 5.8.8 needs this syntax instead of AvARRAY = ary */
2350 AvMAX (av) = 1;
2351 AvFILLp (av) = 0;
2352 ary [0] = newSViv (count);
2353
2354 return newRV_noinc ((SV *)av);
2355}
2356
2357/* semaphore */
2358
2359static void
2360coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2361{
2362 SV *count_sv = AvARRAY (av)[0];
2363 IV count = SvIVX (count_sv);
2364
2365 count += adjust;
2366 SvIVX (count_sv) = count;
2367
2368 /* now wake up as many waiters as are expected to lock */
2369 while (count > 0 && AvFILLp (av) > 0)
2370 {
2371 SV *cb;
2372
2373 /* swap first two elements so we can shift a waiter */
2374 AvARRAY (av)[0] = AvARRAY (av)[1];
2375 AvARRAY (av)[1] = count_sv;
2376 cb = av_shift (av);
2377
2378 if (SvOBJECT (cb))
2379 {
2380 api_ready (aTHX_ cb);
2381 --count;
2382 }
2383 else if (SvTYPE (cb) == SVt_PVCV)
2384 {
2385 dSP;
2386 PUSHMARK (SP);
2387 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2388 PUTBACK;
2389 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2390 }
2391
2392 SvREFCNT_dec (cb);
2393 }
2394}
2395
2396static void
2397coro_semaphore_on_destroy (pTHX_ struct coro *coro)
2398{
2399 /* call $sem->adjust (0) to possibly wake up some other waiters */
2400 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2401}
2402
2403static int
2404slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2405{
2406 AV *av = (AV *)frame->data;
2407 SV *count_sv = AvARRAY (av)[0];
2408
2409 /* if we are about to throw, don't actually acquire the lock, just throw */
2410 if (CORO_THROW)
2411 return 0;
2412 else if (SvIVX (count_sv) > 0)
2413 {
2414 SvSTATE_current->on_destroy = 0;
2415
2416 if (acquire)
2417 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2418 else
2419 coro_semaphore_adjust (aTHX_ av, 0);
2420
2421 return 0;
2422 }
2423 else
2424 {
2425 int i;
2426 /* if we were woken up but can't down, we look through the whole */
2427 /* waiters list and only add us if we aren't in there already */
2428 /* this avoids some degenerate memory usage cases */
2429
2430 for (i = 1; i <= AvFILLp (av); ++i)
2431 if (AvARRAY (av)[i] == SvRV (coro_current))
2432 return 1;
2433
2434 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2435 return 1;
2436 }
2437}
2438
2439static int
2440slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2441{
2442 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2443}
2444
2445static int
2446slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2447{
2448 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2449}
2450
2451static void
2452slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2453{
2454 AV *av = (AV *)SvRV (arg [0]);
2455
2456 if (SvIVX (AvARRAY (av)[0]) > 0)
2457 {
2458 frame->data = (void *)av;
2459 frame->prepare = prepare_nop;
2460 }
2461 else
2462 {
2463 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2464
2465 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av));
2466 frame->prepare = prepare_schedule;
2467
2468 /* to avoid race conditions when a woken-up coro gets terminated */
2469 /* we arrange for a temporary on_destroy that calls adjust (0) */
2470 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2471 }
2472}
2473
2474static void
2475slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2476{
2477 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2478 frame->check = slf_check_semaphore_down;
2479}
2480
2481static void
2482slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2483{
2484 if (items >= 2)
2485 {
2486 /* callback form */
2487 AV *av = (AV *)SvRV (arg [0]);
2488 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2489
2490 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
2491
2492 if (SvIVX (AvARRAY (av)[0]) > 0)
2493 coro_semaphore_adjust (aTHX_ av, 0);
2494
2495 frame->prepare = prepare_nop;
2496 frame->check = slf_check_nop;
2497 }
2498 else
2499 {
2500 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2501 frame->check = slf_check_semaphore_wait;
2502 }
2503}
2504
2505/* signal */
2506
2507static void
2508coro_signal_wake (pTHX_ AV *av, int count)
2509{
2510 SvIVX (AvARRAY (av)[0]) = 0;
2511
2512 /* now signal count waiters */
2513 while (count > 0 && AvFILLp (av) > 0)
2514 {
2515 SV *cb;
2516
2517 /* swap first two elements so we can shift a waiter */
2518 cb = AvARRAY (av)[0];
2519 AvARRAY (av)[0] = AvARRAY (av)[1];
2520 AvARRAY (av)[1] = cb;
2521
2522 cb = av_shift (av);
2523
2524 api_ready (aTHX_ cb);
2525 sv_setiv (cb, 0); /* signal waiter */
2526 SvREFCNT_dec (cb);
2527
2528 --count;
2529 }
2530}
2531
2532static int
2533slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2534{
2535 /* if we are about to throw, also stop waiting */
2536 return SvROK ((SV *)frame->data) && !CORO_THROW;
2537}
2538
2539static void
2540slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2541{
2542 AV *av = (AV *)SvRV (arg [0]);
2543
2544 if (SvIVX (AvARRAY (av)[0]))
2545 {
2546 SvIVX (AvARRAY (av)[0]) = 0;
2547 frame->prepare = prepare_nop;
2548 frame->check = slf_check_nop;
2549 }
2550 else
2551 {
2552 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2553
2554 av_push (av, waiter);
2555
2556 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2557 frame->prepare = prepare_schedule;
2558 frame->check = slf_check_signal_wait;
2559 }
2560}
2561
2562/*****************************************************************************/
2563/* Coro::AIO */
2564
2565#define CORO_MAGIC_type_aio PERL_MAGIC_ext
2566
2567/* helper storage struct */
2568struct io_state
2569{
2570 int errorno;
2571 I32 laststype; /* U16 in 5.10.0 */
2572 int laststatval;
2573 Stat_t statcache;
2574};
2575
2576static void
2577coro_aio_callback (pTHX_ CV *cv)
2578{
2579 dXSARGS;
2580 AV *state = (AV *)GENSUB_ARG;
2581 SV *coro = av_pop (state);
2582 SV *data_sv = newSV (sizeof (struct io_state));
2583
2584 av_extend (state, items - 1);
2585
2586 sv_upgrade (data_sv, SVt_PV);
2587 SvCUR_set (data_sv, sizeof (struct io_state));
2588 SvPOK_only (data_sv);
2589
2590 {
2591 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2592
2593 data->errorno = errno;
2594 data->laststype = PL_laststype;
2595 data->laststatval = PL_laststatval;
2596 data->statcache = PL_statcache;
2597 }
2598
2599 /* now build the result vector out of all the parameters and the data_sv */
2600 {
2601 int i;
2602
2603 for (i = 0; i < items; ++i)
2604 av_push (state, SvREFCNT_inc_NN (ST (i)));
2605 }
2606
2607 av_push (state, data_sv);
2608
2609 api_ready (aTHX_ coro);
2610 SvREFCNT_dec (coro);
2611 SvREFCNT_dec ((AV *)state);
2612}
2613
2614static int
2615slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2616{
2617 AV *state = (AV *)frame->data;
2618
2619 /* if we are about to throw, return early */
2620 /* this does not cancel the aio request, but at least */
2621 /* it quickly returns */
2622 if (CORO_THROW)
2623 return 0;
2624
2625 /* one element that is an RV? repeat! */
2626 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2627 return 1;
2628
2629 /* restore status */
2630 {
2631 SV *data_sv = av_pop (state);
2632 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2633
2634 errno = data->errorno;
2635 PL_laststype = data->laststype;
2636 PL_laststatval = data->laststatval;
2637 PL_statcache = data->statcache;
2638
2639 SvREFCNT_dec (data_sv);
2640 }
2641
2642 /* push result values */
2643 {
2644 dSP;
2645 int i;
2646
2647 EXTEND (SP, AvFILLp (state) + 1);
2648 for (i = 0; i <= AvFILLp (state); ++i)
2649 PUSHs (sv_2mortal (SvREFCNT_inc_NN (AvARRAY (state)[i])));
2650
2651 PUTBACK;
2652 }
2653
2654 return 0;
2655}
2656
2657static void
2658slf_init_aio_req (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2659{
2660 AV *state = (AV *)sv_2mortal ((SV *)newAV ());
2661 SV *coro_hv = SvRV (coro_current);
2662 struct coro *coro = SvSTATE_hv (coro_hv);
2663
2664 /* put our coroutine id on the state arg */
2665 av_push (state, SvREFCNT_inc_NN (coro_hv));
2666
2667 /* first see whether we have a non-zero priority and set it as AIO prio */
2668 if (coro->prio)
2669 {
2670 dSP;
2671
2672 static SV *prio_cv;
2673 static SV *prio_sv;
2674
2675 if (expect_false (!prio_cv))
2676 {
2677 prio_cv = (SV *)get_cv ("IO::AIO::aioreq_pri", 0);
2678 prio_sv = newSViv (0);
2679 }
2680
2681 PUSHMARK (SP);
2682 sv_setiv (prio_sv, coro->prio);
2683 XPUSHs (prio_sv);
2684
2685 PUTBACK;
2686 call_sv (prio_cv, G_VOID | G_DISCARD);
2687 }
2688
2689 /* now call the original request */
2690 {
2691 dSP;
2692 CV *req = (CV *)CORO_MAGIC_NN ((SV *)cv, CORO_MAGIC_type_aio)->mg_obj;
2693 int i;
2694
2695 PUSHMARK (SP);
2696
2697 /* first push all args to the stack */
2698 EXTEND (SP, items + 1);
2699
2700 for (i = 0; i < items; ++i)
2701 PUSHs (arg [i]);
2702
2703 /* now push the callback closure */
2704 PUSHs (sv_2mortal (gensub (aTHX_ coro_aio_callback, (void *)SvREFCNT_inc_NN ((SV *)state))));
2705
2706 /* now call the AIO function - we assume our request is uncancelable */
2707 PUTBACK;
2708 call_sv ((SV *)req, G_VOID | G_DISCARD);
2709 }
2710
2711 /* now that the requets is going, we loop toll we have a result */
2712 frame->data = (void *)state;
2713 frame->prepare = prepare_schedule;
2714 frame->check = slf_check_aio_req;
2715}
2716
2717static void
2718coro_aio_req_xs (pTHX_ CV *cv)
2719{
2720 dXSARGS;
2721
2722 CORO_EXECUTE_SLF_XS (slf_init_aio_req);
2723
2724 XSRETURN_EMPTY;
2725}
2726
2727/*****************************************************************************/
2728
2729MODULE = Coro::State PACKAGE = Coro::State PREFIX = api_
2730
2731PROTOTYPES: DISABLE
2732
2733BOOT:
2734{
2735#ifdef USE_ITHREADS
2736# if CORO_PTHREAD
2737 coro_thx = PERL_GET_CONTEXT;
2738# endif
2739#endif
2740 BOOT_PAGESIZE;
2741
2742 irsgv = gv_fetchpv ("/" , GV_ADD|GV_NOTQUAL, SVt_PV);
2743 stdoutgv = gv_fetchpv ("STDOUT", GV_ADD|GV_NOTQUAL, SVt_PVIO);
2744
2745 orig_sigelem_get = PL_vtbl_sigelem.svt_get; PL_vtbl_sigelem.svt_get = coro_sigelem_get;
2746 orig_sigelem_set = PL_vtbl_sigelem.svt_set; PL_vtbl_sigelem.svt_set = coro_sigelem_set;
2747 orig_sigelem_clr = PL_vtbl_sigelem.svt_clear; PL_vtbl_sigelem.svt_clear = coro_sigelem_clr;
2748
2749 hv_sig = coro_get_hv (aTHX_ "SIG", TRUE);
2750 rv_diehook = newRV_inc ((SV *)gv_fetchpv ("Coro::State::diehook" , 0, SVt_PVCV));
2751 rv_warnhook = newRV_inc ((SV *)gv_fetchpv ("Coro::State::warnhook", 0, SVt_PVCV));
2752
2753 coro_state_stash = gv_stashpv ("Coro::State", TRUE);
2754
2755 newCONSTSUB (coro_state_stash, "CC_TRACE" , newSViv (CC_TRACE));
2756 newCONSTSUB (coro_state_stash, "CC_TRACE_SUB" , newSViv (CC_TRACE_SUB));
2757 newCONSTSUB (coro_state_stash, "CC_TRACE_LINE", newSViv (CC_TRACE_LINE));
2758 newCONSTSUB (coro_state_stash, "CC_TRACE_ALL" , newSViv (CC_TRACE_ALL));
2759
2760 main_mainstack = PL_mainstack;
2761 main_top_env = PL_top_env;
2762
2763 while (main_top_env->je_prev)
2764 main_top_env = main_top_env->je_prev;
2765
2766 {
2767 SV *slf = sv_2mortal (newSViv (PTR2IV (pp_slf)));
2768
2769 if (!PL_custom_op_names) PL_custom_op_names = newHV ();
2770 hv_store_ent (PL_custom_op_names, slf,
2771 newSVpv ("coro_slf", 0), 0);
2772
2773 if (!PL_custom_op_descs) PL_custom_op_descs = newHV ();
2774 hv_store_ent (PL_custom_op_descs, slf,
2775 newSVpv ("coro schedule like function", 0), 0);
2776 }
2777
2778 coroapi.ver = CORO_API_VERSION;
2779 coroapi.rev = CORO_API_REVISION;
2780
2781 coroapi.transfer = api_transfer;
2782
2783 coroapi.sv_state = SvSTATE_;
2784 coroapi.execute_slf = api_execute_slf;
2785 coroapi.prepare_nop = prepare_nop;
2786 coroapi.prepare_schedule = prepare_schedule;
2787 coroapi.prepare_cede = prepare_cede;
2788 coroapi.prepare_cede_notself = prepare_cede_notself;
2789
2790 {
2791 SV **svp = hv_fetch (PL_modglobal, "Time::NVtime", 12, 0);
2792
2793 if (!svp) croak ("Time::HiRes is required");
2794 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer");
2795
2796 nvtime = INT2PTR (double (*)(), SvIV (*svp));
2797 }
2798
2799 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL));
2800}
2801
2802SV *
2803new (char *klass, ...)
2804 ALIAS:
2805 Coro::new = 1
2806 CODE:
2807{
2808 struct coro *coro;
2809 MAGIC *mg;
2810 HV *hv;
2811 CV *cb;
2812 int i;
2813
2814 if (items > 1)
2815 {
2816 cb = coro_sv_2cv (aTHX_ ST (1));
2817
2818 if (!ix)
235 { 2819 {
236 /* I never used formats, so how should I know how these are implemented? */ 2820 if (CvISXSUB (cb))
237 /* my bold guess is as a simple, plain sub... */ 2821 croak ("Coro::State doesn't support XS functions as coroutine start, caught");
238 croak ("CXt_FORMAT not yet handled. Don't switch coroutines from within formats"); 2822
2823 if (!CvROOT (cb))
2824 croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught");
239 } 2825 }
240 } 2826 }
241 2827
242 if (top_si->si_type == PERLSI_MAIN)
243 break;
244
245 top_si = top_si->si_prev;
246 ccstk = top_si->si_cxstack;
247 cxix = top_si->si_cxix;
248 }
249
250 PUTBACK;
251 }
252
253 c->dowarn = PL_dowarn;
254 c->defav = GvAV (PL_defgv);
255 c->curstackinfo = PL_curstackinfo;
256 c->curstack = PL_curstack;
257 c->mainstack = PL_mainstack;
258 c->stack_sp = PL_stack_sp;
259 c->op = PL_op;
260 c->curpad = PL_curpad;
261 c->stack_base = PL_stack_base;
262 c->stack_max = PL_stack_max;
263 c->tmps_stack = PL_tmps_stack;
264 c->tmps_floor = PL_tmps_floor;
265 c->tmps_ix = PL_tmps_ix;
266 c->tmps_max = PL_tmps_max;
267 c->markstack = PL_markstack;
268 c->markstack_ptr = PL_markstack_ptr;
269 c->markstack_max = PL_markstack_max;
270 c->scopestack = PL_scopestack;
271 c->scopestack_ix = PL_scopestack_ix;
272 c->scopestack_max = PL_scopestack_max;
273 c->savestack = PL_savestack;
274 c->savestack_ix = PL_savestack_ix;
275 c->savestack_max = PL_savestack_max;
276 c->retstack = PL_retstack;
277 c->retstack_ix = PL_retstack_ix;
278 c->retstack_max = PL_retstack_max;
279 c->curcop = PL_curcop;
280}
281
282#define LOAD(state) do { load_state(aTHX_ state); SPAGAIN; } while (0)
283#define SAVE(state) do { PUTBACK; save_state(aTHX_ state); } while (0)
284
285static void
286load_state(pTHX_ Coro__State c)
287{
288 PL_dowarn = c->dowarn;
289 GvAV (PL_defgv) = c->defav;
290 PL_curstackinfo = c->curstackinfo;
291 PL_curstack = c->curstack;
292 PL_mainstack = c->mainstack;
293 PL_stack_sp = c->stack_sp;
294 PL_op = c->op;
295 PL_curpad = c->curpad;
296 PL_stack_base = c->stack_base;
297 PL_stack_max = c->stack_max;
298 PL_tmps_stack = c->tmps_stack;
299 PL_tmps_floor = c->tmps_floor;
300 PL_tmps_ix = c->tmps_ix;
301 PL_tmps_max = c->tmps_max;
302 PL_markstack = c->markstack;
303 PL_markstack_ptr = c->markstack_ptr;
304 PL_markstack_max = c->markstack_max;
305 PL_scopestack = c->scopestack;
306 PL_scopestack_ix = c->scopestack_ix;
307 PL_scopestack_max = c->scopestack_max;
308 PL_savestack = c->savestack;
309 PL_savestack_ix = c->savestack_ix;
310 PL_savestack_max = c->savestack_max;
311 PL_retstack = c->retstack;
312 PL_retstack_ix = c->retstack_ix;
313 PL_retstack_max = c->retstack_max;
314 PL_curcop = c->curcop;
315
316 {
317 dSP;
318 CV *cv;
319
320 /* now do the ugly restore mess */
321 while ((cv = (CV *)POPs))
322 {
323 AV *padlist = (AV *)POPs;
324
325 put_padlist (cv);
326 CvPADLIST(cv) = padlist;
327 CvDEPTH(cv) = (I32)POPs;
328
329#ifdef USE_THREADS
330 CvOWNER(cv) = (struct perl_thread *)POPs;
331 error does not work either
332#endif
333 }
334
335 PUTBACK;
336 }
337}
338
339/* this is an EXACT copy of S_nuke_stacks in perl.c, which is unfortunately static */
340STATIC void
341destroy_stacks(pTHX)
342{
343 /* die does this while calling POPSTACK, but I just don't see why. */
344 /* OTOH, die does not have a memleak, but we do... */
345 dounwind(-1);
346
347 /* is this ugly, I ask? */
348 while (PL_scopestack_ix)
349 LEAVE;
350
351 while (PL_curstackinfo->si_next)
352 PL_curstackinfo = PL_curstackinfo->si_next;
353
354 while (PL_curstackinfo)
355 {
356 PERL_SI *p = PL_curstackinfo->si_prev;
357
358 SvREFCNT_dec(PL_curstackinfo->si_stack);
359 Safefree(PL_curstackinfo->si_cxstack);
360 Safefree(PL_curstackinfo);
361 PL_curstackinfo = p;
362 }
363
364 if (PL_scopestack_ix != 0)
365 Perl_warner(aTHX_ WARN_INTERNAL,
366 "Unbalanced scopes: %ld more ENTERs than LEAVEs\n",
367 (long)PL_scopestack_ix);
368 if (PL_savestack_ix != 0)
369 Perl_warner(aTHX_ WARN_INTERNAL,
370 "Unbalanced saves: %ld more saves than restores\n",
371 (long)PL_savestack_ix);
372 if (PL_tmps_floor != -1)
373 Perl_warner(aTHX_ WARN_INTERNAL,"Unbalanced tmps: %ld more allocs than frees\n",
374 (long)PL_tmps_floor + 1);
375 /*
376 */
377 Safefree(PL_tmps_stack);
378 Safefree(PL_markstack);
379 Safefree(PL_scopestack);
380 Safefree(PL_savestack);
381 Safefree(PL_retstack);
382}
383
384#define SUB_INIT "Coro::State::_newcoro"
385
386MODULE = Coro::State PACKAGE = Coro::State
387
388PROTOTYPES: ENABLE
389
390BOOT:
391 if (!padlist_cache)
392 padlist_cache = newHV ();
393
394Coro::State
395_newprocess(args)
396 SV * args
397 PROTOTYPE: $
398 CODE:
399 Coro__State coro;
400
401 if (!SvROK (args) || SvTYPE (SvRV (args)) != SVt_PVAV)
402 croak ("Coro::State::newprocess expects an arrayref");
403
404 New (0, coro, 1, struct coro); 2828 Newz (0, coro, 1, struct coro);
2829 coro->args = newAV ();
2830 coro->flags = CF_NEW;
405 2831
406 coro->mainstack = 0; /* actual work is done inside transfer */ 2832 if (coro_first) coro_first->prev = coro;
407 coro->args = (AV *)SvREFCNT_inc (SvRV (args)); 2833 coro->next = coro_first;
2834 coro_first = coro;
408 2835
409 RETVAL = coro; 2836 coro->hv = hv = newHV ();
2837 mg = sv_magicext ((SV *)hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)coro, 0);
2838 mg->mg_flags |= MGf_DUP;
2839 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));
2840
2841 if (items > 1)
2842 {
2843 av_extend (coro->args, items - 1 + ix - 1);
2844
2845 if (ix)
2846 {
2847 av_push (coro->args, SvREFCNT_inc_NN ((SV *)cb));
2848 cb = cv_coro_run;
2849 }
2850
2851 coro->startcv = (CV *)SvREFCNT_inc_NN ((SV *)cb);
2852
2853 for (i = 2; i < items; i++)
2854 av_push (coro->args, newSVsv (ST (i)));
2855 }
2856}
410 OUTPUT: 2857 OUTPUT:
411 RETVAL 2858 RETVAL
412 2859
413void 2860void
414transfer(prev,next) 2861transfer (...)
415 Coro::State_or_hashref prev 2862 PROTOTYPE: $$
416 Coro::State_or_hashref next 2863 CODE:
417 CODE: 2864 CORO_EXECUTE_SLF_XS (slf_init_transfer);
418 2865
419 if (prev != next) 2866bool
2867_destroy (SV *coro_sv)
2868 CODE:
2869 RETVAL = coro_state_destroy (aTHX_ SvSTATE (coro_sv));
2870 OUTPUT:
2871 RETVAL
2872
2873void
2874_exit (int code)
2875 PROTOTYPE: $
2876 CODE:
2877 _exit (code);
2878
2879int
2880cctx_stacksize (int new_stacksize = 0)
2881 PROTOTYPE: ;$
2882 CODE:
2883 RETVAL = cctx_stacksize;
2884 if (new_stacksize)
420 { 2885 {
2886 cctx_stacksize = new_stacksize;
2887 ++cctx_gen;
421 /* 2888 }
422 * this could be done in newprocess which would lead to 2889 OUTPUT:
423 * extremely elegant and fast (just SAVE/LOAD) 2890 RETVAL
424 * code here, but lazy allocation of stacks has also 2891
425 * some virtues and the overhead of the if() is nil. 2892int
2893cctx_max_idle (int max_idle = 0)
2894 PROTOTYPE: ;$
2895 CODE:
2896 RETVAL = cctx_max_idle;
2897 if (max_idle > 1)
2898 cctx_max_idle = max_idle;
2899 OUTPUT:
2900 RETVAL
2901
2902int
2903cctx_count ()
2904 PROTOTYPE:
2905 CODE:
2906 RETVAL = cctx_count;
2907 OUTPUT:
2908 RETVAL
2909
2910int
2911cctx_idle ()
2912 PROTOTYPE:
2913 CODE:
2914 RETVAL = cctx_idle;
2915 OUTPUT:
2916 RETVAL
2917
2918void
2919list ()
2920 PROTOTYPE:
2921 PPCODE:
2922{
2923 struct coro *coro;
2924 for (coro = coro_first; coro; coro = coro->next)
2925 if (coro->hv)
2926 XPUSHs (sv_2mortal (newRV_inc ((SV *)coro->hv)));
2927}
2928
2929void
2930call (Coro::State coro, SV *coderef)
2931 ALIAS:
2932 eval = 1
2933 CODE:
2934{
2935 if (coro->mainstack && ((coro->flags & CF_RUNNING) || coro->slot))
426 */ 2936 {
427 if (next->mainstack) 2937 struct coro temp;
2938
2939 if (!(coro->flags & CF_RUNNING))
428 { 2940 {
429 SAVE (prev); 2941 PUTBACK;
430 LOAD (next); 2942 save_perl (aTHX_ &temp);
431 /* mark this state as in-use */ 2943 load_perl (aTHX_ coro);
432 next->mainstack = 0;
433 next->tmps_ix = -2;
434 } 2944 }
435 else if (next->tmps_ix == -2) 2945
2946 {
2947 dSP;
2948 ENTER;
2949 SAVETMPS;
2950 PUTBACK;
2951 PUSHSTACK;
2952 PUSHMARK (SP);
2953
2954 if (ix)
2955 eval_sv (coderef, 0);
2956 else
2957 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
2958
2959 POPSTACK;
2960 SPAGAIN;
2961 FREETMPS;
2962 LEAVE;
2963 PUTBACK;
2964 }
2965
2966 if (!(coro->flags & CF_RUNNING))
436 { 2967 {
437 croak ("tried to transfer to running coroutine"); 2968 save_perl (aTHX_ coro);
438 } 2969 load_perl (aTHX_ &temp);
439 else
440 {
441 SAVE (prev);
442
443 /*
444 * emulate part of the perl startup here.
445 */
446 UNOP myop;
447
448 init_stacks (); /* from perl.c */
449 PL_op = (OP *)&myop;
450 /*PL_curcop = 0;*/
451 GvAV (PL_defgv) = (AV *)SvREFCNT_inc ((SV *)next->args);
452
453 SPAGAIN; 2970 SPAGAIN;
454 Zero(&myop, 1, UNOP);
455 myop.op_next = Nullop;
456 myop.op_flags = OPf_WANT_VOID;
457
458 PUSHMARK(SP);
459 XPUSHs ((SV*)get_cv(SUB_INIT, TRUE));
460 PUTBACK;
461 /*
462 * the next line is slightly wrong, as PL_op->op_next
463 * is actually being executed so we skip the first op.
464 * that doesn't matter, though, since it is only
465 * pp_nextstate and we never return...
466 */
467 PL_op = Perl_pp_entersub(aTHX);
468 SPAGAIN;
469
470 ENTER;
471 } 2971 }
472 } 2972 }
2973}
2974
2975SV *
2976is_ready (Coro::State coro)
2977 PROTOTYPE: $
2978 ALIAS:
2979 is_ready = CF_READY
2980 is_running = CF_RUNNING
2981 is_new = CF_NEW
2982 is_destroyed = CF_DESTROYED
2983 CODE:
2984 RETVAL = boolSV (coro->flags & ix);
2985 OUTPUT:
2986 RETVAL
473 2987
474void 2988void
475DESTROY(coro) 2989throw (Coro::State self, SV *throw = &PL_sv_undef)
476 Coro::State coro 2990 PROTOTYPE: $;$
477 CODE: 2991 CODE:
2992{
2993 struct coro *current = SvSTATE_current;
2994 SV **throwp = self == current ? &CORO_THROW : &self->except;
2995 SvREFCNT_dec (*throwp);
2996 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2997}
478 2998
479 if (coro->mainstack) 2999void
3000api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
3001 PROTOTYPE: $;$
3002 C_ARGS: aTHX_ coro, flags
3003
3004SV *
3005has_cctx (Coro::State coro)
3006 PROTOTYPE: $
3007 CODE:
3008 RETVAL = boolSV (!!coro->cctx);
3009 OUTPUT:
3010 RETVAL
3011
3012int
3013is_traced (Coro::State coro)
3014 PROTOTYPE: $
3015 CODE:
3016 RETVAL = (coro->cctx ? coro->cctx->flags : 0) & CC_TRACE_ALL;
3017 OUTPUT:
3018 RETVAL
3019
3020UV
3021rss (Coro::State coro)
3022 PROTOTYPE: $
3023 ALIAS:
3024 usecount = 1
3025 CODE:
3026 switch (ix)
3027 {
3028 case 0: RETVAL = coro_rss (aTHX_ coro); break;
3029 case 1: RETVAL = coro->usecount; break;
3030 }
3031 OUTPUT:
3032 RETVAL
3033
3034void
3035force_cctx ()
3036 PROTOTYPE:
3037 CODE:
3038 SvSTATE_current->cctx->idle_sp = 0;
3039
3040void
3041swap_defsv (Coro::State self)
3042 PROTOTYPE: $
3043 ALIAS:
3044 swap_defav = 1
3045 CODE:
3046 if (!self->slot)
3047 croak ("cannot swap state with coroutine that has no saved state,");
3048 else
480 { 3049 {
481 struct coro temp; 3050 SV **src = ix ? (SV **)&GvAV (PL_defgv) : &GvSV (PL_defgv);
3051 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv;
482 3052
483 SAVE(aTHX_ (&temp)); 3053 SV *tmp = *src; *src = *dst; *dst = tmp;
484 LOAD(aTHX_ coro);
485
486 destroy_stacks ();
487 SvREFCNT_dec ((SV *)GvAV (PL_defgv));
488
489 LOAD((&temp));
490 } 3054 }
491 3055
3056
3057MODULE = Coro::State PACKAGE = Coro
3058
3059BOOT:
3060{
3061 int i;
3062
3063 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
3064 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
3065 cv_coro_run = get_cv ( "Coro::_terminate", GV_ADD);
3066 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD);
3067 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current);
3068 av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE);
3069 av_destroy = coro_get_av (aTHX_ "Coro::destroy" , TRUE);
3070 sv_manager = coro_get_sv (aTHX_ "Coro::manager" , TRUE);
3071
3072 sv_async_pool_idle = newSVpv ("[async pool idle]", 0); SvREADONLY_on (sv_async_pool_idle);
3073 sv_Coro = newSVpv ("Coro", 0); SvREADONLY_on (sv_Coro);
3074 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler);
3075 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new);
3076
3077 coro_stash = gv_stashpv ("Coro", TRUE);
3078
3079 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX));
3080 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH));
3081 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (PRIO_NORMAL));
3082 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (PRIO_LOW));
3083 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (PRIO_IDLE));
3084 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (PRIO_MIN));
3085
3086 for (i = PRIO_MAX - PRIO_MIN + 1; i--; )
3087 coro_ready[i] = newAV ();
3088
3089 {
3090 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE);
3091
3092 coroapi.schedule = api_schedule;
3093 coroapi.schedule_to = api_schedule_to;
3094 coroapi.cede = api_cede;
3095 coroapi.cede_notself = api_cede_notself;
3096 coroapi.ready = api_ready;
3097 coroapi.is_ready = api_is_ready;
3098 coroapi.nready = coro_nready;
3099 coroapi.current = coro_current;
3100
3101 /*GCoroAPI = &coroapi;*/
3102 sv_setiv (sv, (IV)&coroapi);
3103 SvREADONLY_on (sv);
3104 }
3105}
3106
3107void
3108terminate (...)
3109 CODE:
3110 CORO_EXECUTE_SLF_XS (slf_init_terminate);
3111
3112void
3113schedule (...)
3114 CODE:
3115 CORO_EXECUTE_SLF_XS (slf_init_schedule);
3116
3117void
3118schedule_to (...)
3119 CODE:
3120 CORO_EXECUTE_SLF_XS (slf_init_schedule_to);
3121
3122void
3123cede_to (...)
3124 CODE:
3125 CORO_EXECUTE_SLF_XS (slf_init_cede_to);
3126
3127void
3128cede (...)
3129 CODE:
3130 CORO_EXECUTE_SLF_XS (slf_init_cede);
3131
3132void
3133cede_notself (...)
3134 CODE:
3135 CORO_EXECUTE_SLF_XS (slf_init_cede_notself);
3136
3137void
3138_cancel (Coro::State self)
3139 CODE:
3140 coro_state_destroy (aTHX_ self);
3141 coro_call_on_destroy (aTHX_ self);
3142
3143void
3144_set_current (SV *current)
3145 PROTOTYPE: $
3146 CODE:
3147 SvREFCNT_dec (SvRV (coro_current));
3148 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current)));
3149
3150void
3151_set_readyhook (SV *hook)
3152 PROTOTYPE: $
3153 CODE:
492 SvREFCNT_dec (coro->args); 3154 SvREFCNT_dec (coro_readyhook);
493 Safefree (coro); 3155 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0;
494 3156
3157int
3158prio (Coro::State coro, int newprio = 0)
3159 PROTOTYPE: $;$
3160 ALIAS:
3161 nice = 1
3162 CODE:
3163{
3164 RETVAL = coro->prio;
495 3165
3166 if (items > 1)
3167 {
3168 if (ix)
3169 newprio = coro->prio - newprio;
3170
3171 if (newprio < PRIO_MIN) newprio = PRIO_MIN;
3172 if (newprio > PRIO_MAX) newprio = PRIO_MAX;
3173
3174 coro->prio = newprio;
3175 }
3176}
3177 OUTPUT:
3178 RETVAL
3179
3180SV *
3181ready (SV *self)
3182 PROTOTYPE: $
3183 CODE:
3184 RETVAL = boolSV (api_ready (aTHX_ self));
3185 OUTPUT:
3186 RETVAL
3187
3188int
3189nready (...)
3190 PROTOTYPE:
3191 CODE:
3192 RETVAL = coro_nready;
3193 OUTPUT:
3194 RETVAL
3195
3196void
3197_pool_handler (...)
3198 CODE:
3199 CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
3200
3201void
3202async_pool (SV *cv, ...)
3203 PROTOTYPE: &@
3204 PPCODE:
3205{
3206 HV *hv = (HV *)av_pop (av_async_pool);
3207 AV *av = newAV ();
3208 SV *cb = ST (0);
3209 int i;
3210
3211 av_extend (av, items - 2);
3212 for (i = 1; i < items; ++i)
3213 av_push (av, SvREFCNT_inc_NN (ST (i)));
3214
3215 if ((SV *)hv == &PL_sv_undef)
3216 {
3217 PUSHMARK (SP);
3218 EXTEND (SP, 2);
3219 PUSHs (sv_Coro);
3220 PUSHs ((SV *)cv_pool_handler);
3221 PUTBACK;
3222 call_sv ((SV *)cv_coro_state_new, G_SCALAR);
3223 SPAGAIN;
3224
3225 hv = (HV *)SvREFCNT_inc_NN (SvRV (POPs));
3226 }
3227
3228 {
3229 struct coro *coro = SvSTATE_hv (hv);
3230
3231 assert (!coro->invoke_cb);
3232 assert (!coro->invoke_av);
3233 coro->invoke_cb = SvREFCNT_inc (cb);
3234 coro->invoke_av = av;
3235 }
3236
3237 api_ready (aTHX_ (SV *)hv);
3238
3239 if (GIMME_V != G_VOID)
3240 XPUSHs (sv_2mortal (newRV_noinc ((SV *)hv)));
3241 else
3242 SvREFCNT_dec (hv);
3243}
3244
3245SV *
3246rouse_cb ()
3247 PROTOTYPE:
3248 CODE:
3249 RETVAL = coro_new_rouse_cb (aTHX);
3250 OUTPUT:
3251 RETVAL
3252
3253void
3254rouse_wait (...)
3255 PROTOTYPE: ;$
3256 PPCODE:
3257 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
3258
3259
3260MODULE = Coro::State PACKAGE = PerlIO::cede
3261
3262BOOT:
3263 PerlIO_define_layer (aTHX_ &PerlIO_cede);
3264
3265
3266MODULE = Coro::State PACKAGE = Coro::Semaphore
3267
3268SV *
3269new (SV *klass, SV *count = 0)
3270 CODE:
3271 RETVAL = sv_bless (
3272 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
3273 GvSTASH (CvGV (cv))
3274 );
3275 OUTPUT:
3276 RETVAL
3277
3278# helper for Coro::Channel
3279SV *
3280_alloc (int count)
3281 CODE:
3282 RETVAL = coro_waitarray_new (aTHX_ count);
3283 OUTPUT:
3284 RETVAL
3285
3286SV *
3287count (SV *self)
3288 CODE:
3289 RETVAL = newSVsv (AvARRAY ((AV *)SvRV (self))[0]);
3290 OUTPUT:
3291 RETVAL
3292
3293void
3294up (SV *self, int adjust = 1)
3295 ALIAS:
3296 adjust = 1
3297 CODE:
3298 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
3299
3300void
3301down (...)
3302 CODE:
3303 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3304
3305void
3306wait (...)
3307 CODE:
3308 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
3309
3310void
3311try (SV *self)
3312 PPCODE:
3313{
3314 AV *av = (AV *)SvRV (self);
3315 SV *count_sv = AvARRAY (av)[0];
3316 IV count = SvIVX (count_sv);
3317
3318 if (count > 0)
3319 {
3320 --count;
3321 SvIVX (count_sv) = count;
3322 XSRETURN_YES;
3323 }
3324 else
3325 XSRETURN_NO;
3326}
3327
3328void
3329waiters (SV *self)
3330 PPCODE:
3331{
3332 AV *av = (AV *)SvRV (self);
3333 int wcount = AvFILLp (av) + 1 - 1;
3334
3335 if (GIMME_V == G_SCALAR)
3336 XPUSHs (sv_2mortal (newSViv (wcount)));
3337 else
3338 {
3339 int i;
3340 EXTEND (SP, wcount);
3341 for (i = 1; i <= wcount; ++i)
3342 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
3343 }
3344}
3345
3346MODULE = Coro::State PACKAGE = Coro::Signal
3347
3348SV *
3349new (SV *klass)
3350 CODE:
3351 RETVAL = sv_bless (
3352 coro_waitarray_new (aTHX_ 0),
3353 GvSTASH (CvGV (cv))
3354 );
3355 OUTPUT:
3356 RETVAL
3357
3358void
3359wait (...)
3360 CODE:
3361 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3362
3363void
3364broadcast (SV *self)
3365 CODE:
3366{
3367 AV *av = (AV *)SvRV (self);
3368 coro_signal_wake (aTHX_ av, AvFILLp (av));
3369}
3370
3371void
3372send (SV *self)
3373 CODE:
3374{
3375 AV *av = (AV *)SvRV (self);
3376
3377 if (AvFILLp (av))
3378 coro_signal_wake (aTHX_ av, 1);
3379 else
3380 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
3381}
3382
3383IV
3384awaited (SV *self)
3385 CODE:
3386 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3387 OUTPUT:
3388 RETVAL
3389
3390
3391MODULE = Coro::State PACKAGE = Coro::AnyEvent
3392
3393BOOT:
3394 sv_activity = coro_get_sv (aTHX_ "Coro::AnyEvent::ACTIVITY", TRUE);
3395
3396void
3397_schedule (...)
3398 CODE:
3399{
3400 static int incede;
3401
3402 api_cede_notself (aTHX);
3403
3404 ++incede;
3405 while (coro_nready >= incede && api_cede (aTHX))
3406 ;
3407
3408 sv_setsv (sv_activity, &PL_sv_undef);
3409 if (coro_nready >= incede)
3410 {
3411 PUSHMARK (SP);
3412 PUTBACK;
3413 call_pv ("Coro::AnyEvent::_activity", G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3414 }
3415
3416 --incede;
3417}
3418
3419
3420MODULE = Coro::State PACKAGE = Coro::AIO
3421
3422void
3423_register (char *target, char *proto, SV *req)
3424 CODE:
3425{
3426 CV *req_cv = coro_sv_2cv (aTHX_ req);
3427 /* newXSproto doesn't return the CV on 5.8 */
3428 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3429 sv_setpv ((SV *)slf_cv, proto);
3430 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
3431}
3432

Diff Legend

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