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

Diff Legend

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