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.4 by root, Tue Jul 17 02:21:56 2001 UTC vs.
Revision 1.283 by root, Sun Nov 16 11:12:57 2008 UTC

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

Diff Legend

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