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.7 by root, Tue Jul 14 14:18:10 2009 UTC vs.
Revision 1.21 by root, Tue Apr 17 19:24:32 2018 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines