[ViewVC] Diff of: cvs/Coro/Coro/State.xs

Comparing Coro/Coro/State.xs (file contents):
Revision 1.297 by root, Tue Nov 18 11:04:42 2008 UTC vs.
Revision 1.303 by root, Wed Nov 19 05:52:42 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
…		…
258	int usecount; /* number of transfers to this coro */	258	int usecount; /* number of transfers to this coro */
259		259
260	/* coro process data */	260	/* coro process data */
261	int prio;	261	int prio;
262	SV except; / exception to be thrown */	262	SV except; / exception to be thrown */
		263	SV *rouse_cb;
263		264
264	/* async_pool */	265	/* async_pool */
265	SV *saved_deffh;	266	SV *saved_deffh;
266		267
267	/* linked list */	268	/* linked list */
…		…
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
929	INLINE void	931	INLINE 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 */
		1512	static SV *
		1513	gensub (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
1504	INLINE void	1529	INLINE void
1505	coro_enq (pTHX_ struct coro *coro)	1530	coro_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
		1740	static void
		1741	coro_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	/* better take a full copy of the arguments */
		1758	while (items--)
		1759	av_store (av, items, newSVsv (ST (items)));
		1760	}
		1761
		1762	XSRETURN_EMPTY;
		1763	}
		1764
		1765	static int
		1766	slf_check_rouse_wait (pTHX_ struct CoroSLF *frame)
		1767	{
		1768	SV data = (SV )frame->data;
		1769
		1770	if (CORO_THROW)
		1771	return 0;
		1772
		1773	if (SvTYPE (SvRV (data)) != SVt_PVAV)
		1774	return 1;
		1775
		1776	/* now push all results on the stack */
		1777	{
		1778	dSP;
		1779	AV av = (AV )SvRV (data);
		1780	int i;
		1781
		1782	EXTEND (SP, AvFILLp (av) + 1);
		1783	for (i = 0; i <= AvFILLp (av); ++i)
		1784	PUSHs (sv_2mortal (AvARRAY (av)[i]));
		1785
		1786	/* we have stolen the elements, so ste length to zero and free */
		1787	AvFILLp (av) = -1;
		1788	av_undef (av);
		1789
		1790	PUTBACK;
		1791	}
		1792
		1793	return 0;
		1794	}
		1795
		1796	static void
		1797	slf_init_rouse_wait (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
		1798	{
		1799	SV *cb;
		1800
		1801	if (items)
		1802	cb = arg [0];
		1803	else
		1804	{
		1805	struct coro *coro = SvSTATE_current;
		1806
		1807	if (!coro->rouse_cb)
		1808	croak ("Coro::rouse_wait called without rouse callback, and no default rouse callback found either,");
		1809
		1810	cb = sv_2mortal (coro->rouse_cb);
		1811	coro->rouse_cb = 0;
		1812	}
		1813
		1814	if (!SvROK (cb)
		1815	\|\| SvTYPE (SvRV (cb)) != SVt_PVCV
		1816	\|\| CvXSUB ((CV *)SvRV (cb)) != coro_rouse_callback)
		1817	croak ("Coro::rouse_wait called with illegal callback argument,");
		1818
		1819	{
		1820	CV cv = (CV )SvRV (cb); /* for GENSUB_ARG */
		1821	SV data = (SV )GENSUB_ARG;
		1822
		1823	frame->data = (void *)data;
		1824	frame->prepare = SvTYPE (SvRV (data)) == SVt_PVAV ? prepare_nop : prepare_schedule;
		1825	frame->check = slf_check_rouse_wait;
		1826	}
		1827	}
		1828
		1829	static SV *
		1830	coro_new_rouse_cb (pTHX)
		1831	{
		1832	HV hv = (HV )SvRV (coro_current);
		1833	struct coro *coro = SvSTATE_hv (hv);
		1834	SV data = newRV_inc ((SV )hv);
		1835	SV cb = gensub (aTHX_ coro_rouse_callback, (void )data);
		1836
		1837	sv_magicext (SvRV (cb), data, CORO_MAGIC_type_rouse, 0, 0, 0);
		1838	SvREFCNT_dec (data); /* magicext increases the refcount */
		1839
		1840	SvREFCNT_dec (coro->rouse_cb);
		1841	coro->rouse_cb = SvREFCNT_inc_NN (cb);
		1842
		1843	return cb;
1708	}	1844	}
1709		1845
1710	/*****************************************************************************/	1846	/*****************************************************************************/
1711	/* schedule-like-function opcode (SLF) */	1847	/* schedule-like-function opcode (SLF) */
1712		1848
…		…
1995	PerlIOBuf_get_cnt,	2131	PerlIOBuf_get_cnt,
1996	PerlIOBuf_set_ptrcnt,	2132	PerlIOBuf_set_ptrcnt,
1997	};	2133	};
1998		2134
1999	/*****************************************************************************/	2135	/*****************************************************************************/
		2136	/* Coro::Semaphore & Coro::Signal */
		2137
		2138	static SV *
		2139	coro_waitarray_new (pTHX_ int count)
		2140	{
		2141	/* a semaphore contains a counter IV in $sem->[0] and any waiters after that */
		2142	AV *av = newAV ();
		2143	SV **ary;
		2144
		2145	/* unfortunately, building manually saves memory */
		2146	Newx (ary, 2, SV *);
		2147	AvALLOC (av) = ary;
		2148	AvARRAY (av) = ary;
		2149	AvMAX (av) = 1;
		2150	AvFILLp (av) = 0;
		2151	ary [0] = newSViv (count);
		2152
		2153	return newRV_noinc ((SV *)av);
		2154	}
		2155
2000	/* Coro::Semaphore */	2156	/* semaphore */
2001		2157
2002	static void	2158	static void
2003	coro_semaphore_adjust (pTHX_ AV *av, IV adjust)	2159	coro_semaphore_adjust (pTHX_ AV *av, IV adjust)
2004	{	2160	{
2005	SV *count_sv = AvARRAY (av)[0];	2161	SV *count_sv = AvARRAY (av)[0];
…		…
2035	/* call $sem->adjust (0) to possibly wake up some other waiters */	2191	/* call $sem->adjust (0) to possibly wake up some other waiters */
2036	coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);	2192	coro_semaphore_adjust (aTHX_ (AV *)coro->slf_frame.data, 0);
2037	}	2193	}
2038		2194
2039	static int	2195	static int
2040	slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)	2196	slf_check_semaphore_down_or_wait (pTHX_ struct CoroSLF *frame, int acquire)
2041	{	2197	{
2042	AV av = (AV )frame->data;	2198	AV av = (AV )frame->data;
2043	SV *count_sv = AvARRAY (av)[0];	2199	SV *count_sv = AvARRAY (av)[0];
2044		2200
2045	/* if we are about to throw, don't actually acquire the lock, just throw */	2201	/* if we are about to throw, don't actually acquire the lock, just throw */
2046	if (CORO_THROW)	2202	if (CORO_THROW)
2047	return 0;	2203	return 0;
2048	else if (SvIVX (count_sv) > 0)	2204	else if (SvIVX (count_sv) > 0)
2049	{	2205	{
2050	SvSTATE_current->on_destroy = 0;	2206	SvSTATE_current->on_destroy = 0;
		2207
		2208	if (acquire)
2051	SvIVX (count_sv) = SvIVX (count_sv) - 1;	2209	SvIVX (count_sv) = SvIVX (count_sv) - 1;
		2210	else
		2211	coro_semaphore_adjust (aTHX_ av, 0);
		2212
2052	return 0;	2213	return 0;
2053	}	2214	}
2054	else	2215	else
2055	{	2216	{
2056	int i;	2217	int i;
…		…
2065	av_push (av, SvREFCNT_inc (SvRV (coro_current)));	2226	av_push (av, SvREFCNT_inc (SvRV (coro_current)));
2066	return 1;	2227	return 1;
2067	}	2228	}
2068	}	2229	}
2069		2230
2070	static void	2231	static int
		2232	slf_check_semaphore_down (pTHX_ struct CoroSLF *frame)
		2233	{
		2234	return slf_check_semaphore_down_or_wait (aTHX_ frame, 1);
		2235	}
		2236
		2237	static int
		2238	slf_check_semaphore_wait (pTHX_ struct CoroSLF *frame)
		2239	{
		2240	return slf_check_semaphore_down_or_wait (aTHX_ frame, 0);
		2241	}
		2242
		2243	static void
2071	slf_init_semaphore_down (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)	2244	slf_init_semaphore_down_or_wait (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
2072	{	2245	{
2073	AV av = (AV )SvRV (arg [0]);	2246	AV av = (AV )SvRV (arg [0]);
2074		2247
2075	if (SvIVX (AvARRAY (av)[0]) > 0)	2248	if (SvIVX (AvARRAY (av)[0]) > 0)
2076	{	2249	{
…		…
2086		2259
2087	/* to avoid race conditions when a woken-up coro gets terminated */	2260	/* to avoid race conditions when a woken-up coro gets terminated */
2088	/* we arrange for a temporary on_destroy that calls adjust (0) */	2261	/* we arrange for a temporary on_destroy that calls adjust (0) */
2089	SvSTATE_current->on_destroy = coro_semaphore_on_destroy;	2262	SvSTATE_current->on_destroy = coro_semaphore_on_destroy;
2090	}	2263	}
		2264	}
2091		2265
		2266	static void
		2267	slf_init_semaphore_down (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
		2268	{
		2269	slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2092	frame->check = slf_check_semaphore_down;	2270	frame->check = slf_check_semaphore_down;
2093
2094	}	2271	}
2095		2272
2096	/*****************************************************************************/	2273	static void
2097	/* gensub: simple closure generation utility */	2274	slf_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 */
2104	static SV *
2105	gensub (pTHX_ void (xsub)(pTHX_ CV ), void *arg)
2106	{	2275	{
2107	CV cv = (CV )newSV (0);	2276	slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
		2277	frame->check = slf_check_semaphore_wait;
		2278	}
2108		2279
2109	sv_upgrade ((SV *)cv, SVt_PVCV);	2280	/* signal */
2110		2281
2111	CvANON_on (cv);	2282	static void
2112	CvISXSUB_on (cv);	2283	coro_signal_wake (pTHX_ AV *av, int count)
2113	CvXSUB (cv) = xsub;	2284	{
2114	GENSUB_ARG = arg;	2285	SvIVX (AvARRAY (av)[0]) = 0;
2115		2286
2116	return newRV_noinc ((SV *)cv);	2287	/* now signal count waiters */
		2288	while (count > 0 && AvFILLp (av) > 0)
		2289	{
		2290	SV *cb;
		2291
		2292	/* swap first two elements so we can shift a waiter */
		2293	cb = AvARRAY (av)[0];
		2294	AvARRAY (av)[0] = AvARRAY (av)[1];
		2295	AvARRAY (av)[1] = cb;
		2296
		2297	cb = av_shift (av);
		2298
		2299	api_ready (aTHX_ cb);
		2300	sv_setiv (cb, 0); /* signal waiter */
		2301	SvREFCNT_dec (cb);
		2302
		2303	--count;
		2304	}
		2305	}
		2306
		2307	static int
		2308	slf_check_signal_wait (pTHX_ struct CoroSLF *frame)
		2309	{
		2310	/* if we are about to throw, also stop waiting */
		2311	return SvROK ((SV *)frame->data) && !CORO_THROW;
		2312	}
		2313
		2314	static void
		2315	slf_init_signal_wait (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
		2316	{
		2317	AV av = (AV )SvRV (arg [0]);
		2318
		2319	if (SvIVX (AvARRAY (av)[0]))
		2320	{
		2321	SvIVX (AvARRAY (av)[0]) = 0;
		2322	frame->prepare = prepare_nop;
		2323	frame->check = slf_check_nop;
		2324	}
		2325	else
		2326	{
		2327	SV waiter = newRV_inc (SvRV (coro_current)); / owned by signal av */
		2328
		2329	av_push (av, waiter);
		2330
		2331	frame->data = (void )sv_2mortal (SvREFCNT_inc_NN (waiter)); / owned by process */
		2332	frame->prepare = prepare_schedule;
		2333	frame->check = slf_check_signal_wait;
		2334	}
2117	}	2335	}
2118		2336
2119	/*****************************************************************************/	2337	/*****************************************************************************/
2120	/* Coro::AIO */	2338	/* Coro::AIO */
2121		2339
…		…
2762	api_trace (aTHX_ coro_current, 0);	2980	api_trace (aTHX_ coro_current, 0);
2763		2981
2764	av_push (av_async_pool, newSVsv (coro_current));	2982	av_push (av_async_pool, newSVsv (coro_current));
2765	}	2983	}
2766		2984
		2985	SV *
		2986	rouse_cb ()
		2987	PROTOTYPE:
		2988	CODE:
		2989	RETVAL = coro_new_rouse_cb (aTHX);
		2990	OUTPUT:
		2991	RETVAL
		2992
		2993	void
		2994	rouse_wait (SV *cb = 0)
		2995	PROTOTYPE: ;$
		2996	PPCODE:
		2997	CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
		2998
2767		2999
2768	MODULE = Coro::State PACKAGE = PerlIO::cede	3000	MODULE = Coro::State PACKAGE = PerlIO::cede
2769		3001
2770	BOOT:	3002	BOOT:
2771	PerlIO_define_layer (aTHX_ &PerlIO_cede);	3003	PerlIO_define_layer (aTHX_ &PerlIO_cede);
2772		3004
2773		3005
2774	MODULE = Coro::State PACKAGE = Coro::Semaphore	3006	MODULE = Coro::State PACKAGE = Coro::Semaphore
2775		3007
2776	SV *	3008	SV *
2777	new (SV klass, SV count_ = 0)	3009	new (SV klass, SV count = 0)
2778	CODE:	3010	CODE:
2779	{	3011	RETVAL = sv_bless (
2780	/* a semaphore contains a counter IV in $sem->[0] and any waiters after that */	3012	coro_waitarray_new (aTHX_ count && SvOK (count) ? SvIV (count) : 1),
2781	AV *av = newAV ();	3013	GvSTASH (CvGV (cv))
2782	SV **ary;	3014	);
		3015	OUTPUT:
		3016	RETVAL
2783		3017
2784	/* unfortunately, building manually saves memory */	3018	# helper for Coro::Channel
2785	Newx (ary, 2, SV *);	3019	SV *
2786	AvALLOC (av) = ary;	3020	_alloc (int count)
2787	AvARRAY (av) = ary;	3021	CODE:
2788	AvMAX (av) = 1;	3022	RETVAL = coro_waitarray_new (aTHX_ 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:	3023	OUTPUT:
2795	RETVAL	3024	RETVAL
2796		3025
2797	SV *	3026	SV *
2798	count (SV *self)	3027	count (SV *self)
…		…
2810		3039
2811	void	3040	void
2812	down (SV *self)	3041	down (SV *self)
2813	CODE:	3042	CODE:
2814	CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);	3043	CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
		3044
		3045	void
		3046	wait (SV *self)
		3047	CODE:
		3048	CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
2815		3049
2816	void	3050	void
2817	try (SV *self)	3051	try (SV *self)
2818	PPCODE:	3052	PPCODE:
2819	{	3053	{
…		…
2831	XSRETURN_NO;	3065	XSRETURN_NO;
2832	}	3066	}
2833		3067
2834	void	3068	void
2835	waiters (SV *self)	3069	waiters (SV *self)
2836	CODE:	3070	PPCODE:
2837	{	3071	{
2838	AV av = (AV )SvRV (self);	3072	AV av = (AV )SvRV (self);
		3073	int wcount = AvFILLp (av) + 1 - 1;
2839		3074
2840	if (GIMME_V == G_SCALAR)	3075	if (GIMME_V == G_SCALAR)
2841	XPUSHs (sv_2mortal (newSVsv (AvARRAY (av)[0])));	3076	XPUSHs (sv_2mortal (newSViv (wcount)));
2842	else	3077	else
2843	{	3078	{
2844	int i;	3079	int i;
2845	EXTEND (SP, AvFILLp (av) + 1 - 1);	3080	EXTEND (SP, wcount);
2846	for (i = 1; i <= AvFILLp (av); ++i)	3081	for (i = 1; i <= wcount; ++i)
2847	PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));	3082	PUSHs (sv_2mortal (newRV_inc (AvARRAY (av)[i])));
2848	}	3083	}
2849	}	3084	}
		3085
		3086	MODULE = Coro::State PACKAGE = Coro::Signal
		3087
		3088	SV *
		3089	new (SV *klass)
		3090	CODE:
		3091	RETVAL = sv_bless (
		3092	coro_waitarray_new (aTHX_ 0),
		3093	GvSTASH (CvGV (cv))
		3094	);
		3095	OUTPUT:
		3096	RETVAL
		3097
		3098	void
		3099	wait (SV *self)
		3100	CODE:
		3101	CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
		3102
		3103	void
		3104	broadcast (SV *self)
		3105	CODE:
		3106	{
		3107	AV av = (AV )SvRV (self);
		3108	coro_signal_wake (aTHX_ av, AvFILLp (av));
		3109	}
		3110
		3111	void
		3112	send (SV *self)
		3113	CODE:
		3114	{
		3115	AV av = (AV )SvRV (self);
		3116
		3117	if (AvFILLp (av))
		3118	coro_signal_wake (aTHX_ av, 1);
		3119	else
		3120	SvIVX (AvARRAY (av)[0]) = 1; /* remember the signal */
		3121	}
		3122
		3123	IV
		3124	awaited (SV *self)
		3125	CODE:
		3126	RETVAL = AvFILLp ((AV *)SvRV (self)) + 1 - 1;
		3127	OUTPUT:
		3128	RETVAL
2850		3129
2851		3130
2852	MODULE = Coro::State PACKAGE = Coro::AnyEvent	3131	MODULE = Coro::State PACKAGE = Coro::AnyEvent
2853		3132
2854	BOOT:	3133	BOOT:

Diff Legend

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

Comparing Coro/Coro/State.xs (file contents): Revision 1.297 by root, Tue Nov 18 11:04:42 2008 UTC vs. Revision 1.303 by root, Wed Nov 19 05:52:42 2008 UTC

Diff Legend

Comparing Coro/Coro/State.xs (file contents):
Revision 1.297 by root, Tue Nov 18 11:04:42 2008 UTC vs.
Revision 1.303 by root, Wed Nov 19 05:52:42 2008 UTC