ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Async-Interrupt/Interrupt.xs
(Generate patch)

Comparing Async-Interrupt/Interrupt.xs (file contents):
Revision 1.8 by root, Tue Jul 14 19:29:26 2009 UTC vs.
Revision 1.16 by root, Tue Sep 1 16:41:29 2009 UTC

29typedef struct { 29typedef 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
45static AV *asyncs; 48static AV *asyncs;
46static async_t *sig_async [SIG_SIZE]; 49static 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
54static void async_signal (void *signal_arg, int value);
55
56static void
57setsig (int signum, void (*handler)(int))
58{
59#if _WIN32
60 signal (async->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}
68
69static void
70async_sigsend (int signum)
71{
72 async_signal (sig_async [signum], 0);
73}
74
51/* the main workhorse to signal */ 75/* the main workhorse to signal */
52static void 76static void
53async_signal (void *signal_arg, int value) 77async_signal (void *signal_arg, int value)
54{ 78{
55 static char pipedata [8]; 79 static char pipedata [8];
56 80
57 async_t *async = (async_t *)signal_arg; 81 async_t *async = (async_t *)signal_arg;
58 int pending = async->pending; 82 int pending = async->pending;
59 83
60 async->value = value; 84 if (async->hysteresis)
85 setsig (async->signum, SIG_IGN);
86
87 *async->valuep = value ? value : 1;
61 async->pending = 1; 88 async->pending = 1;
62 async_pending = 1; 89 async_pending = 1;
63 psig_pend [9] = 1; 90 psig_pend [9] = 1;
64 *sig_pending = 1; 91 *sig_pending = 1;
65 92
66 { 93 if (!pending && async->fd_enable && async->ep.len)
67 int fd_w = async->fd_w; 94 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} 95}
76 96
77static void 97static void
78handle_async (async_t *async) 98handle_async (async_t *async)
79{ 99{
80 int old_errno = errno; 100 int old_errno = errno;
81 int value = async->value; 101 int value = *async->valuep;
82 102
103 *async->valuep = 0;
83 async->pending = 0; 104 async->pending = 0;
84 105
106 /* restore signal */
107 if (async->hysteresis)
108 setsig (async->signum, async_sigsend);
109
85 /* drain pipe */ 110 /* drain pipe */
86 if (async->fd_r >= 0 && async->fd_enable) 111 if (async->fd_enable && async->ep.len && async->autodrain)
87 { 112 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 113
94 if (async->c_cb) 114 if (async->c_cb)
95 { 115 {
96 dTHX; 116 dTHX;
97 async->c_cb (aTHX_ async->c_arg, value); 117 async->c_cb (aTHX_ async->c_arg, value);
146 166
147 async_pending = 0; 167 async_pending = 0;
148 168
149 for (i = AvFILLp (asyncs); i >= 0; --i) 169 for (i = AvFILLp (asyncs); i >= 0; --i)
150 { 170 {
171 SV *async_sv = AvARRAY (asyncs)[i];
151 async_t *async = SvASYNC_nrv (AvARRAY (asyncs)[i]); 172 async_t *async = SvASYNC_nrv (async_sv);
152 173
153 if (async->pending && !async->blocked) 174 if (async->pending && !async->blocked)
175 {
176 /* temporarily keep a refcount */
177 SvREFCNT_inc (async_sv);
154 handle_async (async); 178 handle_async (async);
179 SvREFCNT_dec (async_sv);
180
181 /* the handler could have deleted any number of asyncs */
182 if (i > AvFILLp (asyncs))
183 i = AvFILLp (asyncs);
184 }
155 } 185 }
156} 186}
157 187
158#if HAS_SA_SIGINFO 188#if HAS_SA_SIGINFO
159static Signal_t async_sighandler (int signum, siginfo_t *si, void *sarg) 189static Signal_t async_sighandler (int signum, siginfo_t *si, void *sarg)
171 else 201 else
172 old_sighandler (signum); 202 old_sighandler (signum);
173} 203}
174#endif 204#endif
175 205
176static void
177async_sigsend (int signum)
178{
179 async_signal (sig_async [signum], 0);
180}
181
182#define block(async) ++(async)->blocked 206#define block(async) ++(async)->blocked
183 207
184static void 208static void
185unblock (async_t *async) 209unblock (async_t *async)
186{ 210{
191 215
192static void 216static void
193scope_block_cb (pTHX_ void *async_sv) 217scope_block_cb (pTHX_ void *async_sv)
194{ 218{
195 async_t *async = SvASYNC_nrv ((SV *)async_sv); 219 async_t *async = SvASYNC_nrv ((SV *)async_sv);
220
221 async->scope_savestack = 0;
196 unblock (async); 222 unblock (async);
197 SvREFCNT_dec (async_sv); 223 SvREFCNT_dec (async_sv);
198} 224}
199 225
226static void
227scope_block (SV *async_sv)
228{
229 async_t *async = SvASYNC_nrv (async_sv);
230
231 /* as a heuristic, we skip the scope block if we already are blocked */
232 /* and the existing scope block used the same savestack */
233
234 if (!async->scope_savestack || async->scope_savestack != PL_savestack)
235 {
236 async->scope_savestack = PL_savestack;
237 block (async);
238
239 LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
240 SAVEDESTRUCTOR_X (scope_block_cb, (void *)SvREFCNT_inc (async_sv));
241 ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
242 }
243}
244
245#endif
200MODULE = Async::Interrupt PACKAGE = Async::Interrupt 246MODULE = Async::Interrupt PACKAGE = Async::Interrupt
201 247
202BOOT: 248BOOT:
203 old_sighandler = PL_sighandlerp; 249 old_sighandler = PL_sighandlerp;
204 PL_sighandlerp = async_sighandler; 250 PL_sighandlerp = async_sighandler;
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
210PROTOTYPES: DISABLE 256PROTOTYPES: DISABLE
211 257
212void 258void
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
311void
312signal_hysteresis (async_t *async, int enable)
313 CODE:
314 async->hysteresis = enable;
253 315
254void 316void
255signal_func (async_t *async) 317signal_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
261void 323void
324scope_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
330IV
331c_var (async_t *async)
332 CODE:
333 RETVAL = PTR2IV (async->valuep);
334 OUTPUT:
335 RETVAL
336
337void
262signal (async_t *async, int value = 0) 338signal (async_t *async, int value = 1)
263 CODE: 339 CODE:
264 async_signal (async, value); 340 async_signal (async, value);
265 341
266void 342void
267block (async_t *async) 343block (async_t *async)
274 unblock (async); 350 unblock (async);
275 351
276void 352void
277scope_block (SV *self) 353scope_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
289void 357void
290pipe_enable (async_t *async) 358pipe_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
365int
366pipe_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
385int
386pipe_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
394void
395post_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
297void 409void
298DESTROY (SV *self) 410DESTROY (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
444SV *
445sig2num (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
464MODULE = Async::Interrupt PACKAGE = Async::Interrupt::EventPipe PREFIX = s_epipe_
465
466void
467new (const char *klass)
468 PPCODE:
469{
470 s_epipe *epp;
471 SV *self;
472
473 Newz (0, epp, 1, s_epipe);
474 XPUSHs (sv_setref_iv (sv_newmortal (), klass, PTR2IV (epp)));
475
476 if (s_epipe_new (epp) < 0)
477 croak ("Async::Interrupt::EventPipe: unable to create new event pipe");
478}
479
480void
481filenos (s_epipe *epp)
482 PPCODE:
483 EXTEND (SP, 2);
484 PUSHs (sv_2mortal (newSViv (epp->fd [0])));
485 PUSHs (sv_2mortal (newSViv (epp->fd [1])));
486
487int
488fileno (s_epipe *epp)
489 ALIAS:
490 fileno = 0
491 fileno_r = 0
492 fileno_w = 1
493 CODE:
494 RETVAL = epp->fd [ix];
495 OUTPUT:
496 RETVAL
497
498int
499type (s_epipe *epp)
500 CODE:
501 RETVAL = epp->len;
502 OUTPUT:
503 RETVAL
504
505void
506s_epipe_signal (s_epipe *epp)
507
508void
509s_epipe_drain (s_epipe *epp)
510
511void
512drain_func (s_epipe *epp)
513 PPCODE:
514 EXTEND (SP, 2);
515 PUSHs (sv_2mortal (newSViv (PTR2IV (s_epipe_drain))));
516 PUSHs (sv_2mortal (newSViv (PTR2IV (epp))));
517
518void
519s_epipe_wait (s_epipe *epp)
520
521void
522DESTROY (s_epipe *epp)
523 CODE:
524 s_epipe_destroy (epp);
525

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines