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.299 by root, Wed Nov 19 00:06:55 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
… …
1997}; 1997};
1998 1998
1999/*****************************************************************************/ 1999/*****************************************************************************/
2000/* Coro::Semaphore */ 2000/* Coro::Semaphore */
2001 2001
2002static SV *
2003coro_semaphore_new (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
2002static void 2020static void
2003coro_semaphore_adjust (pTHX_ AV *av, IV adjust) 2021coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2004{ 2022{
2005 SV *count_sv = AvARRAY (av)[0]; 2023 SV *count_sv = AvARRAY (av)[0];
2006 IV count = SvIVX (count_sv); 2024 IV count = SvIVX (count_sv);
… …
2035 /* call $sem->adjust (0) to possibly wake up some other waiters */ 2053 /* call $sem->adjust (0) to possibly wake up some other waiters */
2036 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0); 2054 coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2037} 2055}
2038 2056
2039static int 2057static int
2040slf_check_semaphore_down (pTHX_ struct CoroSLF *frame) 2058slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2041{ 2059{
2042 AV *av = (AV *)frame->data; 2060 AV *av = (AV *)frame->data;
2043 SV *count_sv = AvARRAY (av)[0]; 2061 SV *count_sv = AvARRAY (av)[0];
2044 2062
2045 /* if we are about to throw, don't actually acquire the lock, just throw */ 2063 /* if we are about to throw, don't actually acquire the lock, just throw */
2046 if (coro_throw && 0)//D 2064 if (CORO_THROW)
2047 return 0; 2065 return 0;
2048 else if (SvIVX (count_sv) > 0) 2066 else if (SvIVX (count_sv) > 0)
2049 { 2067 {
2050 SvSTATE_current->on_destroy = 0; 2068 SvSTATE_current->on_destroy = 0;
2069
2070 if (acquire)
2051 SvIVX (count_sv) = SvIVX (count_sv) - 1; 2071 SvIVX (count_sv) = SvIVX (count_sv) - 1;
2072 else
2073 coro_semaphore_adjust (aTHX_ av, 0);
2074
2052 return 0; 2075 return 0;
2053 } 2076 }
2054 else 2077 else
2055 { 2078 {
2056 int i; 2079 int i;
… …
2065 av_push (av, SvREFCNT_inc (SvRV (coro_current))); 2088 av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2066 return 1; 2089 return 1;
2067 } 2090 }
2068} 2091}
2069 2092
2070static void 2093static int
2094slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
2095{
2096 return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
2097}
2098
2099static int
2100slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
2101{
2102 return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
2103}
2104
2105static void
2071slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items) 2106slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2072{ 2107{
2073 AV *av = (AV *)SvRV (arg [0]); 2108 AV *av = (AV *)SvRV (arg [0]);
2074 2109
2075 if (SvIVX (AvARRAY (av)[0]) > 0) 2110 if (SvIVX (AvARRAY (av)[0]) > 0)
2076 { 2111 {
… …
2086 2121
2087 /* to avoid race conditions when a woken-up coro gets terminated */ 2122 /* to avoid race conditions when a woken-up coro gets terminated */
2088 /* we arrange for a temporary on_destroy that calls adjust (0) */ 2123 /* we arrange for a temporary on_destroy that calls adjust (0) */
2089 SvSTATE_current->on_destroy = coro_semaphore_on_destroy; 2124 SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2090 } 2125 }
2126}
2091 2127
2128static void
2129slf_init_semaphore_down (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2130{
2131 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2092 frame->check = slf_check_semaphore_down; 2132 frame->check = slf_check_semaphore_down;
2133}
2093 2134
2135static void
2136slf_init_semaphore_wait (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
2137{
2138 slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2139 frame->check = slf_check_semaphore_wait;
2094} 2140}
2095 2141
2096/*****************************************************************************/ 2142/*****************************************************************************/
2097/* gensub: simple closure generation utility */ 2143/* gensub: simple closure generation utility */
2098 2144
… …
2170 2216
2171static int 2217static int
2172slf_check_aio_req (pTHX_ struct CoroSLF *frame) 2218slf_check_aio_req (pTHX_ struct CoroSLF *frame)
2173{ 2219{
2174 AV *state = (AV *)frame->data; 2220 AV *state = (AV *)frame->data;
2221
2222 /* if we are about to throw, return early */
2223 /* this does not cancel the aio request, but at least */
2224 /* it quickly returns */
2225 if (CORO_THROW)
2226 return 0;
2175 2227
2176 /* one element that is an RV? repeat! */ 2228 /* one element that is an RV? repeat! */
2177 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0])) 2229 if (AvFILLp (state) == 0 && SvROK (AvARRAY (state)[0]))
2178 return 1; 2230 return 1;
2179 2231
… …
2511throw (Coro::State self, SV *throw = &PL_sv_undef) 2563throw (Coro::State self, SV *throw = &PL_sv_undef)
2512 PROTOTYPE: $;$ 2564 PROTOTYPE: $;$
2513 CODE: 2565 CODE:
2514{ 2566{
2515 struct coro *current = SvSTATE_current; 2567 struct coro *current = SvSTATE_current;
2516 SV **throwp = self == current ? &coro_throw : &self->throw; 2568 SV **throwp = self == current ? &CORO_THROW : &self->except;
2517 SvREFCNT_dec (*throwp); 2569 SvREFCNT_dec (*throwp);
2518 *throwp = SvOK (throw) ? newSVsv (throw) : 0; 2570 *throwp = SvOK (throw) ? newSVsv (throw) : 0;
2519} 2571}
2520 2572
2521void 2573void
… …
2610 coroapi.ready = api_ready; 2662 coroapi.ready = api_ready;
2611 coroapi.is_ready = api_is_ready; 2663 coroapi.is_ready = api_is_ready;
2612 coroapi.nready = coro_nready; 2664 coroapi.nready = coro_nready;
2613 coroapi.current = coro_current; 2665 coroapi.current = coro_current;
2614 2666
2615 GCoroAPI = &coroapi; 2667 /*GCoroAPI = &coroapi;*/
2616 sv_setiv (sv, (IV)&coroapi); 2668 sv_setiv (sv, (IV)&coroapi);
2617 SvREADONLY_on (sv); 2669 SvREADONLY_on (sv);
2618 } 2670 }
2619} 2671}
2620 2672
… …
2766 2818
2767 2819
2768MODULE = Coro::State PACKAGE = Coro::Semaphore 2820MODULE = Coro::State PACKAGE = Coro::Semaphore
2769 2821
2770SV * 2822SV *
2771new (SV *klass, SV *count_ = 0) 2823new (SV *klass, SV *count = 0)
2772 CODE: 2824 CODE:
2773{ 2825 RETVAL = sv_bless (
2774 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2826 coro_semaphore_new (count && SvOK (count) ? SvIV (count) : 1),
2775 AV *av = newAV (); 2827 GvSTASH (CvGV (cv))
2776 SV **ary; 2828 );
2829 OUTPUT:
2830 RETVAL
2777 2831
2778 /* unfortunately, building manually saves memory */ 2832# helper for Coro::Channel
2779 Newx (ary, 2, SV *); 2833SV *
2780 AvALLOC (av) = ary; 2834_alloc (int count)
2781 AvARRAY (av) = ary; 2835 CODE:
2782 AvMAX (av) = 1; 2836 RETVAL = coro_semaphore_new (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: 2837 OUTPUT:
2789 RETVAL 2838 RETVAL
2790 2839
2791SV * 2840SV *
2792count (SV *self) 2841count (SV *self)
… …
2802 CODE: 2851 CODE:
2803 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1); 2852 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
2804 2853
2805void 2854void
2806down (SV *self) 2855down (SV *self)
2856 ALIAS:
2857 Coro::Signal::wait = 0
2807 CODE: 2858 CODE:
2808 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down); 2859 CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
2860
2861void
2862wait (SV *self)
2863 CODE:
2864 CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2809 2865
2810void 2866void
2811try (SV *self) 2867try (SV *self)
2812 PPCODE: 2868 PPCODE:
2813{ 2869{
… …
2825 XSRETURN_NO; 2881 XSRETURN_NO;
2826} 2882}
2827 2883
2828void 2884void
2829waiters (SV *self) 2885waiters (SV *self)
2830 CODE: 2886 PPCODE:
2831{ 2887{
2832 AV *av = (AV *)SvRV (self); 2888 AV *av = (AV *)SvRV (self);
2889 int wcount = AvFILLp (av) + 1 - 1;
2833 2890
2834 if (GIMME_V == G_SCALAR) 2891 if (GIMME_V == G_SCALAR)
2835 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 2892 XPUSHs (sv_2mortal (newSViv (wcount)));
2836 else 2893 else
2837 { 2894 {
2838 int i; 2895 int i;
2839 EXTEND (SP, AvFILLp (av) + 1 - 1); 2896 EXTEND (SP, wcount);
2840 for (i = 1; i <= AvFILLp (av); ++i) 2897 for (i = 1; i <= wcount; ++i)
2841 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 2898 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2842 } 2899 }
2900}
2901
2902MODULE = Coro::State PACKAGE = Coro::Signal
2903
2904SV *
2905new (SV *klass)
2906 CODE:
2907 RETVAL = sv_bless (
2908 coro_semaphore_new (0),
2909 GvSTASH (CvGV (cv))
2910 );
2911 OUTPUT:
2912 RETVAL
2913
2914void
2915broadcast (SV *self, int adjust = 1)
2916 CODE:
2917{
2918 AV *av = (AV *)SvRV (self);
2919 SvIVX (AvARRAY (av)[0]) = 0; /* not necessary, but gives me fuzzy warm feelings */
2920 coro_semaphore_adjust (aTHX_ av, AvFILLp (av) + 1 - 1);
2921 SvIVX (AvARRAY (av)[0]) = 0; /* necessary */
2843} 2922}
2844 2923
2845 2924
2846MODULE = Coro::State PACKAGE = Coro::AnyEvent 2925MODULE = Coro::State PACKAGE = Coro::AnyEvent
2847 2926

Diff Legend

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