ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Coro/Coro/State.xs
(Generate patch)

Comparing Coro/Coro/State.xs (file contents):
Revision 1.5 by root, Tue Jul 17 02:55:29 2001 UTC vs.
Revision 1.358 by root, Mon Jun 29 04:30:25 2009 UTC

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

Diff Legend

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