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.301 by root, Wed Nov 19 02:41:31 2008 UTC vs.
Revision 1.307 by root, Wed Nov 19 14:34:46 2008 UTC

… …
16 16
17#ifdef WIN32 17#ifdef WIN32
18# undef setjmp 18# undef setjmp
19# undef longjmp 19# undef longjmp
20# undef _exit 20# undef _exit
21# define setjmp _setjmp // deep magic, don't ask 21# define setjmp _setjmp /* deep magic */
22#else 22#else
23# include <inttypes.h> /* most portable stdint.h */ 23# include <inttypes.h> /* most portable stdint.h */
24#endif 24#endif
25 25
26#ifdef HAVE_MMAP 26#ifdef HAVE_MMAP
… …
246 /* state data */ 246 /* state data */
247 struct CoroSLF slf_frame; /* saved slf frame */ 247 struct CoroSLF slf_frame; /* saved slf frame */
248 AV *mainstack; 248 AV *mainstack;
249 perl_slots *slot; /* basically the saved sp */ 249 perl_slots *slot; /* basically the saved sp */
250 250
251 CV *startcv; /* the CV to execute */
251 AV *args; /* data associated with this coroutine (initial args) */ 252 AV *args; /* data associated with this coroutine (initial args) */
252 int refcnt; /* coroutines are refcounted, yes */ 253 int refcnt; /* coroutines are refcounted, yes */
253 int flags; /* CF_ flags */ 254 int flags; /* CF_ flags */
254 HV *hv; /* the perl hash associated with this coro, if any */ 255 HV *hv; /* the perl hash associated with this coro, if any */
255 void (*on_destroy)(pTHX_ struct coro *coro); 256 void (*on_destroy)(pTHX_ struct coro *coro);
256 257
257 /* statistics */ 258 /* statistics */
258 int usecount; /* number of transfers to this coro */ 259 int usecount; /* number of transfers to this coro */
259 260
260 /* coro process data */ 261 /* coro process data */
261 int prio; 262 int prio;
262 SV *except; /* exception to be thrown */ 263 SV *except; /* exception to be thrown */
264 SV *rouse_cb;
263 265
264 /* async_pool */ 266 /* async_pool */
265 SV *saved_deffh; 267 SV *saved_deffh;
266 268
267 /* linked list */ 269 /* linked list */
… …
320#if PERL_VERSION_ATLEAST (5,10,0) 322#if PERL_VERSION_ATLEAST (5,10,0)
321 /* silence stupid and wrong 5.10 warning that I am unable to switch off */ 323 /* silence stupid and wrong 5.10 warning that I am unable to switch off */
322 get_hv (name, create); 324 get_hv (name, create);
323#endif 325#endif
324 return get_hv (name, create); 326 return get_hv (name, create);
327}
328
329/* may croak */
330INLINE CV *
331coro_sv_2cv (SV *sv)
332{
333 HV *st;
334 GV *gvp;
335 return sv_2cv (sv, &st, &gvp, 0);
325} 336}
326 337
327static AV * 338static AV *
328coro_clone_padlist (pTHX_ CV *cv) 339coro_clone_padlist (pTHX_ CV *cv)
329{ 340{
… …
448 else 459 else
449 { 460 {
450#if CORO_PREFER_PERL_FUNCTIONS 461#if CORO_PREFER_PERL_FUNCTIONS
451 /* this is probably cleaner? but also slower! */ 462 /* this is probably cleaner? but also slower! */
452 /* in practise, it seems to be less stable */ 463 /* in practise, it seems to be less stable */
453 CV *cp = Perl_cv_clone (cv); 464 CV *cp = Perl_cv_clone (aTHX_ cv);
454 CvPADLIST (cv) = CvPADLIST (cp); 465 CvPADLIST (cv) = CvPADLIST (cp);
455 CvPADLIST (cp) = 0; 466 CvPADLIST (cp) = 0;
456 SvREFCNT_dec (cp); 467 SvREFCNT_dec (cp);
457#else 468#else
458 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv); 469 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv);
… …
602 * of perl.c:init_stacks, except that it uses less memory 613 * of perl.c:init_stacks, except that it uses less memory
603 * on the (sometimes correct) assumption that coroutines do 614 * on the (sometimes correct) assumption that coroutines do
604 * not usually need a lot of stackspace. 615 * not usually need a lot of stackspace.
605 */ 616 */
606#if CORO_PREFER_PERL_FUNCTIONS 617#if CORO_PREFER_PERL_FUNCTIONS
607# define coro_init_stacks init_stacks 618# define coro_init_stacks(thx) init_stacks ()
608#else 619#else
609static void 620static void
610coro_init_stacks (pTHX) 621coro_init_stacks (pTHX)
611{ 622{
612 PL_curstackinfo = new_stackinfo(32, 8); 623 PL_curstackinfo = new_stackinfo(32, 8);
… …
918 SvREFCNT_dec (GvSV (irsgv)); 929 SvREFCNT_dec (GvSV (irsgv));
919 930
920 SvREFCNT_dec (PL_diehook); 931 SvREFCNT_dec (PL_diehook);
921 SvREFCNT_dec (PL_warnhook); 932 SvREFCNT_dec (PL_warnhook);
922 933
934 SvREFCNT_dec (CORO_THROW);
923 SvREFCNT_dec (coro->saved_deffh); 935 SvREFCNT_dec (coro->saved_deffh);
924 SvREFCNT_dec (CORO_THROW); 936 SvREFCNT_dec (coro->rouse_cb);
925 937
926 coro_destruct_stacks (aTHX); 938 coro_destruct_stacks (aTHX);
927} 939}
928 940
929INLINE void 941INLINE void
… …
1016 SAVETMPS; 1028 SAVETMPS;
1017 EXTEND (SP, 3); 1029 EXTEND (SP, 3);
1018 PUSHMARK (SP); 1030 PUSHMARK (SP);
1019 PUSHs (&PL_sv_yes); 1031 PUSHs (&PL_sv_yes);
1020 PUSHs (fullname); 1032 PUSHs (fullname);
1021 PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef); 1033 PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef);
1022 PUTBACK; 1034 PUTBACK;
1023 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0); 1035 cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1024 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD); 1036 if (cb) call_sv (*cb, G_KEEPERR | G_EVAL | G_VOID | G_DISCARD);
1025 SPAGAIN; 1037 SPAGAIN;
1026 FREETMPS; 1038 FREETMPS;
… …
1433 1445
1434 coro->slot = 0; 1446 coro->slot = 0;
1435 } 1447 }
1436 1448
1437 cctx_destroy (coro->cctx); 1449 cctx_destroy (coro->cctx);
1450 SvREFCNT_dec (coro->startcv);
1438 SvREFCNT_dec (coro->args); 1451 SvREFCNT_dec (coro->args);
1439 1452
1440 if (coro->next) coro->next->prev = coro->prev; 1453 if (coro->next) coro->next->prev = coro->prev;
1441 if (coro->prev) coro->prev->next = coro->next; 1454 if (coro->prev) coro->prev->next = coro->next;
1442 if (coro == coro_first) coro_first = coro->next; 1455 if (coro == coro_first) coro_first = coro->next;
… …
1497 1510
1498 prepare_transfer (aTHX_ &ta, prev_sv, next_sv); 1511 prepare_transfer (aTHX_ &ta, prev_sv, next_sv);
1499 TRANSFER (ta, 1); 1512 TRANSFER (ta, 1);
1500} 1513}
1501 1514
1515/*****************************************************************************/
1516/* gensub: simple closure generation utility */
1517
1518#define GENSUB_ARG CvXSUBANY (cv).any_ptr
1519
1520/* create a closure from XS, returns a code reference */
1521/* the arg can be accessed via GENSUB_ARG from the callback */
1522/* the callback must use dXSARGS/XSRETURN */
1523static SV *
1524gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
1525{
1526 CV *cv = (CV *)newSV (0);
1527
1528 sv_upgrade ((SV *)cv, SVt_PVCV);
1529
1530 CvANON_on (cv);
1531 CvISXSUB_on (cv);
1532 CvXSUB (cv) = xsub;
1533 GENSUB_ARG = arg;
1534
1535 return newRV_noinc ((SV *)cv);
1536}
1537
1502/** Coro ********************************************************************/ 1538/** Coro ********************************************************************/
1503 1539
1504INLINE void 1540INLINE void
1505coro_enq (pTHX_ struct coro *coro) 1541coro_enq (pTHX_ struct coro *coro)
1506{ 1542{
… …
1703 if (coro->flags & CF_RUNNING) 1739 if (coro->flags & CF_RUNNING)
1704 PL_runops = RUNOPS_DEFAULT; 1740 PL_runops = RUNOPS_DEFAULT;
1705 else 1741 else
1706 coro->slot->runops = RUNOPS_DEFAULT; 1742 coro->slot->runops = RUNOPS_DEFAULT;
1707 } 1743 }
1744}
1745
1746/*****************************************************************************/
1747/* rouse callback */
1748
1749#define CORO_MAGIC_type_rouse PERL_MAGIC_ext
1750
1751static void
1752coro_rouse_callback (pTHX_ CV *cv)
1753{
1754 dXSARGS;
1755 SV *data = (SV *)GENSUB_ARG;
1756
1757 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1758 {
1759 /* first call, set args */
1760 int i;
1761 AV *av = newAV ();
1762 SV *coro = SvRV (data);
1763
1764 SvRV_set (data, (SV *)av);
1765 api_ready (aTHX_ coro);
1766 SvREFCNT_dec (coro);
1767
1768 /* better take a full copy of the arguments */
1769 while (items--)
1770 av_store (av, items, newSVsv (ST (items)));
1771 }
1772
1773 XSRETURN_EMPTY;
1774}
1775
1776static int
1777slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
1778{
1779 SV *data = (SV *)frame->data;
1780
1781 if (CORO_THROW)
1782 return 0;
1783
1784 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1785 return 1;
1786
1787 /* now push all results on the stack */
1788 {
1789 dSP;
1790 AV *av = (AV *)SvRV (data);
1791 int i;
1792
1793 EXTEND (SP, AvFILLp (av) + 1);
1794 for (i = 0; i <= AvFILLp (av); ++i)
1795 PUSHs (sv_2mortal (AvARRAY (av)[i]));
1796
1797 /* we have stolen the elements, so ste length to zero and free */
1798 AvFILLp (av) = -1;
1799 av_undef (av);
1800
1801 PUTBACK;
1802 }
1803
1804 return 0;
1805}
1806
1807static void
1808slf_init_rouse_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1809{
1810 SV *cb;
1811
1812 if (items)
1813 cb = arg [0];
1814 else
1815 {
1816 struct coro *coro = SvSTATE_current;
1817
1818 if (!coro->rouse_cb)
1819 croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
1820
1821 cb = sv_2mortal (coro->rouse_cb);
1822 coro->rouse_cb = 0;
1823 }
1824
1825 if (!SvROK (cb)
1826 || SvTYPE (SvRV (cb)) != SVt_PVCV
1827 || CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
1828 croak ("Coro::rouse_wait called with illegal callback argument,");
1829
1830 {
1831 CV *cv = (CV *)SvRV (cb); /* for GENSUB_ARG */
1832 SV *data = (SV *)GENSUB_ARG;
1833
1834 frame->data = (void *)data;
1835 frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
1836 frame->check = slf_check_rouse_wait;
1837 }
1838}
1839
1840static SV *
1841coro_new_rouse_cb (pTHX)
1842{
1843 HV *hv = (HV *)SvRV (coro_current);
1844 struct coro *coro = SvSTATE_hv (hv);
1845 SV *data = newRV_inc ((SV *)hv);
1846 SV *cb = gensub (aTHX_ coro_rouse_callback, (void *)data);
1847
1848 sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
1849 SvREFCNT_dec (data); /* magicext increases the refcount */
1850
1851 SvREFCNT_dec (coro->rouse_cb);
1852 coro->rouse_cb = SvREFCNT_inc_NN (cb);
1853
1854 return cb;
1708} 1855}
1709 1856
1710/*****************************************************************************/ 1857/*****************************************************************************/
1711/* schedule-like-function opcode (SLF) */ 1858/* schedule-like-function opcode (SLF) */
1712 1859
… …
2007 SV **ary; 2154 SV **ary;
2008 2155
2009 /* unfortunately, building manually saves memory */ 2156 /* unfortunately, building manually saves memory */
2010 Newx (ary, 2, SV *); 2157 Newx (ary, 2, SV *);
2011 AvALLOC (av) = ary; 2158 AvALLOC (av) = ary;
2012 AvARRAY (av) = ary; 2159 /*AvARRAY (av) = ary;*/
2160 SvPVX ((SV *)av) = (char *)ary; /* 5.8.8 needs this syntax instead of AvARRAY = ary */
2013 AvMAX (av) = 1; 2161 AvMAX (av) = 1;
2014 AvFILLp (av) = 0; 2162 AvFILLp (av) = 0;
2015 ary [0] = newSViv (count); 2163 ary [0] = newSViv (count);
2016 2164
2017 return newRV_noinc ((SV *)av); 2165 return newRV_noinc ((SV *)av);
… …
2037 AvARRAY (av)[0] = AvARRAY (av)[1]; 2185 AvARRAY (av)[0] = AvARRAY (av)[1];
2038 AvARRAY (av)[1] = count_sv; 2186 AvARRAY (av)[1] = count_sv;
2039 cb = av_shift (av); 2187 cb = av_shift (av);
2040 2188
2041 if (SvOBJECT (cb)) 2189 if (SvOBJECT (cb))
2190 {
2042 api_ready (aTHX_ cb); 2191 api_ready (aTHX_ cb);
2043 else 2192 --count;
2044 croak ("callbacks not yet supported"); 2193 }
2194 else if (SvTYPE (cb) == SVt_PVCV)
2195 {
2196 dSP;
2197 PUSHMARK (SP);
2198 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2199 PUTBACK;
2200 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2201 }
2045 2202
2046 SvREFCNT_dec (cb); 2203 SvREFCNT_dec (cb);
2047
2048 --count;
2049 } 2204 }
2050} 2205}
2051 2206
2052static void 2207static void
2053coro_semaphore_on_destroy (pTHX_ struct coro *coro) 2208coro_semaphore_on_destroy (pTHX_ struct coro *coro)
… …
2135} 2290}
2136 2291
2137static void 2292static void
2138slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2293slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2139{ 2294{
2295 if (items >= 2)
2296 {
2297 /* callback form */
2298 AV *av = (AV *)SvRV (arg [0]);
2299 CV *cb_cv = coro_sv_2cv (arg [1]);
2300
2301 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
2302
2303 if (SvIVX (AvARRAY (av)[0]) > 0)
2304 coro_semaphore_adjust (aTHX_ av, 0);
2305
2306 frame->prepare = prepare_nop;
2307 frame->check = slf_check_nop;
2308 }
2309 else
2310 {
2140 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items); 2311 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2141 frame->check = slf_check_semaphore_wait; 2312 frame->check = slf_check_semaphore_wait;
2313 }
2142} 2314}
2143 2315
2144/* signal */ 2316/* signal */
2145 2317
2146static void 2318static void
… …
2194 2366
2195 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */ 2367 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2196 frame->prepare = prepare_schedule; 2368 frame->prepare = prepare_schedule;
2197 frame->check = slf_check_signal_wait; 2369 frame->check = slf_check_signal_wait;
2198 } 2370 }
2199}
2200
2201/*****************************************************************************/
2202/* gensub: simple closure generation utility */
2203
2204#define GENSUB_ARG CvXSUBANY (cv).any_ptr
2205
2206/* create a closure from XS, returns a code reference */
2207/* the arg can be accessed via GENSUB_ARG from the callback */
2208/* the callback must use dXSARGS/XSRETURN */
2209static SV *
2210gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
2211{
2212 CV *cv = (CV *)newSV (0);
2213
2214 sv_upgrade ((SV *)cv, SVt_PVCV);
2215
2216 CvANON_on (cv);
2217 CvISXSUB_on (cv);
2218 CvXSUB (cv) = xsub;
2219 GENSUB_ARG = arg;
2220
2221 return newRV_noinc ((SV *)cv);
2222} 2371}
2223 2372
2224/*****************************************************************************/ 2373/*****************************************************************************/
2225/* Coro::AIO */ 2374/* Coro::AIO */
2226 2375
… …
2466 CODE: 2615 CODE:
2467{ 2616{
2468 struct coro *coro; 2617 struct coro *coro;
2469 MAGIC *mg; 2618 MAGIC *mg;
2470 HV *hv; 2619 HV *hv;
2620 CV *cb;
2471 int i; 2621 int i;
2622
2623 if (items > 1)
2624 {
2625 cb = coro_sv_2cv (ST (1));
2626
2627 if (CvISXSUB (cb))
2628 croak ("Coro::State doesn't support XS functions as coroutine start, caught");
2629
2630 if (!CvROOT (cb))
2631 croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught");
2632 }
2472 2633
2473 Newz (0, coro, 1, struct coro); 2634 Newz (0, coro, 1, struct coro);
2474 coro->args = newAV (); 2635 coro->args = newAV ();
2475 coro->flags = CF_NEW; 2636 coro->flags = CF_NEW;
2476 2637
… …
2481 coro->hv = hv = newHV (); 2642 coro->hv = hv = newHV ();
2482 mg = sv_magicext ((SV *)hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)coro, 0); 2643 mg = sv_magicext ((SV *)hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char *)coro, 0);
2483 mg->mg_flags |= MGf_DUP; 2644 mg->mg_flags |= MGf_DUP;
2484 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1)); 2645 RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));
2485 2646
2647 if (items > 1)
2648 {
2649 coro->startcv = SvREFCNT_inc_NN (cb);
2650
2486 av_extend (coro->args, items - 1); 2651 av_extend (coro->args, items - 1);
2652 av_push (coro->args, SvREFCNT_inc (cb));
2487 for (i = 1; i < items; i++) 2653 for (i = 2; i < items; i++)
2488 av_push (coro->args, newSVsv (ST (i))); 2654 av_push (coro->args, newSVsv (ST (i)));
2655 }
2489} 2656}
2490 OUTPUT: 2657 OUTPUT:
2491 RETVAL 2658 RETVAL
2492 2659
2493void 2660void
… …
2867 api_trace (aTHX_ coro_current, 0); 3034 api_trace (aTHX_ coro_current, 0);
2868 3035
2869 av_push (av_async_pool, newSVsv (coro_current)); 3036 av_push (av_async_pool, newSVsv (coro_current));
2870} 3037}
2871 3038
3039SV *
3040rouse_cb ()
3041 PROTOTYPE:
3042 CODE:
3043 RETVAL = coro_new_rouse_cb (aTHX);
3044 OUTPUT:
3045 RETVAL
3046
3047void
3048rouse_wait (...)
3049 PROTOTYPE: ;$
3050 PPCODE:
3051 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
3052
2872 3053
2873MODULE = Coro::State PACKAGE = PerlIO::cede 3054MODULE = Coro::State PACKAGE = PerlIO::cede
2874 3055
2875BOOT: 3056BOOT:
2876 PerlIO_define_layer (aTHX_ &PerlIO_cede); 3057 PerlIO_define_layer (aTHX_ &PerlIO_cede);
… …
2909 adjust = 1 3090 adjust = 1
2910 CODE: 3091 CODE:
2911 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1); 3092 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
2912 3093
2913void 3094void
2914down (SV *self) 3095down (...)
2915 CODE: 3096 CODE:
2916 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 3097 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
2917 3098
2918void 3099void
2919wait (SV *self) 3100wait (...)
2920 CODE: 3101 CODE:
2921 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait); 3102 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2922 3103
2923void 3104void
2924try (SV *self) 3105try (SV *self)
… …
2967 ); 3148 );
2968 OUTPUT: 3149 OUTPUT:
2969 RETVAL 3150 RETVAL
2970 3151
2971void 3152void
2972wait (SV *self) 3153wait (...)
2973 CODE: 3154 CODE:
2974 CORO_EXECUTE_SLF_XS (slf_init_signal_wait); 3155 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
2975 3156
2976void 3157void
2977broadcast (SV *self) 3158broadcast (SV *self)
… …
3034 3215
3035void 3216void
3036_register (char *target, char *proto, SV *req) 3217_register (char *target, char *proto, SV *req)
3037 CODE: 3218 CODE:
3038{ 3219{
3039 HV *st;
3040 GV *gvp;
3041 CV *req_cv = sv_2cv (req, &st, &gvp, 0); 3220 CV *req_cv = coro_sv_2cv (req);
3042 /* newXSproto doesn't return the CV on 5.8 */ 3221 /* newXSproto doesn't return the CV on 5.8 */
3043 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__); 3222 CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3044 sv_setpv ((SV *)slf_cv, proto); 3223 sv_setpv ((SV *)slf_cv, proto);
3045 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0); 3224 sv_magicext ((SV *)slf_cv, (SV *)req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
3046} 3225}

Diff Legend

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