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.5 by root, Tue Jul 17 02:55:29 2001 UTC vs.
Revision 1.321 by root, Sat Nov 22 02:09:54 2008 UTC

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

Diff Legend

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