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.338 by root, Sun Dec 7 15:33:21 2008 UTC vs.
Revision 1.359 by root, Mon Jun 29 05:07:19 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
… …
156#define GCoroAPI (&coroapi) /* very sneaky */ 156#define GCoroAPI (&coroapi) /* very sneaky */
157 157
158#ifdef USE_ITHREADS 158#ifdef USE_ITHREADS
159# if CORO_PTHREAD 159# if CORO_PTHREAD
160static void *coro_thx; 160static void *coro_thx;
161# endif
162#endif
163
164#ifdef __linux
165# include <time.h> /* for timespec */
166# include <syscall.h> /* for SYS_* */
167# ifdef SYS_clock_gettime
168# define coro_clock_gettime(id, ts) syscall (SYS_clock_gettime, (id), (ts))
169# define CORO_CLOCK_MONOTONIC 1
170# define CORO_CLOCK_THREAD_CPUTIME_ID 3
161# endif 171# endif
162#endif 172#endif
163 173
164static double (*nvtime)(); /* so why doesn't it take void? */ 174static double (*nvtime)(); /* so why doesn't it take void? */
165 175
… …
195static CV *cv_coro_state_new; 205static CV *cv_coro_state_new;
196 206
197/* Coro::AnyEvent */ 207/* Coro::AnyEvent */
198static SV *sv_activity; 208static SV *sv_activity;
199 209
210/* enable processtime/realtime profiling */
211static char profile_times;
212typedef U32 coro_ts[2];
213
200static struct coro_cctx *cctx_first; 214static struct coro_cctx *cctx_first;
201static int cctx_count, cctx_idle; 215static int cctx_count, cctx_idle;
202 216
203enum { 217enum {
204 CC_MAPPED = 0x01, 218 CC_MAPPED = 0x01,
… …
238enum { 252enum {
239 CF_RUNNING = 0x0001, /* coroutine is running */ 253 CF_RUNNING = 0x0001, /* coroutine is running */
240 CF_READY = 0x0002, /* coroutine is ready */ 254 CF_READY = 0x0002, /* coroutine is ready */
241 CF_NEW = 0x0004, /* has never been switched to */ 255 CF_NEW = 0x0004, /* has never been switched to */
242 CF_DESTROYED = 0x0008, /* coroutine data has been freed */ 256 CF_DESTROYED = 0x0008, /* coroutine data has been freed */
257 CF_SUSPENDED = 0x0010, /* coroutine can't be scheduled */
243}; 258};
244 259
245/* the structure where most of the perl state is stored, overlaid on the cxstack */ 260/* the structure where most of the perl state is stored, overlaid on the cxstack */
246typedef struct 261typedef struct
247{ 262{
… …
260/* this is a structure representing a perl-level coroutine */ 275/* this is a structure representing a perl-level coroutine */
261struct coro { 276struct coro {
262 /* the C coroutine allocated to this perl coroutine, if any */ 277 /* the C coroutine allocated to this perl coroutine, if any */
263 coro_cctx *cctx; 278 coro_cctx *cctx;
264 279
280 /* ready queue */
281 struct coro *next_ready;
282
265 /* state data */ 283 /* state data */
266 struct CoroSLF slf_frame; /* saved slf frame */ 284 struct CoroSLF slf_frame; /* saved slf frame */
267 AV *mainstack; 285 AV *mainstack;
268 perl_slots *slot; /* basically the saved sp */ 286 perl_slots *slot; /* basically the saved sp */
269 287
… …
285 /* async_pool */ 303 /* async_pool */
286 SV *saved_deffh; 304 SV *saved_deffh;
287 SV *invoke_cb; 305 SV *invoke_cb;
288 AV *invoke_av; 306 AV *invoke_av;
289 307
308 /* on_enter/on_leave */
309 AV *on_enter;
310 AV *on_leave;
311
290 /* linked list */ 312 /* linked list */
291 struct coro *next, *prev; 313 struct coro *next, *prev;
292}; 314};
293 315
294typedef struct coro *Coro__State; 316typedef struct coro *Coro__State;
… …
299/* the mainr easonw e don't support windows process emulation */ 321/* the mainr easonw e don't support windows process emulation */
300static struct CoroSLF slf_frame; /* the current slf frame */ 322static struct CoroSLF slf_frame; /* the current slf frame */
301 323
302/** Coro ********************************************************************/ 324/** Coro ********************************************************************/
303 325
304#define PRIO_MAX 3 326#define CORO_PRIO_MAX 3
305#define PRIO_HIGH 1 327#define CORO_PRIO_HIGH 1
306#define PRIO_NORMAL 0 328#define CORO_PRIO_NORMAL 0
307#define PRIO_LOW -1 329#define CORO_PRIO_LOW -1
308#define PRIO_IDLE -3 330#define CORO_PRIO_IDLE -3
309#define PRIO_MIN -4 331#define CORO_PRIO_MIN -4
310 332
311/* for Coro.pm */ 333/* for Coro.pm */
312static SV *coro_current; 334static SV *coro_current;
313static SV *coro_readyhook; 335static SV *coro_readyhook;
314static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1]; 336static struct coro *coro_ready [CORO_PRIO_MAX - CORO_PRIO_MIN + 1][2]; /* head|tail */
315static CV *cv_coro_run, *cv_coro_terminate; 337static CV *cv_coro_run, *cv_coro_terminate;
316static struct coro *coro_first; 338static struct coro *coro_first;
317#define coro_nready coroapi.nready 339#define coro_nready coroapi.nready
318 340
319/** lowlevel stuff **********************************************************/ 341/** lowlevel stuff **********************************************************/
… …
352INLINE CV * 374INLINE CV *
353coro_sv_2cv (pTHX_ SV *sv) 375coro_sv_2cv (pTHX_ SV *sv)
354{ 376{
355 HV *st; 377 HV *st;
356 GV *gvp; 378 GV *gvp;
357 return sv_2cv (sv, &st, &gvp, 0); 379 CV *cv = sv_2cv (sv, &st, &gvp, 0);
380
381 if (!cv)
382 croak ("code reference expected");
383
384 return cv;
358} 385}
359 386
360/*****************************************************************************/ 387/*****************************************************************************/
361/* magic glue */ 388/* magic glue */
362 389
… …
462static int 489static int
463coro_cv_free (pTHX_ SV *sv, MAGIC *mg) 490coro_cv_free (pTHX_ SV *sv, MAGIC *mg)
464{ 491{
465 AV *padlist; 492 AV *padlist;
466 AV *av = (AV *)mg->mg_obj; 493 AV *av = (AV *)mg->mg_obj;
494
495 /* perl manages to free our internal AV and _then_ call us */
496 if (IN_DESTRUCT)
497 return;
467 498
468 /* casting is fun. */ 499 /* casting is fun. */
469 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av))) 500 while (&PL_sv_undef != (SV *)(padlist = (AV *)av_pop (av)))
470 free_padlist (aTHX_ padlist); 501 free_padlist (aTHX_ padlist);
471 502
… …
521} 552}
522 553
523/** load & save, init *******************************************************/ 554/** load & save, init *******************************************************/
524 555
525static void 556static void
557on_enterleave_call (pTHX_ SV *cb);
558
559static void
526load_perl (pTHX_ Coro__State c) 560load_perl (pTHX_ Coro__State c)
527{ 561{
528 perl_slots *slot = c->slot; 562 perl_slots *slot = c->slot;
529 c->slot = 0; 563 c->slot = 0;
530 564
… …
556 PUTBACK; 590 PUTBACK;
557 } 591 }
558 592
559 slf_frame = c->slf_frame; 593 slf_frame = c->slf_frame;
560 CORO_THROW = c->except; 594 CORO_THROW = c->except;
595
596 if (expect_false (c->on_enter))
597 {
598 int i;
599
600 for (i = 0; i <= AvFILLp (c->on_enter); ++i)
601 on_enterleave_call (aTHX_ AvARRAY (c->on_enter)[i]);
602 }
561} 603}
562 604
563static void 605static void
564save_perl (pTHX_ Coro__State c) 606save_perl (pTHX_ Coro__State c)
565{ 607{
608 if (expect_false (c->on_leave))
609 {
610 int i;
611
612 for (i = AvFILLp (c->on_leave); i >= 0; --i)
613 on_enterleave_call (aTHX_ AvARRAY (c->on_leave)[i]);
614 }
615
566 c->except = CORO_THROW; 616 c->except = CORO_THROW;
567 c->slf_frame = slf_frame; 617 c->slf_frame = slf_frame;
568 618
569 { 619 {
570 dSP; 620 dSP;
… …
583 { 633 {
584 while (expect_true (cxix >= 0)) 634 while (expect_true (cxix >= 0))
585 { 635 {
586 PERL_CONTEXT *cx = &ccstk[cxix--]; 636 PERL_CONTEXT *cx = &ccstk[cxix--];
587 637
588 if (expect_true (CxTYPE (cx) == CXt_SUB || CxTYPE (cx) == CXt_FORMAT)) 638 if (expect_true (CxTYPE (cx) == CXt_SUB) || expect_false (CxTYPE (cx) == CXt_FORMAT))
589 { 639 {
590 CV *cv = cx->blk_sub.cv; 640 CV *cv = cx->blk_sub.cv;
591 641
592 if (expect_true (CvDEPTH (cv))) 642 if (expect_true (CvDEPTH (cv)))
593 { 643 {
… …
958} 1008}
959 1009
960static void 1010static void
961coro_destruct_perl (pTHX_ struct coro *coro) 1011coro_destruct_perl (pTHX_ struct coro *coro)
962{ 1012{
1013 SV *svf [9];
1014
1015 {
1016 struct coro *current = SvSTATE_current;
1017
1018 assert (("FATAL: tried to destroy currently running coroutine", coro->mainstack != PL_mainstack));
1019
1020 save_perl (aTHX_ current);
1021 load_perl (aTHX_ coro);
1022
963 coro_unwind_stacks (aTHX); 1023 coro_unwind_stacks (aTHX);
1024 coro_destruct_stacks (aTHX);
964 1025
965 SvREFCNT_dec (GvSV (PL_defgv)); 1026 // now save some sv's to be free'd later
966 SvREFCNT_dec (GvAV (PL_defgv)); 1027 svf [0] = GvSV (PL_defgv);
967 SvREFCNT_dec (GvSV (PL_errgv)); 1028 svf [1] = (SV *)GvAV (PL_defgv);
968 SvREFCNT_dec (PL_defoutgv); 1029 svf [2] = GvSV (PL_errgv);
969 SvREFCNT_dec (PL_rs); 1030 svf [3] = (SV *)PL_defoutgv;
970 SvREFCNT_dec (GvSV (irsgv)); 1031 svf [4] = PL_rs;
971 SvREFCNT_dec (GvHV (PL_hintgv)); 1032 svf [5] = GvSV (irsgv);
1033 svf [6] = (SV *)GvHV (PL_hintgv);
1034 svf [7] = PL_diehook;
1035 svf [8] = PL_warnhook;
1036 assert (9 == sizeof (svf) / sizeof (*svf));
972 1037
973 SvREFCNT_dec (PL_diehook); 1038 load_perl (aTHX_ current);
974 SvREFCNT_dec (PL_warnhook);
975 1039 }
1040
1041 {
1042 int i;
1043
1044 for (i = 0; i < sizeof (svf) / sizeof (*svf); ++i)
1045 SvREFCNT_dec (svf [i]);
1046
976 SvREFCNT_dec (coro->saved_deffh); 1047 SvREFCNT_dec (coro->saved_deffh);
977 SvREFCNT_dec (coro->rouse_cb); 1048 SvREFCNT_dec (coro->rouse_cb);
978 SvREFCNT_dec (coro->invoke_cb); 1049 SvREFCNT_dec (coro->invoke_cb);
979 SvREFCNT_dec (coro->invoke_av); 1050 SvREFCNT_dec (coro->invoke_av);
980 1051 }
981 coro_destruct_stacks (aTHX);
982} 1052}
983 1053
984INLINE void 1054INLINE void
985free_coro_mortal (pTHX) 1055free_coro_mortal (pTHX)
986{ 1056{
… …
1354 /* TODO: throwing up here is considered harmful */ 1424 /* TODO: throwing up here is considered harmful */
1355 1425
1356 if (expect_true (prev != next)) 1426 if (expect_true (prev != next))
1357 { 1427 {
1358 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW)))) 1428 if (expect_false (!(prev->flags & (CF_RUNNING | CF_NEW))))
1359 croak ("Coro::State::transfer called with a suspended prev Coro::State, but can only transfer from running or new states,"); 1429 croak ("Coro::State::transfer called with a blocked prev Coro::State, but can only transfer from running or new states,");
1360 1430
1361 if (expect_false (next->flags & CF_RUNNING))
1362 croak ("Coro::State::transfer called with running next Coro::State, but can only transfer to inactive states,");
1363
1364 if (expect_false (next->flags & CF_DESTROYED)) 1431 if (expect_false (next->flags & (CF_RUNNING | CF_DESTROYED | CF_SUSPENDED)))
1365 croak ("Coro::State::transfer called with destroyed next Coro::State, but can only transfer to inactive states,"); 1432 croak ("Coro::State::transfer called with running, destroyed or suspended next Coro::State, but can only transfer to inactive states,");
1366 1433
1367#if !PERL_VERSION_ATLEAST (5,10,0) 1434#if !PERL_VERSION_ATLEAST (5,10,0)
1368 if (expect_false (PL_lex_state != LEX_NOTPARSING)) 1435 if (expect_false (PL_lex_state != LEX_NOTPARSING))
1369 croak ("Coro::State::transfer called while parsing, but this is not supported in your perl version,"); 1436 croak ("Coro::State::transfer called while parsing, but this is not supported in your perl version,");
1370#endif 1437#endif
… …
1458coro_state_destroy (pTHX_ struct coro *coro) 1525coro_state_destroy (pTHX_ struct coro *coro)
1459{ 1526{
1460 if (coro->flags & CF_DESTROYED) 1527 if (coro->flags & CF_DESTROYED)
1461 return 0; 1528 return 0;
1462 1529
1463 if (coro->on_destroy) 1530 if (coro->on_destroy && !PL_dirty)
1464 coro->on_destroy (aTHX_ coro); 1531 coro->on_destroy (aTHX_ coro);
1465 1532
1466 coro->flags |= CF_DESTROYED; 1533 coro->flags |= CF_DESTROYED;
1467 1534
1468 if (coro->flags & CF_READY) 1535 if (coro->flags & CF_READY)
… …
1476 1543
1477 if (coro->mainstack 1544 if (coro->mainstack
1478 && coro->mainstack != main_mainstack 1545 && coro->mainstack != main_mainstack
1479 && coro->slot 1546 && coro->slot
1480 && !PL_dirty) 1547 && !PL_dirty)
1481 {
1482 struct coro temp;
1483
1484 assert (("FATAL: tried to destroy currently running coroutine", coro->mainstack != PL_mainstack));
1485
1486 save_perl (aTHX_ &temp);
1487 load_perl (aTHX_ coro);
1488
1489 coro_destruct_perl (aTHX_ coro); 1548 coro_destruct_perl (aTHX_ coro);
1490
1491 load_perl (aTHX_ &temp);
1492
1493 coro->slot = 0;
1494 }
1495 1549
1496 cctx_destroy (coro->cctx); 1550 cctx_destroy (coro->cctx);
1497 SvREFCNT_dec (coro->startcv); 1551 SvREFCNT_dec (coro->startcv);
1498 SvREFCNT_dec (coro->args); 1552 SvREFCNT_dec (coro->args);
1499 SvREFCNT_dec (CORO_THROW); 1553 SvREFCNT_dec (CORO_THROW);
… …
1586/** Coro ********************************************************************/ 1640/** Coro ********************************************************************/
1587 1641
1588INLINE void 1642INLINE void
1589coro_enq (pTHX_ struct coro *coro) 1643coro_enq (pTHX_ struct coro *coro)
1590{ 1644{
1591 av_push (coro_ready [coro->prio - PRIO_MIN], SvREFCNT_inc_NN (coro->hv)); 1645 struct coro **ready = coro_ready [coro->prio - CORO_PRIO_MIN];
1592}
1593 1646
1594INLINE SV * 1647 SvREFCNT_inc_NN (coro->hv);
1648
1649 coro->next_ready = 0;
1650 *(ready [0] ? &ready [1]->next_ready : &ready [0]) = coro;
1651 ready [1] = coro;
1652}
1653
1654INLINE struct coro *
1595coro_deq (pTHX) 1655coro_deq (pTHX)
1596{ 1656{
1597 int prio; 1657 int prio;
1598 1658
1599 for (prio = PRIO_MAX - PRIO_MIN + 1; --prio >= 0; ) 1659 for (prio = CORO_PRIO_MAX - CORO_PRIO_MIN + 1; --prio >= 0; )
1600 if (AvFILLp (coro_ready [prio]) >= 0) 1660 {
1601 return av_shift (coro_ready [prio]); 1661 struct coro **ready = coro_ready [prio];
1662
1663 if (ready [0])
1664 {
1665 struct coro *coro = ready [0];
1666 ready [0] = coro->next_ready;
1667 return coro;
1668 }
1669 }
1602 1670
1603 return 0; 1671 return 0;
1604} 1672}
1605 1673
1606static int 1674static int
… …
1670static void 1738static void
1671prepare_schedule (pTHX_ struct coro_transfer_args *ta) 1739prepare_schedule (pTHX_ struct coro_transfer_args *ta)
1672{ 1740{
1673 for (;;) 1741 for (;;)
1674 { 1742 {
1675 SV *next_sv = coro_deq (aTHX); 1743 struct coro *next = coro_deq (aTHX);
1676 1744
1677 if (expect_true (next_sv)) 1745 if (expect_true (next))
1678 { 1746 {
1679 struct coro *next = SvSTATE_hv (next_sv);
1680
1681 /* cannot transfer to destroyed coros, skip and look for next */ 1747 /* cannot transfer to destroyed coros, skip and look for next */
1682 if (expect_false (next->flags & CF_DESTROYED)) 1748 if (expect_false (next->flags & (CF_DESTROYED | CF_SUSPENDED)))
1683 SvREFCNT_dec (next_sv); /* coro_nready has already been taken care of by destroy */ 1749 SvREFCNT_dec (next->hv); /* coro_nready has already been taken care of by destroy */
1684 else 1750 else
1685 { 1751 {
1686 next->flags &= ~CF_READY; 1752 next->flags &= ~CF_READY;
1687 --coro_nready; 1753 --coro_nready;
1688 1754
… …
1962 SV *data = (SV *)GENSUB_ARG; 2028 SV *data = (SV *)GENSUB_ARG;
1963 2029
1964 if (SvTYPE (SvRV (data)) != SVt_PVAV) 2030 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1965 { 2031 {
1966 /* first call, set args */ 2032 /* first call, set args */
2033 SV *coro = SvRV (data);
1967 AV *av = newAV (); 2034 AV *av = newAV ();
1968 SV *coro = SvRV (data);
1969 2035
1970 SvRV_set (data, (SV *)av); 2036 SvRV_set (data, (SV *)av);
1971 api_ready (aTHX_ coro);
1972 SvREFCNT_dec (coro);
1973 2037
1974 /* better take a full copy of the arguments */ 2038 /* better take a full copy of the arguments */
1975 while (items--) 2039 while (items--)
1976 av_store (av, items, newSVsv (ST (items))); 2040 av_store (av, items, newSVsv (ST (items)));
2041
2042 api_ready (aTHX_ coro);
2043 SvREFCNT_dec (coro);
1977 } 2044 }
1978 2045
1979 XSRETURN_EMPTY; 2046 XSRETURN_EMPTY;
1980} 2047}
1981 2048
… …
1998 2065
1999 EXTEND (SP, AvFILLp (av) + 1); 2066 EXTEND (SP, AvFILLp (av) + 1);
2000 for (i = 0; i <= AvFILLp (av); ++i) 2067 for (i = 0; i <= AvFILLp (av); ++i)
2001 PUSHs (sv_2mortal (AvARRAY (av)[i])); 2068 PUSHs (sv_2mortal (AvARRAY (av)[i]));
2002 2069
2003 /* we have stolen the elements, so ste length to zero and free */ 2070 /* we have stolen the elements, so set length to zero and free */
2004 AvFILLp (av) = -1; 2071 AvFILLp (av) = -1;
2005 av_undef (av); 2072 av_undef (av);
2006 2073
2007 PUTBACK; 2074 PUTBACK;
2008 } 2075 }
… …
2298 2365
2299 PL_op->op_ppaddr = pp_slf; 2366 PL_op->op_ppaddr = pp_slf;
2300 /*PL_op->op_type = OP_CUSTOM; /* we do behave like entersub still */ 2367 /*PL_op->op_type = OP_CUSTOM; /* we do behave like entersub still */
2301 2368
2302 PL_op = (OP *)&slf_restore; 2369 PL_op = (OP *)&slf_restore;
2370}
2371
2372/*****************************************************************************/
2373/* dynamic wind */
2374
2375static void
2376on_enterleave_call (pTHX_ SV *cb)
2377{
2378 dSP;
2379
2380 PUSHSTACK;
2381
2382 PUSHMARK (SP);
2383 PUTBACK;
2384 call_sv (cb, G_VOID | G_DISCARD);
2385 SPAGAIN;
2386
2387 POPSTACK;
2388}
2389
2390static SV *
2391coro_avp_pop_and_free (pTHX_ AV **avp)
2392{
2393 AV *av = *avp;
2394 SV *res = av_pop (av);
2395
2396 if (AvFILLp (av) < 0)
2397 {
2398 *avp = 0;
2399 SvREFCNT_dec (av);
2400 }
2401
2402 return res;
2403}
2404
2405static void
2406coro_pop_on_enter (pTHX_ void *coro)
2407{
2408 SV *cb = coro_avp_pop_and_free (aTHX_ &((struct coro *)coro)->on_enter);
2409 SvREFCNT_dec (cb);
2410}
2411
2412static void
2413coro_pop_on_leave (pTHX_ void *coro)
2414{
2415 SV *cb = coro_avp_pop_and_free (aTHX_ &((struct coro *)coro)->on_leave);
2416 on_enterleave_call (aTHX_ sv_2mortal (cb));
2303} 2417}
2304 2418
2305/*****************************************************************************/ 2419/*****************************************************************************/
2306/* PerlIO::cede */ 2420/* PerlIO::cede */
2307 2421
… …
2381/* Coro::Semaphore & Coro::Signal */ 2495/* Coro::Semaphore & Coro::Signal */
2382 2496
2383static SV * 2497static SV *
2384coro_waitarray_new (pTHX_ int count) 2498coro_waitarray_new (pTHX_ int count)
2385{ 2499{
2386 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2500 /* a waitarray=semaphore contains a counter IV in $sem->[0] and any waiters after that */
2387 AV *av = newAV (); 2501 AV *av = newAV ();
2388 SV **ary; 2502 SV **ary;
2389 2503
2390 /* unfortunately, building manually saves memory */ 2504 /* unfortunately, building manually saves memory */
2391 Newx (ary, 2, SV *); 2505 Newx (ary, 2, SV *);
… …
2535 { 2649 {
2536 /* callback form */ 2650 /* callback form */
2537 AV *av = (AV *)SvRV (arg [0]); 2651 AV *av = (AV *)SvRV (arg [0]);
2538 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]); 2652 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2539 2653
2540 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv)); 2654 av_push (av, SvREFCNT_inc_NN (cb_cv));
2541 2655
2542 if (SvIVX (AvARRAY (av)[0]) > 0) 2656 if (SvIVX (AvARRAY (av)[0]) > 0)
2543 coro_semaphore_adjust (aTHX_ av, 0); 2657 coro_semaphore_adjust (aTHX_ av, 0);
2544 2658
2545 frame->prepare = prepare_nop; 2659 frame->prepare = prepare_nop;
… …
2569 AvARRAY (av)[0] = AvARRAY (av)[1]; 2683 AvARRAY (av)[0] = AvARRAY (av)[1];
2570 AvARRAY (av)[1] = cb; 2684 AvARRAY (av)[1] = cb;
2571 2685
2572 cb = av_shift (av); 2686 cb = av_shift (av);
2573 2687
2688 if (SvTYPE (cb) == SVt_PVCV)
2689 {
2690 dSP;
2691 PUSHMARK (SP);
2692 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2693 PUTBACK;
2694 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2695 }
2696 else
2697 {
2574 api_ready (aTHX_ cb); 2698 api_ready (aTHX_ cb);
2575 sv_setiv (cb, 0); /* signal waiter */ 2699 sv_setiv (cb, 0); /* signal waiter */
2700 }
2701
2576 SvREFCNT_dec (cb); 2702 SvREFCNT_dec (cb);
2577 2703
2578 --count; 2704 --count;
2579 } 2705 }
2580} 2706}
… …
2589static void 2715static void
2590slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2716slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2591{ 2717{
2592 AV *av = (AV *)SvRV (arg [0]); 2718 AV *av = (AV *)SvRV (arg [0]);
2593 2719
2720 if (items >= 2)
2721 {
2722 CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
2723 av_push (av, SvREFCNT_inc_NN (cb_cv));
2724
2594 if (SvIVX (AvARRAY (av)[0])) 2725 if (SvIVX (AvARRAY (av)[0]))
2726 coro_signal_wake (aTHX_ av, 1); /* ust be the only waiter */
2727
2728 frame->prepare = prepare_nop;
2729 frame->check = slf_check_nop;
2730 }
2731 else if (SvIVX (AvARRAY (av)[0]))
2595 { 2732 {
2596 SvIVX (AvARRAY (av)[0]) = 0; 2733 SvIVX (AvARRAY (av)[0]) = 0;
2597 frame->prepare = prepare_nop; 2734 frame->prepare = prepare_nop;
2598 frame->check = slf_check_nop; 2735 frame->check = slf_check_nop;
2599 } 2736 }
2600 else 2737 else
2601 { 2738 {
2602 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */ 2739 SV *waiter = newSVsv (coro_current); /* owned by signal av */
2603 2740
2604 av_push (av, waiter); 2741 av_push (av, waiter);
2605 2742
2606 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */ 2743 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2607 frame->prepare = prepare_schedule; 2744 frame->prepare = prepare_schedule;
… …
2848 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer"); 2985 if (!SvIOK (*svp)) croak ("Time::NVtime isn't a function pointer");
2849 2986
2850 nvtime = INT2PTR (double (*)(), SvIV (*svp)); 2987 nvtime = INT2PTR (double (*)(), SvIV (*svp));
2851 } 2988 }
2852 2989
2853 assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL)); 2990 assert (("PRIO_NORMAL must be 0", !CORO_PRIO_NORMAL));
2854} 2991}
2855 2992
2856SV * 2993SV *
2857new (char *klass, ...) 2994new (char *klass, ...)
2858 ALIAS: 2995 ALIAS:
… …
3005 eval = 1 3142 eval = 1
3006 CODE: 3143 CODE:
3007{ 3144{
3008 if (coro->mainstack && ((coro->flags & CF_RUNNING) || coro->slot)) 3145 if (coro->mainstack && ((coro->flags & CF_RUNNING) || coro->slot))
3009 { 3146 {
3010 struct coro temp; 3147 struct coro *current = SvSTATE_current;
3011 3148
3012 if (!(coro->flags & CF_RUNNING)) 3149 if (current != coro)
3013 { 3150 {
3014 PUTBACK; 3151 PUTBACK;
3015 save_perl (aTHX_ &temp); 3152 save_perl (aTHX_ current);
3016 load_perl (aTHX_ coro); 3153 load_perl (aTHX_ coro);
3154 SPAGAIN;
3017 } 3155 }
3018 3156
3019 {
3020 dSP;
3021 ENTER;
3022 SAVETMPS;
3023 PUTBACK;
3024 PUSHSTACK; 3157 PUSHSTACK;
3158
3025 PUSHMARK (SP); 3159 PUSHMARK (SP);
3160 PUTBACK;
3026 3161
3027 if (ix) 3162 if (ix)
3028 eval_sv (coderef, 0); 3163 eval_sv (coderef, 0);
3029 else 3164 else
3030 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD); 3165 call_sv (coderef, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
3031 3166
3032 POPSTACK; 3167 POPSTACK;
3033 SPAGAIN; 3168 SPAGAIN;
3034 FREETMPS;
3035 LEAVE;
3036 PUTBACK;
3037 }
3038 3169
3039 if (!(coro->flags & CF_RUNNING)) 3170 if (current != coro)
3040 { 3171 {
3172 PUTBACK;
3041 save_perl (aTHX_ coro); 3173 save_perl (aTHX_ coro);
3042 load_perl (aTHX_ &temp); 3174 load_perl (aTHX_ current);
3043 SPAGAIN; 3175 SPAGAIN;
3044 } 3176 }
3045 } 3177 }
3046} 3178}
3047 3179
… …
3051 ALIAS: 3183 ALIAS:
3052 is_ready = CF_READY 3184 is_ready = CF_READY
3053 is_running = CF_RUNNING 3185 is_running = CF_RUNNING
3054 is_new = CF_NEW 3186 is_new = CF_NEW
3055 is_destroyed = CF_DESTROYED 3187 is_destroyed = CF_DESTROYED
3188 is_suspended = CF_SUSPENDED
3056 CODE: 3189 CODE:
3057 RETVAL = boolSV (coro->flags & ix); 3190 RETVAL = boolSV (coro->flags & ix);
3058 OUTPUT: 3191 OUTPUT:
3059 RETVAL 3192 RETVAL
3060 3193
… …
3064 CODE: 3197 CODE:
3065{ 3198{
3066 struct coro *current = SvSTATE_current; 3199 struct coro *current = SvSTATE_current;
3067 SV **throwp = self == current ? &CORO_THROW : &self->except; 3200 SV **throwp = self == current ? &CORO_THROW : &self->except;
3068 SvREFCNT_dec (*throwp); 3201 SvREFCNT_dec (*throwp);
3202 SvGETMAGIC (throw);
3069 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 3203 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
3070} 3204}
3071 3205
3072void 3206void
3073api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB) 3207api_trace (SV *coro, int flags = CC_TRACE | CC_TRACE_SUB)
… …
3125 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv; 3259 SV **dst = ix ? (SV **)&self->slot->defav : (SV **)&self->slot->defsv;
3126 3260
3127 SV *tmp = *src; *src = *dst; *dst = tmp; 3261 SV *tmp = *src; *src = *dst; *dst = tmp;
3128 } 3262 }
3129 3263
3264void
3265cancel (Coro::State self)
3266 CODE:
3267 coro_state_destroy (aTHX_ self);
3268 coro_call_on_destroy (aTHX_ self); /* actually only for Coro objects */
3269
3130 3270
3131MODULE = Coro::State PACKAGE = Coro 3271MODULE = Coro::State PACKAGE = Coro
3132 3272
3133BOOT: 3273BOOT:
3134{ 3274{
3135 int i;
3136
3137 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE); 3275 sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
3138 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE); 3276 sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
3139 cv_coro_run = get_cv ( "Coro::_coro_run" , GV_ADD); 3277 cv_coro_run = get_cv ( "Coro::_coro_run" , GV_ADD);
3140 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD); 3278 cv_coro_terminate = get_cv ( "Coro::terminate" , GV_ADD);
3141 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current); 3279 coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE); SvREADONLY_on (coro_current);
… …
3149 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler); 3287 cv_pool_handler = get_cv ("Coro::pool_handler", GV_ADD); SvREADONLY_on (cv_pool_handler);
3150 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new); 3288 cv_coro_state_new = get_cv ("Coro::State::new", 0); SvREADONLY_on (cv_coro_state_new);
3151 3289
3152 coro_stash = gv_stashpv ("Coro", TRUE); 3290 coro_stash = gv_stashpv ("Coro", TRUE);
3153 3291
3154 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX)); 3292 newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (CORO_PRIO_MAX));
3155 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH)); 3293 newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (CORO_PRIO_HIGH));
3156 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (PRIO_NORMAL)); 3294 newCONSTSUB (coro_stash, "PRIO_NORMAL", newSViv (CORO_PRIO_NORMAL));
3157 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (PRIO_LOW)); 3295 newCONSTSUB (coro_stash, "PRIO_LOW", newSViv (CORO_PRIO_LOW));
3158 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (PRIO_IDLE)); 3296 newCONSTSUB (coro_stash, "PRIO_IDLE", newSViv (CORO_PRIO_IDLE));
3159 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (PRIO_MIN)); 3297 newCONSTSUB (coro_stash, "PRIO_MIN", newSViv (CORO_PRIO_MIN));
3160
3161 for (i = PRIO_MAX - PRIO_MIN + 1; i--; )
3162 coro_ready[i] = newAV ();
3163 3298
3164 { 3299 {
3165 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE); 3300 SV *sv = coro_get_sv (aTHX_ "Coro::API", TRUE);
3166 3301
3167 coroapi.schedule = api_schedule; 3302 coroapi.schedule = api_schedule;
… …
3208cede_notself (...) 3343cede_notself (...)
3209 CODE: 3344 CODE:
3210 CORO_EXECUTE_SLF_XS (slf_init_cede_notself); 3345 CORO_EXECUTE_SLF_XS (slf_init_cede_notself);
3211 3346
3212void 3347void
3213_cancel (Coro::State self)
3214 CODE:
3215 coro_state_destroy (aTHX_ self);
3216 coro_call_on_destroy (aTHX_ self);
3217
3218void
3219_set_current (SV *current) 3348_set_current (SV *current)
3220 PROTOTYPE: $ 3349 PROTOTYPE: $
3221 CODE: 3350 CODE:
3222 SvREFCNT_dec (SvRV (coro_current)); 3351 SvREFCNT_dec (SvRV (coro_current));
3223 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current))); 3352 SvRV_set (coro_current, SvREFCNT_inc_NN (SvRV (current)));
… …
3225void 3354void
3226_set_readyhook (SV *hook) 3355_set_readyhook (SV *hook)
3227 PROTOTYPE: $ 3356 PROTOTYPE: $
3228 CODE: 3357 CODE:
3229 SvREFCNT_dec (coro_readyhook); 3358 SvREFCNT_dec (coro_readyhook);
3359 SvGETMAGIC (hook);
3230 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0; 3360 coro_readyhook = SvOK (hook) ? newSVsv (hook) : 0;
3231 3361
3232int 3362int
3233prio (Coro::State coro, int newprio = 0) 3363prio (Coro::State coro, int newprio = 0)
3234 PROTOTYPE: $;$ 3364 PROTOTYPE: $;$
… …
3241 if (items > 1) 3371 if (items > 1)
3242 { 3372 {
3243 if (ix) 3373 if (ix)
3244 newprio = coro->prio - newprio; 3374 newprio = coro->prio - newprio;
3245 3375
3246 if (newprio < PRIO_MIN) newprio = PRIO_MIN; 3376 if (newprio < CORO_PRIO_MIN) newprio = CORO_PRIO_MIN;
3247 if (newprio > PRIO_MAX) newprio = PRIO_MAX; 3377 if (newprio > CORO_PRIO_MAX) newprio = CORO_PRIO_MAX;
3248 3378
3249 coro->prio = newprio; 3379 coro->prio = newprio;
3250 } 3380 }
3251} 3381}
3252 OUTPUT: 3382 OUTPUT:
… …
3265 PROTOTYPE: 3395 PROTOTYPE:
3266 CODE: 3396 CODE:
3267 RETVAL = coro_nready; 3397 RETVAL = coro_nready;
3268 OUTPUT: 3398 OUTPUT:
3269 RETVAL 3399 RETVAL
3400
3401void
3402suspend (Coro::State self)
3403 PROTOTYPE: $
3404 CODE:
3405 self->flags |= CF_SUSPENDED;
3406
3407void
3408resume (Coro::State self)
3409 PROTOTYPE: $
3410 CODE:
3411 self->flags &= ~CF_SUSPENDED;
3270 3412
3271void 3413void
3272_pool_handler (...) 3414_pool_handler (...)
3273 CODE: 3415 CODE:
3274 CORO_EXECUTE_SLF_XS (slf_init_pool_handler); 3416 CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
… …
3329rouse_wait (...) 3471rouse_wait (...)
3330 PROTOTYPE: ;$ 3472 PROTOTYPE: ;$
3331 PPCODE: 3473 PPCODE:
3332 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait); 3474 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
3333 3475
3476void
3477on_enter (SV *block)
3478 ALIAS:
3479 on_leave = 1
3480 PROTOTYPE: &
3481 CODE:
3482{
3483 struct coro *coro = SvSTATE_current;
3484 AV **avp = ix ? &coro->on_leave : &coro->on_enter;
3485
3486 block = (SV *)coro_sv_2cv (aTHX_ block);
3487
3488 if (!*avp)
3489 *avp = newAV ();
3490
3491 av_push (*avp, SvREFCNT_inc (block));
3492
3493 if (!ix)
3494 on_enterleave_call (aTHX_ block);
3495
3496 LEAVE; /* pp_entersub unfortunately forces an ENTER/LEAVE around XS calls */
3497 SAVEDESTRUCTOR_X (ix ? coro_pop_on_leave : coro_pop_on_enter, (void *)coro);
3498 ENTER; /* pp_entersub unfortunately forces an ENTER/LEAVE around XS calls */
3499}
3500
3334 3501
3335MODULE = Coro::State PACKAGE = PerlIO::cede 3502MODULE = Coro::State PACKAGE = PerlIO::cede
3336 3503
3337BOOT: 3504BOOT:
3338 PerlIO_define_layer (aTHX_ &PerlIO_cede); 3505 PerlIO_define_layer (aTHX_ &PerlIO_cede);
… …
3341MODULE = Coro::State PACKAGE = Coro::Semaphore 3508MODULE = Coro::State PACKAGE = Coro::Semaphore
3342 3509
3343SV * 3510SV *
3344new (SV *klass, SV *count = 0) 3511new (SV *klass, SV *count = 0)
3345 CODE: 3512 CODE:
3513{
3514 int semcnt = 1;
3515
3516 if (count)
3517 {
3518 SvGETMAGIC (count);
3519
3520 if (SvOK (count))
3521 semcnt = SvIV (count);
3522 }
3523
3346 RETVAL = sv_bless ( 3524 RETVAL = sv_bless (
3347 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1), 3525 coro_waitarray_new (aTHX_ semcnt),
3348 GvSTASH (CvGV (cv)) 3526 GvSTASH (CvGV (cv))
3349 ); 3527 );
3528}
3350 OUTPUT: 3529 OUTPUT:
3351 RETVAL 3530 RETVAL
3352 3531
3353# helper for Coro::Channel 3532# helper for Coro::Channel and others
3354SV * 3533SV *
3355_alloc (int count) 3534_alloc (int count)
3356 CODE: 3535 CODE:
3357 RETVAL = coro_waitarray_new (aTHX_ count); 3536 RETVAL = coro_waitarray_new (aTHX_ count);
3358 OUTPUT: 3537 OUTPUT:
… …
3416 for (i = 1; i <= wcount; ++i) 3595 for (i = 1; i <= wcount; ++i)
3417 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 3596 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
3418 } 3597 }
3419} 3598}
3420 3599
3600MODULE = Coro::State PACKAGE = Coro::SemaphoreSet
3601
3602void
3603_may_delete (SV *sem, int count, int extra_refs)
3604 PPCODE:
3605{
3606 AV *av = (AV *)SvRV (sem);
3607
3608 if (SvREFCNT ((SV *)av) == 1 + extra_refs
3609 && AvFILLp (av) == 0 /* no waiters, just count */
3610 && SvIV (AvARRAY (av)[0]) == count)
3611 XSRETURN_YES;
3612
3613 XSRETURN_NO;
3614}
3615
3421MODULE = Coro::State PACKAGE = Coro::Signal 3616MODULE = Coro::State PACKAGE = Coro::Signal
3422 3617
3423SV * 3618SV *
3424new (SV *klass) 3619new (SV *klass)
3425 CODE: 3620 CODE:

Diff Legend

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