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.294 by root, Tue Nov 18 08:31:03 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
2065 /* if we are about to throw, don't actually acquire the lock, just throw */
2066 if (CORO_THROW)
2067 return 0;
2045 if (SvIVX (count_sv) > 0) 2068 else if (SvIVX (count_sv) > 0)
2046 { 2069 {
2047 SvSTATE_current->on_destroy = 0; 2070 SvSTATE_current->on_destroy = 0;
2071
2072 if (acquire)
2048 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
2049 return 0; 2077 return 0;
2050 } 2078 }
2051 else 2079 else
2052 { 2080 {
2053 int i; 2081 int i;
… …
2062 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2090 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2063 return 1; 2091 return 1;
2064 } 2092 }
2065} 2093}
2066 2094
2067static 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
2068slf_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)
2069{ 2109{
2070 AV *av = (AV *)SvRV (arg [0]); 2110 AV *av = (AV *)SvRV (arg [0]);
2071 2111
2072 if (SvIVX (AvARRAY (av)[0]) > 0) 2112 if (SvIVX (AvARRAY (av)[0]) > 0)
2073 { 2113 {
2074 frame->data = (void *)av; 2114 frame->data = (void *)av;
2075 frame->prepare = prepare_nop; 2115 frame->prepare = prepare_nop;
2076 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2077 } 2116 }
2078 else 2117 else
2079 { 2118 {
2080 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2119 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2081 2120
2082 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av)); 2121 frame->data = (void *)sv_2mortal (SvREFCNT_inc ((SV *)av));
2083 frame->prepare = prepare_schedule; 2122 frame->prepare = prepare_schedule;
2084 2123
2085 /* to avoid race conditions when a woken-up coro gets terminated */ 2124 /* to avoid race conditions when a woken-up coro gets terminated */
2086 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2125 /* we arrange for a temporary on_destroy that calls adjust (0) */
2087 assert (!SvSTATE_current->on_destroy);//D
2088 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2126 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2089 } 2127 }
2128}
2090 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);
2091 frame->check = slf_check_semaphore_down; 2134 frame->check = slf_check_semaphore_down;
2135}
2092 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 }
2093} 2199}
2094 2200
2095/*****************************************************************************/ 2201/*****************************************************************************/
2096/* gensub: simple closure generation utility */ 2202/* gensub: simple closure generation utility */
2097 2203
… …
2169 2275
2170static int 2276static int
2171slf_check_aio_req (pTHX_ struct CoroSLF *frame) 2277slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2172{ 2278{
2173 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;
2174 2286
2175 /* one element that is an RV? repeat! */ 2287 /* one element that is an RV? repeat! */
2176 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2288 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2177 return 1; 2289 return 1;
2178 2290
… …
2510throw (Coro::State self, SV *throw = &PL_sv_undef) 2622throw (Coro::State self, SV *throw = &PL_sv_undef)
2511 PROTOTYPE: $;$ 2623 PROTOTYPE: $;$
2512 CODE: 2624 CODE:
2513{ 2625{
2514 struct coro *current = SvSTATE_current; 2626 struct coro *current = SvSTATE_current;
2515 SV **throwp = self == current ? &coro_throw : &self->throw; 2627 SV **throwp = self == current ? &CORO_THROW : &self->except;
2516 SvREFCNT_dec (*throwp); 2628 SvREFCNT_dec (*throwp);
2517 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2629 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2518} 2630}
2519 2631
2520void 2632void
… …
2609 coroapi.ready = api_ready; 2721 coroapi.ready = api_ready;
2610 coroapi.is_ready = api_is_ready; 2722 coroapi.is_ready = api_is_ready;
2611 coroapi.nready = coro_nready; 2723 coroapi.nready = coro_nready;
2612 coroapi.current = coro_current; 2724 coroapi.current = coro_current;
2613 2725
2614 GCoroAPI = &coroapi; 2726 /*GCoroAPI = &coroapi;*/
2615 sv_setiv (sv, (IV)&coroapi); 2727 sv_setiv (sv, (IV)&coroapi);
2616 SvREADONLY_on (sv); 2728 SvREADONLY_on (sv);
2617 } 2729 }
2618} 2730}
2619 2731
… …
2765 2877
2766 2878
2767MODULE = Coro::State PACKAGE = Coro::Semaphore 2879MODULE = Coro::State PACKAGE = Coro::Semaphore
2768 2880
2769SV * 2881SV *
2770new (SV *klass, SV *count_ = 0) 2882new (SV *klass, SV *count = 0)
2771 CODE: 2883 CODE:
2772{ 2884 RETVAL = sv_bless (
2773 /* 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),
2774 AV *av = newAV (); 2886 GvSTASH (CvGV (cv))
2775 SV **ary; 2887 );
2888 OUTPUT:
2889 RETVAL
2776 2890
2777 /* unfortunately, building manually saves memory */ 2891# helper for Coro::Channel
2778 Newx (ary, 2, SV *); 2892SV *
2779 AvALLOC (av) = ary; 2893_alloc (int count)
2780 AvARRAY (av) = ary; 2894 CODE:
2781 AvMAX (av) = 1; 2895 RETVAL = coro_waitarray_new (aTHX_ count);
2782 AvFILLp (av) = 0;
2783 ary [0] = newSViv (count_ && SvOK (count_) ? SvIV (count_) : 1);
2784
2785 RETVAL = sv_bless (newRV_noinc ((SV *)av), GvSTASH (CvGV (cv)));
2786}
2787 OUTPUT: 2896 OUTPUT:
2788 RETVAL 2897 RETVAL
2789 2898
2790SV * 2899SV *
2791count (SV *self) 2900count (SV *self)
… …
2803 2912
2804void 2913void
2805down (SV *self) 2914down (SV *self)
2806 CODE: 2915 CODE:
2807 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);
2808 2922
2809void 2923void
2810try (SV *self) 2924try (SV *self)
2811 PPCODE: 2925 PPCODE:
2812{ 2926{
… …
2824 XSRETURN_NO; 2938 XSRETURN_NO;
2825} 2939}
2826 2940
2827void 2941void
2828waiters (SV *self) 2942waiters (SV *self)
2829 CODE: 2943 PPCODE:
2830{ 2944{
2831 AV *av = (AV *)SvRV (self); 2945 AV *av = (AV *)SvRV (self);
2946 int wcount = AvFILLp (av) + 1 - 1;
2832 2947
2833 if (GIMME_V == G_SCALAR) 2948 if (GIMME_V == G_SCALAR)
2834 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 2949 XPUSHs (sv_2mortal (newSViv (wcount)));
2835 else 2950 else
2836 { 2951 {
2837 int i; 2952 int i;
2838 EXTEND (SP, AvFILLp (av) + 1 - 1); 2953 EXTEND (SP, wcount);
2839 for (i = 1; i <= AvFILLp (av); ++i) 2954 for (i = 1; i <= wcount; ++i)
2840 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 2955 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2841 } 2956 }
2842} 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
2843 3002
2844 3003
2845MODULE = Coro::State PACKAGE = Coro::AnyEvent 3004MODULE = Coro::State PACKAGE = Coro::AnyEvent
2846 3005
2847BOOT: 3006BOOT:

Diff Legend

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