[ViewVC] Diff of: cvs/Async-Interrupt/Interrupt.xs

Comparing Async-Interrupt/Interrupt.xs (file contents):
Revision 1.8 by root, Tue Jul 14 19:29:26 2009 UTC vs.
Revision 1.18 by root, Sat May 15 00:08:48 2010 UTC

…		…
29	typedef struct {	29	typedef struct {
30	SV *cb;	30	SV *cb;
31	void (c_cb)(pTHX_ void c_arg, int value);	31	void (c_cb)(pTHX_ void c_arg, int value);
32	void *c_arg;	32	void *c_arg;
33	SV fh_r, fh_w;	33	SV fh_r, fh_w;
		34	SV *value;
34	int signum;	35	int signum;
		36	int autodrain;
		37	ANY *scope_savestack;
35	volatile int blocked;	38	volatile int blocked;
36		39
37	int fd_r;	40	s_epipe ep;
38	volatile int fd_w;
39	int fd_wlen;	41	int fd_wlen;
40	atomic_t fd_enable;	42	atomic_t fd_enable;
41	atomic_t value;
42	atomic_t pending;	43	atomic_t pending;
		44	volatile IV *valuep;
		45	atomic_t hysteresis;
43	} async_t;	46	} async_t;
44		47
45	static AV *asyncs;	48	static AV *asyncs;
46	static async_t *sig_async [SIG_SIZE];	49	static async_t *sig_async [SIG_SIZE];
47		50
48	#define SvASYNC_nrv(sv) INT2PTR (async_t *, SvIVX (sv))	51	#define SvASYNC_nrv(sv) INT2PTR (async_t *, SvIVX (sv))
49	#define SvASYNC(rv) SvASYNC_nrv (SvRV (rv))	52	#define SvASYNC(rv) SvASYNC_nrv (SvRV (rv))
50		53
		54	static void async_signal (void *signal_arg, int value);
		55
		56	static void
		57	setsig (int signum, void (*handler)(int))
		58	{
		59	#if _WIN32
		60	signal (signum, handler);
		61	#else
		62	struct sigaction sa;
		63	sa.sa_handler = handler;
		64	sigfillset (&sa.sa_mask);
		65	sa.sa_flags = 0; /* if we interrupt a syscall, we might drain the pipe before it became ready */
		66	sigaction (signum, &sa, 0);
		67	#endif
		68	}
		69
		70	static void
		71	async_sigsend (int signum)
		72	{
		73	async_signal (sig_async [signum], 0);
		74	}
		75
51	/* the main workhorse to signal */	76	/* the main workhorse to signal */
52	static void	77	static void
53	async_signal (void *signal_arg, int value)	78	async_signal (void *signal_arg, int value)
54	{	79	{
55	static char pipedata [8];	80	static char pipedata [8];
56		81
57	async_t async = (async_t )signal_arg;	82	async_t async = (async_t )signal_arg;
58	int pending = async->pending;	83	int pending = async->pending;
59		84
60	async->value = value;	85	if (async->hysteresis)
		86	setsig (async->signum, SIG_IGN);
		87
		88	*async->valuep = value ? value : 1;
61	async->pending = 1;	89	async->pending = 1;
62	async_pending = 1;	90	async_pending = 1;
63	psig_pend [9] = 1;	91	psig_pend [9] = 1;
64	*sig_pending = 1;	92	*sig_pending = 1;
65		93
66	{	94	if (!pending && async->fd_enable && async->ep.len)
67	int fd_w = async->fd_w;	95	s_epipe_signal (&async->ep);
68	int fd_enable = async->fd_enable;
69
70	if (!pending && fd_w >= 0 && fd_enable)
71	if (write (fd_w, pipedata, async->fd_wlen) < 0 && errno == EINVAL)
72	/* on EINVAL we assume it's an eventfd */
73	write (fd_w, pipedata, (async->fd_wlen = 8));
74	}
75	}	96	}
76		97
77	static void	98	static void
78	handle_async (async_t *async)	99	handle_async (async_t *async)
79	{	100	{
80	int old_errno = errno;	101	int old_errno = errno;
81	int value = async->value;	102	int value = *async->valuep;
82		103
		104	*async->valuep = 0;
83	async->pending = 0;	105	async->pending = 0;
84		106
		107	/* restore signal */
		108	if (async->hysteresis)
		109	setsig (async->signum, async_sigsend);
		110
85	/* drain pipe */	111	/* drain pipe */
86	if (async->fd_r >= 0 && async->fd_enable)	112	if (async->fd_enable && async->ep.len && async->autodrain)
87	{	113	s_epipe_drain (&async->ep);
88	char dummy [9]; /* 9 is enough for eventfd and normal pipes */
89
90	while (read (async->fd_r, dummy, sizeof (dummy)) == sizeof (dummy))
91	;
92	}
93		114
94	if (async->c_cb)	115	if (async->c_cb)
95	{	116	{
96	dTHX;	117	dTHX;
97	async->c_cb (aTHX_ async->c_arg, value);	118	async->c_cb (aTHX_ async->c_arg, value);
…		…
146		167
147	async_pending = 0;	168	async_pending = 0;
148		169
149	for (i = AvFILLp (asyncs); i >= 0; --i)	170	for (i = AvFILLp (asyncs); i >= 0; --i)
150	{	171	{
		172	SV *async_sv = AvARRAY (asyncs)[i];
151	async_t *async = SvASYNC_nrv (AvARRAY (asyncs)[i]);	173	async_t *async = SvASYNC_nrv (async_sv);
152		174
153	if (async->pending && !async->blocked)	175	if (async->pending && !async->blocked)
		176	{
		177	/* temporarily keep a refcount */
		178	SvREFCNT_inc (async_sv);
154	handle_async (async);	179	handle_async (async);
		180	SvREFCNT_dec (async_sv);
		181
		182	/* the handler could have deleted any number of asyncs */
		183	if (i > AvFILLp (asyncs))
		184	i = AvFILLp (asyncs);
		185	}
155	}	186	}
156	}	187	}
157		188
158	#if HAS_SA_SIGINFO	189	#if HAS_SA_SIGINFO
159	static Signal_t async_sighandler (int signum, siginfo_t si, void sarg)	190	static Signal_t async_sighandler (int signum, siginfo_t si, void sarg)
…		…
171	else	202	else
172	old_sighandler (signum);	203	old_sighandler (signum);
173	}	204	}
174	#endif	205	#endif
175		206
176	static void
177	async_sigsend (int signum)
178	{
179	async_signal (sig_async [signum], 0);
180	}
181
182	#define block(async) ++(async)->blocked	207	#define block(async) ++(async)->blocked
183		208
184	static void	209	static void
185	unblock (async_t *async)	210	unblock (async_t *async)
186	{	211	{
…		…
191		216
192	static void	217	static void
193	scope_block_cb (pTHX_ void *async_sv)	218	scope_block_cb (pTHX_ void *async_sv)
194	{	219	{
195	async_t async = SvASYNC_nrv ((SV )async_sv);	220	async_t async = SvASYNC_nrv ((SV )async_sv);
		221
		222	async->scope_savestack = 0;
196	unblock (async);	223	unblock (async);
197	SvREFCNT_dec (async_sv);	224	SvREFCNT_dec (async_sv);
		225	}
		226
		227	static void
		228	scope_block (SV *async_sv)
		229	{
		230	async_t *async = SvASYNC_nrv (async_sv);
		231
		232	/* as a heuristic, we skip the scope block if we already are blocked */
		233	/* and the existing scope block used the same savestack */
		234
		235	if (!async->scope_savestack \|\| async->scope_savestack != PL_savestack)
		236	{
		237	async->scope_savestack = PL_savestack;
		238	block (async);
		239
		240	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
		241	SAVEDESTRUCTOR_X (scope_block_cb, (void *)SvREFCNT_inc (async_sv));
		242	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
		243	}
198	}	244	}
199		245
200	MODULE = Async::Interrupt PACKAGE = Async::Interrupt	246	MODULE = Async::Interrupt PACKAGE = Async::Interrupt
201		247
202	BOOT:	248	BOOT:
…		…
208	CvNODEBUG_on (get_cv ("Async::Interrupt::scope_block", 0)); /* otherwise calling scope can be the debugger */	254	CvNODEBUG_on (get_cv ("Async::Interrupt::scope_block", 0)); /* otherwise calling scope can be the debugger */
209		255
210	PROTOTYPES: DISABLE	256	PROTOTYPES: DISABLE
211		257
212	void	258	void
213	_alloc (SV cb, void c_cb, void c_arg, SV fh_r, SV fh_w, SV signl)	259	_alloc (SV cb, void c_cb, void c_arg, SV fh_r, SV fh_w, SV signl, SV *pvalue)
214	PPCODE:	260	PPCODE:
215	{	261	{
216	SV *cv = SvOK (cb) ? SvREFCNT_inc (s_get_cv_croak (cb)) : 0;	262	SV *cv = SvOK (cb) ? SvREFCNT_inc (s_get_cv_croak (cb)) : 0;
217	int fd_r = SvOK (fh_r) ? s_fileno_croak (fh_r, 0) : -1;
218	int fd_w = SvOK (fh_w) ? s_fileno_croak (fh_w, 1) : -1;
219	async_t *async;	263	async_t *async;
220		264
221	Newz (0, async, 1, async_t);	265	Newz (0, async, 1, async_t);
222		266
223	XPUSHs (sv_2mortal (newSViv (PTR2IV (async))));	267	XPUSHs (sv_2mortal (newSViv (PTR2IV (async))));
		268	/* TODO: need to bless right now to ensure deallocation */
224	av_push (asyncs, TOPs);	269	av_push (asyncs, TOPs);
225		270
226	async->fh_r = fd_r >= 0 ? newSVsv (fh_r) : 0; async->fd_r = fd_r;	271	SvGETMAGIC (fh_r); SvGETMAGIC (fh_w);
227	async->fh_w = fd_w >= 0 ? newSVsv (fh_w) : 0; async->fd_w = fd_w;	272	if (SvOK (fh_r) \|\| SvOK (fh_w))
		273	{
		274	int fd_r = s_fileno_croak (fh_r, 0);
		275	int fd_w = s_fileno_croak (fh_w, 1);
		276
		277	async->fh_r = newSVsv (fh_r);
		278	async->fh_w = newSVsv (fh_w);
		279	async->ep.fd [0] = fd_r;
		280	async->ep.fd [1] = fd_w;
228	async->fd_wlen = 1;	281	async->ep.len = 1;
229	async->fd_enable = 1;	282	async->fd_enable = 1;
		283	}
		284
		285	async->value = SvROK (pvalue)
		286	? SvREFCNT_inc_NN (SvRV (pvalue))
		287	: NEWSV (0, 0);
		288
		289	sv_setiv (async->value, 0);
		290	SvIOK_only (async->value); /* just to be sure */
		291	SvREADONLY_on (async->value);
		292
		293	async->valuep = &(SvIVX (async->value));
		294
		295	async->autodrain = 1;
230	async->cb = cv;	296	async->cb = cv;
231	async->c_cb = c_cb;	297	async->c_cb = c_cb;
232	async->c_arg = c_arg;	298	async->c_arg = c_arg;
233	async->signum = SvOK (signl) ? s_signum_croak (signl) : 0;	299	async->signum = SvOK (signl) ? s_signum_croak (signl) : 0;
234		300
…		…
236	{	302	{
237	if (async->signum < 0)	303	if (async->signum < 0)
238	croak ("Async::Interrupt::new got passed illegal signal name or number: %s", SvPV_nolen (signl));	304	croak ("Async::Interrupt::new got passed illegal signal name or number: %s", SvPV_nolen (signl));
239		305
240	sig_async [async->signum] = async;	306	sig_async [async->signum] = async;
241	#if _WIN32
242	signal (async->signum, async_sigsend);	307	setsig (async->signum, async_sigsend);
243	#else
244	{
245	struct sigaction sa = { };
246	sa.sa_handler = async_sigsend;
247	sigfillset (&sa.sa_mask);
248	sigaction (async->signum, &sa, 0);
249	}
250	#endif
251	}	308	}
252	}	309	}
		310
		311	void
		312	signal_hysteresis (async_t *async, int enable)
		313	CODE:
		314	async->hysteresis = enable;
253		315
254	void	316	void
255	signal_func (async_t *async)	317	signal_func (async_t *async)
256	PPCODE:	318	PPCODE:
257	EXTEND (SP, 2);	319	EXTEND (SP, 2);
258	PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));	320	PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
259	PUSHs (sv_2mortal (newSViv (PTR2IV (async))));	321	PUSHs (sv_2mortal (newSViv (PTR2IV (async))));
260		322
261	void	323	void
		324	scope_block_func (SV *self)
		325	PPCODE:
		326	EXTEND (SP, 2);
		327	PUSHs (sv_2mortal (newSViv (PTR2IV (scope_block))));
		328	PUSHs (sv_2mortal (newSViv (PTR2IV (SvRV (self)))));
		329
		330	IV
		331	c_var (async_t *async)
		332	CODE:
		333	RETVAL = PTR2IV (async->valuep);
		334	OUTPUT:
		335	RETVAL
		336
		337	void
262	signal (async_t *async, int value = 0)	338	signal (async_t *async, int value = 1)
263	CODE:	339	CODE:
264	async_signal (async, value);	340	async_signal (async, value);
265		341
266	void	342	void
267	block (async_t *async)	343	block (async_t *async)
…		…
274	unblock (async);	350	unblock (async);
275		351
276	void	352	void
277	scope_block (SV *self)	353	scope_block (SV *self)
278	CODE:	354	CODE:
279	{	355	scope_block (SvRV (self));
280	SV *async_sv = SvRV (self);
281	async_t *async = SvASYNC_nrv (async_sv);
282	block (async);
283
284	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
285	SAVEDESTRUCTOR_X (scope_block_cb, (void *)SvREFCNT_inc (async_sv));
286	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
287	}
288		356
289	void	357	void
290	pipe_enable (async_t *async)	358	pipe_enable (async_t *async)
291	ALIAS:	359	ALIAS:
292	pipe_enable = 1	360	pipe_enable = 1
293	pipe_disable = 0	361	pipe_disable = 0
294	CODE:	362	CODE:
295	async->fd_enable = ix;	363	async->fd_enable = ix;
296		364
		365	int
		366	pipe_fileno (async_t *async)
		367	CODE:
		368	if (!async->ep.len)
		369	{
		370	int res;
		371
		372	/block (async);//TODO/
		373	res = s_epipe_new (&async->ep);
		374	async->fd_enable = 1;
		375	/unblock (async);//TODO/
		376
		377	if (res < 0)
		378	croak ("Async::Interrupt: unable to initialize event pipe");
		379	}
		380
		381	RETVAL = async->ep.fd [0];
		382	OUTPUT:
		383	RETVAL
		384
		385	int
		386	pipe_autodrain (async_t *async, int enable = -1)
		387	CODE:
		388	RETVAL = async->autodrain;
		389	if (enable >= 0)
		390	async->autodrain = enable;
		391	OUTPUT:
		392	RETVAL
		393
		394	void
		395	post_fork (async_t *async)
		396	CODE:
		397	if (async->ep.len)
		398	{
		399	int res;
		400
		401	/block (async);//TODO/
		402	res = s_epipe_renew (&async->ep);
		403	/unblock (async);//TODO/
		404
		405	if (res < 0)
		406	croak ("Async::Interrupt: unable to initialize event pipe after fork");
		407	}
		408
297	void	409	void
298	DESTROY (SV *self)	410	DESTROY (SV *self)
299	CODE:	411	CODE:
300	{	412	{
301	int i;	413	int i;
…		…
303	async_t *async = SvASYNC_nrv (async_sv);	415	async_t *async = SvASYNC_nrv (async_sv);
304		416
305	for (i = AvFILLp (asyncs); i >= 0; --i)	417	for (i = AvFILLp (asyncs); i >= 0; --i)
306	if (AvARRAY (asyncs)[i] == async_sv)	418	if (AvARRAY (asyncs)[i] == async_sv)
307	{	419	{
308	if (i < AvFILLp (asyncs))
309	AvARRAY (asyncs)[i] = AvARRAY (asyncs)[AvFILLp (asyncs)];	420	AvARRAY (asyncs)[i] = AvARRAY (asyncs)[AvFILLp (asyncs)];
310		421	av_pop (asyncs);
311	assert (av_pop (asyncs) == async_sv);
312	goto found;	422	goto found;
313	}	423	}
314		424
315	if (!PL_dirty)	425	if (!PL_dirty)
316	warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report");	426	warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report");
317		427
318	found:	428	found:
319		429
320	if (async->signum)	430	if (async->signum)
321	{
322	#if _WIN32
323	signal (async->signum, SIG_DFL);	431	setsig (async->signum, SIG_DFL);
324	#else	432
325	{	433	if (!async->fh_r && async->ep.len)
326	struct sigaction sa = { };	434	s_epipe_destroy (&async->ep);
327	sa.sa_handler = SIG_DFL;
328	sigaction (async->signum, &sa, 0);
329	}
330	#endif
331	}
332		435
333	SvREFCNT_dec (async->fh_r);	436	SvREFCNT_dec (async->fh_r);
334	SvREFCNT_dec (async->fh_w);	437	SvREFCNT_dec (async->fh_w);
335	SvREFCNT_dec (async->cb);	438	SvREFCNT_dec (async->cb);
		439	SvREFCNT_dec (async->value);
336		440
337	Safefree (async);	441	Safefree (async);
338	}	442	}
339		443
		444	SV *
		445	sig2num (SV *signame_or_number)
		446	ALIAS:
		447	sig2num = 0
		448	sig2name = 1
		449	PROTOTYPE: $
		450	CODE:
		451	{
		452	int signum = s_signum (signame_or_number);
		453
		454	if (signum < 0)
		455	RETVAL = &PL_sv_undef;
		456	else if (ix)
		457	RETVAL = newSVpv (PL_sig_name [signum], 0);
		458	else
		459	RETVAL = newSViv (signum);
		460	}
		461	OUTPUT:
		462	RETVAL
		463
		464	MODULE = Async::Interrupt PACKAGE = Async::Interrupt::EventPipe PREFIX = s_epipe_
		465
		466	void
		467	new (const char *klass)
		468	PPCODE:
		469	{
		470	s_epipe *epp;
		471
		472	Newz (0, epp, 1, s_epipe);
		473	XPUSHs (sv_setref_iv (sv_newmortal (), klass, PTR2IV (epp)));
		474
		475	if (s_epipe_new (epp) < 0)
		476	croak ("Async::Interrupt::EventPipe: unable to create new event pipe");
		477	}
		478
		479	void
		480	filenos (s_epipe *epp)
		481	PPCODE:
		482	EXTEND (SP, 2);
		483	PUSHs (sv_2mortal (newSViv (epp->fd [0])));
		484	PUSHs (sv_2mortal (newSViv (epp->fd [1])));
		485
		486	int
		487	fileno (s_epipe *epp)
		488	ALIAS:
		489	fileno = 0
		490	fileno_r = 0
		491	fileno_w = 1
		492	CODE:
		493	RETVAL = epp->fd [ix];
		494	OUTPUT:
		495	RETVAL
		496
		497	int
		498	type (s_epipe *epp)
		499	CODE:
		500	RETVAL = epp->len;
		501	OUTPUT:
		502	RETVAL
		503
		504	void
		505	s_epipe_signal (s_epipe *epp)
		506
		507	void
		508	s_epipe_drain (s_epipe *epp)
		509
		510	void
		511	signal_func (s_epipe *epp)
		512	ALIAS:
		513	drain_func = 1
		514	PPCODE:
		515	EXTEND (SP, 2);
		516	PUSHs (sv_2mortal (newSViv (PTR2IV (ix ? s_epipe_drain : s_epipe_signal))));
		517	PUSHs (sv_2mortal (newSViv (PTR2IV (epp))));
		518
		519	void
		520	s_epipe_wait (s_epipe *epp)
		521
		522	void
		523	DESTROY (s_epipe *epp)
		524	CODE:
		525	s_epipe_destroy (epp);
		526

Diff Legend

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

Comparing Async-Interrupt/Interrupt.xs (file contents): Revision 1.8 by root, Tue Jul 14 19:29:26 2009 UTC vs. Revision 1.18 by root, Sat May 15 00:08:48 2010 UTC

Diff Legend

Comparing Async-Interrupt/Interrupt.xs (file contents):
Revision 1.8 by root, Tue Jul 14 19:29:26 2009 UTC vs.
Revision 1.18 by root, Sat May 15 00:08:48 2010 UTC