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

Comparing Coro/Coro/State.xs (file contents):
Revision 1.305 by root, Wed Nov 19 10:44:41 2008 UTC vs.
Revision 1.312 by root, Thu Nov 20 03:10:30 2008 UTC

…		…
55		55
56	#define PERL_VERSION_ATLEAST(a,b,c) \	56	#define PERL_VERSION_ATLEAST(a,b,c) \
57	(PERL_REVISION > (a) \	57	(PERL_REVISION > (a) \
58	\|\| (PERL_REVISION == (a) \	58	\|\| (PERL_REVISION == (a) \
59	&& (PERL_VERSION > (b) \	59	&& (PERL_VERSION > (b) \
60	\|\| (PERL_VERSION == (b) && PERLSUBVERSION >= (c)))))	60	\|\| (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))
61		61
62	#if !PERL_VERSION_ATLEAST (5,6,0)	62	#if !PERL_VERSION_ATLEAST (5,6,0)
63	# ifndef PL_ppaddr	63	# ifndef PL_ppaddr
64	# define PL_ppaddr ppaddr	64	# define PL_ppaddr ppaddr
65	# endif	65	# endif
…		…
176	static HV hv_sig; / %SIG */	176	static HV hv_sig; / %SIG */
177		177
178	/* async_pool helper stuff */	178	/* async_pool helper stuff */
179	static SV *sv_pool_rss;	179	static SV *sv_pool_rss;
180	static SV *sv_pool_size;	180	static SV *sv_pool_size;
		181	static SV *sv_async_pool_idle;
181	static AV *av_async_pool;	182	static AV *av_async_pool;
		183	static SV *sv_Coro;
		184	static CV *cv_pool_handler;
		185	static CV *cv_coro_new;
182		186
183	/* Coro::AnyEvent */	187	/* Coro::AnyEvent */
184	static SV *sv_activity;	188	static SV *sv_activity;
185		189
186	static struct coro_cctx *cctx_first;	190	static struct coro_cctx *cctx_first;
…		…
246	/* state data */	250	/* state data */
247	struct CoroSLF slf_frame; /* saved slf frame */	251	struct CoroSLF slf_frame; /* saved slf frame */
248	AV *mainstack;	252	AV *mainstack;
249	perl_slots slot; / basically the saved sp */	253	perl_slots slot; / basically the saved sp */
250		254
		255	CV startcv; / the CV to execute */
251	AV args; / data associated with this coroutine (initial args) */	256	AV args; / data associated with this coroutine (initial args) */
252	int refcnt; /* coroutines are refcounted, yes */	257	int refcnt; /* coroutines are refcounted, yes */
253	int flags; /* CF_ flags */	258	int flags; /* CF_ flags */
254	HV hv; / the perl hash associated with this coro, if any */	259	HV hv; / the perl hash associated with this coro, if any */
255	void (on_destroy)(pTHX_ struct coro coro);	260	void (on_destroy)(pTHX_ struct coro coro);
256		261
257	/* statistics */	262	/* statistics */
258	int usecount; /* number of transfers to this coro */	263	int usecount; /* number of transfers to this coro */
259		264
…		…
262	SV except; / exception to be thrown */	267	SV except; / exception to be thrown */
263	SV *rouse_cb;	268	SV *rouse_cb;
264		269
265	/* async_pool */	270	/* async_pool */
266	SV *saved_deffh;	271	SV *saved_deffh;
		272	SV *invoke_cb;
		273	AV *invoke_av;
267		274
268	/* linked list */	275	/* linked list */
269	struct coro next, prev;	276	struct coro next, prev;
270	};	277	};
271		278
…		…
288		295
289	/* for Coro.pm */	296	/* for Coro.pm */
290	static SV *coro_current;	297	static SV *coro_current;
291	static SV *coro_readyhook;	298	static SV *coro_readyhook;
292	static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1];	299	static AV *coro_ready [PRIO_MAX - PRIO_MIN + 1];
		300	static CV cv_coro_run, cv_coro_terminate;
293	static struct coro *coro_first;	301	static struct coro *coro_first;
294	#define coro_nready coroapi.nready	302	#define coro_nready coroapi.nready
295		303
296	/ lowlevel stuff ********************************************************/	304	/ lowlevel stuff ********************************************************/
297		305
…		…
321	#if PERL_VERSION_ATLEAST (5,10,0)	329	#if PERL_VERSION_ATLEAST (5,10,0)
322	/* silence stupid and wrong 5.10 warning that I am unable to switch off */	330	/* silence stupid and wrong 5.10 warning that I am unable to switch off */
323	get_hv (name, create);	331	get_hv (name, create);
324	#endif	332	#endif
325	return get_hv (name, create);	333	return get_hv (name, create);
		334	}
		335
		336	/* may croak */
		337	INLINE CV *
		338	coro_sv_2cv (pTHX_ SV *sv)
		339	{
		340	HV *st;
		341	GV *gvp;
		342	return sv_2cv (sv, &st, &gvp, 0);
326	}	343	}
327		344
328	static AV *	345	static AV *
329	coro_clone_padlist (pTHX_ CV *cv)	346	coro_clone_padlist (pTHX_ CV *cv)
330	{	347	{
…		…
676	#if !PERL_VERSION_ATLEAST (5,10,0)	693	#if !PERL_VERSION_ATLEAST (5,10,0)
677	Safefree (PL_retstack);	694	Safefree (PL_retstack);
678	#endif	695	#endif
679	}	696	}
680		697
		698	#define CORO_RSS \
		699	rss += sizeof (SYM (curstackinfo)); \
		700	rss += (SYM (curstackinfo->si_cxmax) + 1) * sizeof (PERL_CONTEXT); \
		701	rss += sizeof (SV) + sizeof (struct xpvav) + (1 + AvMAX (SYM (curstack))) * sizeof (SV *); \
		702	rss += SYM (tmps_max) * sizeof (SV *); \
		703	rss += (SYM (markstack_max) - SYM (markstack_ptr)) * sizeof (I32); \
		704	rss += SYM (scopestack_max) * sizeof (I32); \
		705	rss += SYM (savestack_max) * sizeof (ANY);
		706
681	static size_t	707	static size_t
682	coro_rss (pTHX_ struct coro *coro)	708	coro_rss (pTHX_ struct coro *coro)
683	{	709	{
684	size_t rss = sizeof (*coro);	710	size_t rss = sizeof (*coro);
685		711
686	if (coro->mainstack)	712	if (coro->mainstack)
687	{	713	{
688	perl_slots tmp_slot;
689	perl_slots *slot;
690
691	if (coro->flags & CF_RUNNING)	714	if (coro->flags & CF_RUNNING)
692	{	715	{
693	slot = &tmp_slot;	716	#define SYM(sym) PL_ ## sym
694		717	CORO_RSS;
695	#define VAR(name,type) slot->name = PL_ ## name;
696	# include "state.h"
697	#undef VAR	718	#undef SYM
698	}	719	}
699	else	720	else
700	slot = coro->slot;
701
702	if (slot)
703	{	721	{
704	rss += sizeof (slot->curstackinfo);	722	#define SYM(sym) coro->slot->sym
705	rss += (slot->curstackinfo->si_cxmax + 1) * sizeof (PERL_CONTEXT);	723	CORO_RSS;
706	rss += sizeof (SV) + sizeof (struct xpvav) + (1 + AvMAX (slot->curstack)) * sizeof (SV *);	724	#undef SYM
707	rss += slot->tmps_max * sizeof (SV *);
708	rss += (slot->markstack_max - slot->markstack_ptr) * sizeof (I32);
709	rss += slot->scopestack_max * sizeof (I32);
710	rss += slot->savestack_max * sizeof (ANY);
711
712	#if !PERL_VERSION_ATLEAST (5,10,0)
713	rss += slot->retstack_max * sizeof (OP *);
714	#endif
715	}	725	}
716	}	726	}
717		727
718	return rss;	728	return rss;
719	}	729	}
…		…
861	dSP;	871	dSP;
862	UNOP myop;	872	UNOP myop;
863		873
864	Zero (&myop, 1, UNOP);	874	Zero (&myop, 1, UNOP);
865	myop.op_next = Nullop;	875	myop.op_next = Nullop;
		876	myop.op_type = OP_ENTERSUB;
866	myop.op_flags = OPf_WANT_VOID;	877	myop.op_flags = OPf_WANT_VOID;
867		878
868	PUSHMARK (SP);	879	PUSHMARK (SP);
869	XPUSHs (sv_2mortal (av_shift (GvAV (PL_defgv))));	880	PUSHs ((SV *)coro->startcv);
870	PUTBACK;	881	PUTBACK;
871	PL_op = (OP *)&myop;	882	PL_op = (OP *)&myop;
872	PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);	883	PL_op = PL_ppaddr[OP_ENTERSUB](aTHX);
873	SPAGAIN;
874	}	884	}
875		885
876	/* this newly created coroutine might be run on an existing cctx which most	886	/* this newly created coroutine might be run on an existing cctx which most
877	* likely was suspended in pp_slf, so we have to emulate entering pp_slf here.	887	* likely was suspended in pp_slf, so we have to emulate entering pp_slf here.
878	*/	888	*/
…		…
881		891
882	/* and we have to provide the pp_slf op in any case, so pp_slf can skip it */	892	/* and we have to provide the pp_slf op in any case, so pp_slf can skip it */
883	coro_setup_op.op_next = PL_op;	893	coro_setup_op.op_next = PL_op;
884	coro_setup_op.op_type = OP_CUSTOM;	894	coro_setup_op.op_type = OP_CUSTOM;
885	coro_setup_op.op_ppaddr = pp_slf;	895	coro_setup_op.op_ppaddr = pp_slf;
886	/* no flags required, as an init function won't be called */	896	/* no flags etc. required, as an init function won't be called */
887		897
888	PL_op = (OP *)&coro_setup_op;	898	PL_op = (OP *)&coro_setup_op;
889		899
890	/* copy throw, in case it was set before coro_setup */	900	/* copy throw, in case it was set before coro_setup */
891	CORO_THROW = coro->except;	901	CORO_THROW = coro->except;
…		…
919	SvREFCNT_dec (GvSV (irsgv));	929	SvREFCNT_dec (GvSV (irsgv));
920		930
921	SvREFCNT_dec (PL_diehook);	931	SvREFCNT_dec (PL_diehook);
922	SvREFCNT_dec (PL_warnhook);	932	SvREFCNT_dec (PL_warnhook);
923		933
924	SvREFCNT_dec (CORO_THROW);
925	SvREFCNT_dec (coro->saved_deffh);	934	SvREFCNT_dec (coro->saved_deffh);
926	SvREFCNT_dec (coro->rouse_cb);	935	SvREFCNT_dec (coro->rouse_cb);
		936	SvREFCNT_dec (coro->invoke_cb);
		937	SvREFCNT_dec (coro->invoke_av);
927		938
928	coro_destruct_stacks (aTHX);	939	coro_destruct_stacks (aTHX);
929	}	940	}
930		941
931	INLINE void	942	INLINE void
…		…
1018	SAVETMPS;	1029	SAVETMPS;
1019	EXTEND (SP, 3);	1030	EXTEND (SP, 3);
1020	PUSHMARK (SP);	1031	PUSHMARK (SP);
1021	PUSHs (&PL_sv_yes);	1032	PUSHs (&PL_sv_yes);
1022	PUSHs (fullname);	1033	PUSHs (fullname);
1023	PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef);	1034	PUSHs (CxHASARGS (cx) ? sv_2mortal (newRV_inc ((SV *)cx->blk_sub.argarray)) : &PL_sv_undef);
1024	PUTBACK;	1035	PUTBACK;
1025	cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);	1036	cb = hv_fetch ((HV *)SvRV (coro_current), "_trace_sub_cb", sizeof ("_trace_sub_cb") - 1, 0);
1026	if (cb) call_sv (*cb, G_KEEPERR \| G_EVAL \| G_VOID \| G_DISCARD);	1037	if (cb) call_sv (*cb, G_KEEPERR \| G_EVAL \| G_VOID \| G_DISCARD);
1027	SPAGAIN;	1038	SPAGAIN;
1028	FREETMPS;	1039	FREETMPS;
…		…
1134	transfer_tail (aTHX);	1145	transfer_tail (aTHX);
1135		1146
1136	/* somebody or something will hit me for both perl_run and PL_restartop */	1147	/* somebody or something will hit me for both perl_run and PL_restartop */
1137	PL_restartop = PL_op;	1148	PL_restartop = PL_op;
1138	perl_run (PL_curinterp);	1149	perl_run (PL_curinterp);
		1150	/*
		1151	* Unfortunately, there is no way to get at the return values of the
		1152	* coro body here, as perl_run destroys these
		1153	*/
1139		1154
1140	/*	1155	/*
1141	* If perl-run returns we assume exit() was being called or the coro	1156	* If perl-run returns we assume exit() was being called or the coro
1142	* fell off the end, which seems to be the only valid (non-bug)	1157	* fell off the end, which seems to be the only valid (non-bug)
1143	* reason for perl_run to return. We try to exit by jumping to the	1158	* reason for perl_run to return. We try to exit by jumping to the
…		…
1435		1450
1436	coro->slot = 0;	1451	coro->slot = 0;
1437	}	1452	}
1438		1453
1439	cctx_destroy (coro->cctx);	1454	cctx_destroy (coro->cctx);
		1455	SvREFCNT_dec (coro->startcv);
1440	SvREFCNT_dec (coro->args);	1456	SvREFCNT_dec (coro->args);
		1457	SvREFCNT_dec (CORO_THROW);
1441		1458
1442	if (coro->next) coro->next->prev = coro->prev;	1459	if (coro->next) coro->next->prev = coro->prev;
1443	if (coro->prev) coro->prev->next = coro->next;	1460	if (coro->prev) coro->prev->next = coro->next;
1444	if (coro == coro_first) coro_first = coro->next;	1461	if (coro == coro_first) coro_first = coro->next;
1445		1462
…		…
1728	if (coro->flags & CF_RUNNING)	1745	if (coro->flags & CF_RUNNING)
1729	PL_runops = RUNOPS_DEFAULT;	1746	PL_runops = RUNOPS_DEFAULT;
1730	else	1747	else
1731	coro->slot->runops = RUNOPS_DEFAULT;	1748	coro->slot->runops = RUNOPS_DEFAULT;
1732	}	1749	}
		1750	}
		1751
		1752	/*****************************************************************************/
		1753	/* async pool handler */
		1754
		1755	static int
		1756	slf_check_pool_handler (pTHX_ struct CoroSLF *frame)
		1757	{
		1758	HV hv = (HV )SvRV (coro_current);
		1759	struct coro coro = (struct coro )frame->data;
		1760
		1761	if (!coro->invoke_cb)
		1762	return 1; /* loop till we have invoke */
		1763	else
		1764	{
		1765	int i, len;
		1766
		1767	hv_store (hv, "desc", sizeof ("desc") - 1,
		1768	newSVpvn ("[async_pool]", sizeof ("[async_pool]") - 1), 0);
		1769
		1770	coro->saved_deffh = SvREFCNT_inc_NN ((SV *)PL_defoutgv);
		1771
		1772	len = av_len (coro->invoke_av);
		1773
		1774	{
		1775	dSP;
		1776	XPUSHs (sv_2mortal (coro->invoke_cb)); coro->invoke_cb = 0;
		1777	PUTBACK;
		1778	}
		1779
		1780	SvREFCNT_dec (GvAV (PL_defgv));
		1781	GvAV (PL_defgv) = coro->invoke_av;
		1782	coro->invoke_av = 0;
		1783
		1784	return 0;
		1785	}
		1786	}
		1787
		1788	static void
		1789	slf_init_pool_handler (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
		1790	{
		1791	HV hv = (HV )SvRV (coro_current);
		1792	struct coro coro = SvSTATE_hv ((SV )hv);
		1793
		1794	if (expect_true (coro->saved_deffh))
		1795	{
		1796	/* subsequent iteration */
		1797	SvREFCNT_dec ((SV )PL_defoutgv); PL_defoutgv = (GV )coro->saved_deffh;
		1798	coro->saved_deffh = 0;
		1799
		1800	if (coro_rss (aTHX_ coro) > SvUV (sv_pool_rss)
		1801	\|\| av_len (av_async_pool) + 1 >= SvIV (sv_pool_size))
		1802	{
		1803	coro->invoke_cb = SvREFCNT_inc_NN ((SV *)cv_coro_terminate);
		1804	coro->invoke_av = newAV ();
		1805
		1806	frame->prepare = prepare_nop;
		1807	}
		1808	else
		1809	{
		1810	av_clear (GvAV (PL_defgv));
		1811	hv_store (hv, "desc", sizeof ("desc") - 1, SvREFCNT_inc_NN (sv_async_pool_idle), 0);
		1812
		1813	coro->prio = 0;
		1814
		1815	if (coro->cctx && (coro->cctx->flags & CC_TRACE))
		1816	api_trace (aTHX_ coro_current, 0);
		1817
		1818	frame->prepare = prepare_schedule;
		1819	av_push (av_async_pool, SvREFCNT_inc (hv));
		1820	}
		1821	}
		1822	else
		1823	{
		1824	/* first iteration, simply fall through */
		1825	frame->prepare = prepare_nop;
		1826	}
		1827
		1828	frame->check = slf_check_pool_handler;
		1829	frame->data = (void *)coro;
1733	}	1830	}
1734		1831
1735	/*****************************************************************************/	1832	/*****************************************************************************/
1736	/* rouse callback */	1833	/* rouse callback */
1737		1834
…		…
2174	AvARRAY (av)[0] = AvARRAY (av)[1];	2271	AvARRAY (av)[0] = AvARRAY (av)[1];
2175	AvARRAY (av)[1] = count_sv;	2272	AvARRAY (av)[1] = count_sv;
2176	cb = av_shift (av);	2273	cb = av_shift (av);
2177		2274
2178	if (SvOBJECT (cb))	2275	if (SvOBJECT (cb))
		2276	{
2179	api_ready (aTHX_ cb);	2277	api_ready (aTHX_ cb);
2180	else	2278	--count;
2181	croak ("callbacks not yet supported");	2279	}
		2280	else if (SvTYPE (cb) == SVt_PVCV)
		2281	{
		2282	dSP;
		2283	PUSHMARK (SP);
		2284	XPUSHs (sv_2mortal (newRV_inc ((SV *)av)));
		2285	PUTBACK;
		2286	call_sv (cb, G_VOID \| G_DISCARD \| G_EVAL \| G_KEEPERR);
		2287	}
2182		2288
2183	SvREFCNT_dec (cb);	2289	SvREFCNT_dec (cb);
2184
2185	--count;
2186	}	2290	}
2187	}	2291	}
2188		2292
2189	static void	2293	static void
2190	coro_semaphore_on_destroy (pTHX_ struct coro *coro)	2294	coro_semaphore_on_destroy (pTHX_ struct coro *coro)
…		…
2272	}	2376	}
2273		2377
2274	static void	2378	static void
2275	slf_init_semaphore_wait (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)	2379	slf_init_semaphore_wait (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
2276	{	2380	{
		2381	if (items >= 2)
		2382	{
		2383	/* callback form */
		2384	AV av = (AV )SvRV (arg [0]);
		2385	CV *cb_cv = coro_sv_2cv (aTHX_ arg [1]);
		2386
		2387	av_push (av, (SV *)SvREFCNT_inc_NN (cb_cv));
		2388
		2389	if (SvIVX (AvARRAY (av)[0]) > 0)
		2390	coro_semaphore_adjust (aTHX_ av, 0);
		2391
		2392	frame->prepare = prepare_nop;
		2393	frame->check = slf_check_nop;
		2394	}
		2395	else
		2396	{
2277	slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);	2397	slf_init_semaphore_down_or_wait (aTHX_ frame, cv, arg, items);
2278	frame->check = slf_check_semaphore_wait;	2398	frame->check = slf_check_semaphore_wait;
		2399	}
2279	}	2400	}
2280		2401
2281	/* signal */	2402	/* signal */
2282		2403
2283	static void	2404	static void
…		…
2355	dXSARGS;	2476	dXSARGS;
2356	AV state = (AV )GENSUB_ARG;	2477	AV state = (AV )GENSUB_ARG;
2357	SV *coro = av_pop (state);	2478	SV *coro = av_pop (state);
2358	SV *data_sv = newSV (sizeof (struct io_state));	2479	SV *data_sv = newSV (sizeof (struct io_state));
2359		2480
2360	av_extend (state, items);	2481	av_extend (state, items - 1);
2361		2482
2362	sv_upgrade (data_sv, SVt_PV);	2483	sv_upgrade (data_sv, SVt_PV);
2363	SvCUR_set (data_sv, sizeof (struct io_state));	2484	SvCUR_set (data_sv, sizeof (struct io_state));
2364	SvPOK_only (data_sv);	2485	SvPOK_only (data_sv);
2365		2486
…		…
2575	assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL));	2696	assert (("PRIO_NORMAL must be 0", !PRIO_NORMAL));
2576	}	2697	}
2577		2698
2578	SV *	2699	SV *
2579	new (char *klass, ...)	2700	new (char *klass, ...)
		2701	ALIAS:
		2702	Coro::new = 1
2580	CODE:	2703	CODE:
2581	{	2704	{
2582	struct coro *coro;	2705	struct coro *coro;
2583	MAGIC *mg;	2706	MAGIC *mg;
2584	HV *hv;	2707	HV *hv;
		2708	CV *cb;
2585	int i;	2709	int i;
		2710
		2711	if (items > 1)
		2712	{
		2713	cb = coro_sv_2cv (aTHX_ ST (1));
		2714
		2715	if (!ix)
		2716	{
		2717	if (CvISXSUB (cb))
		2718	croak ("Coro::State doesn't support XS functions as coroutine start, caught");
		2719
		2720	if (!CvROOT (cb))
		2721	croak ("Coro::State doesn't support autoloaded or undefined functions as coroutine start, caught");
		2722	}
		2723	}
2586		2724
2587	Newz (0, coro, 1, struct coro);	2725	Newz (0, coro, 1, struct coro);
2588	coro->args = newAV ();	2726	coro->args = newAV ();
2589	coro->flags = CF_NEW;	2727	coro->flags = CF_NEW;
2590		2728
…		…
2595	coro->hv = hv = newHV ();	2733	coro->hv = hv = newHV ();
2596	mg = sv_magicext ((SV )hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char )coro, 0);	2734	mg = sv_magicext ((SV )hv, 0, CORO_MAGIC_type_state, &coro_state_vtbl, (char )coro, 0);
2597	mg->mg_flags \|= MGf_DUP;	2735	mg->mg_flags \|= MGf_DUP;
2598	RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));	2736	RETVAL = sv_bless (newRV_noinc ((SV *)hv), gv_stashpv (klass, 1));
2599		2737
		2738	if (items > 1)
		2739	{
2600	av_extend (coro->args, items - 1);	2740	av_extend (coro->args, items - 1 + ix - 1);
		2741
		2742	if (ix)
		2743	{
		2744	av_push (coro->args, SvREFCNT_inc_NN ((SV *)cb));
		2745	cb = cv_coro_run;
		2746	}
		2747
		2748	coro->startcv = (CV )SvREFCNT_inc_NN ((SV )cb);
		2749
2601	for (i = 1; i < items; i++)	2750	for (i = 2; i < items; i++)
2602	av_push (coro->args, newSVsv (ST (i)));	2751	av_push (coro->args, newSVsv (ST (i)));
		2752	}
2603	}	2753	}
2604	OUTPUT:	2754	OUTPUT:
2605	RETVAL	2755	RETVAL
2606		2756
2607	void	2757	void
…		…
2805		2955
2806	BOOT:	2956	BOOT:
2807	{	2957	{
2808	int i;	2958	int i;
2809		2959
2810	av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE);	2960	av_async_pool = coro_get_av (aTHX_ "Coro::async_pool", TRUE);
2811	sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);	2961	sv_pool_rss = coro_get_sv (aTHX_ "Coro::POOL_RSS" , TRUE);
2812	sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);	2962	sv_pool_size = coro_get_sv (aTHX_ "Coro::POOL_SIZE" , TRUE);
2813		2963	cv_coro_run = get_cv ( "Coro::_terminate", GV_ADD);
		2964	cv_coro_terminate = get_cv ( "Coro::terminate", GV_ADD);
2814	coro_current = coro_get_sv (aTHX_ "Coro::current", FALSE);	2965	coro_current = coro_get_sv (aTHX_ "Coro::current" , FALSE);
2815	SvREADONLY_on (coro_current);	2966	SvREADONLY_on (coro_current);
		2967
		2968	sv_async_pool_idle = newSVpv ("[async pool idle]", 0); SvREADONLY_on (sv_async_pool_idle);
		2969	sv_Coro = newSVpv ("Coro", 0); SvREADONLY_on (sv_Coro);
		2970	cv_pool_handler = get_cv ("Coro::_pool_handler", 0); SvREADONLY_on (cv_pool_handler);
		2971	cv_coro_new = get_cv ("Coro::new", 0); SvREADONLY_on (cv_coro_new);
2816		2972
2817	coro_stash = gv_stashpv ("Coro", TRUE);	2973	coro_stash = gv_stashpv ("Coro", TRUE);
2818		2974
2819	newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX));	2975	newCONSTSUB (coro_stash, "PRIO_MAX", newSViv (PRIO_MAX));
2820	newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH));	2976	newCONSTSUB (coro_stash, "PRIO_HIGH", newSViv (PRIO_HIGH));
…		…
2909	CODE:	3065	CODE:
2910	RETVAL = coro_nready;	3066	RETVAL = coro_nready;
2911	OUTPUT:	3067	OUTPUT:
2912	RETVAL	3068	RETVAL
2913		3069
2914	# for async_pool speedup
2915	void	3070	void
2916	_pool_1 (SV *cb)	3071	_pool_handler (...)
2917	CODE:	3072	CODE:
2918	{	3073	CORO_EXECUTE_SLF_XS (slf_init_pool_handler);
2919	HV hv = (HV )SvRV (coro_current);
2920	struct coro coro = SvSTATE_hv ((SV )hv);
2921	AV *defav = GvAV (PL_defgv);
2922	SV *invoke = hv_delete (hv, "_invoke", sizeof ("_invoke") - 1, 0);
2923	AV *invoke_av;
2924	int i, len;
2925		3074
2926	if (!invoke)	3075	void
		3076	async_pool (SV *cv, ...)
		3077	PROTOTYPE: &@
		3078	PPCODE:
		3079	{
		3080	HV hv = (HV )av_pop (av_async_pool);
		3081	AV *av = newAV ();
		3082	SV *cb = ST (0);
		3083	int i;
		3084
		3085	av_extend (av, items - 2);
		3086	for (i = 1; i < items; ++i)
		3087	av_push (av, SvREFCNT_inc_NN (ST (i)));
		3088
		3089	if ((SV *)hv == &PL_sv_undef)
2927	{	3090	{
2928	SV *old = PL_diehook;	3091	PUSHMARK (SP);
2929	PL_diehook = 0;	3092	EXTEND (SP, 2);
2930	SvREFCNT_dec (old);	3093	PUSHs (sv_Coro);
2931	croak ("\3async_pool terminate\2\n");	3094	PUSHs ((SV *)cv_pool_handler);
		3095	PUTBACK;
		3096	call_sv (cv_coro_new, G_SCALAR);
		3097	SPAGAIN;
		3098
		3099	hv = (HV *)SvREFCNT_inc_NN (SvRV (POPs));
2932	}	3100	}
2933		3101
2934	SvREFCNT_dec (coro->saved_deffh);
2935	coro->saved_deffh = SvREFCNT_inc_NN ((SV *)PL_defoutgv);
2936
2937	hv_store (hv, "desc", sizeof ("desc") - 1,
2938	newSVpvn ("[async_pool]", sizeof ("[async_pool]") - 1), 0);
2939
2940	invoke_av = (AV *)SvRV (invoke);
2941	len = av_len (invoke_av);
2942
2943	sv_setsv (cb, AvARRAY (invoke_av)[0]);
2944
2945	if (len > 0)
2946	{	3102	{
2947	av_fill (defav, len - 1);	3103	struct coro *coro = SvSTATE_hv (hv);
2948	for (i = 0; i < len; ++i)	3104
2949	av_store (defav, i, SvREFCNT_inc_NN (AvARRAY (invoke_av)[i + 1]));	3105	assert (!coro->invoke_cb);
		3106	assert (!coro->invoke_av);
		3107	coro->invoke_cb = SvREFCNT_inc (cb);
		3108	coro->invoke_av = av;
2950	}	3109	}
2951	}
2952		3110
2953	void	3111	api_ready ((SV *)hv);
2954	_pool_2 (SV *cb)
2955	CODE:
2956	{
2957	HV hv = (HV )SvRV (coro_current);
2958	struct coro coro = SvSTATE_hv ((SV )hv);
2959		3112
2960	sv_setsv (cb, &PL_sv_undef);	3113	if (GIMME_V != G_VOID)
2961		3114	XPUSHs (sv_2mortal (newRV_noinc ((SV *)hv)));
2962	SvREFCNT_dec ((SV )PL_defoutgv); PL_defoutgv = (GV )coro->saved_deffh;	3115	else
2963	coro->saved_deffh = 0;
2964
2965	if (coro_rss (aTHX_ coro) > SvUV (sv_pool_rss)
2966	\|\| av_len (av_async_pool) + 1 >= SvIV (sv_pool_size))
2967	{
2968	SV *old = PL_diehook;
2969	PL_diehook = 0;
2970	SvREFCNT_dec (old);	3116	SvREFCNT_dec (hv);
2971	croak ("\3async_pool terminate\2\n");
2972	}
2973
2974	av_clear (GvAV (PL_defgv));
2975	hv_store (hv, "desc", sizeof ("desc") - 1,
2976	newSVpvn ("[async_pool idle]", sizeof ("[async_pool idle]") - 1), 0);
2977
2978	coro->prio = 0;
2979
2980	if (coro->cctx && (coro->cctx->flags & CC_TRACE))
2981	api_trace (aTHX_ coro_current, 0);
2982
2983	av_push (av_async_pool, newSVsv (coro_current));
2984	}	3117	}
2985		3118
2986	SV *	3119	SV *
2987	rouse_cb ()	3120	rouse_cb ()
2988	PROTOTYPE:	3121	PROTOTYPE:
…		…
2990	RETVAL = coro_new_rouse_cb (aTHX);	3123	RETVAL = coro_new_rouse_cb (aTHX);
2991	OUTPUT:	3124	OUTPUT:
2992	RETVAL	3125	RETVAL
2993		3126
2994	void	3127	void
2995	rouse_wait (SV *cb = 0)	3128	rouse_wait (...)
2996	PROTOTYPE: ;$	3129	PROTOTYPE: ;$
2997	PPCODE:	3130	PPCODE:
2998	CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);	3131	CORO_EXECUTE_SLF_XS (slf_init_rouse_wait);
2999		3132
3000		3133
…		…
3037	adjust = 1	3170	adjust = 1
3038	CODE:	3171	CODE:
3039	coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);	3172	coro_semaphore_adjust (aTHX_ (AV *)SvRV (self), ix ? adjust : 1);
3040		3173
3041	void	3174	void
3042	down (SV *self)	3175	down (...)
3043	CODE:	3176	CODE:
3044	CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);	3177	CORO_EXECUTE_SLF_XS (slf_init_semaphore_down);
3045		3178
3046	void	3179	void
3047	wait (SV *self)	3180	wait (...)
3048	CODE:	3181	CODE:
3049	CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);	3182	CORO_EXECUTE_SLF_XS (slf_init_semaphore_wait);
3050		3183
3051	void	3184	void
3052	try (SV *self)	3185	try (SV *self)
…		…
3095	);	3228	);
3096	OUTPUT:	3229	OUTPUT:
3097	RETVAL	3230	RETVAL
3098		3231
3099	void	3232	void
3100	wait (SV *self)	3233	wait (...)
3101	CODE:	3234	CODE:
3102	CORO_EXECUTE_SLF_XS (slf_init_signal_wait);	3235	CORO_EXECUTE_SLF_XS (slf_init_signal_wait);
3103		3236
3104	void	3237	void
3105	broadcast (SV *self)	3238	broadcast (SV *self)
…		…
3162		3295
3163	void	3296	void
3164	_register (char target, char proto, SV *req)	3297	_register (char target, char proto, SV *req)
3165	CODE:	3298	CODE:
3166	{	3299	{
3167	HV *st;	3300	CV *req_cv = coro_sv_2cv (aTHX_ req);
3168	GV *gvp;
3169	CV *req_cv = sv_2cv (req, &st, &gvp, 0);
3170	/* newXSproto doesn't return the CV on 5.8 */	3301	/* newXSproto doesn't return the CV on 5.8 */
3171	CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);	3302	CV *slf_cv = newXS (target, coro_aio_req_xs, __FILE__);
3172	sv_setpv ((SV *)slf_cv, proto);	3303	sv_setpv ((SV *)slf_cv, proto);
3173	sv_magicext ((SV )slf_cv, (SV )req_cv, CORO_MAGIC_type_aio, 0, 0, 0);	3304	sv_magicext ((SV )slf_cv, (SV )req_cv, CORO_MAGIC_type_aio, 0, 0, 0);
3174	}	3305	}

Diff Legend

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

Comparing Coro/Coro/State.xs (file contents): Revision 1.305 by root, Wed Nov 19 10:44:41 2008 UTC vs. Revision 1.312 by root, Thu Nov 20 03:10:30 2008 UTC

Diff Legend

Comparing Coro/Coro/State.xs (file contents):
Revision 1.305 by root, Wed Nov 19 10:44:41 2008 UTC vs.
Revision 1.312 by root, Thu Nov 20 03:10:30 2008 UTC