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.296 by root, Tue Nov 18 10:39:52 2008 UTC vs.
Revision 1.306 by root, Wed Nov 19 11:11:10 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
… …
145#define expect_true(expr) expect ((expr) != 0, 1) 145#define expect_true(expr) expect ((expr) != 0, 1)
146 146
147#define NOINLINE attribute ((noinline)) 147#define NOINLINE attribute ((noinline))
148 148
149#include "CoroAPI.h" 149#include "CoroAPI.h"
150#define GCoroAPI (&coroapi) /* very sneaky */
150 151
151#ifdef USE_ITHREADS 152#ifdef USE_ITHREADS
152# if CORO_PTHREAD 153# if CORO_PTHREAD
153static void *coro_thx; 154static void *coro_thx;
154# endif 155# endif
… …
256 /* statistics */ 257 /* statistics */
257 int usecount; /* number of transfers to this coro */ 258 int usecount; /* number of transfers to this coro */
258 259
259 /* coro process data */ 260 /* coro process data */
260 int prio; 261 int prio;
261 SV *throw; /* exception to be thrown */ 262 SV *except; /* exception to be thrown */
263 SV *rouse_cb;
262 264
263 /* async_pool */ 265 /* async_pool */
264 SV *saved_deffh; 266 SV *saved_deffh;
265 267
266 /* linked list */ 268 /* linked list */
… …
272 274
273/* the following variables are effectively part of the perl context */ 275/* the following variables are effectively part of the perl context */
274/* and get copied between struct coro and these variables */ 276/* and get copied between struct coro and these variables */
275/* the mainr easonw e don't support windows process emulation */ 277/* the mainr easonw e don't support windows process emulation */
276static struct CoroSLF slf_frame; /* the current slf frame */ 278static struct CoroSLF slf_frame; /* the current slf frame */
277static SV *coro_throw;
278 279
279/** Coro ********************************************************************/ 280/** Coro ********************************************************************/
280 281
281#define PRIO_MAX 3 282#define PRIO_MAX 3
282#define PRIO_HIGH 1 283#define PRIO_HIGH 1
… …
448 else 449 else
449 { 450 {
450#if CORO_PREFER_PERL_FUNCTIONS 451#if CORO_PREFER_PERL_FUNCTIONS
451 /* this is probably cleaner? but also slower! */ 452 /* this is probably cleaner? but also slower! */
452 /* in practise, it seems to be less stable */ 453 /* in practise, it seems to be less stable */
453 CV *cp = Perl_cv_clone (cv); 454 CV *cp = Perl_cv_clone (aTHX_ cv);
454 CvPADLIST (cv) = CvPADLIST (cp); 455 CvPADLIST (cv) = CvPADLIST (cp);
455 CvPADLIST (cp) = 0; 456 CvPADLIST (cp) = 0;
456 SvREFCNT_dec (cp); 457 SvREFCNT_dec (cp);
457#else 458#else
458 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv); 459 CvPADLIST (cv) = coro_clone_padlist (aTHX_ cv);
… …
511 512
512 PUTBACK; 513 PUTBACK;
513 } 514 }
514 515
515 slf_frame = c->slf_frame; 516 slf_frame = c->slf_frame;
516 coro_throw = c->throw; 517 CORO_THROW = c->except;
517} 518}
518 519
519static void 520static void
520save_perl (pTHX_ Coro__State c) 521save_perl (pTHX_ Coro__State c)
521{ 522{
522 c->throw = coro_throw; 523 c->except = CORO_THROW;
523 c->slf_frame = slf_frame; 524 c->slf_frame = slf_frame;
524 525
525 { 526 {
526 dSP; 527 dSP;
527 I32 cxix = cxstack_ix; 528 I32 cxix = cxstack_ix;
… …
602 * of perl.c:init_stacks, except that it uses less memory 603 * of perl.c:init_stacks, except that it uses less memory
603 * on the (sometimes correct) assumption that coroutines do 604 * on the (sometimes correct) assumption that coroutines do
604 * not usually need a lot of stackspace. 605 * not usually need a lot of stackspace.
605 */ 606 */
606#if CORO_PREFER_PERL_FUNCTIONS 607#if CORO_PREFER_PERL_FUNCTIONS
607# define coro_init_stacks init_stacks 608# define coro_init_stacks(thx) init_stacks ()
608#else 609#else
609static void 610static void
610coro_init_stacks (pTHX) 611coro_init_stacks (pTHX)
611{ 612{
612 PL_curstackinfo = new_stackinfo(32, 8); 613 PL_curstackinfo = new_stackinfo(32, 8);
… …
885 /* no flags required, as an init function won't be called */ 886 /* no flags required, as an init function won't be called */
886 887
887 PL_op = (OP *)&coro_setup_op; 888 PL_op = (OP *)&coro_setup_op;
888 889
889 /* copy throw, in case it was set before coro_setup */ 890 /* copy throw, in case it was set before coro_setup */
890 coro_throw = coro->throw; 891 CORO_THROW = coro->except;
891} 892}
892 893
893static void 894static void
894coro_destruct (pTHX_ struct coro *coro) 895coro_destruct (pTHX_ struct coro *coro)
895{ 896{
… …
918 SvREFCNT_dec (GvSV (irsgv)); 919 SvREFCNT_dec (GvSV (irsgv));
919 920
920 SvREFCNT_dec (PL_diehook); 921 SvREFCNT_dec (PL_diehook);
921 SvREFCNT_dec (PL_warnhook); 922 SvREFCNT_dec (PL_warnhook);
922 923
924 SvREFCNT_dec (CORO_THROW);
923 SvREFCNT_dec (coro->saved_deffh); 925 SvREFCNT_dec (coro->saved_deffh);
924 SvREFCNT_dec (coro_throw); 926 SvREFCNT_dec (coro->rouse_cb);
925 927
926 coro_destruct_stacks (aTHX); 928 coro_destruct_stacks (aTHX);
927} 929}
928 930
929INLINE void 931INLINE void
… …
1497 1499
1498 prepare_transfer (aTHX_ &ta, prev_sv, next_sv); 1500 prepare_transfer (aTHX_ &ta, prev_sv, next_sv);
1499 TRANSFER (ta, 1); 1501 TRANSFER (ta, 1);
1500} 1502}
1501 1503
1504/*****************************************************************************/
1505/* gensub: simple closure generation utility */
1506
1507#define GENSUB_ARG CvXSUBANY (cv).any_ptr
1508
1509/* create a closure from XS, returns a code reference */
1510/* the arg can be accessed via GENSUB_ARG from the callback */
1511/* the callback must use dXSARGS/XSRETURN */
1512static SV *
1513gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
1514{
1515 CV *cv = (CV *)newSV (0);
1516
1517 sv_upgrade ((SV *)cv, SVt_PVCV);
1518
1519 CvANON_on (cv);
1520 CvISXSUB_on (cv);
1521 CvXSUB (cv) = xsub;
1522 GENSUB_ARG = arg;
1523
1524 return newRV_noinc ((SV *)cv);
1525}
1526
1502/** Coro ********************************************************************/ 1527/** Coro ********************************************************************/
1503 1528
1504INLINE void 1529INLINE void
1505coro_enq (pTHX_ struct coro *coro) 1530coro_enq (pTHX_ struct coro *coro)
1506{ 1531{
… …
1703 if (coro->flags & CF_RUNNING) 1728 if (coro->flags & CF_RUNNING)
1704 PL_runops = RUNOPS_DEFAULT; 1729 PL_runops = RUNOPS_DEFAULT;
1705 else 1730 else
1706 coro->slot->runops = RUNOPS_DEFAULT; 1731 coro->slot->runops = RUNOPS_DEFAULT;
1707 } 1732 }
1733}
1734
1735/*****************************************************************************/
1736/* rouse callback */
1737
1738#define CORO_MAGIC_type_rouse PERL_MAGIC_ext
1739
1740static void
1741coro_rouse_callback (pTHX_ CV *cv)
1742{
1743 dXSARGS;
1744 SV *data = (SV *)GENSUB_ARG;
1745
1746 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1747 {
1748 /* first call, set args */
1749 int i;
1750 AV *av = newAV ();
1751 SV *coro = SvRV (data);
1752
1753 SvRV_set (data, (SV *)av);
1754 api_ready (aTHX_ coro);
1755 SvREFCNT_dec (coro);
1756
1757 /* better take a full copy of the arguments */
1758 while (items--)
1759 av_store (av, items, newSVsv (ST (items)));
1760 }
1761
1762 XSRETURN_EMPTY;
1763}
1764
1765static int
1766slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
1767{
1768 SV *data = (SV *)frame->data;
1769
1770 if (CORO_THROW)
1771 return 0;
1772
1773 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1774 return 1;
1775
1776 /* now push all results on the stack */
1777 {
1778 dSP;
1779 AV *av = (AV *)SvRV (data);
1780 int i;
1781
1782 EXTEND (SP, AvFILLp (av) + 1);
1783 for (i = 0; i <= AvFILLp (av); ++i)
1784 PUSHs (sv_2mortal (AvARRAY (av)[i]));
1785
1786 /* we have stolen the elements, so ste length to zero and free */
1787 AvFILLp (av) = -1;
1788 av_undef (av);
1789
1790 PUTBACK;
1791 }
1792
1793 return 0;
1794}
1795
1796static void
1797slf_init_rouse_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1798{
1799 SV *cb;
1800
1801 if (items)
1802 cb = arg [0];
1803 else
1804 {
1805 struct coro *coro = SvSTATE_current;
1806
1807 if (!coro->rouse_cb)
1808 croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
1809
1810 cb = sv_2mortal (coro->rouse_cb);
1811 coro->rouse_cb = 0;
1812 }
1813
1814 if (!SvROK (cb)
1815 || SvTYPE (SvRV (cb)) != SVt_PVCV
1816 || CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
1817 croak ("Coro::rouse_wait called with illegal callback argument,");
1818
1819 {
1820 CV *cv = (CV *)SvRV (cb); /* for GENSUB_ARG */
1821 SV *data = (SV *)GENSUB_ARG;
1822
1823 frame->data = (void *)data;
1824 frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
1825 frame->check = slf_check_rouse_wait;
1826 }
1827}
1828
1829static SV *
1830coro_new_rouse_cb (pTHX)
1831{
1832 HV *hv = (HV *)SvRV (coro_current);
1833 struct coro *coro = SvSTATE_hv (hv);
1834 SV *data = newRV_inc ((SV *)hv);
1835 SV *cb = gensub (aTHX_ coro_rouse_callback, (void *)data);
1836
1837 sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
1838 SvREFCNT_dec (data); /* magicext increases the refcount */
1839
1840 SvREFCNT_dec (coro->rouse_cb);
1841 coro->rouse_cb = SvREFCNT_inc_NN (cb);
1842
1843 return cb;
1708} 1844}
1709 1845
1710/*****************************************************************************/ 1846/*****************************************************************************/
1711/* schedule-like-function opcode (SLF) */ 1847/* schedule-like-function opcode (SLF) */
1712 1848
… …
1840 while (slf_frame.check (aTHX_ &slf_frame)); 1976 while (slf_frame.check (aTHX_ &slf_frame));
1841 1977
1842 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */ 1978 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
1843 1979
1844 /* exception handling */ 1980 /* exception handling */
1845 if (expect_false (coro_throw)) 1981 if (expect_false (CORO_THROW))
1846 { 1982 {
1847 SV *exception = sv_2mortal (coro_throw); 1983 SV *exception = sv_2mortal (CORO_THROW);
1848 1984
1849 coro_throw = 0; 1985 CORO_THROW = 0;
1850 sv_setsv (ERRSV, exception); 1986 sv_setsv (ERRSV, exception);
1851 croak (0); 1987 croak (0);
1852 } 1988 }
1853 1989
1854 /* return value handling - mostly like entersub */ 1990 /* return value handling - mostly like entersub */
… …
1995 PerlIOBuf_get_cnt, 2131 PerlIOBuf_get_cnt,
1996 PerlIOBuf_set_ptrcnt, 2132 PerlIOBuf_set_ptrcnt,
1997}; 2133};
1998 2134
1999/*****************************************************************************/ 2135/*****************************************************************************/
2136/* Coro::Semaphore & Coro::Signal */
2137
2138static SV *
2139coro_waitarray_new (pTHX_ int count)
2140{
2141 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2142 AV *av = newAV ();
2143 SV **ary;
2144
2145 /* unfortunately, building manually saves memory */
2146 Newx (ary, 2, SV *);
2147 AvALLOC (av) = ary;
2148 /*AvARRAY (av) = ary;*/
2149 SvPVX ((SV *)av) = (char *)ary; /* 5.8.8 needs this syntax instead of AvARRAY = ary */
2150 AvMAX (av) = 1;
2151 AvFILLp (av) = 0;
2152 ary [0] = newSViv (count);
2153
2154 return newRV_noinc ((SV *)av);
2155}
2156
2000/* Coro::Semaphore */ 2157/* semaphore */
2001 2158
2002static void 2159static void
2003coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2160coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2004{ 2161{
2005 SV *count_sv = AvARRAY (av)[0]; 2162 SV *count_sv = AvARRAY (av)[0];
… …
2017 AvARRAY (av)[0] = AvARRAY (av)[1]; 2174 AvARRAY (av)[0] = AvARRAY (av)[1];
2018 AvARRAY (av)[1] = count_sv; 2175 AvARRAY (av)[1] = count_sv;
2019 cb = av_shift (av); 2176 cb = av_shift (av);
2020 2177
2021 if (SvOBJECT (cb)) 2178 if (SvOBJECT (cb))
2179 {
2022 api_ready (aTHX_ cb); 2180 api_ready (aTHX_ cb);
2023 else 2181 --count;
2024 croak ("callbacks not yet supported"); 2182 }
2183 else if (SvTYPE (cb) == SVt_PVCV)
2184 {
2185 dSP;
2186 PUSHMARK (SP);
2187 XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
2188 PUTBACK;
2189 call_sv (cb, G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);
2190 }
2025 2191
2026 SvREFCNT_dec (cb); 2192 SvREFCNT_dec (cb);
2027
2028 --count;
2029 } 2193 }
2030} 2194}
2031 2195
2032static void 2196static void
2033coro_semaphore_on_destroy (pTHX_ struct coro *coro) 2197coro_semaphore_on_destroy (pTHX_ struct coro *coro)
… …
2035 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2199 /* call $sem->adjust (0) to possibly wake up some other waiters */
2036 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2200 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2037} 2201}
2038 2202
2039static int 2203static int
2040slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2204slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2041{ 2205{
2042 AV *av = (AV *)frame->data; 2206 AV *av = (AV *)frame->data;
2043 SV *count_sv = AvARRAY (av)[0]; 2207 SV *count_sv = AvARRAY (av)[0];
2044 2208
2045 /* if we are about to throw, don't actually acquire the lock, just throw */ 2209 /* if we are about to throw, don't actually acquire the lock, just throw */
2046 if (coro_throw) 2210 if (CORO_THROW)
2047 return 0; 2211 return 0;
2048 else if (SvIVX (count_sv) > 0) 2212 else if (SvIVX (count_sv) > 0)
2049 { 2213 {
2050 SvSTATE_current->on_destroy = 0; 2214 SvSTATE_current->on_destroy = 0;
2215
2216 if (acquire)
2051 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2217 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2218 else
2219 coro_semaphore_adjust (aTHX_ av, 0);
2220
2052 return 0; 2221 return 0;
2053 } 2222 }
2054 else 2223 else
2055 { 2224 {
2056 int i; 2225 int i;
… …
2065 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2234 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2066 return 1; 2235 return 1;
2067 } 2236 }
2068} 2237}
2069 2238
2070static void 2239static int
2240slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2241{
2242 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2243}
2244
2245static int
2246slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2247{
2248 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2249}
2250
2251static void
2071slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2252slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2072{ 2253{
2073 AV *av = (AV *)SvRV (arg [0]); 2254 AV *av = (AV *)SvRV (arg [0]);
2074 2255
2075 if (SvIVX (AvARRAY (av)[0]) > 0) 2256 if (SvIVX (AvARRAY (av)[0]) > 0)
2076 { 2257 {
… …
2086 2267
2087 /* to avoid race conditions when a woken-up coro gets terminated */ 2268 /* to avoid race conditions when a woken-up coro gets terminated */
2088 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2269 /* we arrange for a temporary on_destroy that calls adjust (0) */
2089 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2270 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2090 } 2271 }
2272}
2091 2273
2274static void
2275slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2276{
2277 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2092 frame->check = slf_check_semaphore_down; 2278 frame->check = slf_check_semaphore_down;
2093
2094} 2279}
2095 2280
2096/*****************************************************************************/ 2281static void
2097/* gensub: simple closure generation utility */ 2282slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2098
2099#define GENSUB_ARG CvXSUBANY (cv).any_ptr
2100
2101/* create a closure from XS, returns a code reference */
2102/* the arg can be accessed via GENSUB_ARG from the callback */
2103/* the callback must use dXSARGS/XSRETURN */
2104static SV *
2105gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
2106{ 2283{
2107 CV *cv = (CV *)newSV (0); 2284 if (items >= 2)
2285 {
2286 /* callback form */
2287 AV *av = (AV *)SvRV (arg [0]);
2288 HV *st;
2289 GV *gvp;
2290 CV *cb_cv = sv_2cv (arg [1], &st, &gvp, 0);
2108 2291
2109 sv_upgrade ((SV *)cv, SVt_PVCV); 2292 av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
2110 2293
2111 CvANON_on (cv); 2294 if (SvIVX (AvARRAY (av)[0]) > 0)
2112 CvISXSUB_on (cv); 2295 coro_semaphore_adjust (aTHX_ av, 0);
2113 CvXSUB (cv) = xsub;
2114 GENSUB_ARG = arg;
2115 2296
2116 return newRV_noinc ((SV *)cv); 2297 frame->prepare = prepare_nop;
2298 frame->check = slf_check_nop;
2299 }
2300 else
2301 {
2302 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2303 frame->check = slf_check_semaphore_wait;
2304 }
2305}
2306
2307/* signal */
2308
2309static void
2310coro_signal_wake (pTHX_ AV *av, int count)
2311{
2312 SvIVX (AvARRAY (av)[0]) = 0;
2313
2314 /* now signal count waiters */
2315 while (count > 0 && AvFILLp (av) > 0)
2316 {
2317 SV *cb;
2318
2319 /* swap first two elements so we can shift a waiter */
2320 cb = AvARRAY (av)[0];
2321 AvARRAY (av)[0] = AvARRAY (av)[1];
2322 AvARRAY (av)[1] = cb;
2323
2324 cb = av_shift (av);
2325
2326 api_ready (aTHX_ cb);
2327 sv_setiv (cb, 0); /* signal waiter */
2328 SvREFCNT_dec (cb);
2329
2330 --count;
2331 }
2332}
2333
2334static int
2335slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2336{
2337 /* if we are about to throw, also stop waiting */
2338 return SvROK ((SV *)frame->data) && !CORO_THROW;
2339}
2340
2341static void
2342slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2343{
2344 AV *av = (AV *)SvRV (arg [0]);
2345
2346 if (SvIVX (AvARRAY (av)[0]))
2347 {
2348 SvIVX (AvARRAY (av)[0]) = 0;
2349 frame->prepare = prepare_nop;
2350 frame->check = slf_check_nop;
2351 }
2352 else
2353 {
2354 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2355
2356 av_push (av, waiter);
2357
2358 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2359 frame->prepare = prepare_schedule;
2360 frame->check = slf_check_signal_wait;
2361 }
2117} 2362}
2118 2363
2119/*****************************************************************************/ 2364/*****************************************************************************/
2120/* Coro::AIO */ 2365/* Coro::AIO */
2121 2366
… …
2174 AV *state = (AV *)frame->data; 2419 AV *state = (AV *)frame->data;
2175 2420
2176 /* if we are about to throw, return early */ 2421 /* if we are about to throw, return early */
2177 /* this does not cancel the aio request, but at least */ 2422 /* this does not cancel the aio request, but at least */
2178 /* it quickly returns */ 2423 /* it quickly returns */
2179 if (coro_throw) 2424 if (CORO_THROW)
2180 return 0; 2425 return 0;
2181 2426
2182 /* one element that is an RV? repeat! */ 2427 /* one element that is an RV? repeat! */
2183 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2428 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2184 return 1; 2429 return 1;
… …
2517throw (Coro::State self, SV *throw = &PL_sv_undef) 2762throw (Coro::State self, SV *throw = &PL_sv_undef)
2518 PROTOTYPE: $;$ 2763 PROTOTYPE: $;$
2519 CODE: 2764 CODE:
2520{ 2765{
2521 struct coro *current = SvSTATE_current; 2766 struct coro *current = SvSTATE_current;
2522 SV **throwp = self == current ? &coro_throw : &self->throw; 2767 SV **throwp = self == current ? &CORO_THROW : &self->except;
2523 SvREFCNT_dec (*throwp); 2768 SvREFCNT_dec (*throwp);
2524 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2769 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2525} 2770}
2526 2771
2527void 2772void
… …
2616 coroapi.ready = api_ready; 2861 coroapi.ready = api_ready;
2617 coroapi.is_ready = api_is_ready; 2862 coroapi.is_ready = api_is_ready;
2618 coroapi.nready = coro_nready; 2863 coroapi.nready = coro_nready;
2619 coroapi.current = coro_current; 2864 coroapi.current = coro_current;
2620 2865
2621 GCoroAPI = &coroapi; 2866 /*GCoroAPI = &coroapi;*/
2622 sv_setiv (sv, (IV)&coroapi); 2867 sv_setiv (sv, (IV)&coroapi);
2623 SvREADONLY_on (sv); 2868 SvREADONLY_on (sv);
2624 } 2869 }
2625} 2870}
2626 2871
… …
2762 api_trace (aTHX_ coro_current, 0); 3007 api_trace (aTHX_ coro_current, 0);
2763 3008
2764 av_push (av_async_pool, newSVsv (coro_current)); 3009 av_push (av_async_pool, newSVsv (coro_current));
2765} 3010}
2766 3011
3012SV *
3013rouse_cb ()
3014 PROTOTYPE:
3015 CODE:
3016 RETVAL = coro_new_rouse_cb (aTHX);
3017 OUTPUT:
3018 RETVAL
3019
3020void
3021rouse_wait (...)
3022 PROTOTYPE: ;$
3023 PPCODE:
3024 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
3025
2767 3026
2768MODULE = Coro::State PACKAGE = PerlIO::cede 3027MODULE = Coro::State PACKAGE = PerlIO::cede
2769 3028
2770BOOT: 3029BOOT:
2771 PerlIO_define_layer (aTHX_ &PerlIO_cede); 3030 PerlIO_define_layer (aTHX_ &PerlIO_cede);
2772 3031
2773 3032
2774MODULE = Coro::State PACKAGE = Coro::Semaphore 3033MODULE = Coro::State PACKAGE = Coro::Semaphore
2775 3034
2776SV * 3035SV *
2777new (SV *klass, SV *count_ = 0) 3036new (SV *klass, SV *count = 0)
2778 CODE: 3037 CODE:
2779{ 3038 RETVAL = sv_bless (
2780 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 3039 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2781 AV *av = newAV (); 3040 GvSTASH (CvGV (cv))
2782 SV **ary; 3041 );
3042 OUTPUT:
3043 RETVAL
2783 3044
2784 /* unfortunately, building manually saves memory */ 3045# helper for Coro::Channel
2785 Newx (ary, 2, SV *); 3046SV *
2786 AvALLOC (av) = ary; 3047_alloc (int count)
2787 AvARRAY (av) = ary; 3048 CODE:
2788 AvMAX (av) = 1; 3049 RETVAL = coro_waitarray_new (aTHX_ count);
2789 AvFILLp (av) = 0;
2790 ary [0] = newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1);
2791
2792 RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv)));
2793}
2794 OUTPUT: 3050 OUTPUT:
2795 RETVAL 3051 RETVAL
2796 3052
2797SV * 3053SV *
2798count (SV *self) 3054count (SV *self)
… …
2807 adjust = 1 3063 adjust = 1
2808 CODE: 3064 CODE:
2809 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1); 3065 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
2810 3066
2811void 3067void
2812down (SV *self) 3068down (...)
2813 CODE: 3069 CODE:
2814 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 3070 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3071
3072void
3073wait (...)
3074 CODE:
3075 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2815 3076
2816void 3077void
2817try (SV *self) 3078try (SV *self)
2818 PPCODE: 3079 PPCODE:
2819{ 3080{
… …
2831 XSRETURN_NO; 3092 XSRETURN_NO;
2832} 3093}
2833 3094
2834void 3095void
2835waiters (SV *self) 3096waiters (SV *self)
2836 CODE: 3097 PPCODE:
2837{ 3098{
2838 AV *av = (AV *)SvRV (self); 3099 AV *av = (AV *)SvRV (self);
3100 int wcount = AvFILLp (av) + 1 - 1;
2839 3101
2840 if (GIMME_V == G_SCALAR) 3102 if (GIMME_V == G_SCALAR)
2841 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 3103 XPUSHs (sv_2mortal (newSViv (wcount)));
2842 else 3104 else
2843 { 3105 {
2844 int i; 3106 int i;
2845 EXTEND (SP, AvFILLp (av) + 1 - 1); 3107 EXTEND (SP, wcount);
2846 for (i = 1; i <= AvFILLp (av); ++i) 3108 for (i = 1; i <= wcount; ++i)
2847 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 3109 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2848 } 3110 }
2849} 3111}
3112
3113MODULE = Coro::State PACKAGE = Coro::Signal
3114
3115SV *
3116new (SV *klass)
3117 CODE:
3118 RETVAL = sv_bless (
3119 coro_waitarray_new (aTHX_ 0),
3120 GvSTASH (CvGV (cv))
3121 );
3122 OUTPUT:
3123 RETVAL
3124
3125void
3126wait (...)
3127 CODE:
3128 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3129
3130void
3131broadcast (SV *self)
3132 CODE:
3133{
3134 AV *av = (AV *)SvRV (self);
3135 coro_signal_wake (aTHX_ av, AvFILLp (av));
3136}
3137
3138void
3139send (SV *self)
3140 CODE:
3141{
3142 AV *av = (AV *)SvRV (self);
3143
3144 if (AvFILLp (av))
3145 coro_signal_wake (aTHX_ av, 1);
3146 else
3147 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
3148}
3149
3150IV
3151awaited (SV *self)
3152 CODE:
3153 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3154 OUTPUT:
3155 RETVAL
2850 3156
2851 3157
2852MODULE = Coro::State PACKAGE = Coro::AnyEvent 3158MODULE = Coro::State PACKAGE = Coro::AnyEvent
2853 3159
2854BOOT: 3160BOOT:

Diff Legend

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