Async-Interrupt/Interrupt.xs

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.h"

#include "schmorp.h"

typedef volatile sig_atomic_t atomic_t;

static int *sig_pending, *psig_pend; /* make local copies because of missing THX */
static Sighandler_t old_sighandler;
static atomic_t async_pending;

#define PERL_VERSION_ATLEAST(a,b,c)                             \
  (PERL_REVISION > (a)                                          \
   || (PERL_REVISION == (a)                                     \
       && (PERL_VERSION > (b)                                   \
           || (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))

#if defined(HAS_SIGACTION) && defined(SA_SIGINFO)
# define HAS_SA_SIGINFO 1
#endif

#if !PERL_VERSION_ATLEAST(5,10,0)
# undef HAS_SA_SIGINFO
#endif

/*****************************************************************************/

typedef struct {
  SV *cb;
  void (*c_cb)(pTHX_ void *c_arg, int value);
  void *c_arg;
  SV *fh_r, *fh_w;
  SV *value;
  int signum;
  volatile int blocked;

  s_epipe ep;
  int fd_wlen;
  atomic_t fd_enable;
  atomic_t pending;
  volatile IV *valuep;
} async_t;

static AV *asyncs;
static async_t *sig_async [SIG_SIZE];

#define SvASYNC_nrv(sv) INT2PTR (async_t *, SvIVX (sv))
#define SvASYNC(rv)     SvASYNC_nrv (SvRV (rv))

/* the main workhorse to signal */
static void
async_signal (void *signal_arg, int value)
{
  static char pipedata [8];

  async_t *async = (async_t *)signal_arg;
  int pending = async->pending;

  *async->valuep = value ? value : 1;
  async->pending = 1;
  async_pending  = 1;
  psig_pend [9]  = 1;
  *sig_pending   = 1;

  {
    int fd_enable = async->fd_enable;

    if (!pending && fd_enable && async->ep.fd [1] >= 0)
      s_epipe_signal (&async->ep);
  }
}

static void
handle_async (async_t *async)
{
  int old_errno = errno;
  int value = *async->valuep;

  *async->valuep = 0;
  async->pending = 0;

  /* drain pipe */
  if (async->fd_enable && async->ep.fd [0] >= 0)
    s_epipe_drain (&async->ep);

  if (async->c_cb)
    {
      dTHX;
      async->c_cb (aTHX_ async->c_arg, value);
    }

  if (async->cb)
    {
      dSP;

      SV *saveerr = SvOK (ERRSV) ? sv_mortalcopy (ERRSV) : 0;
      SV *savedie = PL_diehook;

      PL_diehook = 0;

      PUSHSTACKi (PERLSI_SIGNAL);

      PUSHMARK (SP);
      XPUSHs (sv_2mortal (newSViv (value)));
      PUTBACK;
      call_sv (async->cb, G_VOID | G_DISCARD | G_EVAL);

      if (SvTRUE (ERRSV))
        {
          SPAGAIN;

          PUSHMARK (SP);
          PUTBACK;
          call_sv (get_sv ("Async::Interrupt::DIED", 1), G_VOID | G_DISCARD | G_EVAL | G_KEEPERR);

          sv_setpvn (ERRSV, "", 0);
        }

      if (saveerr)
        sv_setsv (ERRSV, saveerr);

      {
        SV *oldhook = PL_diehook;
        PL_diehook = savedie;
        SvREFCNT_dec (oldhook);
      }

      POPSTACK;
    }

  errno = old_errno;
}

static void
handle_asyncs (void)
{
  int i;

  async_pending = 0;

  for (i = AvFILLp (asyncs); i >= 0; --i)
    {
      async_t *async = SvASYNC_nrv (AvARRAY (asyncs)[i]);

      if (async->pending && !async->blocked)
        handle_async (async);
    }
}

#if HAS_SA_SIGINFO
static Signal_t async_sighandler (int signum, siginfo_t *si, void *sarg)
{
  if (signum == 9)
    handle_asyncs ();
  else
    old_sighandler (signum, si, sarg);
}
#else
static Signal_t async_sighandler (int signum)
{
  if (signum == 9)
    handle_asyncs ();
  else
    old_sighandler (signum);
}
#endif

static void
async_sigsend (int signum)
{
  async_signal (sig_async [signum], 0);
}

#define block(async) ++(async)->blocked

static void
unblock (async_t *async)
{
  --async->blocked;
  if (async->pending && !async->blocked)
    handle_async (async);
}

static void
scope_block_cb (pTHX_ void *async_sv)
{
  async_t *async = SvASYNC_nrv ((SV *)async_sv);
  unblock (async);
  SvREFCNT_dec (async_sv);
}

MODULE = Async::Interrupt               PACKAGE = Async::Interrupt

BOOT:
        old_sighandler = PL_sighandlerp;
        PL_sighandlerp = async_sighandler;
        sig_pending = &PL_sig_pending;
        psig_pend   = PL_psig_pend;
        asyncs      = newAV ();
        CvNODEBUG_on (get_cv ("Async::Interrupt::scope_block", 0)); /* otherwise calling scope can be the debugger */

PROTOTYPES: DISABLE

void
_alloc (SV *cb, void *c_cb, void *c_arg, SV *fh_r, SV *fh_w, SV *signl, SV *pvalue)
        PPCODE:
{
        SV *cv   = SvOK (cb) ? SvREFCNT_inc (s_get_cv_croak (cb)) : 0;
        async_t *async;

        Newz (0, async, 1, async_t);

        XPUSHs (sv_2mortal (newSViv (PTR2IV (async))));
        /* TODO: need to bless right now to ensure deallocation */
        av_push (asyncs, TOPs);

        SvGETMAGIC (fh_r); SvGETMAGIC (fh_w);
        if (SvOK (fh_r) || SvOK (fh_w))
          {
            int fd_r = s_fileno_croak (fh_r, 0);
            int fd_w = s_fileno_croak (fh_w, 1);

            async->fh_r      = newSVsv (fh_r);
            async->fh_w      = newSVsv (fh_w);
            async->ep.fd [0] = fd_r;
            async->ep.fd [1] = fd_w;
            async->ep.len    = 1;
            async->fd_enable = 1;
          }

        async->value     = SvROK (pvalue)
                           ? SvREFCNT_inc_NN (SvRV (pvalue))
                           : NEWSV (0, 0);

        sv_setiv (async->value, 0);
        SvIOK_only (async->value); /* just to be sure */
        SvREADONLY_on (async->value);

        async->valuep    = &(SvIVX (async->value));

        async->cb        = cv;
        async->c_cb      = c_cb;
        async->c_arg     = c_arg;
        async->signum    = SvOK (signl) ? s_signum_croak (signl) : 0;

        if (async->signum)
          {
            if (async->signum < 0)
              croak ("Async::Interrupt::new got passed illegal signal name or number: %s", SvPV_nolen (signl));

            sig_async [async->signum] = async;
#if _WIN32
            signal (async->signum, async_sigsend);
#else
            {
              struct sigaction sa = { };
              sa.sa_handler = async_sigsend;
              sigfillset (&sa.sa_mask);
              sigaction (async->signum, &sa, 0);
            }
#endif
          }
}

void
signal_func (async_t *async)
        PPCODE:
        EXTEND (SP, 2);
        PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
        PUSHs (sv_2mortal (newSViv (PTR2IV (async))));

IV
c_var (async_t *async)
        CODE:
        RETVAL = PTR2IV (async->valuep);
        OUTPUT:
        RETVAL

void
signal (async_t *async, int value = 1)
        CODE:
        async_signal (async, value);

void
block (async_t *async)
        CODE:
        block (async);

void
unblock (async_t *async)
        CODE:
        unblock (async);

void
scope_block (SV *self)
        CODE:
{
        SV *async_sv = SvRV (self);
        async_t *async = SvASYNC_nrv (async_sv);
        block (async);

        LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
        SAVEDESTRUCTOR_X (scope_block_cb, (void *)SvREFCNT_inc (async_sv));
        ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
}

void
pipe_enable (async_t *async)
        ALIAS:
        pipe_enable = 1
        pipe_disable = 0
        CODE:
        async->fd_enable = ix;

int
pipe_fileno (async_t *async)
        CODE:
        if (!async->ep.len)
          {
            int res;

            /*block (async);*//*TODO*/
            res = s_epipe_new (&async->ep);
            async->fd_enable = 1;
            /*unblock (async);*//*TODO*/

            if (res < 0)
              croak ("Async::Interrupt: unable to initialize event pipe");
          }

        RETVAL = async->ep.fd [0];
        OUTPUT:
        RETVAL


void
post_fork (async_t *async)
        CODE:
        if (async->ep.len)
          {
            int res;

            /*block (async);*//*TODO*/
            res = s_epipe_renew (&async->ep);
            /*unblock (async);*//*TODO*/

            if (res < 0)
              croak ("Async::Interrupt: unable to initialize event pipe after fork");
          }

void
DESTROY (SV *self)
        CODE:
{
        int i;
        SV *async_sv = SvRV (self);
        async_t *async = SvASYNC_nrv (async_sv);

        for (i = AvFILLp (asyncs); i >= 0; --i)
          if (AvARRAY (asyncs)[i] == async_sv)
            {
              if (i < AvFILLp (asyncs))
                AvARRAY (asyncs)[i] = AvARRAY (asyncs)[AvFILLp (asyncs)];

              assert (av_pop (asyncs) == async_sv);
              goto found;
            }

        if (!PL_dirty)
          warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report");

        found:

        if (async->signum)
          {
#if _WIN32
            signal (async->signum, SIG_DFL);
#else
            {
              struct sigaction sa = { };
              sa.sa_handler = SIG_DFL;
              sigaction (async->signum, &sa, 0);
            }
#endif
          }

        if (!async->fh_r && async->ep.len)
          s_epipe_destroy (&async->ep);

        SvREFCNT_dec (async->fh_r);
        SvREFCNT_dec (async->fh_w);
        SvREFCNT_dec (async->cb);
        SvREFCNT_dec (async->value);

        Safefree (async);
}

Revision:	1.9
Committed:	Fri Jul 17 01:51:32 2009 UTC (14 years, 10 months ago) by root
Branch:	MAIN
CVS Tags:	rel-0_5
Changes since 1.8:	+84 -25 lines
Log Message:	* empty log message *
#	Content
1	#include "EXTERN.h"
2	#include "perl.h"
3	#include "XSUB.h"
4
5	#include "schmorp.h"
6
7	typedef volatile sig_atomic_t atomic_t;
8
9	static int sig_pending, psig_pend; /* make local copies because of missing THX */
10	static Sighandler_t old_sighandler;
11	static atomic_t async_pending;
12
13	#define PERL_VERSION_ATLEAST(a,b,c) \
14	(PERL_REVISION > (a) \
15	\|\| (PERL_REVISION == (a) \
16	&& (PERL_VERSION > (b) \
17	\|\| (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))
18
19	#if defined(HAS_SIGACTION) && defined(SA_SIGINFO)
20	# define HAS_SA_SIGINFO 1
21	#endif
22
23	#if !PERL_VERSION_ATLEAST(5,10,0)
24	# undef HAS_SA_SIGINFO
25	#endif
26
27	/*****************************************************************************/
28
29	typedef struct {
30	SV *cb;
31	void (c_cb)(pTHX_ void c_arg, int value);
32	void *c_arg;
33	SV fh_r, fh_w;
34	SV *value;
35	int signum;
36	volatile int blocked;
37
38	s_epipe ep;
39	int fd_wlen;
40	atomic_t fd_enable;
41	atomic_t pending;
42	volatile IV *valuep;
43	} async_t;
44
45	static AV *asyncs;
46	static async_t *sig_async [SIG_SIZE];
47
48	#define SvASYNC_nrv(sv) INT2PTR (async_t *, SvIVX (sv))
49	#define SvASYNC(rv) SvASYNC_nrv (SvRV (rv))
50
51	/* the main workhorse to signal */
52	static void
53	async_signal (void *signal_arg, int value)
54	{
55	static char pipedata [8];
56
57	async_t async = (async_t )signal_arg;
58	int pending = async->pending;
59
60	*async->valuep = value ? value : 1;
61	async->pending = 1;
62	async_pending = 1;
63	psig_pend [9] = 1;
64	*sig_pending = 1;
65
66	{
67	int fd_enable = async->fd_enable;
68
69	if (!pending && fd_enable && async->ep.fd [1] >= 0)
70	s_epipe_signal (&async->ep);
71	}
72	}
73
74	static void
75	handle_async (async_t *async)
76	{
77	int old_errno = errno;
78	int value = *async->valuep;
79
80	*async->valuep = 0;
81	async->pending = 0;
82
83	/* drain pipe */
84	if (async->fd_enable && async->ep.fd [0] >= 0)
85	s_epipe_drain (&async->ep);
86
87	if (async->c_cb)
88	{
89	dTHX;
90	async->c_cb (aTHX_ async->c_arg, value);
91	}
92
93	if (async->cb)
94	{
95	dSP;
96
97	SV *saveerr = SvOK (ERRSV) ? sv_mortalcopy (ERRSV) : 0;
98	SV *savedie = PL_diehook;
99
100	PL_diehook = 0;
101
102	PUSHSTACKi (PERLSI_SIGNAL);
103
104	PUSHMARK (SP);
105	XPUSHs (sv_2mortal (newSViv (value)));
106	PUTBACK;
107	call_sv (async->cb, G_VOID \| G_DISCARD \| G_EVAL);
108
109	if (SvTRUE (ERRSV))
110	{
111	SPAGAIN;
112
113	PUSHMARK (SP);
114	PUTBACK;
115	call_sv (get_sv ("Async::Interrupt::DIED", 1), G_VOID \| G_DISCARD \| G_EVAL \| G_KEEPERR);
116
117	sv_setpvn (ERRSV, "", 0);
118	}
119
120	if (saveerr)
121	sv_setsv (ERRSV, saveerr);
122
123	{
124	SV *oldhook = PL_diehook;
125	PL_diehook = savedie;
126	SvREFCNT_dec (oldhook);
127	}
128
129	POPSTACK;
130	}
131
132	errno = old_errno;
133	}
134
135	static void
136	handle_asyncs (void)
137	{
138	int i;
139
140	async_pending = 0;
141
142	for (i = AvFILLp (asyncs); i >= 0; --i)
143	{
144	async_t *async = SvASYNC_nrv (AvARRAY (asyncs)[i]);
145
146	if (async->pending && !async->blocked)
147	handle_async (async);
148	}
149	}
150
151	#if HAS_SA_SIGINFO
152	static Signal_t async_sighandler (int signum, siginfo_t si, void sarg)
153	{
154	if (signum == 9)
155	handle_asyncs ();
156	else
157	old_sighandler (signum, si, sarg);
158	}
159	#else
160	static Signal_t async_sighandler (int signum)
161	{
162	if (signum == 9)
163	handle_asyncs ();
164	else
165	old_sighandler (signum);
166	}
167	#endif
168
169	static void
170	async_sigsend (int signum)
171	{
172	async_signal (sig_async [signum], 0);
173	}
174
175	#define block(async) ++(async)->blocked
176
177	static void
178	unblock (async_t *async)
179	{
180	--async->blocked;
181	if (async->pending && !async->blocked)
182	handle_async (async);
183	}
184
185	static void
186	scope_block_cb (pTHX_ void *async_sv)
187	{
188	async_t async = SvASYNC_nrv ((SV )async_sv);
189	unblock (async);
190	SvREFCNT_dec (async_sv);
191	}
192
193	MODULE = Async::Interrupt PACKAGE = Async::Interrupt
194
195	BOOT:
196	old_sighandler = PL_sighandlerp;
197	PL_sighandlerp = async_sighandler;
198	sig_pending = &PL_sig_pending;
199	psig_pend = PL_psig_pend;
200	asyncs = newAV ();
201	CvNODEBUG_on (get_cv ("Async::Interrupt::scope_block", 0)); /* otherwise calling scope can be the debugger */
202
203	PROTOTYPES: DISABLE
204
205	void
206	_alloc (SV cb, void c_cb, void c_arg, SV fh_r, SV fh_w, SV signl, SV *pvalue)
207	PPCODE:
208	{
209	SV *cv = SvOK (cb) ? SvREFCNT_inc (s_get_cv_croak (cb)) : 0;
210	async_t *async;
211
212	Newz (0, async, 1, async_t);
213
214	XPUSHs (sv_2mortal (newSViv (PTR2IV (async))));
215	/* TODO: need to bless right now to ensure deallocation */
216	av_push (asyncs, TOPs);
217
218	SvGETMAGIC (fh_r); SvGETMAGIC (fh_w);
219	if (SvOK (fh_r) \|\| SvOK (fh_w))
220	{
221	int fd_r = s_fileno_croak (fh_r, 0);
222	int fd_w = s_fileno_croak (fh_w, 1);
223
224	async->fh_r = newSVsv (fh_r);
225	async->fh_w = newSVsv (fh_w);
226	async->ep.fd [0] = fd_r;
227	async->ep.fd [1] = fd_w;
228	async->ep.len = 1;
229	async->fd_enable = 1;
230	}
231
232	async->value = SvROK (pvalue)
233	? SvREFCNT_inc_NN (SvRV (pvalue))
234	: NEWSV (0, 0);
235
236	sv_setiv (async->value, 0);
237	SvIOK_only (async->value); /* just to be sure */
238	SvREADONLY_on (async->value);
239
240	async->valuep = &(SvIVX (async->value));
241
242	async->cb = cv;
243	async->c_cb = c_cb;
244	async->c_arg = c_arg;
245	async->signum = SvOK (signl) ? s_signum_croak (signl) : 0;
246
247	if (async->signum)
248	{
249	if (async->signum < 0)
250	croak ("Async::Interrupt::new got passed illegal signal name or number: %s", SvPV_nolen (signl));
251
252	sig_async [async->signum] = async;
253	#if _WIN32
254	signal (async->signum, async_sigsend);
255	#else
256	{
257	struct sigaction sa = { };
258	sa.sa_handler = async_sigsend;
259	sigfillset (&sa.sa_mask);
260	sigaction (async->signum, &sa, 0);
261	}
262	#endif
263	}
264	}
265
266	void
267	signal_func (async_t *async)
268	PPCODE:
269	EXTEND (SP, 2);
270	PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
271	PUSHs (sv_2mortal (newSViv (PTR2IV (async))));
272
273	IV
274	c_var (async_t *async)
275	CODE:
276	RETVAL = PTR2IV (async->valuep);
277	OUTPUT:
278	RETVAL
279
280	void
281	signal (async_t *async, int value = 1)
282	CODE:
283	async_signal (async, value);
284
285	void
286	block (async_t *async)
287	CODE:
288	block (async);
289
290	void
291	unblock (async_t *async)
292	CODE:
293	unblock (async);
294
295	void
296	scope_block (SV *self)
297	CODE:
298	{
299	SV *async_sv = SvRV (self);
300	async_t *async = SvASYNC_nrv (async_sv);
301	block (async);
302
303	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
304	SAVEDESTRUCTOR_X (scope_block_cb, (void *)SvREFCNT_inc (async_sv));
305	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
306	}
307
308	void
309	pipe_enable (async_t *async)
310	ALIAS:
311	pipe_enable = 1
312	pipe_disable = 0
313	CODE:
314	async->fd_enable = ix;
315
316	int
317	pipe_fileno (async_t *async)
318	CODE:
319	if (!async->ep.len)
320	{
321	int res;
322
323	/block (async);//TODO/
324	res = s_epipe_new (&async->ep);
325	async->fd_enable = 1;
326	/unblock (async);//TODO/
327
328	if (res < 0)
329	croak ("Async::Interrupt: unable to initialize event pipe");
330	}
331
332	RETVAL = async->ep.fd [0];
333	OUTPUT:
334	RETVAL
335
336
337	void
338	post_fork (async_t *async)
339	CODE:
340	if (async->ep.len)
341	{
342	int res;
343
344	/block (async);//TODO/
345	res = s_epipe_renew (&async->ep);
346	/unblock (async);//TODO/
347
348	if (res < 0)
349	croak ("Async::Interrupt: unable to initialize event pipe after fork");
350	}
351
352	void
353	DESTROY (SV *self)
354	CODE:
355	{
356	int i;
357	SV *async_sv = SvRV (self);
358	async_t *async = SvASYNC_nrv (async_sv);
359
360	for (i = AvFILLp (asyncs); i >= 0; --i)
361	if (AvARRAY (asyncs)[i] == async_sv)
362	{
363	if (i < AvFILLp (asyncs))
364	AvARRAY (asyncs)[i] = AvARRAY (asyncs)[AvFILLp (asyncs)];
365
366	assert (av_pop (asyncs) == async_sv);
367	goto found;
368	}
369
370	if (!PL_dirty)
371	warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report");
372
373	found:
374
375	if (async->signum)
376	{
377	#if _WIN32
378	signal (async->signum, SIG_DFL);
379	#else
380	{
381	struct sigaction sa = { };
382	sa.sa_handler = SIG_DFL;
383	sigaction (async->signum, &sa, 0);
384	}
385	#endif
386	}
387
388	if (!async->fh_r && async->ep.len)
389	s_epipe_destroy (&async->ep);
390
391	SvREFCNT_dec (async->fh_r);
392	SvREFCNT_dec (async->fh_w);
393	SvREFCNT_dec (async->cb);
394	SvREFCNT_dec (async->value);
395
396	Safefree (async);
397	}
398