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.339 by root, Mon Dec 15 00:28:30 2008 UTC vs.
Revision 1.360 by root, Mon Jun 29 06:14:23 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 UV tv[2];
408
409 u2time (aTHX_ &tv);
410 time_real [0] = tv [0];
411 time_real [1] = tv [1] * 1000;
412#endif
413}
414
415INLINE void
416coro_times_add (struct coro *c)
417{
418 c->t_real [1] += time_real [1];
419 if (c->t_real [1] > 1000000000) { c->t_real [1] -= 1000000000; ++c->t_real [0]; }
420 c->t_real [0] += time_real [0];
421
422 c->t_cpu [1] += time_cpu [1];
423 if (c->t_cpu [1] > 1000000000) { c->t_cpu [1] -= 1000000000; ++c->t_cpu [0]; }
424 c->t_cpu [0] += time_cpu [0];
425}
426
427INLINE void
428coro_times_sub (struct coro *c)
429{
430 if (c->t_real [1] < time_real [1]) { c->t_real [1] += 1000000000; --c->t_real [0]; }
431 c->t_real [1] -= time_real [1];
432 c->t_real [0] -= time_real [0];
433
434 if (c->t_cpu [1] < time_cpu [1]) { c->t_cpu [1] += 1000000000; --c->t_cpu [0]; }
435 c->t_cpu [1] -= time_cpu [1];
436 c->t_cpu [0] -= time_cpu [0];
367} 437}
368 438
369/*****************************************************************************/ 439/*****************************************************************************/
370/* magic glue */ 440/* magic glue */
371 441
… …
471static int 541static int
472coro_cv_free (pTHX_ SV *sv, MAGIC *mg) 542coro_cv_free (pTHX_ SV *sv, MAGIC *mg)
473{ 543{
474 AV *padlist; 544 AV *padlist;
475 AV *av = (AV *)mg->mg_obj; 545 AV *av = (AV *)mg->mg_obj;
546
547 /* perl manages to free our internal AV and _then_ call us */
548 if (IN_DESTRUCT)
549 return;
476 550
477 /* casting is fun. */ 551 /* casting is fun. */
478 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av))) 552 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av)))
479 free_padlist (aTHX_ padlist); 553 free_padlist (aTHX_ padlist);
480 554
… …
569 } 643 }
570 644
571 slf_frame = c->slf_frame; 645 slf_frame = c->slf_frame;
572 CORO_THROW = c->except; 646 CORO_THROW = c->except;
573 647
648 if (expect_false (enable_times))
649 {
650 if (expect_false (!times_valid))
651 coro_times_update ();
652
653 coro_times_sub (c);
654 }
655
574 if (expect_false (c->on_enter)) 656 if (expect_false (c->on_enter))
575 { 657 {
576 int i; 658 int i;
577 659
578 for (i = 0; i <= AvFILLp (c->on_enter); ++i) 660 for (i = 0; i <= AvFILLp (c->on_enter); ++i)
579 on_enterleave_call (AvARRAY (c->on_enter)[i]); 661 on_enterleave_call (aTHX_ AvARRAY (c->on_enter)[i]);
580 } 662 }
581} 663}
582 664
583static void 665static void
584save_perl (pTHX_ Coro__State c) 666save_perl (pTHX_ Coro__State c)
… …
586 if (expect_false (c->on_leave)) 668 if (expect_false (c->on_leave))
587 { 669 {
588 int i; 670 int i;
589 671
590 for (i = AvFILLp (c->on_leave); i >= 0; --i) 672 for (i = AvFILLp (c->on_leave); i >= 0; --i)
591 on_enterleave_call (AvARRAY (c->on_leave)[i]); 673 on_enterleave_call (aTHX_ AvARRAY (c->on_leave)[i]);
674 }
675
676 times_valid = 0;
677
678 if (expect_false (enable_times))
679 {
680 coro_times_update (); times_valid = 1;
681 coro_times_add (c);
592 } 682 }
593 683
594 c->except = CORO_THROW; 684 c->except = CORO_THROW;
595 c->slf_frame = slf_frame; 685 c->slf_frame = slf_frame;
596 686
… …
611 { 701 {
612 while (expect_true (cxix >= 0)) 702 while (expect_true (cxix >= 0))
613 { 703 {
614 PERL_CONTEXT *cx = &ccstk[cxix--]; 704 PERL_CONTEXT *cx = &ccstk[cxix--];
615 705
616 if (expect_true (CxTYPE (cx) == CXt_SUB || CxTYPE (cx) == CXt_FORMAT)) 706 if (expect_true (CxTYPE (cx) == CXt_SUB) || expect_false (CxTYPE (cx) == CXt_FORMAT))
617 { 707 {
618 CV *cv = cx->blk_sub.cv; 708 CV *cv = cx->blk_sub.cv;
619 709
620 if (expect_true (CvDEPTH (cv))) 710 if (expect_true (CvDEPTH (cv)))
621 { 711 {
… …
960 1050
961 PL_op = (OP *)&coro_setup_op; 1051 PL_op = (OP *)&coro_setup_op;
962 1052
963 /* copy throw, in case it was set before coro_setup */ 1053 /* copy throw, in case it was set before coro_setup */
964 CORO_THROW = coro->except; 1054 CORO_THROW = coro->except;
1055
1056 if (expect_false (enable_times))
1057 {
1058 coro_times_update ();
1059 coro_times_sub (coro);
1060 }
965} 1061}
966 1062
967static void 1063static void
968coro_unwind_stacks (pTHX) 1064coro_unwind_stacks (pTHX)
969{ 1065{
… …
986} 1082}
987 1083
988static void 1084static void
989coro_destruct_perl (pTHX_ struct coro *coro) 1085coro_destruct_perl (pTHX_ struct coro *coro)
990{ 1086{
1087 SV *svf [9];
1088
1089 {
1090 struct coro *current = SvSTATE_current;
1091
1092 assert (("FATAL: tried to destroy currently running coroutine", coro->mainstack != PL_mainstack));
1093
1094 save_perl (aTHX_ current);
1095 load_perl (aTHX_ coro);
1096
991 coro_unwind_stacks (aTHX); 1097 coro_unwind_stacks (aTHX);
1098 coro_destruct_stacks (aTHX);
992 1099
993 SvREFCNT_dec (GvSV (PL_defgv)); 1100 // now save some sv's to be free'd later
994 SvREFCNT_dec (GvAV (PL_defgv)); 1101 svf [0] = GvSV (PL_defgv);
995 SvREFCNT_dec (GvSV (PL_errgv)); 1102 svf [1] = (SV *)GvAV (PL_defgv);
996 SvREFCNT_dec (PL_defoutgv); 1103 svf [2] = GvSV (PL_errgv);
997 SvREFCNT_dec (PL_rs); 1104 svf [3] = (SV *)PL_defoutgv;
998 SvREFCNT_dec (GvSV (irsgv)); 1105 svf [4] = PL_rs;
999 SvREFCNT_dec (GvHV (PL_hintgv)); 1106 svf [5] = GvSV (irsgv);
1107 svf [6] = (SV *)GvHV (PL_hintgv);
1108 svf [7] = PL_diehook;
1109 svf [8] = PL_warnhook;
1110 assert (9 == sizeof (svf) / sizeof (*svf));
1000 1111
1001 SvREFCNT_dec (PL_diehook); 1112 load_perl (aTHX_ current);
1002 SvREFCNT_dec (PL_warnhook);
1003 1113 }
1114
1115 {
1116 int i;
1117
1118 for (i = 0; i < sizeof (svf) / sizeof (*svf); ++i)
1119 SvREFCNT_dec (svf [i]);
1120
1004 SvREFCNT_dec (coro->saved_deffh); 1121 SvREFCNT_dec (coro->saved_deffh);
1005 SvREFCNT_dec (coro->rouse_cb); 1122 SvREFCNT_dec (coro->rouse_cb);
1006 SvREFCNT_dec (coro->invoke_cb); 1123 SvREFCNT_dec (coro->invoke_cb);
1007 SvREFCNT_dec (coro->invoke_av); 1124 SvREFCNT_dec (coro->invoke_av);
1008 1125 }
1009 coro_destruct_stacks (aTHX);
1010} 1126}
1011 1127
1012INLINE void 1128INLINE void
1013free_coro_mortal (pTHX) 1129free_coro_mortal (pTHX)
1014{ 1130{
… …
1382 /* TODO: throwing up here is considered harmful */ 1498 /* TODO: throwing up here is considered harmful */
1383 1499
1384 if (expect_true (prev != next)) 1500 if (expect_true (prev != next))
1385 { 1501 {
1386 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW)))) 1502 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,"); 1503 croak ("Coro::State::transfer called with a blocked prev Coro::State, but can only transfer from running or new states,");
1388 1504
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)) 1505 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,"); 1506 croak ("Coro::State::transfer called with running, destroyed or suspended next Coro::State, but can only transfer to inactive states,");
1394 1507
1395#if !PERL_VERSION_ATLEAST (5,10,0) 1508#if !PERL_VERSION_ATLEAST (5,10,0)
1396 if (expect_false (PL_lex_state != LEX_NOTPARSING)) 1509 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,"); 1510 croak ("Coro::State::transfer called while parsing, but this is not supported in your perl version,");
1398#endif 1511#endif
… …
1486coro_state_destroy (pTHX_ struct coro *coro) 1599coro_state_destroy (pTHX_ struct coro *coro)
1487{ 1600{
1488 if (coro->flags & CF_DESTROYED) 1601 if (coro->flags & CF_DESTROYED)
1489 return 0; 1602 return 0;
1490 1603
1491 if (coro->on_destroy) 1604 if (coro->on_destroy && !PL_dirty)
1492 coro->on_destroy (aTHX_ coro); 1605 coro->on_destroy (aTHX_ coro);
1493 1606
1494 coro->flags |= CF_DESTROYED; 1607 coro->flags |= CF_DESTROYED;
1495 1608
1496 if (coro->flags & CF_READY) 1609 if (coro->flags & CF_READY)
… …
1504 1617
1505 if (coro->mainstack 1618 if (coro->mainstack
1506 && coro->mainstack != main_mainstack 1619 && coro->mainstack != main_mainstack
1507 && coro->slot 1620 && coro->slot
1508 && !PL_dirty) 1621 && !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); 1622 coro_destruct_perl (aTHX_ coro);
1518
1519 load_perl (aTHX_ current);
1520
1521 coro->slot = 0;
1522 }
1523 1623
1524 cctx_destroy (coro->cctx); 1624 cctx_destroy (coro->cctx);
1525 SvREFCNT_dec (coro->startcv); 1625 SvREFCNT_dec (coro->startcv);
1526 SvREFCNT_dec (coro->args); 1626 SvREFCNT_dec (coro->args);
1527 SvREFCNT_dec (CORO_THROW); 1627 SvREFCNT_dec (CORO_THROW);
… …
1614/** Coro ********************************************************************/ 1714/** Coro ********************************************************************/
1615 1715
1616INLINE void 1716INLINE void
1617coro_enq (pTHX_ struct coro *coro) 1717coro_enq (pTHX_ struct coro *coro)
1618{ 1718{
1619 av_push (coro_ready [coro->prio - PRIO_MIN], SvREFCNT_inc_NN (coro->hv)); 1719 struct coro **ready = coro_ready [coro->prio - CORO_PRIO_MIN];
1620}
1621 1720
1622INLINE SV * 1721 SvREFCNT_inc_NN (coro->hv);
1722
1723 coro->next_ready = 0;
1724 *(ready [0] ? &ready [1]->next_ready : &ready [0]) = coro;
1725 ready [1] = coro;
1726}
1727
1728INLINE struct coro *
1623coro_deq (pTHX) 1729coro_deq (pTHX)
1624{ 1730{
1625 int prio; 1731 int prio;
1626 1732
1627 for (prio = PRIO_MAX - PRIO_MIN + 1; --prio >= 0; ) 1733 for (prio = CORO_PRIO_MAX - CORO_PRIO_MIN + 1; --prio >= 0; )
1628 if (AvFILLp (coro_ready [prio]) >= 0) 1734 {
1629 return av_shift (coro_ready [prio]); 1735 struct coro **ready = coro_ready [prio];
1736
1737 if (ready [0])
1738 {
1739 struct coro *coro = ready [0];
1740 ready [0] = coro->next_ready;
1741 return coro;
1742 }
1743 }
1630 1744
1631 return 0; 1745 return 0;
1632} 1746}
1633 1747
1634static int 1748static int
… …
1698static void 1812static void
1699prepare_schedule (pTHX_ struct coro_transfer_args *ta) 1813prepare_schedule (pTHX_ struct coro_transfer_args *ta)
1700{ 1814{
1701 for (;;) 1815 for (;;)
1702 { 1816 {
1703 SV *next_sv = coro_deq (aTHX); 1817 struct coro *next = coro_deq (aTHX);
1704 1818
1705 if (expect_true (next_sv)) 1819 if (expect_true (next))
1706 { 1820 {
1707 struct coro *next = SvSTATE_hv (next_sv);
1708
1709 /* cannot transfer to destroyed coros, skip and look for next */ 1821 /* cannot transfer to destroyed coros, skip and look for next */
1710 if (expect_false (next->flags & CF_DESTROYED)) 1822 if (expect_false (next->flags & (CF_DESTROYED | CF_SUSPENDED)))
1711 SvREFCNT_dec (next_sv); /* coro_nready has already been taken care of by destroy */ 1823 SvREFCNT_dec (next->hv); /* coro_nready has already been taken care of by destroy */
1712 else 1824 else
1713 { 1825 {
1714 next->flags &= ~CF_READY; 1826 next->flags &= ~CF_READY;
1715 --coro_nready; 1827 --coro_nready;
1716 1828
… …
1990 SV *data = (SV *)GENSUB_ARG; 2102 SV *data = (SV *)GENSUB_ARG;
1991 2103
1992 if (SvTYPE (SvRV (data)) != SVt_PVAV) 2104 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1993 { 2105 {
1994 /* first call, set args */ 2106 /* first call, set args */
2107 SV *coro = SvRV (data);
1995 AV *av = newAV (); 2108 AV *av = newAV ();
1996 SV *coro = SvRV (data);
1997 2109
1998 SvRV_set (data, (SV *)av); 2110 SvRV_set (data, (SV *)av);
1999 api_ready (aTHX_ coro);
2000 SvREFCNT_dec (coro);
2001 2111
2002 /* better take a full copy of the arguments */ 2112 /* better take a full copy of the arguments */
2003 while (items--) 2113 while (items--)
2004 av_store (av, items, newSVsv (ST (items))); 2114 av_store (av, items, newSVsv (ST (items)));
2115
2116 api_ready (aTHX_ coro);
2117 SvREFCNT_dec (coro);
2005 } 2118 }
2006 2119
2007 XSRETURN_EMPTY; 2120 XSRETURN_EMPTY;
2008} 2121}
2009 2122
… …
2026 2139
2027 EXTEND (SP, AvFILLp (av) + 1); 2140 EXTEND (SP, AvFILLp (av) + 1);
2028 for (i = 0; i <= AvFILLp (av); ++i) 2141 for (i = 0; i <= AvFILLp (av); ++i)
2029 PUSHs (sv_2mortal (AvARRAY (av)[i])); 2142 PUSHs (sv_2mortal (AvARRAY (av)[i]));
2030 2143
2031 /* we have stolen the elements, so ste length to zero and free */ 2144 /* we have stolen the elements, so set length to zero and free */
2032 AvFILLp (av) = -1; 2145 AvFILLp (av) = -1;
2033 av_undef (av); 2146 av_undef (av);
2034 2147
2035 PUTBACK; 2148 PUTBACK;
2036 } 2149 }
… …
2347 2460
2348 POPSTACK; 2461 POPSTACK;
2349} 2462}
2350 2463
2351static SV * 2464static SV *
2352coro_avp_pop_and_free (AV **avp) 2465coro_avp_pop_and_free (pTHX_ AV **avp)
2353{ 2466{
2354 AV *av = *avp; 2467 AV *av = *avp;
2355 SV *res = av_pop (av); 2468 SV *res = av_pop (av);
2356 2469
2357 if (AvFILLp (av) < 0) 2470 if (AvFILLp (av) < 0)
… …
2364} 2477}
2365 2478
2366static void 2479static void
2367coro_pop_on_enter (pTHX_ void *coro) 2480coro_pop_on_enter (pTHX_ void *coro)
2368{ 2481{
2369 SV *cb = coro_avp_pop_and_free (&((struct coro *)coro)->on_enter); 2482 SV *cb = coro_avp_pop_and_free (aTHX_ &((struct coro *)coro)->on_enter);
2370 SvREFCNT_dec (cb); 2483 SvREFCNT_dec (cb);
2371} 2484}
2372 2485
2373static void 2486static void
2374coro_pop_on_leave (pTHX_ void *coro) 2487coro_pop_on_leave (pTHX_ void *coro)
2375{ 2488{
2376 SV *cb = coro_avp_pop_and_free (&((struct coro *)coro)->on_leave); 2489 SV *cb = coro_avp_pop_and_free (aTHX_ &((struct coro *)coro)->on_leave);
2377 on_enterleave_call (sv_2mortal (cb)); 2490 on_enterleave_call (aTHX_ sv_2mortal (cb));
2378} 2491}
2379 2492
2380/*****************************************************************************/ 2493/*****************************************************************************/
2381/* PerlIO::cede */ 2494/* PerlIO::cede */
2382 2495
… …
2456/* Coro::Semaphore & Coro::Signal */ 2569/* Coro::Semaphore & Coro::Signal */
2457 2570
2458static SV * 2571static SV *
2459coro_waitarray_new (pTHX_ int count) 2572coro_waitarray_new (pTHX_ int count)
2460{ 2573{
2461 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2574 /* a waitarray=semaphore contains a counter IV in $sem->[0] and any waiters after that */
2462 AV *av = newAV (); 2575 AV *av = newAV ();
2463 SV **ary; 2576 SV **ary;
2464 2577
2465 /* unfortunately, building manually saves memory */ 2578 /* unfortunately, building manually saves memory */
2466 Newx (ary, 2, SV *); 2579 Newx (ary, 2, SV *);
… …
2610 { 2723 {
2611 /* callback form */ 2724 /* callback form */
2612 AV *av = (AV *)SvRV (arg [0]); 2725 AV *av = (AV *)SvRV (arg [0]);
2613 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]); 2726 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2614 2727
2615 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv)); 2728 av_push (av, SvREFCNT_inc_NN (cb_cv));
2616 2729
2617 if (SvIVX (AvARRAY (av)[0]) > 0) 2730 if (SvIVX (AvARRAY (av)[0]) > 0)
2618 coro_semaphore_adjust (aTHX_ av, 0); 2731 coro_semaphore_adjust (aTHX_ av, 0);
2619 2732
2620 frame->prepare = prepare_nop; 2733 frame->prepare = prepare_nop;
… …
2644 AvARRAY (av)[0] = AvARRAY (av)[1]; 2757 AvARRAY (av)[0] = AvARRAY (av)[1];
2645 AvARRAY (av)[1] = cb; 2758 AvARRAY (av)[1] = cb;
2646 2759
2647 cb = av_shift (av); 2760 cb = av_shift (av);
2648 2761
2762 if (SvTYPE (cb) == SVt_PVCV)
2763 {
2764 dSP;
2765 PUSHMARK (SP);
2766 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2767 PUTBACK;
2768 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2769 }
2770 else
2771 {
2649 api_ready (aTHX_ cb); 2772 api_ready (aTHX_ cb);
2650 sv_setiv (cb, 0); /* signal waiter */ 2773 sv_setiv (cb, 0); /* signal waiter */
2774 }
2775
2651 SvREFCNT_dec (cb); 2776 SvREFCNT_dec (cb);
2652 2777
2653 --count; 2778 --count;
2654 } 2779 }
2655} 2780}
… …
2664static void 2789static void
2665slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2790slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2666{ 2791{
2667 AV *av = (AV *)SvRV (arg [0]); 2792 AV *av = (AV *)SvRV (arg [0]);
2668 2793
2794 if (items >= 2)
2795 {
2796 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2797 av_push (av, SvREFCNT_inc_NN (cb_cv));
2798
2669 if (SvIVX (AvARRAY (av)[0])) 2799 if (SvIVX (AvARRAY (av)[0]))
2800 coro_signal_wake (aTHX_ av, 1); /* ust be the only waiter */
2801
2802 frame->prepare = prepare_nop;
2803 frame->check = slf_check_nop;
2804 }
2805 else if (SvIVX (AvARRAY (av)[0]))
2670 { 2806 {
2671 SvIVX (AvARRAY (av)[0]) = 0; 2807 SvIVX (AvARRAY (av)[0]) = 0;
2672 frame->prepare = prepare_nop; 2808 frame->prepare = prepare_nop;
2673 frame->check = slf_check_nop; 2809 frame->check = slf_check_nop;
2674 } 2810 }
2675 else 2811 else
2676 { 2812 {
2677 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */ 2813 SV *waiter = newSVsv (coro_current); /* owned by signal av */
2678 2814
2679 av_push (av, waiter); 2815 av_push (av, waiter);
2680 2816
2681 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */ 2817 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2682 frame->prepare = prepare_schedule; 2818 frame->prepare = prepare_schedule;
… …
2921 3057
2922 if (!svp) croak ("Time::HiRes is required"); 3058 if (!svp) croak ("Time::HiRes is required");
2923 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer"); 3059 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer");
2924 3060
2925 nvtime = INT2PTR (double (*)(), SvIV (*svp)); 3061 nvtime = INT2PTR (double (*)(), SvIV (*svp));
3062
3063 svp = hv_fetch (PL_modglobal, "Time::U2time", 12, 0);
3064 u2time = INT2PTR (double (*)(), SvIV (*svp));
2926 } 3065 }
2927 3066
2928 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL)); 3067 assert (("PRIO_NORMAL must be 0", !CORO_PRIO_NORMAL));
2929} 3068}
2930 3069
2931SV * 3070SV *
2932new (char *klass, ...) 3071new (char *klass, ...)
2933 ALIAS: 3072 ALIAS:
… …
3100 if (ix) 3239 if (ix)
3101 eval_sv (coderef, 0); 3240 eval_sv (coderef, 0);
3102 else 3241 else
3103 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD); 3242 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3104 3243
3244 POPSTACK;
3105 SPAGAIN; 3245 SPAGAIN;
3106 POPSTACK;
3107 3246
3108 if (current != coro) 3247 if (current != coro)
3109 { 3248 {
3110 PUTBACK; 3249 PUTBACK;
3111 save_perl (aTHX_ coro); 3250 save_perl (aTHX_ coro);
… …
3121 ALIAS: 3260 ALIAS:
3122 is_ready = CF_READY 3261 is_ready = CF_READY
3123 is_running = CF_RUNNING 3262 is_running = CF_RUNNING
3124 is_new = CF_NEW 3263 is_new = CF_NEW
3125 is_destroyed = CF_DESTROYED 3264 is_destroyed = CF_DESTROYED
3265 is_suspended = CF_SUSPENDED
3126 CODE: 3266 CODE:
3127 RETVAL = boolSV (coro->flags & ix); 3267 RETVAL = boolSV (coro->flags & ix);
3128 OUTPUT: 3268 OUTPUT:
3129 RETVAL 3269 RETVAL
3130 3270
… …
3134 CODE: 3274 CODE:
3135{ 3275{
3136 struct coro *current = SvSTATE_current; 3276 struct coro *current = SvSTATE_current;
3137 SV **throwp = self == current ? &CORO_THROW : &self->except; 3277 SV **throwp = self == current ? &CORO_THROW : &self->except;
3138 SvREFCNT_dec (*throwp); 3278 SvREFCNT_dec (*throwp);
3279 SvGETMAGIC (throw);
3139 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 3280 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
3140} 3281}
3141 3282
3142void 3283void
3143api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB) 3284api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
… …
3195 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv; 3336 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv;
3196 3337
3197 SV *tmp = *src; *src = *dst; *dst = tmp; 3338 SV *tmp = *src; *src = *dst; *dst = tmp;
3198 } 3339 }
3199 3340
3341void
3342cancel (Coro::State self)
3343 CODE:
3344 coro_state_destroy (aTHX_ self);
3345 coro_call_on_destroy (aTHX_ self); /* actually only for Coro objects */
3346
3347
3348SV *
3349enable_times (int enabled = enable_times)
3350 CODE:
3351{
3352 RETVAL = boolSV (enable_times);
3353
3354 if (enabled != enable_times)
3355 {
3356 enable_times = enabled;
3357
3358 coro_times_update ();
3359 (enabled ? coro_times_sub : coro_times_add)(SvSTATE (coro_current));
3360 }
3361}
3362 OUTPUT:
3363 RETVAL
3364
3365void
3366times (Coro::State self)
3367 PPCODE:
3368{
3369 struct coro *current = SvSTATE (coro_current);
3370
3371 if (expect_false (current == self))
3372 {
3373 coro_times_update ();
3374 coro_times_add (SvSTATE (coro_current));
3375 }
3376
3377 EXTEND (SP, 2);
3378 PUSHs (sv_2mortal (newSVnv (self->t_real [0] + self->t_real [1] * 1e-9)));
3379 PUSHs (sv_2mortal (newSVnv (self->t_cpu [0] + self->t_cpu [1] * 1e-9)));
3380
3381 if (expect_false (current == self))
3382 coro_times_sub (SvSTATE (coro_current));
3383}
3200 3384
3201MODULE = Coro::State PACKAGE = Coro 3385MODULE = Coro::State PACKAGE = Coro
3202 3386
3203BOOT: 3387BOOT:
3204{ 3388{
3205 int i;
3206
3207 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE); 3389 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
3208 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE); 3390 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
3209 cv_coro_run = get_cv ( "Coro::_coro_run" , GV_ADD); 3391 cv_coro_run = get_cv ( "Coro::_coro_run" , GV_ADD);
3210 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD); 3392 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD);
3211 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current); 3393 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); 3401 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); 3402 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new);
3221 3403
3222 coro_stash = gv_stashpv ("Coro", TRUE); 3404 coro_stash = gv_stashpv ("Coro", TRUE);
3223 3405
3224 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX)); 3406 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (CORO_PRIO_MAX));
3225 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH)); 3407 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (CORO_PRIO_HIGH));
3226 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (PRIO_NORMAL)); 3408 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (CORO_PRIO_NORMAL));
3227 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (PRIO_LOW)); 3409 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (CORO_PRIO_LOW));
3228 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (PRIO_IDLE)); 3410 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (CORO_PRIO_IDLE));
3229 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (PRIO_MIN)); 3411 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 3412
3234 { 3413 {
3235 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE); 3414 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE);
3236 3415
3237 coroapi.schedule = api_schedule; 3416 coroapi.schedule = api_schedule;
… …
3278cede_notself (...) 3457cede_notself (...)
3279 CODE: 3458 CODE:
3280 CORO_EXECUTE_SLF_XS (slf_init_cede_notself); 3459 CORO_EXECUTE_SLF_XS (slf_init_cede_notself);
3281 3460
3282void 3461void
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) 3462_set_current (SV *current)
3290 PROTOTYPE: $ 3463 PROTOTYPE: $
3291 CODE: 3464 CODE:
3292 SvREFCNT_dec (SvRV (coro_current)); 3465 SvREFCNT_dec (SvRV (coro_current));
3293 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current))); 3466 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current)));
… …
3295void 3468void
3296_set_readyhook (SV *hook) 3469_set_readyhook (SV *hook)
3297 PROTOTYPE: $ 3470 PROTOTYPE: $
3298 CODE: 3471 CODE:
3299 SvREFCNT_dec (coro_readyhook); 3472 SvREFCNT_dec (coro_readyhook);
3473 SvGETMAGIC (hook);
3300 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0; 3474 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0;
3301 3475
3302int 3476int
3303prio (Coro::State coro, int newprio = 0) 3477prio (Coro::State coro, int newprio = 0)
3304 PROTOTYPE: $;$ 3478 PROTOTYPE: $;$
… …
3311 if (items > 1) 3485 if (items > 1)
3312 { 3486 {
3313 if (ix) 3487 if (ix)
3314 newprio = coro->prio - newprio; 3488 newprio = coro->prio - newprio;
3315 3489
3316 if (newprio < PRIO_MIN) newprio = PRIO_MIN; 3490 if (newprio < CORO_PRIO_MIN) newprio = CORO_PRIO_MIN;
3317 if (newprio > PRIO_MAX) newprio = PRIO_MAX; 3491 if (newprio > CORO_PRIO_MAX) newprio = CORO_PRIO_MAX;
3318 3492
3319 coro->prio = newprio; 3493 coro->prio = newprio;
3320 } 3494 }
3321} 3495}
3322 OUTPUT: 3496 OUTPUT:
… …
3335 PROTOTYPE: 3509 PROTOTYPE:
3336 CODE: 3510 CODE:
3337 RETVAL = coro_nready; 3511 RETVAL = coro_nready;
3338 OUTPUT: 3512 OUTPUT:
3339 RETVAL 3513 RETVAL
3514
3515void
3516suspend (Coro::State self)
3517 PROTOTYPE: $
3518 CODE:
3519 self->flags |= CF_SUSPENDED;
3520
3521void
3522resume (Coro::State self)
3523 PROTOTYPE: $
3524 CODE:
3525 self->flags &= ~CF_SUSPENDED;
3340 3526
3341void 3527void
3342_pool_handler (...) 3528_pool_handler (...)
3343 CODE: 3529 CODE:
3344 CORO_EXECUTE_SLF_XS (slf_init_pool_handler); 3530 CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
… …
3409 CODE: 3595 CODE:
3410{ 3596{
3411 struct coro *coro = SvSTATE_current; 3597 struct coro *coro = SvSTATE_current;
3412 AV **avp = ix ? &coro->on_leave : &coro->on_enter; 3598 AV **avp = ix ? &coro->on_leave : &coro->on_enter;
3413 3599
3414 block = (SV *)coro_sv_2cv (block); 3600 block = (SV *)coro_sv_2cv (aTHX_ block);
3415 3601
3416 if (!*avp) 3602 if (!*avp)
3417 *avp = newAV (); 3603 *avp = newAV ();
3418 3604
3419 av_push (*avp, SvREFCNT_inc (block)); 3605 av_push (*avp, SvREFCNT_inc (block));
3420 3606
3421 if (!ix) 3607 if (!ix)
3422 on_enterleave_call (aTHX_ block); 3608 on_enterleave_call (aTHX_ block);
3423 3609
3424 LEAVE; /* pp_entersub unfortunately forces an ENTER/LEAVE around xs calls */ 3610 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); 3611 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 */ 3612 ENTER; /* pp_entersub unfortunately forces an ENTER/LEAVE around XS calls */
3427} 3613}
3428 3614
3429 3615
3430MODULE = Coro::State PACKAGE = PerlIO::cede 3616MODULE = Coro::State PACKAGE = PerlIO::cede
3431 3617
… …
3436MODULE = Coro::State PACKAGE = Coro::Semaphore 3622MODULE = Coro::State PACKAGE = Coro::Semaphore
3437 3623
3438SV * 3624SV *
3439new (SV *klass, SV *count = 0) 3625new (SV *klass, SV *count = 0)
3440 CODE: 3626 CODE:
3627{
3628 int semcnt = 1;
3629
3630 if (count)
3631 {
3632 SvGETMAGIC (count);
3633
3634 if (SvOK (count))
3635 semcnt = SvIV (count);
3636 }
3637
3441 RETVAL = sv_bless ( 3638 RETVAL = sv_bless (
3442 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1), 3639 coro_waitarray_new (aTHX_ semcnt),
3443 GvSTASH (CvGV (cv)) 3640 GvSTASH (CvGV (cv))
3444 ); 3641 );
3642}
3445 OUTPUT: 3643 OUTPUT:
3446 RETVAL 3644 RETVAL
3447 3645
3448# helper for Coro::Channel 3646# helper for Coro::Channel and others
3449SV * 3647SV *
3450_alloc (int count) 3648_alloc (int count)
3451 CODE: 3649 CODE:
3452 RETVAL = coro_waitarray_new (aTHX_ count); 3650 RETVAL = coro_waitarray_new (aTHX_ count);
3453 OUTPUT: 3651 OUTPUT:
… …
3511 for (i = 1; i <= wcount; ++i) 3709 for (i = 1; i <= wcount; ++i)
3512 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 3710 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
3513 } 3711 }
3514} 3712}
3515 3713
3714MODULE = Coro::State PACKAGE = Coro::SemaphoreSet
3715
3716void
3717_may_delete (SV *sem, int count, int extra_refs)
3718 PPCODE:
3719{
3720 AV *av = (AV *)SvRV (sem);
3721
3722 if (SvREFCNT ((SV *)av) == 1 + extra_refs
3723 && AvFILLp (av) == 0 /* no waiters, just count */
3724 && SvIV (AvARRAY (av)[0]) == count)
3725 XSRETURN_YES;
3726
3727 XSRETURN_NO;
3728}
3729
3516MODULE = Coro::State PACKAGE = Coro::Signal 3730MODULE = Coro::State PACKAGE = Coro::Signal
3517 3731
3518SV * 3732SV *
3519new (SV *klass) 3733new (SV *klass)
3520 CODE: 3734 CODE:

Diff Legend

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