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.300 by root, Wed Nov 19 02:07:48 2008 UTC

… …
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 */
262 263
263 /* async_pool */ 264 /* async_pool */
264 SV *saved_deffh; 265 SV *saved_deffh;
265 266
266 /* linked list */ 267 /* linked list */
… …
272 273
273/* the following variables are effectively part of the perl context */ 274/* the following variables are effectively part of the perl context */
274/* and get copied between struct coro and these variables */ 275/* and get copied between struct coro and these variables */
275/* the mainr easonw e don't support windows process emulation */ 276/* the mainr easonw e don't support windows process emulation */
276static struct CoroSLF slf_frame; /* the current slf frame */ 277static struct CoroSLF slf_frame; /* the current slf frame */
277static SV *coro_throw;
278 278
279/** Coro ********************************************************************/ 279/** Coro ********************************************************************/
280 280
281#define PRIO_MAX 3 281#define PRIO_MAX 3
282#define PRIO_HIGH 1 282#define PRIO_HIGH 1
… …
511 511
512 PUTBACK; 512 PUTBACK;
513 } 513 }
514 514
515 slf_frame = c->slf_frame; 515 slf_frame = c->slf_frame;
516 coro_throw = c->throw; 516 CORO_THROW = c->except;
517} 517}
518 518
519static void 519static void
520save_perl (pTHX_ Coro__State c) 520save_perl (pTHX_ Coro__State c)
521{ 521{
522 c->throw = coro_throw; 522 c->except = CORO_THROW;
523 c->slf_frame = slf_frame; 523 c->slf_frame = slf_frame;
524 524
525 { 525 {
526 dSP; 526 dSP;
527 I32 cxix = cxstack_ix; 527 I32 cxix = cxstack_ix;
… …
885 /* no flags required, as an init function won't be called */ 885 /* no flags required, as an init function won't be called */
886 886
887 PL_op = (OP *)&coro_setup_op; 887 PL_op = (OP *)&coro_setup_op;
888 888
889 /* copy throw, in case it was set before coro_setup */ 889 /* copy throw, in case it was set before coro_setup */
890 coro_throw = coro->throw; 890 CORO_THROW = coro->except;
891} 891}
892 892
893static void 893static void
894coro_destruct (pTHX_ struct coro *coro) 894coro_destruct (pTHX_ struct coro *coro)
895{ 895{
… …
919 919
920 SvREFCNT_dec (PL_diehook); 920 SvREFCNT_dec (PL_diehook);
921 SvREFCNT_dec (PL_warnhook); 921 SvREFCNT_dec (PL_warnhook);
922 922
923 SvREFCNT_dec (coro->saved_deffh); 923 SvREFCNT_dec (coro->saved_deffh);
924 SvREFCNT_dec (coro_throw); 924 SvREFCNT_dec (CORO_THROW);
925 925
926 coro_destruct_stacks (aTHX); 926 coro_destruct_stacks (aTHX);
927} 927}
928 928
929INLINE void 929INLINE void
… …
1839 } 1839 }
1840 while (slf_frame.check (aTHX_ &slf_frame)); 1840 while (slf_frame.check (aTHX_ &slf_frame));
1841 1841
1842 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */ 1842 slf_frame.prepare = 0; /* invalidate the frame, we are done processing it */
1843 1843
1844 /* exception handling */
1845 if (expect_false (CORO_THROW))
1846 {
1847 SV *exception = sv_2mortal (CORO_THROW);
1848
1849 CORO_THROW = 0;
1850 sv_setsv (ERRSV, exception);
1851 croak (0);
1852 }
1853
1844 /* return value handling - mostly like entersub */ 1854 /* return value handling - mostly like entersub */
1845 /* make sure we put something on the stack in scalar context */ 1855 /* make sure we put something on the stack in scalar context */
1846 if (GIMME_V == G_SCALAR) 1856 if (GIMME_V == G_SCALAR)
1847 { 1857 {
1848 dSP; 1858 dSP;
… …
1854 bot [1] = *sp; 1864 bot [1] = *sp;
1855 1865
1856 SP = bot + 1; 1866 SP = bot + 1;
1857 1867
1858 PUTBACK; 1868 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 } 1869 }
1870 1870
1871 return NORMAL; 1871 return NORMAL;
1872} 1872}
1873 1873
… …
1995 PerlIOBuf_get_cnt, 1995 PerlIOBuf_get_cnt,
1996 PerlIOBuf_set_ptrcnt, 1996 PerlIOBuf_set_ptrcnt,
1997}; 1997};
1998 1998
1999/*****************************************************************************/ 1999/*****************************************************************************/
2000/* Coro::Semaphore & Coro::Signal */
2001
2002static SV *
2003coro_waitarray_new (pTHX_ int count)
2004{
2005 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
2006 AV *av = newAV ();
2007 SV **ary;
2008
2009 /* unfortunately, building manually saves memory */
2010 Newx (ary, 2, SV *);
2011 AvALLOC (av) = ary;
2012 AvARRAY (av) = ary;
2013 AvMAX (av) = 1;
2014 AvFILLp (av) = 0;
2015 ary [0] = newSViv (count);
2016
2017 return newRV_noinc ((SV *)av);
2018}
2019
2000/* Coro::Semaphore */ 2020/* semaphore */
2001 2021
2002static void 2022static void
2003coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2023coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2004{ 2024{
2005 SV *count_sv = AvARRAY (av)[0]; 2025 SV *count_sv = AvARRAY (av)[0];
… …
2035 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2055 /* call $sem->adjust (0) to possibly wake up some other waiters */
2036 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2056 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2037} 2057}
2038 2058
2039static int 2059static int
2040slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2060slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2041{ 2061{
2042 AV *av = (AV *)frame->data; 2062 AV *av = (AV *)frame->data;
2043 SV *count_sv = AvARRAY (av)[0]; 2063 SV *count_sv = AvARRAY (av)[0];
2044 2064
2045 /* if we are about to throw, don't actually acquire the lock, just throw */ 2065 /* if we are about to throw, don't actually acquire the lock, just throw */
2046 if (coro_throw && 0)//D 2066 if (CORO_THROW)
2047 return 0; 2067 return 0;
2048 else if (SvIVX (count_sv) > 0) 2068 else if (SvIVX (count_sv) > 0)
2049 { 2069 {
2050 SvSTATE_current->on_destroy = 0; 2070 SvSTATE_current->on_destroy = 0;
2071
2072 if (acquire)
2051 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2073 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2074 else
2075 coro_semaphore_adjust (aTHX_ av, 0);
2076
2052 return 0; 2077 return 0;
2053 } 2078 }
2054 else 2079 else
2055 { 2080 {
2056 int i; 2081 int i;
… …
2065 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2090 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2066 return 1; 2091 return 1;
2067 } 2092 }
2068} 2093}
2069 2094
2070static void 2095static int
2096slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2097{
2098 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2099}
2100
2101static int
2102slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2103{
2104 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2105}
2106
2107static void
2071slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2108slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2072{ 2109{
2073 AV *av = (AV *)SvRV (arg [0]); 2110 AV *av = (AV *)SvRV (arg [0]);
2074 2111
2075 if (SvIVX (AvARRAY (av)[0]) > 0) 2112 if (SvIVX (AvARRAY (av)[0]) > 0)
2076 { 2113 {
… …
2086 2123
2087 /* to avoid race conditions when a woken-up coro gets terminated */ 2124 /* to avoid race conditions when a woken-up coro gets terminated */
2088 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2125 /* we arrange for a temporary on_destroy that calls adjust (0) */
2089 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2126 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2090 } 2127 }
2128}
2091 2129
2130static void
2131slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2132{
2133 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2092 frame->check = slf_check_semaphore_down; 2134 frame->check = slf_check_semaphore_down;
2135}
2093 2136
2137static void
2138slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2139{
2140 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2141 frame->check = slf_check_semaphore_wait;
2142}
2143
2144/* signal */
2145
2146static void
2147coro_signal_wake (pTHX_ AV *av, int count)
2148{
2149 SvIVX (AvARRAY (av)[0]) = 0;
2150
2151 /* now signal count waiters */
2152 while (count > 0 && AvFILLp (av) > 0)
2153 {
2154 SV *cb;
2155
2156 /* swap first two elements so we can shift a waiter */
2157 cb = AvARRAY (av)[0];
2158 AvARRAY (av)[0] = AvARRAY (av)[1];
2159 AvARRAY (av)[1] = cb;
2160
2161 cb = av_shift (av);
2162
2163 api_ready (cb);
2164 sv_setiv (cb, 0); /* signal waiter */
2165 SvREFCNT_dec (cb);
2166
2167 --count;
2168 }
2169}
2170
2171static int
2172slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
2173{
2174 /* if we are about to throw, also stop waiting */
2175 return SvROK ((SV *)frame->data) && !CORO_THROW;
2176}
2177
2178static void
2179slf_init_signal_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2180{
2181 AV *av = (AV *)SvRV (arg [0]);
2182
2183 if (SvIVX (AvARRAY (av)[0]))
2184 {
2185 SvIVX (AvARRAY (av)[0]) = 0;
2186 frame->prepare = prepare_nop;
2187 frame->check = slf_check_nop;
2188 }
2189 else
2190 {
2191 SV *waiter = newRV_inc (SvRV (coro_current)); /* owned by signal av */
2192
2193 av_push (av, waiter);
2194
2195 frame->data = (void *)sv_2mortal (SvREFCNT_inc_NN (waiter)); /* owned by process */
2196 frame->prepare = prepare_schedule;
2197 frame->check = slf_check_signal_wait;
2198 }
2094} 2199}
2095 2200
2096/*****************************************************************************/ 2201/*****************************************************************************/
2097/* gensub: simple closure generation utility */ 2202/* gensub: simple closure generation utility */
2098 2203
… …
2170 2275
2171static int 2276static int
2172slf_check_aio_req (pTHX_ struct CoroSLF *frame) 2277slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2173{ 2278{
2174 AV *state = (AV *)frame->data; 2279 AV *state = (AV *)frame->data;
2280
2281 /* if we are about to throw, return early */
2282 /* this does not cancel the aio request, but at least */
2283 /* it quickly returns */
2284 if (CORO_THROW)
2285 return 0;
2175 2286
2176 /* one element that is an RV? repeat! */ 2287 /* one element that is an RV? repeat! */
2177 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2288 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2178 return 1; 2289 return 1;
2179 2290
… …
2511throw (Coro::State self, SV *throw = &PL_sv_undef) 2622throw (Coro::State self, SV *throw = &PL_sv_undef)
2512 PROTOTYPE: $;$ 2623 PROTOTYPE: $;$
2513 CODE: 2624 CODE:
2514{ 2625{
2515 struct coro *current = SvSTATE_current; 2626 struct coro *current = SvSTATE_current;
2516 SV **throwp = self == current ? &coro_throw : &self->throw; 2627 SV **throwp = self == current ? &CORO_THROW : &self->except;
2517 SvREFCNT_dec (*throwp); 2628 SvREFCNT_dec (*throwp);
2518 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2629 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2519} 2630}
2520 2631
2521void 2632void
… …
2610 coroapi.ready = api_ready; 2721 coroapi.ready = api_ready;
2611 coroapi.is_ready = api_is_ready; 2722 coroapi.is_ready = api_is_ready;
2612 coroapi.nready = coro_nready; 2723 coroapi.nready = coro_nready;
2613 coroapi.current = coro_current; 2724 coroapi.current = coro_current;
2614 2725
2615 GCoroAPI = &coroapi; 2726 /*GCoroAPI = &coroapi;*/
2616 sv_setiv (sv, (IV)&coroapi); 2727 sv_setiv (sv, (IV)&coroapi);
2617 SvREADONLY_on (sv); 2728 SvREADONLY_on (sv);
2618 } 2729 }
2619} 2730}
2620 2731
… …
2766 2877
2767 2878
2768MODULE = Coro::State PACKAGE = Coro::Semaphore 2879MODULE = Coro::State PACKAGE = Coro::Semaphore
2769 2880
2770SV * 2881SV *
2771new (SV *klass, SV *count_ = 0) 2882new (SV *klass, SV *count = 0)
2772 CODE: 2883 CODE:
2773{ 2884 RETVAL = sv_bless (
2774 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2885 coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2775 AV *av = newAV (); 2886 GvSTASH (CvGV (cv))
2776 SV **ary; 2887 );
2888 OUTPUT:
2889 RETVAL
2777 2890
2778 /* unfortunately, building manually saves memory */ 2891# helper for Coro::Channel
2779 Newx (ary, 2, SV *); 2892SV *
2780 AvALLOC (av) = ary; 2893_alloc (int count)
2781 AvARRAY (av) = ary; 2894 CODE:
2782 AvMAX (av) = 1; 2895 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: 2896 OUTPUT:
2789 RETVAL 2897 RETVAL
2790 2898
2791SV * 2899SV *
2792count (SV *self) 2900count (SV *self)
… …
2804 2912
2805void 2913void
2806down (SV *self) 2914down (SV *self)
2807 CODE: 2915 CODE:
2808 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 2916 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
2917
2918void
2919wait (SV *self)
2920 CODE:
2921 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2809 2922
2810void 2923void
2811try (SV *self) 2924try (SV *self)
2812 PPCODE: 2925 PPCODE:
2813{ 2926{
… …
2825 XSRETURN_NO; 2938 XSRETURN_NO;
2826} 2939}
2827 2940
2828void 2941void
2829waiters (SV *self) 2942waiters (SV *self)
2830 CODE: 2943 PPCODE:
2831{ 2944{
2832 AV *av = (AV *)SvRV (self); 2945 AV *av = (AV *)SvRV (self);
2946 int wcount = AvFILLp (av) + 1 - 1;
2833 2947
2834 if (GIMME_V == G_SCALAR) 2948 if (GIMME_V == G_SCALAR)
2835 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 2949 XPUSHs (sv_2mortal (newSViv (wcount)));
2836 else 2950 else
2837 { 2951 {
2838 int i; 2952 int i;
2839 EXTEND (SP, AvFILLp (av) + 1 - 1); 2953 EXTEND (SP, wcount);
2840 for (i = 1; i <= AvFILLp (av); ++i) 2954 for (i = 1; i <= wcount; ++i)
2841 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 2955 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2842 } 2956 }
2843} 2957}
2958
2959MODULE = Coro::State PACKAGE = Coro::Signal
2960
2961SV *
2962new (SV *klass)
2963 CODE:
2964 RETVAL = sv_bless (
2965 coro_waitarray_new (aTHX_ 0),
2966 GvSTASH (CvGV (cv))
2967 );
2968 OUTPUT:
2969 RETVAL
2970
2971void
2972wait (SV *self)
2973 CODE:
2974 CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
2975
2976void
2977broadcast (SV *self)
2978 CODE:
2979{
2980 AV *av = (AV *)SvRV (self);
2981 coro_signal_wake (aTHX_ av, AvFILLp (av));
2982}
2983
2984void
2985send (SV *self)
2986 CODE:
2987{
2988 AV *av = (AV *)SvRV (self);
2989
2990 if (AvFILLp (av))
2991 coro_signal_wake (av, 1);
2992 else
2993 SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
2994}
2995
2996IV
2997awaited (SV *self)
2998 CODE:
2999 RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
3000 OUTPUT:
3001 RETVAL
2844 3002
2845 3003
2846MODULE = Coro::State PACKAGE = Coro::AnyEvent 3004MODULE = Coro::State PACKAGE = Coro::AnyEvent
2847 3005
2848BOOT: 3006BOOT:

Diff Legend

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