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.297 by root, Tue Nov 18 11:04:42 2008 UTC vs.
Revision 1.299 by root, Wed Nov 19 00:06:55 2008 UTC

… …
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) 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
… …
2772 2818
2773 2819
2774MODULE = Coro::State PACKAGE = Coro::Semaphore 2820MODULE = Coro::State PACKAGE = Coro::Semaphore
2775 2821
2776SV * 2822SV *
2777new (SV *klass, SV *count_ = 0) 2823new (SV *klass, SV *count = 0)
2778 CODE: 2824 CODE:
2779{ 2825 RETVAL = sv_bless (
2780 /* a semaphore contains a counter IV in $sem->[0] and any waiters after that */ 2826 coro_semaphore_new (count && SvOK (count) ? SvIV (count) : 1),
2781 AV *av = newAV (); 2827 GvSTASH (CvGV (cv))
2782 SV **ary; 2828 );
2829 OUTPUT:
2830 RETVAL
2783 2831
2784 /* unfortunately, building manually saves memory */ 2832# helper for Coro::Channel
2785 Newx (ary, 2, SV *); 2833SV *
2786 AvALLOC (av) = ary; 2834_alloc (int count)
2787 AvARRAY (av) = ary; 2835 CODE:
2788 AvMAX (av) = 1; 2836 RETVAL = coro_semaphore_new (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: 2837 OUTPUT:
2795 RETVAL 2838 RETVAL
2796 2839
2797SV * 2840SV *
2798count (SV *self) 2841count (SV *self)
… …
2808 CODE: 2851 CODE:
2809 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1); 2852 coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
2810 2853
2811void 2854void
2812down (SV *self) 2855down (SV *self)
2856 ALIAS:
2857 Coro::Signal::wait = 0
2813 CODE: 2858 CODE:
2814 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);
2815 2865
2816void 2866void
2817try (SV *self) 2867try (SV *self)
2818 PPCODE: 2868 PPCODE:
2819{ 2869{
… …
2831 XSRETURN_NO; 2881 XSRETURN_NO;
2832} 2882}
2833 2883
2834void 2884void
2835waiters (SV *self) 2885waiters (SV *self)
2836 CODE: 2886 PPCODE:
2837{ 2887{
2838 AV *av = (AV *)SvRV (self); 2888 AV *av = (AV *)SvRV (self);
2889 int wcount = AvFILLp (av) + 1 - 1;
2839 2890
2840 if (GIMME_V == G_SCALAR) 2891 if (GIMME_V == G_SCALAR)
2841 XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0]))); 2892 XPUSHs (sv_2mortal (newSViv (wcount)));
2842 else 2893 else
2843 { 2894 {
2844 int i; 2895 int i;
2845 EXTEND (SP, AvFILLp (av) + 1 - 1); 2896 EXTEND (SP, wcount);
2846 for (i = 1; i <= AvFILLp (av); ++i) 2897 for (i = 1; i <= wcount; ++i)
2847 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i]))); 2898 PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2848 } 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 */
2849} 2922}
2850 2923
2851 2924
2852MODULE = Coro::State PACKAGE = Coro::AnyEvent 2925MODULE = Coro::State PACKAGE = Coro::AnyEvent
2853 2926

Diff Legend

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