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.295 by root, Tue Nov 18 10:35:01 2008 UTC vs.
Revision 1.302 by root, Wed Nov 19 04:48:24 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
… …
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;
… …
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 while (items--)
1758 av_store (av, items, SvREFCNT_inc_NN (ST (items)));
1759 }
1760
1761 XSRETURN_EMPTY;
1762}
1763
1764static int
1765slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
1766{
1767 SV *data = (SV *)frame->data;
1768
1769 if (CORO_THROW)
1770 return 0;
1771
1772 if (SvTYPE (SvRV (data)) != SVt_PVAV)
1773 return 1;
1774
1775 /* now push all results on the stack */
1776 {
1777 dSP;
1778 AV *av = (AV *)SvRV (data);
1779 int i;
1780
1781 EXTEND (SP, AvFILLp (av) + 1);
1782 for (i = 0; i <= AvFILLp (av); ++i)
1783 PUSHs (AvARRAY (av)[i]);
1784
1785 PUTBACK;
1786 }
1787
1788 return 0;
1789}
1790
1791static void
1792slf_init_rouse_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
1793{
1794 SV *cb;
1795
1796 if (items)
1797 cb = arg [0];
1798 else
1799 {
1800 struct coro *coro = SvSTATE_current;
1801
1802 if (!coro->rouse_cb)
1803 croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
1804
1805 cb = sv_2mortal (coro->rouse_cb);
1806 coro->rouse_cb = 0;
1807 }
1808
1809 if (!SvROK (cb)
1810 || SvTYPE (SvRV (cb)) != SVt_PVCV
1811 || CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
1812 croak ("Coro::rouse_wait called with illegal callback argument,");
1813
1814 {
1815 CV *cv = (CV *)SvRV (cb); /* for GENSUB_ARG */
1816 SV *data = (SV *)GENSUB_ARG;
1817
1818 frame->data = (void *)data;
1819 frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
1820 frame->check = slf_check_rouse_wait;
1821 }
1822}
1823
1824static SV *
1825coro_new_rouse_cb (pTHX)
1826{
1827 HV *hv = (HV *)SvRV (coro_current);
1828 struct coro *coro = SvSTATE_hv (hv);
1829 SV *data = newRV_inc ((SV *)hv);
1830 SV *cb = gensub (aTHX_ coro_rouse_callback, (void *)data);
1831
1832 sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
1833
1834 SvREFCNT_dec (coro->rouse_cb);
1835 coro->rouse_cb = SvREFCNT_inc_NN (cb);
1836
1837 return cb;
1708} 1838}
1709 1839
1710/*****************************************************************************/ 1840/*****************************************************************************/
1711/* schedule-like-function opcode (SLF) */ 1841/* schedule-like-function opcode (SLF) */
1712 1842
… …
1839 } 1969 }
1840 while (slf_frame.check (aTHX_ &slf_frame)); 1970 while (slf_frame.check (aTHX_ &slf_frame));
1841 1971
1842 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */ 1972 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
1843 1973
1974 /* exception handling */
1975 if (expect_false (CORO_THROW))
1976 {
1977 SV *exception = sv_2mortal (CORO_THROW);
1978
1979 CORO_THROW = 0;
1980 sv_setsv (ERRSV, exception);
1981 croak (0);
1982 }
1983
1844 /* return value handling - mostly like entersub */ 1984 /* return value handling - mostly like entersub */
1845 /* make sure we put something on the stack in scalar context */ 1985 /* make sure we put something on the stack in scalar context */
1846 if (GIMME_V == G_SCALAR) 1986 if (GIMME_V == G_SCALAR)
1847 { 1987 {
1848 dSP; 1988 dSP;
… …
1854 bot [1] = *sp; 1994 bot [1] = *sp;
1855 1995
1856 SP = bot + 1; 1996 SP = bot + 1;
1857 1997
1858 PUTBACK; 1998 PUTBACK;
1859 }
1860
1861 /* exception handling */
1862 if (expect_false (coro_throw))
1863 {
1864 SV *exception = sv_2mortal (coro_throw);
1865
1866 coro_throw = 0;
1867 sv_setsv (ERRSV, exception);
1868 croak (0);
1869 } 1999 }
1870 2000
1871 return NORMAL; 2001 return NORMAL;
1872} 2002}
1873 2003
… …
1995 PerlIOBuf_get_cnt, 2125 PerlIOBuf_get_cnt,
1996 PerlIOBuf_set_ptrcnt, 2126 PerlIOBuf_set_ptrcnt,
1997}; 2127};
1998 2128
1999/*****************************************************************************/ 2129/*****************************************************************************/
2130/* Coro::Semaphore & Coro::Signal */
2131
2132static SV *
2133coro_waitarray_new (pTHX_ int count)
2134{
2135 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2136 AV *av = newAV ();
2137 SV **ary;
2138
2139 /* unfortunately, building manually saves memory */
2140 Newx (ary, 2, SV *);
2141 AvALLOC (av) = ary;
2142 AvARRAY (av) = ary;
2143 AvMAX (av) = 1;
2144 AvFILLp (av) = 0;
2145 ary [0] = newSViv (count);
2146
2147 return newRV_noinc ((SV *)av);
2148}
2149
2000/* Coro::Semaphore */ 2150/* semaphore */
2001 2151
2002static void 2152static void
2003coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2153coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2004{ 2154{
2005 SV *count_sv = AvARRAY (av)[0]; 2155 SV *count_sv = AvARRAY (av)[0];
… …
2035 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2185 /* call $sem->adjust (0) to possibly wake up some other waiters */
2036 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2186 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2037} 2187}
2038 2188
2039static int 2189static int
2040slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2190slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2041{ 2191{
2042 AV *av = (AV *)frame->data; 2192 AV *av = (AV *)frame->data;
2043 SV *count_sv = AvARRAY (av)[0]; 2193 SV *count_sv = AvARRAY (av)[0];
2044 2194
2045 /* if we are about to throw, don't actually acquire the lock, just throw */ 2195 /* if we are about to throw, don't actually acquire the lock, just throw */
2046 if (coro_throw && 0)//D 2196 if (CORO_THROW)
2047 return 0; 2197 return 0;
2048 else if (SvIVX (count_sv) > 0) 2198 else if (SvIVX (count_sv) > 0)
2049 { 2199 {
2050 SvSTATE_current->on_destroy = 0; 2200 SvSTATE_current->on_destroy = 0;
2201
2202 if (acquire)
2051 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2203 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2204 else
2205 coro_semaphore_adjust (aTHX_ av, 0);
2206
2052 return 0; 2207 return 0;
2053 } 2208 }
2054 else 2209 else
2055 { 2210 {
2056 int i; 2211 int i;
… …
2065 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2220 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2066 return 1; 2221 return 1;
2067 } 2222 }
2068} 2223}
2069 2224
2070static void 2225static int
2226slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2227{
2228 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2229}
2230
2231static int
2232slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2233{
2234 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2235}
2236
2237static void
2071slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2238slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2072{ 2239{
2073 AV *av = (AV *)SvRV (arg [0]); 2240 AV *av = (AV *)SvRV (arg [0]);
2074 2241
2075 if (SvIVX (AvARRAY (av)[0]) > 0) 2242 if (SvIVX (AvARRAY (av)[0]) > 0)
2076 { 2243 {
… …
2086 2253
2087 /* to avoid race conditions when a woken-up coro gets terminated */ 2254 /* to avoid race conditions when a woken-up coro gets terminated */
2088 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2255 /* we arrange for a temporary on_destroy that calls adjust (0) */
2089 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2256 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2090 } 2257 }
2258}
2091 2259
2260static void
2261slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2262{
2263 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2092 frame->check = slf_check_semaphore_down; 2264 frame->check = slf_check_semaphore_down;
2093
2094} 2265}
2095 2266
2096/*****************************************************************************/ 2267static void
2097/* gensub: simple closure generation utility */ 2268slf_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{ 2269{
2107 CV *cv = (CV *)newSV (0); 2270 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2271 frame->check = slf_check_semaphore_wait;
2272}
2108 2273
2109 sv_upgrade ((SV *)cv, SVt_PVCV); 2274/* signal */
2110 2275
2111 CvANON_on (cv); 2276static void
2112 CvISXSUB_on (cv); 2277coro_signal_wake (pTHX_ AV *av, int count)
2113 CvXSUB (cv) = xsub; 2278{
2114 GENSUB_ARG = arg; 2279 SvIVX (AvARRAY (av)[0]) = 0;
2115 2280
2116 return newRV_noinc ((SV *)cv); 2281 /* now signal count waiters */
2282 while (count > 0 && AvFILLp (av) > 0)
2283 {
2284 SV *cb;
2285
2286 /* swap first two elements so we can shift a waiter */
2287 cb = AvARRAY (av)[0];
2288 AvARRAY (av)[0] = AvARRAY (av)[1];
2289 AvARRAY (av)[1] = cb;
2290
2291 cb = av_shift (av);
2292
2293 api_ready (aTHX_ cb);
2294 sv_setiv (cb, 0); /* signal waiter */
2295 SvREFCNT_dec (cb);
2296
2297 --count;
2298 }
2299}
2300
2301static int
2302slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2303{
2304 /* if we are about to throw, also stop waiting */
2305 return SvROK ((SV *)frame->data) && !CORO_THROW;
2306}
2307
2308static void
2309slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2310{
2311 AV *av = (AV *)SvRV (arg [0]);
2312
2313 if (SvIVX (AvARRAY (av)[0]))
2314 {
2315 SvIVX (AvARRAY (av)[0]) = 0;
2316 frame->prepare = prepare_nop;
2317 frame->check = slf_check_nop;
2318 }
2319 else
2320 {
2321 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2322
2323 av_push (av, waiter);
2324
2325 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2326 frame->prepare = prepare_schedule;
2327 frame->check = slf_check_signal_wait;
2328 }
2117} 2329}
2118 2330
2119/*****************************************************************************/ 2331/*****************************************************************************/
2120/* Coro::AIO */ 2332/* Coro::AIO */
2121 2333
… …
2170 2382
2171static int 2383static int
2172slf_check_aio_req (pTHX_ struct CoroSLF *frame) 2384slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2173{ 2385{
2174 AV *state = (AV *)frame->data; 2386 AV *state = (AV *)frame->data;
2387
2388 /* if we are about to throw, return early */
2389 /* this does not cancel the aio request, but at least */
2390 /* it quickly returns */
2391 if (CORO_THROW)
2392 return 0;
2175 2393
2176 /* one element that is an RV? repeat! */ 2394 /* one element that is an RV? repeat! */
2177 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2395 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2178 return 1; 2396 return 1;
2179 2397
… …
2511throw (Coro::State self, SV *throw = &PL_sv_undef) 2729throw (Coro::State self, SV *throw = &PL_sv_undef)
2512 PROTOTYPE: $;$ 2730 PROTOTYPE: $;$
2513 CODE: 2731 CODE:
2514{ 2732{
2515 struct coro *current = SvSTATE_current; 2733 struct coro *current = SvSTATE_current;
2516 SV **throwp = self == current ? &coro_throw : &self->throw; 2734 SV **throwp = self == current ? &CORO_THROW : &self->except;
2517 SvREFCNT_dec (*throwp); 2735 SvREFCNT_dec (*throwp);
2518 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2736 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2519} 2737}
2520 2738
2521void 2739void
… …
2610 coroapi.ready = api_ready; 2828 coroapi.ready = api_ready;
2611 coroapi.is_ready = api_is_ready; 2829 coroapi.is_ready = api_is_ready;
2612 coroapi.nready = coro_nready; 2830 coroapi.nready = coro_nready;
2613 coroapi.current = coro_current; 2831 coroapi.current = coro_current;
2614 2832
2615 GCoroAPI = &coroapi; 2833 /*GCoroAPI = &coroapi;*/
2616 sv_setiv (sv, (IV)&coroapi); 2834 sv_setiv (sv, (IV)&coroapi);
2617 SvREADONLY_on (sv); 2835 SvREADONLY_on (sv);
2618 } 2836 }
2619} 2837}
2620 2838
… …
2756 api_trace (aTHX_ coro_current, 0); 2974 api_trace (aTHX_ coro_current, 0);
2757 2975
2758 av_push (av_async_pool, newSVsv (coro_current)); 2976 av_push (av_async_pool, newSVsv (coro_current));
2759} 2977}
2760 2978
2979SV *
2980rouse_cb ()
2981 PROTOTYPE:
2982 CODE:
2983 RETVAL = coro_new_rouse_cb (aTHX);
2984 OUTPUT:
2985 RETVAL
2986
2987void
2988rouse_wait (SV *cb = 0)
2989 PROTOTYPE: ;$
2990 PPCODE:
2991 CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
2992
2761 2993
2762MODULE = Coro::State PACKAGE = PerlIO::cede 2994MODULE = Coro::State PACKAGE = PerlIO::cede
2763 2995
2764BOOT: 2996BOOT:
2765 PerlIO_define_layer (aTHX_ &PerlIO_cede); 2997 PerlIO_define_layer (aTHX_ &PerlIO_cede);
2766 2998
2767 2999
2768MODULE = Coro::State PACKAGE = Coro::Semaphore 3000MODULE = Coro::State PACKAGE = Coro::Semaphore
2769 3001
2770SV * 3002SV *
2771new (SV *klass, SV *count_ = 0) 3003new (SV *klass, SV *count = 0)
2772 CODE: 3004 CODE:
2773{ 3005 RETVAL = sv_bless (
2774 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 3006 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2775 AV *av = newAV (); 3007 GvSTASH (CvGV (cv))
2776 SV **ary; 3008 );
3009 OUTPUT:
3010 RETVAL
2777 3011
2778 /* unfortunately, building manually saves memory */ 3012# helper for Coro::Channel
2779 Newx (ary, 2, SV *); 3013SV *
2780 AvALLOC (av) = ary; 3014_alloc (int count)
2781 AvARRAY (av) = ary; 3015 CODE:
2782 AvMAX (av) = 1; 3016 RETVAL = coro_waitarray_new (aTHX_ count);
2783 AvFILLp (av) = 0;
2784 ary [0] = newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1);
2785
2786 RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv)));
2787}
2788 OUTPUT: 3017 OUTPUT:
2789 RETVAL 3018 RETVAL
2790 3019
2791SV * 3020SV *
2792count (SV *self) 3021count (SV *self)
… …
2804 3033
2805void 3034void
2806down (SV *self) 3035down (SV *self)
2807 CODE: 3036 CODE:
2808 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 3037 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3038
3039void
3040wait (SV *self)
3041 CODE:
3042 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2809 3043
2810void 3044void
2811try (SV *self) 3045try (SV *self)
2812 PPCODE: 3046 PPCODE:
2813{ 3047{
… …
2825 XSRETURN_NO; 3059 XSRETURN_NO;
2826} 3060}
2827 3061
2828void 3062void
2829waiters (SV *self) 3063waiters (SV *self)
2830 CODE: 3064 PPCODE:
2831{ 3065{
2832 AV *av = (AV *)SvRV (self); 3066 AV *av = (AV *)SvRV (self);
3067 int wcount = AvFILLp (av) + 1 - 1;
2833 3068
2834 if (GIMME_V == G_SCALAR) 3069 if (GIMME_V == G_SCALAR)
2835 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 3070 XPUSHs (sv_2mortal (newSViv (wcount)));
2836 else 3071 else
2837 { 3072 {
2838 int i; 3073 int i;
2839 EXTEND (SP, AvFILLp (av) + 1 - 1); 3074 EXTEND (SP, wcount);
2840 for (i = 1; i <= AvFILLp (av); ++i) 3075 for (i = 1; i <= wcount; ++i)
2841 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 3076 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2842 } 3077 }
2843} 3078}
3079
3080MODULE = Coro::State PACKAGE = Coro::Signal
3081
3082SV *
3083new (SV *klass)
3084 CODE:
3085 RETVAL = sv_bless (
3086 coro_waitarray_new (aTHX_ 0),
3087 GvSTASH (CvGV (cv))
3088 );
3089 OUTPUT:
3090 RETVAL
3091
3092void
3093wait (SV *self)
3094 CODE:
3095 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3096
3097void
3098broadcast (SV *self)
3099 CODE:
3100{
3101 AV *av = (AV *)SvRV (self);
3102 coro_signal_wake (aTHX_ av, AvFILLp (av));
3103}
3104
3105void
3106send (SV *self)
3107 CODE:
3108{
3109 AV *av = (AV *)SvRV (self);
3110
3111 if (AvFILLp (av))
3112 coro_signal_wake (aTHX_ av, 1);
3113 else
3114 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
3115}
3116
3117IV
3118awaited (SV *self)
3119 CODE:
3120 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3121 OUTPUT:
3122 RETVAL
2844 3123
2845 3124
2846MODULE = Coro::State PACKAGE = Coro::AnyEvent 3125MODULE = Coro::State PACKAGE = Coro::AnyEvent
2847 3126
2848BOOT: 3127BOOT:

Diff Legend

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