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.340 by root, Mon Dec 15 00:30:40 2008 UTC vs.
Revision 1.361 by root, Wed Jul 1 07:26:40 2009 UTC

… …
137 137
138#define IN_DESTRUCT PL_dirty 138#define IN_DESTRUCT PL_dirty
139 139
140#if __GNUC__ >= 3 140#if __GNUC__ >= 3
141# define attribute(x) __attribute__(x) 141# define attribute(x) __attribute__(x)
142# define expect(expr,value) __builtin_expect ((expr),(value)) 142# define expect(expr,value) __builtin_expect ((expr), (value))
143# define INLINE static inline 143# define INLINE static inline
144#else 144#else
145# define attribute(x) 145# define attribute(x)
146# define expect(expr,value) (expr) 146# define expect(expr,value) (expr)
147# define INLINE static 147# define INLINE static
… …
159# if CORO_PTHREAD 159# if CORO_PTHREAD
160static void *coro_thx; 160static void *coro_thx;
161# endif 161# endif
162#endif 162#endif
163 163
164#ifdef __linux
165# include <time.h> /* for timespec */
166# include <syscall.h> /* for SYS_* */
167# ifdef SYS_clock_gettime
168# define coro_clock_gettime(id, ts) syscall (SYS_clock_gettime, (id), (ts))
169# define CORO_CLOCK_MONOTONIC 1
170# define CORO_CLOCK_THREAD_CPUTIME_ID 3
171# endif
172#endif
173
164static double (*nvtime)(); /* so why doesn't it take void? */ 174static double (*nvtime)(); /* so why doesn't it take void? */
175static void (*u2time)(pTHX_ UV ret[2]);
165 176
166/* we hijack an hopefully unused CV flag for our purposes */ 177/* we hijack an hopefully unused CV flag for our purposes */
167#define CVf_SLF 0x4000 178#define CVf_SLF 0x4000
168static OP *pp_slf (pTHX); 179static OP *pp_slf (pTHX);
169 180
… …
195static CV *cv_coro_state_new; 206static CV *cv_coro_state_new;
196 207
197/* Coro::AnyEvent */ 208/* Coro::AnyEvent */
198static SV *sv_activity; 209static SV *sv_activity;
199 210
211/* enable processtime/realtime profiling */
212static char enable_times;
213typedef U32 coro_ts[2];
214static coro_ts time_real, time_cpu;
215static char times_valid;
216
200static struct coro_cctx *cctx_first; 217static struct coro_cctx *cctx_first;
201static int cctx_count, cctx_idle; 218static int cctx_count, cctx_idle;
202 219
203enum { 220enum {
204 CC_MAPPED = 0x01, 221 CC_MAPPED = 0x01,
… …
238enum { 255enum {
239 CF_RUNNING = 0x0001, /* coroutine is running */ 256 CF_RUNNING = 0x0001, /* coroutine is running */
240 CF_READY = 0x0002, /* coroutine is ready */ 257 CF_READY = 0x0002, /* coroutine is ready */
241 CF_NEW = 0x0004, /* has never been switched to */ 258 CF_NEW = 0x0004, /* has never been switched to */
242 CF_DESTROYED = 0x0008, /* coroutine data has been freed */ 259 CF_DESTROYED = 0x0008, /* coroutine data has been freed */
260 CF_SUSPENDED = 0x0010, /* coroutine can't be scheduled */
243}; 261};
244 262
245/* the structure where most of the perl state is stored, overlaid on the cxstack */ 263/* the structure where most of the perl state is stored, overlaid on the cxstack */
246typedef struct 264typedef struct
247{ 265{
… …
260/* this is a structure representing a perl-level coroutine */ 278/* this is a structure representing a perl-level coroutine */
261struct coro { 279struct coro {
262 /* the C coroutine allocated to this perl coroutine, if any */ 280 /* the C coroutine allocated to this perl coroutine, if any */
263 coro_cctx *cctx; 281 coro_cctx *cctx;
264 282
283 /* ready queue */
284 struct coro *next_ready;
285
265 /* state data */ 286 /* state data */
266 struct CoroSLF slf_frame; /* saved slf frame */ 287 struct CoroSLF slf_frame; /* saved slf frame */
267 AV *mainstack; 288 AV *mainstack;
268 perl_slots *slot; /* basically the saved sp */ 289 perl_slots *slot; /* basically the saved sp */
269 290
… …
289 310
290 /* on_enter/on_leave */ 311 /* on_enter/on_leave */
291 AV *on_enter; 312 AV *on_enter;
292 AV *on_leave; 313 AV *on_leave;
293 314
315 /* times */
316 coro_ts t_cpu, t_real;
317
294 /* linked list */ 318 /* linked list */
295 struct coro *next, *prev; 319 struct coro *next, *prev;
296}; 320};
297 321
298typedef struct coro *Coro__State; 322typedef struct coro *Coro__State;
… …
303/* the mainr easonw e don't support windows process emulation */ 327/* the mainr easonw e don't support windows process emulation */
304static struct CoroSLF slf_frame; /* the current slf frame */ 328static struct CoroSLF slf_frame; /* the current slf frame */
305 329
306/** Coro ********************************************************************/ 330/** Coro ********************************************************************/
307 331
308#define PRIO_MAX 3 332#define CORO_PRIO_MAX 3
309#define PRIO_HIGH 1 333#define CORO_PRIO_HIGH 1
310#define PRIO_NORMAL 0 334#define CORO_PRIO_NORMAL 0
311#define PRIO_LOW -1 335#define CORO_PRIO_LOW -1
312#define PRIO_IDLE -3 336#define CORO_PRIO_IDLE -3
313#define PRIO_MIN -4 337#define CORO_PRIO_MIN -4
314 338
315/* for Coro.pm */ 339/* for Coro.pm */
316static SV *coro_current; 340static SV *coro_current;
317static SV *coro_readyhook; 341static SV *coro_readyhook;
318static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1]; 342static struct coro *coro_ready [CORO_PRIO_MAX - CORO_PRIO_MIN + 1][2]; /* head|tail */
319static CV *cv_coro_run, *cv_coro_terminate; 343static CV *cv_coro_run, *cv_coro_terminate;
320static struct coro *coro_first; 344static struct coro *coro_first;
321#define coro_nready coroapi.nready 345#define coro_nready coroapi.nready
322 346
323/** lowlevel stuff **********************************************************/ 347/** lowlevel stuff **********************************************************/
… …
362 386
363 if (!cv) 387 if (!cv)
364 croak ("code reference expected"); 388 croak ("code reference expected");
365 389
366 return cv; 390 return cv;
391}
392
393INLINE void
394coro_times_update ()
395{
396#ifdef coro_clock_gettime
397 struct timespec ts;
398
399 ts.tv_sec = ts.tv_nsec = 0;
400 coro_clock_gettime (CORO_CLOCK_THREAD_CPUTIME_ID, &ts);
401 time_cpu [0] = ts.tv_sec; time_cpu [1] = ts.tv_nsec;
402
403 ts.tv_sec = ts.tv_nsec = 0;
404 coro_clock_gettime (CORO_CLOCK_MONOTONIC, &ts);
405 time_real [0] = ts.tv_sec; time_real [1] = ts.tv_nsec;
406#else
407 dTHX;
408 UV tv[2];
409
410 u2time (aTHX_ &tv);
411 time_real [0] = tv [0];
412 time_real [1] = tv [1] * 1000;
413#endif
414}
415
416INLINE void
417coro_times_add (struct coro *c)
418{
419 c->t_real [1] += time_real [1];
420 if (c->t_real [1] > 1000000000) { c->t_real [1] -= 1000000000; ++c->t_real [0]; }
421 c->t_real [0] += time_real [0];
422
423 c->t_cpu [1] += time_cpu [1];
424 if (c->t_cpu [1] > 1000000000) { c->t_cpu [1] -= 1000000000; ++c->t_cpu [0]; }
425 c->t_cpu [0] += time_cpu [0];
426}
427
428INLINE void
429coro_times_sub (struct coro *c)
430{
431 if (c->t_real [1] < time_real [1]) { c->t_real [1] += 1000000000; --c->t_real [0]; }
432 c->t_real [1] -= time_real [1];
433 c->t_real [0] -= time_real [0];
434
435 if (c->t_cpu [1] < time_cpu [1]) { c->t_cpu [1] += 1000000000; --c->t_cpu [0]; }
436 c->t_cpu [1] -= time_cpu [1];
437 c->t_cpu [0] -= time_cpu [0];
367} 438}
368 439
369/*****************************************************************************/ 440/*****************************************************************************/
370/* magic glue */ 441/* magic glue */
371 442
… …
471static int 542static int
472coro_cv_free (pTHX_ SV *sv, MAGIC *mg) 543coro_cv_free (pTHX_ SV *sv, MAGIC *mg)
473{ 544{
474 AV *padlist; 545 AV *padlist;
475 AV *av = (AV *)mg->mg_obj; 546 AV *av = (AV *)mg->mg_obj;
547
548 /* perl manages to free our internal AV and _then_ call us */
549 if (IN_DESTRUCT)
550 return;
476 551
477 /* casting is fun. */ 552 /* casting is fun. */
478 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av))) 553 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av)))
479 free_padlist (aTHX_ padlist); 554 free_padlist (aTHX_ padlist);
480 555
… …
569 } 644 }
570 645
571 slf_frame = c->slf_frame; 646 slf_frame = c->slf_frame;
572 CORO_THROW = c->except; 647 CORO_THROW = c->except;
573 648
649 if (expect_false (enable_times))
650 {
651 if (expect_false (!times_valid))
652 coro_times_update ();
653
654 coro_times_sub (c);
655 }
656
574 if (expect_false (c->on_enter)) 657 if (expect_false (c->on_enter))
575 { 658 {
576 int i; 659 int i;
577 660
578 for (i = 0; i <= AvFILLp (c->on_enter); ++i) 661 for (i = 0; i <= AvFILLp (c->on_enter); ++i)
… …
587 { 670 {
588 int i; 671 int i;
589 672
590 for (i = AvFILLp (c->on_leave); i >= 0; --i) 673 for (i = AvFILLp (c->on_leave); i >= 0; --i)
591 on_enterleave_call (aTHX_ AvARRAY (c->on_leave)[i]); 674 on_enterleave_call (aTHX_ AvARRAY (c->on_leave)[i]);
675 }
676
677 times_valid = 0;
678
679 if (expect_false (enable_times))
680 {
681 coro_times_update (); times_valid = 1;
682 coro_times_add (c);
592 } 683 }
593 684
594 c->except = CORO_THROW; 685 c->except = CORO_THROW;
595 c->slf_frame = slf_frame; 686 c->slf_frame = slf_frame;
596 687
… …
611 { 702 {
612 while (expect_true (cxix >= 0)) 703 while (expect_true (cxix >= 0))
613 { 704 {
614 PERL_CONTEXT *cx = &ccstk[cxix--]; 705 PERL_CONTEXT *cx = &ccstk[cxix--];
615 706
616 if (expect_true (CxTYPE (cx) == CXt_SUB || CxTYPE (cx) == CXt_FORMAT)) 707 if (expect_true (CxTYPE (cx) == CXt_SUB) || expect_false (CxTYPE (cx) == CXt_FORMAT))
617 { 708 {
618 CV *cv = cx->blk_sub.cv; 709 CV *cv = cx->blk_sub.cv;
619 710
620 if (expect_true (CvDEPTH (cv))) 711 if (expect_true (CvDEPTH (cv)))
621 { 712 {
… …
960 1051
961 PL_op = (OP *)&coro_setup_op; 1052 PL_op = (OP *)&coro_setup_op;
962 1053
963 /* copy throw, in case it was set before coro_setup */ 1054 /* copy throw, in case it was set before coro_setup */
964 CORO_THROW = coro->except; 1055 CORO_THROW = coro->except;
1056
1057 if (expect_false (enable_times))
1058 {
1059 coro_times_update ();
1060 coro_times_sub (coro);
1061 }
965} 1062}
966 1063
967static void 1064static void
968coro_unwind_stacks (pTHX) 1065coro_unwind_stacks (pTHX)
969{ 1066{
… …
986} 1083}
987 1084
988static void 1085static void
989coro_destruct_perl (pTHX_ struct coro *coro) 1086coro_destruct_perl (pTHX_ struct coro *coro)
990{ 1087{
1088 SV *svf [9];
1089
1090 {
1091 struct coro *current = SvSTATE_current;
1092
1093 assert (("FATAL: tried to destroy currently running coroutine", coro->mainstack != PL_mainstack));
1094
1095 save_perl (aTHX_ current);
1096 load_perl (aTHX_ coro);
1097
991 coro_unwind_stacks (aTHX); 1098 coro_unwind_stacks (aTHX);
1099 coro_destruct_stacks (aTHX);
992 1100
993 SvREFCNT_dec (GvSV (PL_defgv)); 1101 // now save some sv's to be free'd later
994 SvREFCNT_dec (GvAV (PL_defgv)); 1102 svf [0] = GvSV (PL_defgv);
995 SvREFCNT_dec (GvSV (PL_errgv)); 1103 svf [1] = (SV *)GvAV (PL_defgv);
996 SvREFCNT_dec (PL_defoutgv); 1104 svf [2] = GvSV (PL_errgv);
997 SvREFCNT_dec (PL_rs); 1105 svf [3] = (SV *)PL_defoutgv;
998 SvREFCNT_dec (GvSV (irsgv)); 1106 svf [4] = PL_rs;
999 SvREFCNT_dec (GvHV (PL_hintgv)); 1107 svf [5] = GvSV (irsgv);
1108 svf [6] = (SV *)GvHV (PL_hintgv);
1109 svf [7] = PL_diehook;
1110 svf [8] = PL_warnhook;
1111 assert (9 == sizeof (svf) / sizeof (*svf));
1000 1112
1001 SvREFCNT_dec (PL_diehook); 1113 load_perl (aTHX_ current);
1002 SvREFCNT_dec (PL_warnhook);
1003 1114 }
1115
1116 {
1117 int i;
1118
1119 for (i = 0; i < sizeof (svf) / sizeof (*svf); ++i)
1120 SvREFCNT_dec (svf [i]);
1121
1004 SvREFCNT_dec (coro->saved_deffh); 1122 SvREFCNT_dec (coro->saved_deffh);
1005 SvREFCNT_dec (coro->rouse_cb); 1123 SvREFCNT_dec (coro->rouse_cb);
1006 SvREFCNT_dec (coro->invoke_cb); 1124 SvREFCNT_dec (coro->invoke_cb);
1007 SvREFCNT_dec (coro->invoke_av); 1125 SvREFCNT_dec (coro->invoke_av);
1008 1126 }
1009 coro_destruct_stacks (aTHX);
1010} 1127}
1011 1128
1012INLINE void 1129INLINE void
1013free_coro_mortal (pTHX) 1130free_coro_mortal (pTHX)
1014{ 1131{
… …
1382 /* TODO: throwing up here is considered harmful */ 1499 /* TODO: throwing up here is considered harmful */
1383 1500
1384 if (expect_true (prev != next)) 1501 if (expect_true (prev != next))
1385 { 1502 {
1386 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW)))) 1503 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW))))
1387 croak ("Coro::State::transfer called with a suspended prev Coro::State, but can only transfer from running or new states,"); 1504 croak ("Coro::State::transfer called with a blocked prev Coro::State, but can only transfer from running or new states,");
1388 1505
1389 if (expect_false (next->flags & CF_RUNNING))
1390 croak ("Coro::State::transfer called with running next Coro::State, but can only transfer to inactive states,");
1391
1392 if (expect_false (next->flags & CF_DESTROYED)) 1506 if (expect_false (next->flags & (CF_RUNNING | CF_DESTROYED | CF_SUSPENDED)))
1393 croak ("Coro::State::transfer called with destroyed next Coro::State, but can only transfer to inactive states,"); 1507 croak ("Coro::State::transfer called with running, destroyed or suspended next Coro::State, but can only transfer to inactive states,");
1394 1508
1395#if !PERL_VERSION_ATLEAST (5,10,0) 1509#if !PERL_VERSION_ATLEAST (5,10,0)
1396 if (expect_false (PL_lex_state != LEX_NOTPARSING)) 1510 if (expect_false (PL_lex_state != LEX_NOTPARSING))
1397 croak ("Coro::State::transfer called while parsing, but this is not supported in your perl version,"); 1511 croak ("Coro::State::transfer called while parsing, but this is not supported in your perl version,");
1398#endif 1512#endif
… …
1486coro_state_destroy (pTHX_ struct coro *coro) 1600coro_state_destroy (pTHX_ struct coro *coro)
1487{ 1601{
1488 if (coro->flags & CF_DESTROYED) 1602 if (coro->flags & CF_DESTROYED)
1489 return 0; 1603 return 0;
1490 1604
1491 if (coro->on_destroy) 1605 if (coro->on_destroy && !PL_dirty)
1492 coro->on_destroy (aTHX_ coro); 1606 coro->on_destroy (aTHX_ coro);
1493 1607
1494 coro->flags |= CF_DESTROYED; 1608 coro->flags |= CF_DESTROYED;
1495 1609
1496 if (coro->flags & CF_READY) 1610 if (coro->flags & CF_READY)
… …
1504 1618
1505 if (coro->mainstack 1619 if (coro->mainstack
1506 && coro->mainstack != main_mainstack 1620 && coro->mainstack != main_mainstack
1507 && coro->slot 1621 && coro->slot
1508 && !PL_dirty) 1622 && !PL_dirty)
1509 {
1510 struct coro *current = SvSTATE_current;
1511
1512 assert (("FATAL: tried to destroy currently running coroutine", coro->mainstack != PL_mainstack));
1513
1514 save_perl (aTHX_ current);
1515 load_perl (aTHX_ coro);
1516
1517 coro_destruct_perl (aTHX_ coro); 1623 coro_destruct_perl (aTHX_ coro);
1518
1519 load_perl (aTHX_ current);
1520
1521 coro->slot = 0;
1522 }
1523 1624
1524 cctx_destroy (coro->cctx); 1625 cctx_destroy (coro->cctx);
1525 SvREFCNT_dec (coro->startcv); 1626 SvREFCNT_dec (coro->startcv);
1526 SvREFCNT_dec (coro->args); 1627 SvREFCNT_dec (coro->args);
1527 SvREFCNT_dec (CORO_THROW); 1628 SvREFCNT_dec (CORO_THROW);
… …
1614/** Coro ********************************************************************/ 1715/** Coro ********************************************************************/
1615 1716
1616INLINE void 1717INLINE void
1617coro_enq (pTHX_ struct coro *coro) 1718coro_enq (pTHX_ struct coro *coro)
1618{ 1719{
1619 av_push (coro_ready [coro->prio - PRIO_MIN], SvREFCNT_inc_NN (coro->hv)); 1720 struct coro **ready = coro_ready [coro->prio - CORO_PRIO_MIN];
1620}
1621 1721
1622INLINE SV * 1722 SvREFCNT_inc_NN (coro->hv);
1723
1724 coro->next_ready = 0;
1725 *(ready [0] ? &ready [1]->next_ready : &ready [0]) = coro;
1726 ready [1] = coro;
1727}
1728
1729INLINE struct coro *
1623coro_deq (pTHX) 1730coro_deq (pTHX)
1624{ 1731{
1625 int prio; 1732 int prio;
1626 1733
1627 for (prio = PRIO_MAX - PRIO_MIN + 1; --prio >= 0; ) 1734 for (prio = CORO_PRIO_MAX - CORO_PRIO_MIN + 1; --prio >= 0; )
1628 if (AvFILLp (coro_ready [prio]) >= 0) 1735 {
1629 return av_shift (coro_ready [prio]); 1736 struct coro **ready = coro_ready [prio];
1737
1738 if (ready [0])
1739 {
1740 struct coro *coro = ready [0];
1741 ready [0] = coro->next_ready;
1742 return coro;
1743 }
1744 }
1630 1745
1631 return 0; 1746 return 0;
1632} 1747}
1633 1748
1634static int 1749static int
… …
1698static void 1813static void
1699prepare_schedule (pTHX_ struct coro_transfer_args *ta) 1814prepare_schedule (pTHX_ struct coro_transfer_args *ta)
1700{ 1815{
1701 for (;;) 1816 for (;;)
1702 { 1817 {
1703 SV *next_sv = coro_deq (aTHX); 1818 struct coro *next = coro_deq (aTHX);
1704 1819
1705 if (expect_true (next_sv)) 1820 if (expect_true (next))
1706 { 1821 {
1707 struct coro *next = SvSTATE_hv (next_sv);
1708
1709 /* cannot transfer to destroyed coros, skip and look for next */ 1822 /* cannot transfer to destroyed coros, skip and look for next */
1710 if (expect_false (next->flags & CF_DESTROYED)) 1823 if (expect_false (next->flags & (CF_DESTROYED | CF_SUSPENDED)))
1711 SvREFCNT_dec (next_sv); /* coro_nready has already been taken care of by destroy */ 1824 SvREFCNT_dec (next->hv); /* coro_nready has already been taken care of by destroy */
1712 else 1825 else
1713 { 1826 {
1714 next->flags &= ~CF_READY; 1827 next->flags &= ~CF_READY;
1715 --coro_nready; 1828 --coro_nready;
1716 1829
… …
1990 SV *data = (SV *)GENSUB_ARG; 2103 SV *data = (SV *)GENSUB_ARG;
1991 2104
1992 if (SvTYPE (SvRV (data)) != SVt_PVAV) 2105 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1993 { 2106 {
1994 /* first call, set args */ 2107 /* first call, set args */
2108 SV *coro = SvRV (data);
1995 AV *av = newAV (); 2109 AV *av = newAV ();
1996 SV *coro = SvRV (data);
1997 2110
1998 SvRV_set (data, (SV *)av); 2111 SvRV_set (data, (SV *)av);
1999 api_ready (aTHX_ coro);
2000 SvREFCNT_dec (coro);
2001 2112
2002 /* better take a full copy of the arguments */ 2113 /* better take a full copy of the arguments */
2003 while (items--) 2114 while (items--)
2004 av_store (av, items, newSVsv (ST (items))); 2115 av_store (av, items, newSVsv (ST (items)));
2116
2117 api_ready (aTHX_ coro);
2118 SvREFCNT_dec (coro);
2005 } 2119 }
2006 2120
2007 XSRETURN_EMPTY; 2121 XSRETURN_EMPTY;
2008} 2122}
2009 2123
… …
2026 2140
2027 EXTEND (SP, AvFILLp (av) + 1); 2141 EXTEND (SP, AvFILLp (av) + 1);
2028 for (i = 0; i <= AvFILLp (av); ++i) 2142 for (i = 0; i <= AvFILLp (av); ++i)
2029 PUSHs (sv_2mortal (AvARRAY (av)[i])); 2143 PUSHs (sv_2mortal (AvARRAY (av)[i]));
2030 2144
2031 /* we have stolen the elements, so ste length to zero and free */ 2145 /* we have stolen the elements, so set length to zero and free */
2032 AvFILLp (av) = -1; 2146 AvFILLp (av) = -1;
2033 av_undef (av); 2147 av_undef (av);
2034 2148
2035 PUTBACK; 2149 PUTBACK;
2036 } 2150 }
… …
2456/* Coro::Semaphore & Coro::Signal */ 2570/* Coro::Semaphore & Coro::Signal */
2457 2571
2458static SV * 2572static SV *
2459coro_waitarray_new (pTHX_ int count) 2573coro_waitarray_new (pTHX_ int count)
2460{ 2574{
2461 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2575 /* a waitarray=semaphore contains a counter IV in $sem->[0] and any waiters after that */
2462 AV *av = newAV (); 2576 AV *av = newAV ();
2463 SV **ary; 2577 SV **ary;
2464 2578
2465 /* unfortunately, building manually saves memory */ 2579 /* unfortunately, building manually saves memory */
2466 Newx (ary, 2, SV *); 2580 Newx (ary, 2, SV *);
… …
2610 { 2724 {
2611 /* callback form */ 2725 /* callback form */
2612 AV *av = (AV *)SvRV (arg [0]); 2726 AV *av = (AV *)SvRV (arg [0]);
2613 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]); 2727 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2614 2728
2615 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv)); 2729 av_push (av, SvREFCNT_inc_NN (cb_cv));
2616 2730
2617 if (SvIVX (AvARRAY (av)[0]) > 0) 2731 if (SvIVX (AvARRAY (av)[0]) > 0)
2618 coro_semaphore_adjust (aTHX_ av, 0); 2732 coro_semaphore_adjust (aTHX_ av, 0);
2619 2733
2620 frame->prepare = prepare_nop; 2734 frame->prepare = prepare_nop;
… …
2644 AvARRAY (av)[0] = AvARRAY (av)[1]; 2758 AvARRAY (av)[0] = AvARRAY (av)[1];
2645 AvARRAY (av)[1] = cb; 2759 AvARRAY (av)[1] = cb;
2646 2760
2647 cb = av_shift (av); 2761 cb = av_shift (av);
2648 2762
2763 if (SvTYPE (cb) == SVt_PVCV)
2764 {
2765 dSP;
2766 PUSHMARK (SP);
2767 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2768 PUTBACK;
2769 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2770 }
2771 else
2772 {
2649 api_ready (aTHX_ cb); 2773 api_ready (aTHX_ cb);
2650 sv_setiv (cb, 0); /* signal waiter */ 2774 sv_setiv (cb, 0); /* signal waiter */
2775 }
2776
2651 SvREFCNT_dec (cb); 2777 SvREFCNT_dec (cb);
2652 2778
2653 --count; 2779 --count;
2654 } 2780 }
2655} 2781}
… …
2664static void 2790static void
2665slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2791slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2666{ 2792{
2667 AV *av = (AV *)SvRV (arg [0]); 2793 AV *av = (AV *)SvRV (arg [0]);
2668 2794
2795 if (items >= 2)
2796 {
2797 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2798 av_push (av, SvREFCNT_inc_NN (cb_cv));
2799
2669 if (SvIVX (AvARRAY (av)[0])) 2800 if (SvIVX (AvARRAY (av)[0]))
2801 coro_signal_wake (aTHX_ av, 1); /* ust be the only waiter */
2802
2803 frame->prepare = prepare_nop;
2804 frame->check = slf_check_nop;
2805 }
2806 else if (SvIVX (AvARRAY (av)[0]))
2670 { 2807 {
2671 SvIVX (AvARRAY (av)[0]) = 0; 2808 SvIVX (AvARRAY (av)[0]) = 0;
2672 frame->prepare = prepare_nop; 2809 frame->prepare = prepare_nop;
2673 frame->check = slf_check_nop; 2810 frame->check = slf_check_nop;
2674 } 2811 }
2675 else 2812 else
2676 { 2813 {
2677 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */ 2814 SV *waiter = newSVsv (coro_current); /* owned by signal av */
2678 2815
2679 av_push (av, waiter); 2816 av_push (av, waiter);
2680 2817
2681 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */ 2818 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2682 frame->prepare = prepare_schedule; 2819 frame->prepare = prepare_schedule;
… …
2921 3058
2922 if (!svp) croak ("Time::HiRes is required"); 3059 if (!svp) croak ("Time::HiRes is required");
2923 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer"); 3060 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer");
2924 3061
2925 nvtime = INT2PTR (double (*)(), SvIV (*svp)); 3062 nvtime = INT2PTR (double (*)(), SvIV (*svp));
3063
3064 svp = hv_fetch (PL_modglobal, "Time::U2time", 12, 0);
3065 u2time = INT2PTR (double (*)(), SvIV (*svp));
2926 } 3066 }
2927 3067
2928 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL)); 3068 assert (("PRIO_NORMAL must be 0", !CORO_PRIO_NORMAL));
2929} 3069}
2930 3070
2931SV * 3071SV *
2932new (char *klass, ...) 3072new (char *klass, ...)
2933 ALIAS: 3073 ALIAS:
… …
3100 if (ix) 3240 if (ix)
3101 eval_sv (coderef, 0); 3241 eval_sv (coderef, 0);
3102 else 3242 else
3103 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD); 3243 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3104 3244
3245 POPSTACK;
3105 SPAGAIN; 3246 SPAGAIN;
3106 POPSTACK;
3107 3247
3108 if (current != coro) 3248 if (current != coro)
3109 { 3249 {
3110 PUTBACK; 3250 PUTBACK;
3111 save_perl (aTHX_ coro); 3251 save_perl (aTHX_ coro);
… …
3121 ALIAS: 3261 ALIAS:
3122 is_ready = CF_READY 3262 is_ready = CF_READY
3123 is_running = CF_RUNNING 3263 is_running = CF_RUNNING
3124 is_new = CF_NEW 3264 is_new = CF_NEW
3125 is_destroyed = CF_DESTROYED 3265 is_destroyed = CF_DESTROYED
3266 is_suspended = CF_SUSPENDED
3126 CODE: 3267 CODE:
3127 RETVAL = boolSV (coro->flags & ix); 3268 RETVAL = boolSV (coro->flags & ix);
3128 OUTPUT: 3269 OUTPUT:
3129 RETVAL 3270 RETVAL
3130 3271
… …
3134 CODE: 3275 CODE:
3135{ 3276{
3136 struct coro *current = SvSTATE_current; 3277 struct coro *current = SvSTATE_current;
3137 SV **throwp = self == current ? &CORO_THROW : &self->except; 3278 SV **throwp = self == current ? &CORO_THROW : &self->except;
3138 SvREFCNT_dec (*throwp); 3279 SvREFCNT_dec (*throwp);
3280 SvGETMAGIC (throw);
3139 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 3281 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
3140} 3282}
3141 3283
3142void 3284void
3143api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB) 3285api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
… …
3195 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv; 3337 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv;
3196 3338
3197 SV *tmp = *src; *src = *dst; *dst = tmp; 3339 SV *tmp = *src; *src = *dst; *dst = tmp;
3198 } 3340 }
3199 3341
3342void
3343cancel (Coro::State self)
3344 CODE:
3345 coro_state_destroy (aTHX_ self);
3346 coro_call_on_destroy (aTHX_ self); /* actually only for Coro objects */
3347
3348
3349SV *
3350enable_times (int enabled = enable_times)
3351 CODE:
3352{
3353 RETVAL = boolSV (enable_times);
3354
3355 if (enabled != enable_times)
3356 {
3357 enable_times = enabled;
3358
3359 coro_times_update ();
3360 (enabled ? coro_times_sub : coro_times_add)(SvSTATE (coro_current));
3361 }
3362}
3363 OUTPUT:
3364 RETVAL
3365
3366void
3367times (Coro::State self)
3368 PPCODE:
3369{
3370 struct coro *current = SvSTATE (coro_current);
3371
3372 if (expect_false (current == self))
3373 {
3374 coro_times_update ();
3375 coro_times_add (SvSTATE (coro_current));
3376 }
3377
3378 EXTEND (SP, 2);
3379 PUSHs (sv_2mortal (newSVnv (self->t_real [0] + self->t_real [1] * 1e-9)));
3380 PUSHs (sv_2mortal (newSVnv (self->t_cpu [0] + self->t_cpu [1] * 1e-9)));
3381
3382 if (expect_false (current == self))
3383 coro_times_sub (SvSTATE (coro_current));
3384}
3200 3385
3201MODULE = Coro::State PACKAGE = Coro 3386MODULE = Coro::State PACKAGE = Coro
3202 3387
3203BOOT: 3388BOOT:
3204{ 3389{
3205 int i;
3206
3207 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE); 3390 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
3208 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE); 3391 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
3209 cv_coro_run = get_cv ( "Coro::_coro_run" , GV_ADD); 3392 cv_coro_run = get_cv ( "Coro::_coro_run" , GV_ADD);
3210 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD); 3393 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD);
3211 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current); 3394 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current);
… …
3219 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler); 3402 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler);
3220 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new); 3403 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new);
3221 3404
3222 coro_stash = gv_stashpv ("Coro", TRUE); 3405 coro_stash = gv_stashpv ("Coro", TRUE);
3223 3406
3224 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX)); 3407 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (CORO_PRIO_MAX));
3225 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH)); 3408 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (CORO_PRIO_HIGH));
3226 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (PRIO_NORMAL)); 3409 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (CORO_PRIO_NORMAL));
3227 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (PRIO_LOW)); 3410 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (CORO_PRIO_LOW));
3228 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (PRIO_IDLE)); 3411 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (CORO_PRIO_IDLE));
3229 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (PRIO_MIN)); 3412 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (CORO_PRIO_MIN));
3230
3231 for (i = PRIO_MAX - PRIO_MIN + 1; i--; )
3232 coro_ready[i] = newAV ();
3233 3413
3234 { 3414 {
3235 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE); 3415 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE);
3236 3416
3237 coroapi.schedule = api_schedule; 3417 coroapi.schedule = api_schedule;
… …
3278cede_notself (...) 3458cede_notself (...)
3279 CODE: 3459 CODE:
3280 CORO_EXECUTE_SLF_XS (slf_init_cede_notself); 3460 CORO_EXECUTE_SLF_XS (slf_init_cede_notself);
3281 3461
3282void 3462void
3283_cancel (Coro::State self)
3284 CODE:
3285 coro_state_destroy (aTHX_ self);
3286 coro_call_on_destroy (aTHX_ self);
3287
3288void
3289_set_current (SV *current) 3463_set_current (SV *current)
3290 PROTOTYPE: $ 3464 PROTOTYPE: $
3291 CODE: 3465 CODE:
3292 SvREFCNT_dec (SvRV (coro_current)); 3466 SvREFCNT_dec (SvRV (coro_current));
3293 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current))); 3467 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current)));
… …
3295void 3469void
3296_set_readyhook (SV *hook) 3470_set_readyhook (SV *hook)
3297 PROTOTYPE: $ 3471 PROTOTYPE: $
3298 CODE: 3472 CODE:
3299 SvREFCNT_dec (coro_readyhook); 3473 SvREFCNT_dec (coro_readyhook);
3474 SvGETMAGIC (hook);
3300 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0; 3475 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0;
3301 3476
3302int 3477int
3303prio (Coro::State coro, int newprio = 0) 3478prio (Coro::State coro, int newprio = 0)
3304 PROTOTYPE: $;$ 3479 PROTOTYPE: $;$
… …
3311 if (items > 1) 3486 if (items > 1)
3312 { 3487 {
3313 if (ix) 3488 if (ix)
3314 newprio = coro->prio - newprio; 3489 newprio = coro->prio - newprio;
3315 3490
3316 if (newprio < PRIO_MIN) newprio = PRIO_MIN; 3491 if (newprio < CORO_PRIO_MIN) newprio = CORO_PRIO_MIN;
3317 if (newprio > PRIO_MAX) newprio = PRIO_MAX; 3492 if (newprio > CORO_PRIO_MAX) newprio = CORO_PRIO_MAX;
3318 3493
3319 coro->prio = newprio; 3494 coro->prio = newprio;
3320 } 3495 }
3321} 3496}
3322 OUTPUT: 3497 OUTPUT:
… …
3335 PROTOTYPE: 3510 PROTOTYPE:
3336 CODE: 3511 CODE:
3337 RETVAL = coro_nready; 3512 RETVAL = coro_nready;
3338 OUTPUT: 3513 OUTPUT:
3339 RETVAL 3514 RETVAL
3515
3516void
3517suspend (Coro::State self)
3518 PROTOTYPE: $
3519 CODE:
3520 self->flags |= CF_SUSPENDED;
3521
3522void
3523resume (Coro::State self)
3524 PROTOTYPE: $
3525 CODE:
3526 self->flags &= ~CF_SUSPENDED;
3340 3527
3341void 3528void
3342_pool_handler (...) 3529_pool_handler (...)
3343 CODE: 3530 CODE:
3344 CORO_EXECUTE_SLF_XS (slf_init_pool_handler); 3531 CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
… …
3419 av_push (*avp, SvREFCNT_inc (block)); 3606 av_push (*avp, SvREFCNT_inc (block));
3420 3607
3421 if (!ix) 3608 if (!ix)
3422 on_enterleave_call (aTHX_ block); 3609 on_enterleave_call (aTHX_ block);
3423 3610
3424 LEAVE; /* pp_entersub unfortunately forces an ENTER/LEAVE around xs calls */ 3611 LEAVE; /* pp_entersub unfortunately forces an ENTER/LEAVE around XS calls */
3425 SAVEDESTRUCTOR_X (ix ? coro_pop_on_leave : coro_pop_on_enter, (void *)coro); 3612 SAVEDESTRUCTOR_X (ix ? coro_pop_on_leave : coro_pop_on_enter, (void *)coro);
3426 ENTER; /* pp_entersub unfortunately forces an ENTER/LEAVE around xs calls */ 3613 ENTER; /* pp_entersub unfortunately forces an ENTER/LEAVE around XS calls */
3427} 3614}
3428 3615
3429 3616
3430MODULE = Coro::State PACKAGE = PerlIO::cede 3617MODULE = Coro::State PACKAGE = PerlIO::cede
3431 3618
… …
3436MODULE = Coro::State PACKAGE = Coro::Semaphore 3623MODULE = Coro::State PACKAGE = Coro::Semaphore
3437 3624
3438SV * 3625SV *
3439new (SV *klass, SV *count = 0) 3626new (SV *klass, SV *count = 0)
3440 CODE: 3627 CODE:
3628{
3629 int semcnt = 1;
3630
3631 if (count)
3632 {
3633 SvGETMAGIC (count);
3634
3635 if (SvOK (count))
3636 semcnt = SvIV (count);
3637 }
3638
3441 RETVAL = sv_bless ( 3639 RETVAL = sv_bless (
3442 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1), 3640 coro_waitarray_new (aTHX_ semcnt),
3443 GvSTASH (CvGV (cv)) 3641 GvSTASH (CvGV (cv))
3444 ); 3642 );
3643}
3445 OUTPUT: 3644 OUTPUT:
3446 RETVAL 3645 RETVAL
3447 3646
3448# helper for Coro::Channel 3647# helper for Coro::Channel and others
3449SV * 3648SV *
3450_alloc (int count) 3649_alloc (int count)
3451 CODE: 3650 CODE:
3452 RETVAL = coro_waitarray_new (aTHX_ count); 3651 RETVAL = coro_waitarray_new (aTHX_ count);
3453 OUTPUT: 3652 OUTPUT:
… …
3511 for (i = 1; i <= wcount; ++i) 3710 for (i = 1; i <= wcount; ++i)
3512 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 3711 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
3513 } 3712 }
3514} 3713}
3515 3714
3715MODULE = Coro::State PACKAGE = Coro::SemaphoreSet
3716
3717void
3718_may_delete (SV *sem, int count, int extra_refs)
3719 PPCODE:
3720{
3721 AV *av = (AV *)SvRV (sem);
3722
3723 if (SvREFCNT ((SV *)av) == 1 + extra_refs
3724 && AvFILLp (av) == 0 /* no waiters, just count */
3725 && SvIV (AvARRAY (av)[0]) == count)
3726 XSRETURN_YES;
3727
3728 XSRETURN_NO;
3729}
3730
3516MODULE = Coro::State PACKAGE = Coro::Signal 3731MODULE = Coro::State PACKAGE = Coro::Signal
3517 3732
3518SV * 3733SV *
3519new (SV *klass) 3734new (SV *klass)
3520 CODE: 3735 CODE:

Diff Legend

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