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

Diff Legend

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