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.327 by root, Mon Nov 24 06:07:16 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 = av_fetch (padlist, i--, FALSE);
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; 394 SV *sv;
95 if (*name == '&') 395 while (&PL_sv_undef != (sv = av_pop ((AV *)*svp)))
96 sv = SvREFCNT_inc (ppad[ix]); 396 SvREFCNT_dec (sv);
97 else if (*name == '@') 397
98 sv = (SV *) newAV (); 398 SvREFCNT_dec (*svp);
99 else if (*name == '%')
100 sv = (SV *) newHV ();
101 else
102 sv = NEWSV (0, 0);
103 if (!SvPADBUSY (sv))
104 SvPADMY_on (sv);
105 npad[ix] = sv;
106 } 399 }
107 } 400 }
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 401
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); 402 SvREFCNT_dec ((SV*)padlist);
403 }
404}
405
406static int
407coro_cv_free (pTHX_ SV *sv, MAGIC *mg)
408{
409 AV *padlist;
410 AV *av = (AV *)mg->mg_obj;
411
412 /* casting is fun. */
413 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av)))
414 free_padlist (aTHX_ padlist);
415
416 SvREFCNT_dec (av); /* sv_magicext increased the refcount */
417
418 return 0;
419}
420
421#define CORO_MAGIC_type_cv 26
422#define CORO_MAGIC_type_state PERL_MAGIC_ext
423
424static MGVTBL coro_cv_vtbl = {
425 0, 0, 0, 0,
426 coro_cv_free
427};
428
429#define CORO_MAGIC_NN(sv, type) \
430 (expect_true (SvMAGIC (sv)->mg_type == type) \
431 ? SvMAGIC (sv) \
432 : mg_find (sv, type))
433
434#define CORO_MAGIC(sv, type) \
435 (expect_true (SvMAGIC (sv)) \
436 ? CORO_MAGIC_NN (sv, type) \
437 : 0)
438
439#define CORO_MAGIC_cv(cv) CORO_MAGIC (((SV *)(cv)), CORO_MAGIC_type_cv)
440#define CORO_MAGIC_state(sv) CORO_MAGIC_NN (((SV *)(sv)), CORO_MAGIC_type_state)
441
442INLINE struct coro *
443SvSTATE_ (pTHX_ SV *coro)
444{
445 HV *stash;
446 MAGIC *mg;
447
448 if (SvROK (coro))
449 coro = SvRV (coro);
450
451 if (expect_false (SvTYPE (coro) != SVt_PVHV))
452 croak ("Coro::State object required");
453
454 stash = SvSTASH (coro);
455 if (expect_false (stash != coro_stash && stash != coro_state_stash))
456 {
457 /* very slow, but rare, check */
458 if (!sv_derived_from (sv_2mortal (newRV_inc (coro)), "Coro::State"))
459 croak ("Coro::State object required");
460 }
461
462 mg = CORO_MAGIC_state (coro);
463 return (struct coro *)mg->mg_ptr;
464}
465
466#define SvSTATE(sv) SvSTATE_ (aTHX_ (sv))
467
468/* faster than SvSTATE, but expects a coroutine hv */
469#define SvSTATE_hv(hv) ((struct coro *)CORO_MAGIC_NN ((SV *)hv, CORO_MAGIC_type_state)->mg_ptr)
470#define SvSTATE_current SvSTATE_hv (SvRV (coro_current))
471
472/* the next two functions merely cache the padlists */
473static void
474get_padlist (pTHX_ CV *cv)
475{
476 MAGIC *mg = CORO_MAGIC_cv (cv);
477 AV *av;
478
479 if (expect_true (mg && AvFILLp ((av = (AV *)mg->mg_obj)) >= 0))
480 CvPADLIST (cv) = (AV *)AvARRAY (av)[AvFILLp (av)--];
481 else
482 {
483#if CORO_PREFER_PERL_FUNCTIONS
484 /* this is probably cleaner? but also slower! */
485 /* in practise, it seems to be less stable */
486 CV *cp = Perl_cv_clone (aTHX_ cv);
487 CvPADLIST (cv) = CvPADLIST (cp);
488 CvPADLIST (cp) = 0;
489 SvREFCNT_dec (cp);
490#else
491 CvPADLIST (cv) = coro_derive_padlist (aTHX_ cv);
492#endif
493 }
494}
495
496static void
497put_padlist (pTHX_ CV *cv)
498{
499 MAGIC *mg = CORO_MAGIC_cv (cv);
500 AV *av;
501
502 if (expect_false (!mg))
503 mg = sv_magicext ((SV *)cv, (SV *)newAV (), CORO_MAGIC_type_cv, &coro_cv_vtbl, 0, 0);
504
505 av = (AV *)mg->mg_obj;
506
507 if (expect_false (AvFILLp (av) >= AvMAX (av)))
508 av_extend (av, AvMAX (av) + 1);
509
510 AvARRAY (av)[++AvFILLp (av)] = (SV *)CvPADLIST (cv);
511}
512
513/** load & save, init *******************************************************/
514
515static void
516load_perl (pTHX_ Coro__State c)
517{
518 perl_slots *slot = c->slot;
519 c->slot = 0;
520
521 PL_mainstack = c->mainstack;
522
523 GvSV (PL_defgv) = slot->defsv;
524 GvAV (PL_defgv) = slot->defav;
525 GvSV (PL_errgv) = slot->errsv;
526 GvSV (irsgv) = slot->irsgv;
527 GvHV (PL_hintgv) = slot->hinthv;
528
529 #define VAR(name,type) PL_ ## name = slot->name;
530 # include "state.h"
531 #undef VAR
532
533 {
534 dSP;
535
536 CV *cv;
537
538 /* now do the ugly restore mess */
539 while (expect_true (cv = (CV *)POPs))
540 {
541 put_padlist (aTHX_ cv); /* mark this padlist as available */
542 CvDEPTH (cv) = PTR2IV (POPs);
543 CvPADLIST (cv) = (AV *)POPs;
544 }
545
546 PUTBACK;
159 } 547 }
160}
161 548
162/* the next tow functions merely cache the padlists */ 549 slf_frame = c->slf_frame;
163STATIC void 550 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} 551}
173 552
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 553static void
189save_state(pTHX_ Coro__State c) 554save_perl (pTHX_ Coro__State c)
190{ 555{
556 c->except = CORO_THROW;
557 c->slf_frame = slf_frame;
558
191 { 559 {
192 dSP; 560 dSP;
193 I32 cxix = cxstack_ix; 561 I32 cxix = cxstack_ix;
562 PERL_CONTEXT *ccstk = cxstack;
194 PERL_SI *top_si = PL_curstackinfo; 563 PERL_SI *top_si = PL_curstackinfo;
195 PERL_CONTEXT *ccstk = cxstack;
196 564
197 /* 565 /*
198 * the worst thing you can imagine happens first - we have to save 566 * the worst thing you can imagine happens first - we have to save
199 * (and reinitialize) all cv's in the whole callchain :( 567 * (and reinitialize) all cv's in the whole callchain :(
200 */ 568 */
201 569
202 PUSHs (Nullsv); 570 XPUSHs (Nullsv);
203 /* this loop was inspired by pp_caller */ 571 /* this loop was inspired by pp_caller */
204 for (;;) 572 for (;;)
205 { 573 {
206 while (cxix >= 0) 574 while (expect_true (cxix >= 0))
207 { 575 {
208 PERL_CONTEXT *cx = &ccstk[cxix--]; 576 PERL_CONTEXT *cx = &ccstk[cxix--];
209 577
210 if (CxTYPE(cx) == CXt_SUB) 578 if (expect_true (CxTYPE (cx) == CXt_SUB || CxTYPE (cx) == CXt_FORMAT))
211 { 579 {
212 CV *cv = cx->blk_sub.cv; 580 CV *cv = cx->blk_sub.cv;
581
213 if (CvDEPTH(cv)) 582 if (expect_true (CvDEPTH (cv)))
214 { 583 {
215#ifdef USE_THREADS
216 XPUSHs ((SV *)CvOWNER(cv));
217#endif
218 EXTEND (SP, 3); 584 EXTEND (SP, 3);
219 PUSHs ((SV *)CvDEPTH(cv));
220 PUSHs ((SV *)CvPADLIST(cv)); 585 PUSHs ((SV *)CvPADLIST (cv));
586 PUSHs (INT2PTR (SV *, (IV)CvDEPTH (cv)));
221 PUSHs ((SV *)cv); 587 PUSHs ((SV *)cv);
222 588
223 get_padlist (cv);
224
225 CvDEPTH(cv) = 0; 589 CvDEPTH (cv) = 0;
226#ifdef USE_THREADS 590 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 } 591 }
233 } 592 }
234 else if (CxTYPE(cx) == CXt_FORMAT) 593 }
594
595 if (expect_true (top_si->si_type == PERLSI_MAIN))
596 break;
597
598 top_si = top_si->si_prev;
599 ccstk = top_si->si_cxstack;
600 cxix = top_si->si_cxix;
601 }
602
603 PUTBACK;
604 }
605
606 /* allocate some space on the context stack for our purposes */
607 /* we manually unroll here, as usually 2 slots is enough */
608 if (SLOT_COUNT >= 1) CXINC;
609 if (SLOT_COUNT >= 2) CXINC;
610 if (SLOT_COUNT >= 3) CXINC;
611 {
612 int i;
613 for (i = 3; i < SLOT_COUNT; ++i)
614 CXINC;
615 }
616 cxstack_ix -= SLOT_COUNT; /* undo allocation */
617
618 c->mainstack = PL_mainstack;
619
620 {
621 perl_slots *slot = c->slot = (perl_slots *)(cxstack + cxstack_ix + 1);
622
623 slot->defav = GvAV (PL_defgv);
624 slot->defsv = DEFSV;
625 slot->errsv = ERRSV;
626 slot->irsgv = GvSV (irsgv);
627 slot->hinthv = GvHV (PL_hintgv);
628
629 #define VAR(name,type) slot->name = PL_ ## name;
630 # include "state.h"
631 #undef VAR
632 }
633}
634
635/*
636 * allocate various perl stacks. This is almost an exact copy
637 * of perl.c:init_stacks, except that it uses less memory
638 * on the (sometimes correct) assumption that coroutines do
639 * not usually need a lot of stackspace.
640 */
641#if CORO_PREFER_PERL_FUNCTIONS
642# define coro_init_stacks(thx) init_stacks ()
643#else
644static void
645coro_init_stacks (pTHX)
646{
647 PL_curstackinfo = new_stackinfo(32, 8);
648 PL_curstackinfo->si_type = PERLSI_MAIN;
649 PL_curstack = PL_curstackinfo->si_stack;
650 PL_mainstack = PL_curstack; /* remember in case we switch stacks */
651
652 PL_stack_base = AvARRAY(PL_curstack);
653 PL_stack_sp = PL_stack_base;
654 PL_stack_max = PL_stack_base + AvMAX(PL_curstack);
655
656 New(50,PL_tmps_stack,32,SV*);
657 PL_tmps_floor = -1;
658 PL_tmps_ix = -1;
659 PL_tmps_max = 32;
660
661 New(54,PL_markstack,16,I32);
662 PL_markstack_ptr = PL_markstack;
663 PL_markstack_max = PL_markstack + 16;
664
665#ifdef SET_MARK_OFFSET
666 SET_MARK_OFFSET;
667#endif
668
669 New(54,PL_scopestack,8,I32);
670 PL_scopestack_ix = 0;
671 PL_scopestack_max = 8;
672
673 New(54,PL_savestack,24,ANY);
674 PL_savestack_ix = 0;
675 PL_savestack_max = 24;
676
677#if !PERL_VERSION_ATLEAST (5,10,0)
678 New(54,PL_retstack,4,OP*);
679 PL_retstack_ix = 0;
680 PL_retstack_max = 4;
681#endif
682}
683#endif
684
685/*
686 * destroy the stacks, the callchain etc...
687 */
688static void
689coro_destruct_stacks (pTHX)
690{
691 while (PL_curstackinfo->si_next)
692 PL_curstackinfo = PL_curstackinfo->si_next;
693
694 while (PL_curstackinfo)
695 {
696 PERL_SI *p = PL_curstackinfo->si_prev;
697
698 if (!IN_DESTRUCT)
699 SvREFCNT_dec (PL_curstackinfo->si_stack);
700
701 Safefree (PL_curstackinfo->si_cxstack);
702 Safefree (PL_curstackinfo);
703 PL_curstackinfo = p;
704 }
705
706 Safefree (PL_tmps_stack);
707 Safefree (PL_markstack);
708 Safefree (PL_scopestack);
709 Safefree (PL_savestack);
710#if !PERL_VERSION_ATLEAST (5,10,0)
711 Safefree (PL_retstack);
712#endif
713}
714
715#define CORO_RSS \
716 rss += sizeof (SYM (curstackinfo)); \
717 rss += (SYM (curstackinfo->si_cxmax) + 1) * sizeof (PERL_CONTEXT); \
718 rss += sizeof (SV) + sizeof (struct xpvav) + (1 + AvMAX (SYM (curstack))) * sizeof (SV *); \
719 rss += SYM (tmps_max) * sizeof (SV *); \
720 rss += (SYM (markstack_max) - SYM (markstack_ptr)) * sizeof (I32); \
721 rss += SYM (scopestack_max) * sizeof (I32); \
722 rss += SYM (savestack_max) * sizeof (ANY);
723
724static size_t
725coro_rss (pTHX_ struct coro *coro)
726{
727 size_t rss = sizeof (*coro);
728
729 if (coro->mainstack)
730 {
731 if (coro->flags & CF_RUNNING)
732 {
733 #define SYM(sym) PL_ ## sym
734 CORO_RSS;
735 #undef SYM
736 }
737 else
738 {
739 #define SYM(sym) coro->slot->sym
740 CORO_RSS;
741 #undef SYM
742 }
743 }
744
745 return rss;
746}
747
748/** coroutine stack handling ************************************************/
749
750static int (*orig_sigelem_get) (pTHX_ SV *sv, MAGIC *mg);
751static int (*orig_sigelem_set) (pTHX_ SV *sv, MAGIC *mg);
752static int (*orig_sigelem_clr) (pTHX_ SV *sv, MAGIC *mg);
753
754/* apparently < 5.8.8 */
755#ifndef MgPV_nolen_const
756#define MgPV_nolen_const(mg) (((((int)(mg)->mg_len)) == HEf_SVKEY) ? \
757 SvPV_nolen((SV*)((mg)->mg_ptr)) : \
758 (const char*)(mg)->mg_ptr)
759#endif
760
761/*
762 * This overrides the default magic get method of %SIG elements.
763 * The original one doesn't provide for reading back of PL_diehook/PL_warnhook
764 * and instead of tryign to save and restore the hash elements, we just provide
765 * readback here.
766 * We only do this when the hook is != 0, as they are often set to 0 temporarily,
767 * not expecting this to actually change the hook. This is a potential problem
768 * when a schedule happens then, but we ignore this.
769 */
770static int
771coro_sigelem_get (pTHX_ SV *sv, MAGIC *mg)
772{
773 const char *s = MgPV_nolen_const (mg);
774
775 if (*s == '_')
776 {
777 SV **svp = 0;
778
779 if (strEQ (s, "__DIE__" )) svp = &PL_diehook;
780 if (strEQ (s, "__WARN__")) svp = &PL_warnhook;
781
782 if (svp)
783 {
784 sv_setsv (sv, *svp ? *svp : &PL_sv_undef);
785 return 0;
786 }
787 }
788
789 return orig_sigelem_get ? orig_sigelem_get (aTHX_ sv, mg) : 0;
790}
791
792static int
793coro_sigelem_clr (pTHX_ SV *sv, MAGIC *mg)
794{
795 const char *s = MgPV_nolen_const (mg);
796
797 if (*s == '_')
798 {
799 SV **svp = 0;
800
801 if (strEQ (s, "__DIE__" )) svp = &PL_diehook;
802 if (strEQ (s, "__WARN__")) svp = &PL_warnhook;
803
804 if (svp)
805 {
806 SV *old = *svp;
807 *svp = 0;
808 SvREFCNT_dec (old);
809 return 0;
810 }
811 }
812
813 return orig_sigelem_clr ? orig_sigelem_clr (aTHX_ sv, mg) : 0;
814}
815
816static int
817coro_sigelem_set (pTHX_ SV *sv, MAGIC *mg)
818{
819 const char *s = MgPV_nolen_const (mg);
820
821 if (*s == '_')
822 {
823 SV **svp = 0;
824
825 if (strEQ (s, "__DIE__" )) svp = &PL_diehook;
826 if (strEQ (s, "__WARN__")) svp = &PL_warnhook;
827
828 if (svp)
829 {
830 SV *old = *svp;
831 *svp = newSVsv (sv);
832 SvREFCNT_dec (old);
833 return 0;
834 }
835 }
836
837 return orig_sigelem_set ? orig_sigelem_set (aTHX_ sv, mg) : 0;
838}
839
840static void
841prepare_nop (pTHX_ struct coro_transfer_args *ta)
842{
843 /* kind of mega-hacky, but works */
844 ta->next = ta->prev = (struct coro *)ta;
845}
846
847static int
848slf_check_nop (pTHX_ struct CoroSLF *frame)
849{
850 return 0;
851}
852
853static int
854slf_check_repeat (pTHX_ struct CoroSLF *frame)
855{
856 return 1;
857}
858
859static UNOP coro_setup_op;
860
861static void NOINLINE /* noinline to keep it out of the transfer fast path */
862coro_setup (pTHX_ struct coro *coro)
863{
864 /*
865 * emulate part of the perl startup here.
866 */
867 coro_init_stacks (aTHX);
868
869 PL_runops = RUNOPS_DEFAULT;
870 PL_curcop = &PL_compiling;
871 PL_in_eval = EVAL_NULL;
872 PL_comppad = 0;
873 PL_comppad_name = 0;
874 PL_comppad_name_fill = 0;
875 PL_comppad_name_floor = 0;
876 PL_curpm = 0;
877 PL_curpad = 0;
878 PL_localizing = 0;
879 PL_dirty = 0;
880 PL_restartop = 0;
881#if PERL_VERSION_ATLEAST (5,10,0)
882 PL_parser = 0;
883#endif
884 PL_hints = 0;
885
886 /* recreate the die/warn hooks */
887 PL_diehook = 0; SvSetMagicSV (*hv_fetch (hv_sig, "__DIE__" , sizeof ("__DIE__" ) - 1, 1), rv_diehook );
888 PL_warnhook = 0; SvSetMagicSV (*hv_fetch (hv_sig, "__WARN__", sizeof ("__WARN__") - 1, 1), rv_warnhook);
889
890 GvSV (PL_defgv) = newSV (0);
891 GvAV (PL_defgv) = coro->args; coro->args = 0;
892 GvSV (PL_errgv) = newSV (0);
893 GvSV (irsgv) = newSVpvn ("\n", 1); sv_magic (GvSV (irsgv), (SV *)irsgv, PERL_MAGIC_sv, "/", 0);
894 GvHV (PL_hintgv) = 0;
895 PL_rs = newSVsv (GvSV (irsgv));
896 PL_defoutgv = (GV *)SvREFCNT_inc_NN (stdoutgv);
897
898 {
899 dSP;
900 UNOP myop;
901
902 Zero (&myop, 1, UNOP);
903 myop.op_next = Nullop;
904 myop.op_type = OP_ENTERSUB;
905 myop.op_flags = OPf_WANT_VOID;
906
907 PUSHMARK (SP);
908 PUSHs ((SV *)coro->startcv);
909 PUTBACK;
910 PL_op = (OP *)&myop;
911 PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
912 }
913
914 /* this newly created coroutine might be run on an existing cctx which most
915 * likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
916 */
917 slf_frame.prepare = prepare_nop; /* provide a nop function for an eventual pp_slf */
918 slf_frame.check = slf_check_nop; /* signal pp_slf to not repeat */
919
920 /* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
921 coro_setup_op.op_next = PL_op;
922 coro_setup_op.op_type = OP_ENTERSUB;
923 coro_setup_op.op_ppaddr = pp_slf;
924 /* no flags etc. required, as an init function won't be called */
925
926 PL_op = (OP *)&coro_setup_op;
927
928 /* copy throw, in case it was set before coro_setup */
929 CORO_THROW = coro->except;
930}
931
932static void
933coro_destruct (pTHX_ struct coro *coro)
934{
935 if (!IN_DESTRUCT)
936 {
937 /* restore all saved variables and stuff */
938 LEAVE_SCOPE (0);
939 assert (PL_tmps_floor == -1);
940
941 /* free all temporaries */
942 FREETMPS;
943 assert (PL_tmps_ix == -1);
944
945 /* unwind all extra stacks */
946 POPSTACK_TO (PL_mainstack);
947
948 /* unwind main stack */
949 dounwind (-1);
950 }
951
952 SvREFCNT_dec (GvSV (PL_defgv));
953 SvREFCNT_dec (GvAV (PL_defgv));
954 SvREFCNT_dec (GvSV (PL_errgv));
955 SvREFCNT_dec (PL_defoutgv);
956 SvREFCNT_dec (PL_rs);
957 SvREFCNT_dec (GvSV (irsgv));
958 SvREFCNT_dec (GvHV (PL_hintgv));
959
960 SvREFCNT_dec (PL_diehook);
961 SvREFCNT_dec (PL_warnhook);
962
963 SvREFCNT_dec (coro->saved_deffh);
964 SvREFCNT_dec (coro->rouse_cb);
965 SvREFCNT_dec (coro->invoke_cb);
966 SvREFCNT_dec (coro->invoke_av);
967
968 coro_destruct_stacks (aTHX);
969}
970
971INLINE void
972free_coro_mortal (pTHX)
973{
974 if (expect_true (coro_mortal))
975 {
976 SvREFCNT_dec (coro_mortal);
977 coro_mortal = 0;
978 }
979}
980
981static int
982runops_trace (pTHX)
983{
984 COP *oldcop = 0;
985 int oldcxix = -2;
986
987 while ((PL_op = CALL_FPTR (PL_op->op_ppaddr) (aTHX)))
988 {
989 PERL_ASYNC_CHECK ();
990
991 if (cctx_current->flags & CC_TRACE_ALL)
992 {
993 if (PL_op->op_type == OP_LEAVESUB && cctx_current->flags & CC_TRACE_SUB)
994 {
995 PERL_CONTEXT *cx = &cxstack[cxstack_ix];
996 SV **bot, **top;
997 AV *av = newAV (); /* return values */
998 SV **cb;
999 dSP;
1000
1001 GV *gv = CvGV (cx->blk_sub.cv);
1002 SV *fullname = sv_2mortal (newSV (0));
1003 if (isGV (gv))
1004 gv_efullname3 (fullname, gv, 0);
1005
1006 bot = PL_stack_base + cx->blk_oldsp + 1;
1007 top = cx->blk_gimme == G_ARRAY ? SP + 1
1008 : cx->blk_gimme == G_SCALAR ? bot + 1
1009 : bot;
1010
1011 av_extend (av, top - bot);
1012 while (bot < top)
1013 av_push (av, SvREFCNT_inc_NN (*bot++));
1014
1015 PL_runops = RUNOPS_DEFAULT;
1016 ENTER;
1017 SAVETMPS;
1018 EXTEND (SP, 3);
1019 PUSHMARK (SP);
1020 PUSHs (&PL_sv_no);
1021 PUSHs (fullname);
1022 PUSHs (sv_2mortal (newRV_noinc ((SV *)av)));
1023 PUTBACK;
1024 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1025 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1026 SPAGAIN;
1027 FREETMPS;
1028 LEAVE;
1029 PL_runops = runops_trace;
1030 }
1031
1032 if (oldcop != PL_curcop)
1033 {
1034 oldcop = PL_curcop;
1035
1036 if (PL_curcop != &PL_compiling)
1037 {
1038 SV **cb;
1039
1040 if (oldcxix != cxstack_ix && cctx_current->flags & CC_TRACE_SUB)
1041 {
1042 PERL_CONTEXT *cx = &cxstack[cxstack_ix];
1043
1044 if (CxTYPE (cx) == CXt_SUB && oldcxix < cxstack_ix)
1045 {
1046 dSP;
1047 GV *gv = CvGV (cx->blk_sub.cv);
1048 SV *fullname = sv_2mortal (newSV (0));
1049
1050 if (isGV (gv))
1051 gv_efullname3 (fullname, gv, 0);
1052
1053 PL_runops = RUNOPS_DEFAULT;
1054 ENTER;
1055 SAVETMPS;
1056 EXTEND (SP, 3);
1057 PUSHMARK (SP);
1058 PUSHs (&PL_sv_yes);
1059 PUSHs (fullname);
1060 PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef);
1061 PUTBACK;
1062 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1063 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1064 SPAGAIN;
1065 FREETMPS;
1066 LEAVE;
1067 PL_runops = runops_trace;
1068 }
1069
1070 oldcxix = cxstack_ix;
1071 }
1072
1073 if (cctx_current->flags & CC_TRACE_LINE)
1074 {
1075 dSP;
1076
1077 PL_runops = RUNOPS_DEFAULT;
1078 ENTER;
1079 SAVETMPS;
1080 EXTEND (SP, 3);
1081 PL_runops = RUNOPS_DEFAULT;
1082 PUSHMARK (SP);
1083 PUSHs (sv_2mortal (newSVpv (OutCopFILE (oldcop), 0)));
1084 PUSHs (sv_2mortal (newSViv (CopLINE (oldcop))));
1085 PUTBACK;
1086 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_line_cb", sizeof ("_trace_line_cb") - 1, 0);
1087 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1088 SPAGAIN;
1089 FREETMPS;
1090 LEAVE;
1091 PL_runops = runops_trace;
1092 }
1093 }
1094 }
1095 }
1096 }
1097
1098 TAINT_NOT;
1099 return 0;
1100}
1101
1102static struct CoroSLF cctx_ssl_frame;
1103
1104static void
1105slf_prepare_set_stacklevel (pTHX_ struct coro_transfer_args *ta)
1106{
1107 ta->prev = 0;
1108}
1109
1110static int
1111slf_check_set_stacklevel (pTHX_ struct CoroSLF *frame)
1112{
1113 *frame = cctx_ssl_frame;
1114
1115 return frame->check (aTHX_ frame); /* execute the restored frame - there must be one */
1116}
1117
1118/* initialises PL_top_env and injects a pseudo-slf-call to set the stacklevel */
1119static void NOINLINE
1120cctx_prepare (pTHX)
1121{
1122 PL_top_env = &PL_start_env;
1123
1124 if (cctx_current->flags & CC_TRACE)
1125 PL_runops = runops_trace;
1126
1127 /* we already must be executing an SLF op, there is no other valid way
1128 * that can lead to creation of a new cctx */
1129 assert (("FATAL: can't prepare slf-less cctx in Coro module (please report)",
1130 slf_frame.prepare && PL_op->op_ppaddr == pp_slf));
1131
1132 /* we must emulate leaving pp_slf, which is done inside slf_check_set_stacklevel */
1133 cctx_ssl_frame = slf_frame;
1134
1135 slf_frame.prepare = slf_prepare_set_stacklevel;
1136 slf_frame.check = slf_check_set_stacklevel;
1137}
1138
1139/* the tail of transfer: execute stuff we can only do after a transfer */
1140INLINE void
1141transfer_tail (pTHX)
1142{
1143 free_coro_mortal (aTHX);
1144}
1145
1146/*
1147 * this is a _very_ stripped down perl interpreter ;)
1148 */
1149static void
1150cctx_run (void *arg)
1151{
1152#ifdef USE_ITHREADS
1153# if CORO_PTHREAD
1154 PERL_SET_CONTEXT (coro_thx);
1155# endif
1156#endif
1157 {
1158 dTHX;
1159
1160 /* normally we would need to skip the entersub here */
1161 /* not doing so will re-execute it, which is exactly what we want */
1162 /* PL_nop = PL_nop->op_next */
1163
1164 /* inject a fake subroutine call to cctx_init */
1165 cctx_prepare (aTHX);
1166
1167 /* cctx_run is the alternative tail of transfer() */
1168 transfer_tail (aTHX);
1169
1170 /* somebody or something will hit me for both perl_run and PL_restartop */
1171 PL_restartop = PL_op;
1172 perl_run (PL_curinterp);
1173 /*
1174 * Unfortunately, there is no way to get at the return values of the
1175 * coro body here, as perl_run destroys these
1176 */
1177
1178 /*
1179 * If perl-run returns we assume exit() was being called or the coro
1180 * fell off the end, which seems to be the only valid (non-bug)
1181 * reason for perl_run to return. We try to exit by jumping to the
1182 * bootstrap-time "top" top_env, as we cannot restore the "main"
1183 * coroutine as Coro has no such concept
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 /*AvARRAY (av) = ary;*/
2374 SvPVX ((SV *)av) = (char *)ary; /* 5.8.8 needs this syntax instead of AvARRAY = ary */
2375 AvMAX (av) = 1;
2376 AvFILLp (av) = 0;
2377 ary [0] = newSViv (count);
2378
2379 return newRV_noinc ((SV *)av);
2380}
2381
2382/* semaphore */
2383
2384static void
2385coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2386{
2387 SV *count_sv = AvARRAY (av)[0];
2388 IV count = SvIVX (count_sv);
2389
2390 count += adjust;
2391 SvIVX (count_sv) = count;
2392
2393 /* now wake up as many waiters as are expected to lock */
2394 while (count > 0 && AvFILLp (av) > 0)
2395 {
2396 SV *cb;
2397
2398 /* swap first two elements so we can shift a waiter */
2399 AvARRAY (av)[0] = AvARRAY (av)[1];
2400 AvARRAY (av)[1] = count_sv;
2401 cb = av_shift (av);
2402
2403 if (SvOBJECT (cb))
2404 {
2405 api_ready (aTHX_ cb);
2406 --count;
2407 }
2408 else if (SvTYPE (cb) == SVt_PVCV)
2409 {
2410 dSP;
2411 PUSHMARK (SP);
2412 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2413 PUTBACK;
2414 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2415 }
2416
2417 SvREFCNT_dec (cb);
2418 }
2419}
2420
2421static void
2422coro_semaphore_on_destroy (pTHX_ struct coro *coro)
2423{
2424 /* call $sem->adjust (0) to possibly wake up some other waiters */
2425 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2426}
2427
2428static int
2429slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2430{
2431 AV *av = (AV *)frame->data;
2432 SV *count_sv = AvARRAY (av)[0];
2433
2434 /* if we are about to throw, don't actually acquire the lock, just throw */
2435 if (CORO_THROW)
2436 return 0;
2437 else if (SvIVX (count_sv) > 0)
2438 {
2439 SvSTATE_current->on_destroy = 0;
2440
2441 if (acquire)
2442 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2443 else
2444 coro_semaphore_adjust (aTHX_ av, 0);
2445
2446 return 0;
2447 }
2448 else
2449 {
2450 int i;
2451 /* if we were woken up but can't down, we look through the whole */
2452 /* waiters list and only add us if we aren't in there already */
2453 /* this avoids some degenerate memory usage cases */
2454
2455 for (i = 1; i <= AvFILLp (av); ++i)
2456 if (AvARRAY (av)[i] == SvRV (coro_current))
2457 return 1;
2458
2459 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2460 return 1;
2461 }
2462}
2463
2464static int
2465slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2466{
2467 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2468}
2469
2470static int
2471slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2472{
2473 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2474}
2475
2476static void
2477slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2478{
2479 AV *av = (AV *)SvRV (arg [0]);
2480
2481 if (SvIVX (AvARRAY (av)[0]) > 0)
2482 {
2483 frame->data = (void *)av;
2484 frame->prepare = prepare_nop;
2485 }
2486 else
2487 {
2488 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2489
2490 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av));
2491 frame->prepare = prepare_schedule;
2492
2493 /* to avoid race conditions when a woken-up coro gets terminated */
2494 /* we arrange for a temporary on_destroy that calls adjust (0) */
2495 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2496 }
2497}
2498
2499static void
2500slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2501{
2502 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2503 frame->check = slf_check_semaphore_down;
2504}
2505
2506static void
2507slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2508{
2509 if (items >= 2)
2510 {
2511 /* callback form */
2512 AV *av = (AV *)SvRV (arg [0]);
2513 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2514
2515 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
2516
2517 if (SvIVX (AvARRAY (av)[0]) > 0)
2518 coro_semaphore_adjust (aTHX_ av, 0);
2519
2520 frame->prepare = prepare_nop;
2521 frame->check = slf_check_nop;
2522 }
2523 else
2524 {
2525 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2526 frame->check = slf_check_semaphore_wait;
2527 }
2528}
2529
2530/* signal */
2531
2532static void
2533coro_signal_wake (pTHX_ AV *av, int count)
2534{
2535 SvIVX (AvARRAY (av)[0]) = 0;
2536
2537 /* now signal count waiters */
2538 while (count > 0 && AvFILLp (av) > 0)
2539 {
2540 SV *cb;
2541
2542 /* swap first two elements so we can shift a waiter */
2543 cb = AvARRAY (av)[0];
2544 AvARRAY (av)[0] = AvARRAY (av)[1];
2545 AvARRAY (av)[1] = cb;
2546
2547 cb = av_shift (av);
2548
2549 api_ready (aTHX_ cb);
2550 sv_setiv (cb, 0); /* signal waiter */
2551 SvREFCNT_dec (cb);
2552
2553 --count;
2554 }
2555}
2556
2557static int
2558slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2559{
2560 /* if we are about to throw, also stop waiting */
2561 return SvROK ((SV *)frame->data) && !CORO_THROW;
2562}
2563
2564static void
2565slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2566{
2567 AV *av = (AV *)SvRV (arg [0]);
2568
2569 if (SvIVX (AvARRAY (av)[0]))
2570 {
2571 SvIVX (AvARRAY (av)[0]) = 0;
2572 frame->prepare = prepare_nop;
2573 frame->check = slf_check_nop;
2574 }
2575 else
2576 {
2577 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2578
2579 av_push (av, waiter);
2580
2581 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2582 frame->prepare = prepare_schedule;
2583 frame->check = slf_check_signal_wait;
2584 }
2585}
2586
2587/*****************************************************************************/
2588/* Coro::AIO */
2589
2590#define CORO_MAGIC_type_aio PERL_MAGIC_ext
2591
2592/* helper storage struct */
2593struct io_state
2594{
2595 int errorno;
2596 I32 laststype; /* U16 in 5.10.0 */
2597 int laststatval;
2598 Stat_t statcache;
2599};
2600
2601static void
2602coro_aio_callback (pTHX_ CV *cv)
2603{
2604 dXSARGS;
2605 AV *state = (AV *)GENSUB_ARG;
2606 SV *coro = av_pop (state);
2607 SV *data_sv = newSV (sizeof (struct io_state));
2608
2609 av_extend (state, items - 1);
2610
2611 sv_upgrade (data_sv, SVt_PV);
2612 SvCUR_set (data_sv, sizeof (struct io_state));
2613 SvPOK_only (data_sv);
2614
2615 {
2616 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2617
2618 data->errorno = errno;
2619 data->laststype = PL_laststype;
2620 data->laststatval = PL_laststatval;
2621 data->statcache = PL_statcache;
2622 }
2623
2624 /* now build the result vector out of all the parameters and the data_sv */
2625 {
2626 int i;
2627
2628 for (i = 0; i < items; ++i)
2629 av_push (state, SvREFCNT_inc_NN (ST (i)));
2630 }
2631
2632 av_push (state, data_sv);
2633
2634 api_ready (aTHX_ coro);
2635 SvREFCNT_dec (coro);
2636 SvREFCNT_dec ((AV *)state);
2637}
2638
2639static int
2640slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2641{
2642 AV *state = (AV *)frame->data;
2643
2644 /* if we are about to throw, return early */
2645 /* this does not cancel the aio request, but at least */
2646 /* it quickly returns */
2647 if (CORO_THROW)
2648 return 0;
2649
2650 /* one element that is an RV? repeat! */
2651 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2652 return 1;
2653
2654 /* restore status */
2655 {
2656 SV *data_sv = av_pop (state);
2657 struct io_state *data = (struct io_state *)SvPVX (data_sv);
2658
2659 errno = data->errorno;
2660 PL_laststype = data->laststype;
2661 PL_laststatval = data->laststatval;
2662 PL_statcache = data->statcache;
2663
2664 SvREFCNT_dec (data_sv);
2665 }
2666
2667 /* push result values */
2668 {
2669 dSP;
2670 int i;
2671
2672 EXTEND (SP, AvFILLp (state) + 1);
2673 for (i = 0; i <= AvFILLp (state); ++i)
2674 PUSHs (sv_2mortal (SvREFCNT_inc_NN (AvARRAY (state)[i])));
2675
2676 PUTBACK;
2677 }
2678
2679 return 0;
2680}
2681
2682static void
2683slf_init_aio_req (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2684{
2685 AV *state = (AV *)sv_2mortal ((SV *)newAV ());
2686 SV *coro_hv = SvRV (coro_current);
2687 struct coro *coro = SvSTATE_hv (coro_hv);
2688
2689 /* put our coroutine id on the state arg */
2690 av_push (state, SvREFCNT_inc_NN (coro_hv));
2691
2692 /* first see whether we have a non-zero priority and set it as AIO prio */
2693 if (coro->prio)
2694 {
2695 dSP;
2696
2697 static SV *prio_cv;
2698 static SV *prio_sv;
2699
2700 if (expect_false (!prio_cv))
2701 {
2702 prio_cv = (SV *)get_cv ("IO::AIO::aioreq_pri", 0);
2703 prio_sv = newSViv (0);
2704 }
2705
2706 PUSHMARK (SP);
2707 sv_setiv (prio_sv, coro->prio);
2708 XPUSHs (prio_sv);
2709
2710 PUTBACK;
2711 call_sv (prio_cv, G_VOID | G_DISCARD);
2712 }
2713
2714 /* now call the original request */
2715 {
2716 dSP;
2717 CV *req = (CV *)CORO_MAGIC_NN ((SV *)cv, CORO_MAGIC_type_aio)->mg_obj;
2718 int i;
2719
2720 PUSHMARK (SP);
2721
2722 /* first push all args to the stack */
2723 EXTEND (SP, items + 1);
2724
2725 for (i = 0; i < items; ++i)
2726 PUSHs (arg [i]);
2727
2728 /* now push the callback closure */
2729 PUSHs (sv_2mortal (gensub (aTHX_ coro_aio_callback, (void *)SvREFCNT_inc_NN ((SV *)state))));
2730
2731 /* now call the AIO function - we assume our request is uncancelable */
2732 PUTBACK;
2733 call_sv ((SV *)req, G_VOID | G_DISCARD);
2734 }
2735
2736 /* now that the requets is going, we loop toll we have a result */
2737 frame->data = (void *)state;
2738 frame->prepare = prepare_schedule;
2739 frame->check = slf_check_aio_req;
2740}
2741
2742static void
2743coro_aio_req_xs (pTHX_ CV *cv)
2744{
2745 dXSARGS;
2746
2747 CORO_EXECUTE_SLF_XS (slf_init_aio_req);
2748
2749 XSRETURN_EMPTY;
2750}
2751
2752/*****************************************************************************/
2753
2754#if CORO_CLONE
2755# include "clone.c"
2756#endif
2757
2758MODULE = Coro::State PACKAGE = Coro::State PREFIX = api_
2759
2760PROTOTYPES: DISABLE
2761
2762BOOT:
2763{
2764#ifdef USE_ITHREADS
2765# if CORO_PTHREAD
2766 coro_thx = PERL_GET_CONTEXT;
2767# endif
2768#endif
2769 BOOT_PAGESIZE;
2770
2771 cctx_current = cctx_new_empty ();
2772
2773 irsgv = gv_fetchpv ("/" , GV_ADD|GV_NOTQUAL, SVt_PV);
2774 stdoutgv = gv_fetchpv ("STDOUT", GV_ADD|GV_NOTQUAL, SVt_PVIO);
2775
2776 orig_sigelem_get = PL_vtbl_sigelem.svt_get; PL_vtbl_sigelem.svt_get = coro_sigelem_get;
2777 orig_sigelem_set = PL_vtbl_sigelem.svt_set; PL_vtbl_sigelem.svt_set = coro_sigelem_set;
2778 orig_sigelem_clr = PL_vtbl_sigelem.svt_clear; PL_vtbl_sigelem.svt_clear = coro_sigelem_clr;
2779
2780 hv_sig = coro_get_hv (aTHX_ "SIG", TRUE);
2781 rv_diehook = newRV_inc ((SV *)gv_fetchpv ("Coro::State::diehook" , 0, SVt_PVCV));
2782 rv_warnhook = newRV_inc ((SV *)gv_fetchpv ("Coro::State::warnhook", 0, SVt_PVCV));
2783
2784 coro_state_stash = gv_stashpv ("Coro::State", TRUE);
2785
2786 newCONSTSUB (coro_state_stash, "CC_TRACE" , newSViv (CC_TRACE));
2787 newCONSTSUB (coro_state_stash, "CC_TRACE_SUB" , newSViv (CC_TRACE_SUB));
2788 newCONSTSUB (coro_state_stash, "CC_TRACE_LINE", newSViv (CC_TRACE_LINE));
2789 newCONSTSUB (coro_state_stash, "CC_TRACE_ALL" , newSViv (CC_TRACE_ALL));
2790
2791 main_mainstack = PL_mainstack;
2792 main_top_env = PL_top_env;
2793
2794 while (main_top_env->je_prev)
2795 main_top_env = main_top_env->je_prev;
2796
2797 {
2798 SV *slf = sv_2mortal (newSViv (PTR2IV (pp_slf)));
2799
2800 if (!PL_custom_op_names) PL_custom_op_names = newHV ();
2801 hv_store_ent (PL_custom_op_names, slf, newSVpv ("coro_slf", 0), 0);
2802
2803 if (!PL_custom_op_descs) PL_custom_op_descs = newHV ();
2804 hv_store_ent (PL_custom_op_descs, slf, newSVpv ("coro schedule like function", 0), 0);
2805 }
2806
2807 coroapi.ver = CORO_API_VERSION;
2808 coroapi.rev = CORO_API_REVISION;
2809
2810 coroapi.transfer = api_transfer;
2811
2812 coroapi.sv_state = SvSTATE_;
2813 coroapi.execute_slf = api_execute_slf;
2814 coroapi.prepare_nop = prepare_nop;
2815 coroapi.prepare_schedule = prepare_schedule;
2816 coroapi.prepare_cede = prepare_cede;
2817 coroapi.prepare_cede_notself = prepare_cede_notself;
2818
2819 {
2820 SV **svp = hv_fetch (PL_modglobal, "Time::NVtime", 12, 0);
2821
2822 if (!svp) croak ("Time::HiRes is required");
2823 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer");
2824
2825 nvtime = INT2PTR (double (*)(), SvIV (*svp));
2826 }
2827
2828 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL));
2829}
2830
2831SV *
2832new (char *klass, ...)
2833 ALIAS:
2834 Coro::new = 1
2835 CODE:
2836{
2837 struct coro *coro;
2838 MAGIC *mg;
2839 HV *hv;
2840 CV *cb;
2841 int i;
2842
2843 if (items > 1)
2844 {
2845 cb = coro_sv_2cv (aTHX_ ST (1));
2846
2847 if (!ix)
235 { 2848 {
236 /* I never used formats, so how should I know how these are implemented? */ 2849 if (CvISXSUB (cb))
237 /* my bold guess is as a simple, plain sub... */ 2850 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"); 2851
2852 if (!CvROOT (cb))
2853 croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught");
239 } 2854 }
240 } 2855 }
241 2856
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); 2857 Newz (0, coro, 1, struct coro);
2858 coro->args = newAV ();
2859 coro->flags = CF_NEW;
405 2860
406 coro->mainstack = 0; /* actual work is done inside transfer */ 2861 if (coro_first) coro_first->prev = coro;
407 coro->args = (AV *)SvREFCNT_inc (SvRV (args)); 2862 coro->next = coro_first;
2863 coro_first = coro;
408 2864
409 RETVAL = coro; 2865 coro->hv = hv = newHV ();
2866 mg = sv_magicext ((SV *)hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)coro, 0);
2867 mg->mg_flags |= MGf_DUP;
2868 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));
2869
2870 if (items > 1)
2871 {
2872 av_extend (coro->args, items - 1 + ix - 1);
2873
2874 if (ix)
2875 {
2876 av_push (coro->args, SvREFCNT_inc_NN ((SV *)cb));
2877 cb = cv_coro_run;
2878 }
2879
2880 coro->startcv = (CV *)SvREFCNT_inc_NN ((SV *)cb);
2881
2882 for (i = 2; i < items; i++)
2883 av_push (coro->args, newSVsv (ST (i)));
2884 }
2885}
410 OUTPUT: 2886 OUTPUT:
411 RETVAL 2887 RETVAL
412 2888
413void 2889void
414transfer(prev,next) 2890transfer (...)
415 Coro::State_or_hashref prev 2891 PROTOTYPE: $$
416 Coro::State_or_hashref next 2892 CODE:
417 CODE: 2893 CORO_EXECUTE_SLF_XS (slf_init_transfer);
418 2894
419 if (prev != next) 2895bool
2896_destroy (SV *coro_sv)
2897 CODE:
2898 RETVAL = coro_state_destroy (aTHX_ SvSTATE (coro_sv));
2899 OUTPUT:
2900 RETVAL
2901
2902void
2903_exit (int code)
2904 PROTOTYPE: $
2905 CODE:
2906 _exit (code);
2907
2908SV *
2909clone (Coro::State coro)
2910 CODE:
2911{
2912#if CORO_CLONE
2913 struct coro *ncoro = coro_clone (coro);
2914 MAGIC *mg;
2915 /* TODO: too much duplication */
2916 ncoro->hv = newHV ();
2917 mg = sv_magicext ((SV *)ncoro->hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)ncoro, 0);
2918 mg->mg_flags |= MGf_DUP;
2919 RETVAL = sv_bless (newRV_noinc ((SV *)ncoro->hv), SvSTASH (coro->hv));
2920#else
2921 croak ("Coro::State->clone has not been configured into this installation of Coro, realised");
2922#endif
2923}
2924 OUTPUT:
2925 RETVAL
2926
2927int
2928cctx_stacksize (int new_stacksize = 0)
2929 PROTOTYPE: ;$
2930 CODE:
2931 RETVAL = cctx_stacksize;
2932 if (new_stacksize)
420 { 2933 {
2934 cctx_stacksize = new_stacksize;
2935 ++cctx_gen;
421 /* 2936 }
422 * this could be done in newprocess which would lead to 2937 OUTPUT:
423 * extremely elegant and fast (just SAVE/LOAD) 2938 RETVAL
424 * code here, but lazy allocation of stacks has also 2939
425 * some virtues and the overhead of the if() is nil. 2940int
2941cctx_max_idle (int max_idle = 0)
2942 PROTOTYPE: ;$
2943 CODE:
2944 RETVAL = cctx_max_idle;
2945 if (max_idle > 1)
2946 cctx_max_idle = max_idle;
2947 OUTPUT:
2948 RETVAL
2949
2950int
2951cctx_count ()
2952 PROTOTYPE:
2953 CODE:
2954 RETVAL = cctx_count;
2955 OUTPUT:
2956 RETVAL
2957
2958int
2959cctx_idle ()
2960 PROTOTYPE:
2961 CODE:
2962 RETVAL = cctx_idle;
2963 OUTPUT:
2964 RETVAL
2965
2966void
2967list ()
2968 PROTOTYPE:
2969 PPCODE:
2970{
2971 struct coro *coro;
2972 for (coro = coro_first; coro; coro = coro->next)
2973 if (coro->hv)
2974 XPUSHs (sv_2mortal (newRV_inc ((SV *)coro->hv)));
2975}
2976
2977void
2978call (Coro::State coro, SV *coderef)
2979 ALIAS:
2980 eval = 1
2981 CODE:
2982{
2983 if (coro->mainstack && ((coro->flags & CF_RUNNING) || coro->slot))
426 */ 2984 {
427 if (next->mainstack) 2985 struct coro temp;
2986
2987 if (!(coro->flags & CF_RUNNING))
428 { 2988 {
429 SAVE (prev); 2989 PUTBACK;
430 LOAD (next); 2990 save_perl (aTHX_ &temp);
431 /* mark this state as in-use */ 2991 load_perl (aTHX_ coro);
432 next->mainstack = 0;
433 next->tmps_ix = -2;
434 } 2992 }
435 else if (next->tmps_ix == -2) 2993
2994 {
2995 dSP;
2996 ENTER;
2997 SAVETMPS;
2998 PUTBACK;
2999 PUSHSTACK;
3000 PUSHMARK (SP);
3001
3002 if (ix)
3003 eval_sv (coderef, 0);
3004 else
3005 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3006
3007 POPSTACK;
3008 SPAGAIN;
3009 FREETMPS;
3010 LEAVE;
3011 PUTBACK;
3012 }
3013
3014 if (!(coro->flags & CF_RUNNING))
436 { 3015 {
437 croak ("tried to transfer to running coroutine"); 3016 save_perl (aTHX_ coro);
438 } 3017 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; 3018 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 } 3019 }
472 } 3020 }
3021}
3022
3023SV *
3024is_ready (Coro::State coro)
3025 PROTOTYPE: $
3026 ALIAS:
3027 is_ready = CF_READY
3028 is_running = CF_RUNNING
3029 is_new = CF_NEW
3030 is_destroyed = CF_DESTROYED
3031 CODE:
3032 RETVAL = boolSV (coro->flags & ix);
3033 OUTPUT:
3034 RETVAL
473 3035
474void 3036void
475DESTROY(coro) 3037throw (Coro::State self, SV *throw = &PL_sv_undef)
476 Coro::State coro 3038 PROTOTYPE: $;$
477 CODE: 3039 CODE:
3040{
3041 struct coro *current = SvSTATE_current;
3042 SV **throwp = self == current ? &CORO_THROW : &self->except;
3043 SvREFCNT_dec (*throwp);
3044 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
3045}
478 3046
479 if (coro->mainstack) 3047void
3048api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
3049 PROTOTYPE: $;$
3050 C_ARGS: aTHX_ coro, flags
3051
3052SV *
3053has_cctx (Coro::State coro)
3054 PROTOTYPE: $
3055 CODE:
3056 /* maybe manage the running flag differently */
3057 RETVAL = boolSV (!!coro->cctx || (coro->flags & CF_RUNNING));
3058 OUTPUT:
3059 RETVAL
3060
3061int
3062is_traced (Coro::State coro)
3063 PROTOTYPE: $
3064 CODE:
3065 RETVAL = (coro->cctx ? coro->cctx->flags : 0) & CC_TRACE_ALL;
3066 OUTPUT:
3067 RETVAL
3068
3069UV
3070rss (Coro::State coro)
3071 PROTOTYPE: $
3072 ALIAS:
3073 usecount = 1
3074 CODE:
3075 switch (ix)
3076 {
3077 case 0: RETVAL = coro_rss (aTHX_ coro); break;
3078 case 1: RETVAL = coro->usecount; break;
3079 }
3080 OUTPUT:
3081 RETVAL
3082
3083void
3084force_cctx ()
3085 PROTOTYPE:
3086 CODE:
3087 cctx_current->idle_sp = 0;
3088
3089void
3090swap_defsv (Coro::State self)
3091 PROTOTYPE: $
3092 ALIAS:
3093 swap_defav = 1
3094 CODE:
3095 if (!self->slot)
3096 croak ("cannot swap state with coroutine that has no saved state,");
3097 else
480 { 3098 {
481 struct coro temp; 3099 SV **src = ix ? (SV **)&GvAV (PL_defgv) : &GvSV (PL_defgv);
3100 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv;
482 3101
483 SAVE(aTHX_ (&temp)); 3102 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 } 3103 }
491 3104
3105
3106MODULE = Coro::State PACKAGE = Coro
3107
3108BOOT:
3109{
3110 int i;
3111
3112 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
3113 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
3114 cv_coro_run = get_cv ( "Coro::_terminate", GV_ADD);
3115 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD);
3116 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current);
3117 av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE);
3118 av_destroy = coro_get_av (aTHX_ "Coro::destroy" , TRUE);
3119 sv_manager = coro_get_sv (aTHX_ "Coro::manager" , TRUE);
3120 sv_idle = coro_get_sv (aTHX_ "Coro::idle" , TRUE);
3121
3122 sv_async_pool_idle = newSVpv ("[async pool idle]", 0); SvREADONLY_on (sv_async_pool_idle);
3123 sv_Coro = newSVpv ("Coro", 0); SvREADONLY_on (sv_Coro);
3124 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler);
3125 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new);
3126
3127 coro_stash = gv_stashpv ("Coro", TRUE);
3128
3129 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX));
3130 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH));
3131 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (PRIO_NORMAL));
3132 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (PRIO_LOW));
3133 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (PRIO_IDLE));
3134 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (PRIO_MIN));
3135
3136 for (i = PRIO_MAX - PRIO_MIN + 1; i--; )
3137 coro_ready[i] = newAV ();
3138
3139 {
3140 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE);
3141
3142 coroapi.schedule = api_schedule;
3143 coroapi.schedule_to = api_schedule_to;
3144 coroapi.cede = api_cede;
3145 coroapi.cede_notself = api_cede_notself;
3146 coroapi.ready = api_ready;
3147 coroapi.is_ready = api_is_ready;
3148 coroapi.nready = coro_nready;
3149 coroapi.current = coro_current;
3150
3151 /*GCoroAPI = &coroapi;*/
3152 sv_setiv (sv, (IV)&coroapi);
3153 SvREADONLY_on (sv);
3154 }
3155}
3156
3157void
3158terminate (...)
3159 CODE:
3160 CORO_EXECUTE_SLF_XS (slf_init_terminate);
3161
3162void
3163schedule (...)
3164 CODE:
3165 CORO_EXECUTE_SLF_XS (slf_init_schedule);
3166
3167void
3168schedule_to (...)
3169 CODE:
3170 CORO_EXECUTE_SLF_XS (slf_init_schedule_to);
3171
3172void
3173cede_to (...)
3174 CODE:
3175 CORO_EXECUTE_SLF_XS (slf_init_cede_to);
3176
3177void
3178cede (...)
3179 CODE:
3180 CORO_EXECUTE_SLF_XS (slf_init_cede);
3181
3182void
3183cede_notself (...)
3184 CODE:
3185 CORO_EXECUTE_SLF_XS (slf_init_cede_notself);
3186
3187void
3188_cancel (Coro::State self)
3189 CODE:
3190 coro_state_destroy (aTHX_ self);
3191 coro_call_on_destroy (aTHX_ self);
3192
3193void
3194_set_current (SV *current)
3195 PROTOTYPE: $
3196 CODE:
3197 SvREFCNT_dec (SvRV (coro_current));
3198 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current)));
3199
3200void
3201_set_readyhook (SV *hook)
3202 PROTOTYPE: $
3203 CODE:
492 SvREFCNT_dec (coro->args); 3204 SvREFCNT_dec (coro_readyhook);
493 Safefree (coro); 3205 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0;
494 3206
3207int
3208prio (Coro::State coro, int newprio = 0)
3209 PROTOTYPE: $;$
3210 ALIAS:
3211 nice = 1
3212 CODE:
3213{
3214 RETVAL = coro->prio;
495 3215
3216 if (items > 1)
3217 {
3218 if (ix)
3219 newprio = coro->prio - newprio;
3220
3221 if (newprio < PRIO_MIN) newprio = PRIO_MIN;
3222 if (newprio > PRIO_MAX) newprio = PRIO_MAX;
3223
3224 coro->prio = newprio;
3225 }
3226}
3227 OUTPUT:
3228 RETVAL
3229
3230SV *
3231ready (SV *self)
3232 PROTOTYPE: $
3233 CODE:
3234 RETVAL = boolSV (api_ready (aTHX_ self));
3235 OUTPUT:
3236 RETVAL
3237
3238int
3239nready (...)
3240 PROTOTYPE:
3241 CODE:
3242 RETVAL = coro_nready;
3243 OUTPUT:
3244 RETVAL
3245
3246void
3247_pool_handler (...)
3248 CODE:
3249 CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
3250
3251void
3252async_pool (SV *cv, ...)
3253 PROTOTYPE: &@
3254 PPCODE:
3255{
3256 HV *hv = (HV *)av_pop (av_async_pool);
3257 AV *av = newAV ();
3258 SV *cb = ST (0);
3259 int i;
3260
3261 av_extend (av, items - 2);
3262 for (i = 1; i < items; ++i)
3263 av_push (av, SvREFCNT_inc_NN (ST (i)));
3264
3265 if ((SV *)hv == &PL_sv_undef)
3266 {
3267 PUSHMARK (SP);
3268 EXTEND (SP, 2);
3269 PUSHs (sv_Coro);
3270 PUSHs ((SV *)cv_pool_handler);
3271 PUTBACK;
3272 call_sv ((SV *)cv_coro_state_new, G_SCALAR);
3273 SPAGAIN;
3274
3275 hv = (HV *)SvREFCNT_inc_NN (SvRV (POPs));
3276 }
3277
3278 {
3279 struct coro *coro = SvSTATE_hv (hv);
3280
3281 assert (!coro->invoke_cb);
3282 assert (!coro->invoke_av);
3283 coro->invoke_cb = SvREFCNT_inc (cb);
3284 coro->invoke_av = av;
3285 }
3286
3287 api_ready (aTHX_ (SV *)hv);
3288
3289 if (GIMME_V != G_VOID)
3290 XPUSHs (sv_2mortal (newRV_noinc ((SV *)hv)));
3291 else
3292 SvREFCNT_dec (hv);
3293}
3294
3295SV *
3296rouse_cb ()
3297 PROTOTYPE:
3298 CODE:
3299 RETVAL = coro_new_rouse_cb (aTHX);
3300 OUTPUT:
3301 RETVAL
3302
3303void
3304rouse_wait (...)
3305 PROTOTYPE: ;$
3306 PPCODE:
3307 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
3308
3309
3310MODULE = Coro::State PACKAGE = PerlIO::cede
3311
3312BOOT:
3313 PerlIO_define_layer (aTHX_ &PerlIO_cede);
3314
3315
3316MODULE = Coro::State PACKAGE = Coro::Semaphore
3317
3318SV *
3319new (SV *klass, SV *count = 0)
3320 CODE:
3321 RETVAL = sv_bless (
3322 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
3323 GvSTASH (CvGV (cv))
3324 );
3325 OUTPUT:
3326 RETVAL
3327
3328# helper for Coro::Channel
3329SV *
3330_alloc (int count)
3331 CODE:
3332 RETVAL = coro_waitarray_new (aTHX_ count);
3333 OUTPUT:
3334 RETVAL
3335
3336SV *
3337count (SV *self)
3338 CODE:
3339 RETVAL = newSVsv (AvARRAY ((AV *)SvRV (self))[0]);
3340 OUTPUT:
3341 RETVAL
3342
3343void
3344up (SV *self, int adjust = 1)
3345 ALIAS:
3346 adjust = 1
3347 CODE:
3348 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
3349
3350void
3351down (...)
3352 CODE:
3353 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3354
3355void
3356wait (...)
3357 CODE:
3358 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
3359
3360void
3361try (SV *self)
3362 PPCODE:
3363{
3364 AV *av = (AV *)SvRV (self);
3365 SV *count_sv = AvARRAY (av)[0];
3366 IV count = SvIVX (count_sv);
3367
3368 if (count > 0)
3369 {
3370 --count;
3371 SvIVX (count_sv) = count;
3372 XSRETURN_YES;
3373 }
3374 else
3375 XSRETURN_NO;
3376}
3377
3378void
3379waiters (SV *self)
3380 PPCODE:
3381{
3382 AV *av = (AV *)SvRV (self);
3383 int wcount = AvFILLp (av) + 1 - 1;
3384
3385 if (GIMME_V == G_SCALAR)
3386 XPUSHs (sv_2mortal (newSViv (wcount)));
3387 else
3388 {
3389 int i;
3390 EXTEND (SP, wcount);
3391 for (i = 1; i <= wcount; ++i)
3392 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
3393 }
3394}
3395
3396MODULE = Coro::State PACKAGE = Coro::Signal
3397
3398SV *
3399new (SV *klass)
3400 CODE:
3401 RETVAL = sv_bless (
3402 coro_waitarray_new (aTHX_ 0),
3403 GvSTASH (CvGV (cv))
3404 );
3405 OUTPUT:
3406 RETVAL
3407
3408void
3409wait (...)
3410 CODE:
3411 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3412
3413void
3414broadcast (SV *self)
3415 CODE:
3416{
3417 AV *av = (AV *)SvRV (self);
3418 coro_signal_wake (aTHX_ av, AvFILLp (av));
3419}
3420
3421void
3422send (SV *self)
3423 CODE:
3424{
3425 AV *av = (AV *)SvRV (self);
3426
3427 if (AvFILLp (av))
3428 coro_signal_wake (aTHX_ av, 1);
3429 else
3430 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
3431}
3432
3433IV
3434awaited (SV *self)
3435 CODE:
3436 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3437 OUTPUT:
3438 RETVAL
3439
3440
3441MODULE = Coro::State PACKAGE = Coro::AnyEvent
3442
3443BOOT:
3444 sv_activity = coro_get_sv (aTHX_ "Coro::AnyEvent::ACTIVITY", TRUE);
3445
3446void
3447_schedule (...)
3448 CODE:
3449{
3450 static int incede;
3451
3452 api_cede_notself (aTHX);
3453
3454 ++incede;
3455 while (coro_nready >= incede && api_cede (aTHX))
3456 ;
3457
3458 sv_setsv (sv_activity, &PL_sv_undef);
3459 if (coro_nready >= incede)
3460 {
3461 PUSHMARK (SP);
3462 PUTBACK;
3463 call_pv ("Coro::AnyEvent::_activity", G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3464 }
3465
3466 --incede;
3467}
3468
3469
3470MODULE = Coro::State PACKAGE = Coro::AIO
3471
3472void
3473_register (char *target, char *proto, SV *req)
3474 CODE:
3475{
3476 CV *req_cv = coro_sv_2cv (aTHX_ req);
3477 /* newXSproto doesn't return the CV on 5.8 */
3478 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3479 sv_setpv ((SV *)slf_cv, proto);
3480 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
3481}
3482

Diff Legend

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