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

Diff Legend

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