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

Comparing Coro/Coro/State.xs (file contents):
Revision 1.4 by root, Tue Jul 17 02:21:56 2001 UTC vs.
Revision 1.409 by root, Sun Jun 12 04:29:15 2011 UTC

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

Diff Legend

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