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;
  int signum;
  volatile int blocked;

  int fd_r;
  volatile int fd_w;
  int fd_wlen;
  atomic_t fd_enable;
  atomic_t value;
  atomic_t pending;
} 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->value   = value;
  async->pending = 1;
  async_pending  = 1;
  psig_pend [9]  = 1;
  *sig_pending   = 1;

  {
    int fd_w      = async->fd_w;
    int fd_enable = async->fd_enable;

    if (!pending && fd_w >= 0 && fd_enable)
      if (write (fd_w, pipedata, async->fd_wlen) < 0 && errno == EINVAL)
        /* on EINVAL we assume it's an eventfd */
        write (fd_w, pipedata, (async->fd_wlen = 8));
  }
}

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

  async->pending = 0;

  /* drain pipe */
  if (async->fd_r >= 0 && async->fd_enable)
    {
      char dummy [9]; /* 9 is enough for eventfd and normal pipes */

      while (read (async->fd_r, dummy, sizeof (dummy)) == sizeof (dummy))
        ;
    }

  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)
        PPCODE:
{
        SV *cv   = SvOK (cb) ? SvREFCNT_inc (s_get_cv_croak (cb)) : 0;
        int fd_r = SvOK (fh_r) ? s_fileno_croak (fh_r, 0) : -1;
        int fd_w = SvOK (fh_w) ? s_fileno_croak (fh_w, 1) : -1;
        async_t *async;

        Newz (0, async, 1, async_t);

        XPUSHs (sv_2mortal (newSViv (PTR2IV (async))));
        av_push (asyncs, TOPs);

        async->fh_r      = fd_r >= 0 ? newSVsv (fh_r) : 0; async->fd_r = fd_r;
        async->fh_w      = fd_w >= 0 ? newSVsv (fh_w) : 0; async->fd_w = fd_w;
        async->fd_wlen   = 1;
        async->fd_enable = 1;
        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))));

void
signal (async_t *async, int value = 0)
        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;

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
          }

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

        Safefree (async);
}

Revision:	1.8
Committed:	Tue Jul 14 19:29:26 2009 UTC (14 years, 10 months ago) by root
Branch:	MAIN
CVS Tags:	rel-0_042
Changes since 1.7:	+6 -62 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#include "EXTERN.h"
2			#include "perl.h"
3			#include "XSUB.h"
4
5	root	1.8	#include "schmorp.h"
6
7	root	1.1	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	root	1.5	#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	root	1.6	/*****************************************************************************/
28	root	1.1
29	root	1.6	typedef struct {
30	root	1.1	SV *cb;
31	root	1.2	void (c_cb)(pTHX_ void c_arg, int value);
32			void *c_arg;
33			SV fh_r, fh_w;
34	root	1.6	int signum;
35	root	1.7	volatile int blocked;
36	root	1.1
37	root	1.7	int fd_r;
38			volatile int fd_w;
39	root	1.6	int fd_wlen;
40			atomic_t fd_enable;
41	root	1.1	atomic_t value;
42	root	1.2	atomic_t pending;
43	root	1.6	} 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	root	1.1
51			/* the main workhorse to signal */
52			static void
53			async_signal (void *signal_arg, int value)
54			{
55	root	1.6	static char pipedata [8];
56
57			async_t async = (async_t )signal_arg;
58	root	1.2	int pending = async->pending;
59	root	1.1
60	root	1.2	async->value = value;
61			async->pending = 1;
62			async_pending = 1;
63			psig_pend [9] = 1;
64			*sig_pending = 1;
65	root	1.1
66	root	1.7	{
67			int fd_w = async->fd_w;
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	root	1.2	}
76
77			static void
78	root	1.6	handle_async (async_t *async)
79	root	1.2	{
80			int old_errno = errno;
81			int value = async->value;
82
83			async->pending = 0;
84
85			/* drain pipe */
86	root	1.6	if (async->fd_r >= 0 && async->fd_enable)
87	root	1.1	{
88	root	1.6	char dummy [9]; /* 9 is enough for eventfd and normal pipes */
89	root	1.2
90			while (read (async->fd_r, dummy, sizeof (dummy)) == sizeof (dummy))
91			;
92			}
93
94			if (async->c_cb)
95			{
96			dTHX;
97			async->c_cb (aTHX_ async->c_arg, value);
98			}
99
100			if (async->cb)
101			{
102			dSP;
103
104			SV *saveerr = SvOK (ERRSV) ? sv_mortalcopy (ERRSV) : 0;
105			SV *savedie = PL_diehook;
106
107			PL_diehook = 0;
108
109			PUSHSTACKi (PERLSI_SIGNAL);
110
111			PUSHMARK (SP);
112			XPUSHs (sv_2mortal (newSViv (value)));
113			PUTBACK;
114			call_sv (async->cb, G_VOID \| G_DISCARD \| G_EVAL);
115
116			if (SvTRUE (ERRSV))
117			{
118			SPAGAIN;
119
120			PUSHMARK (SP);
121			PUTBACK;
122			call_sv (get_sv ("Async::Interrupt::DIED", 1), G_VOID \| G_DISCARD \| G_EVAL \| G_KEEPERR);
123
124			sv_setpvn (ERRSV, "", 0);
125			}
126
127			if (saveerr)
128			sv_setsv (ERRSV, saveerr);
129	root	1.1
130	root	1.2	{
131			SV *oldhook = PL_diehook;
132			PL_diehook = savedie;
133			SvREFCNT_dec (oldhook);
134			}
135
136			POPSTACK;
137	root	1.1	}
138
139	root	1.2	errno = old_errno;
140	root	1.1	}
141
142			static void
143	root	1.2	handle_asyncs (void)
144	root	1.1	{
145	root	1.2	int i;
146
147	root	1.1	async_pending = 0;
148	root	1.2
149			for (i = AvFILLp (asyncs); i >= 0; --i)
150			{
151	root	1.6	async_t *async = SvASYNC_nrv (AvARRAY (asyncs)[i]);
152	root	1.2
153			if (async->pending && !async->blocked)
154			handle_async (async);
155			}
156	root	1.1	}
157
158	root	1.5	#if HAS_SA_SIGINFO
159	root	1.1	static Signal_t async_sighandler (int signum, siginfo_t si, void sarg)
160			{
161			if (signum == 9)
162	root	1.2	handle_asyncs ();
163	root	1.1	else
164			old_sighandler (signum, si, sarg);
165			}
166			#else
167	root	1.3	static Signal_t async_sighandler (int signum)
168	root	1.1	{
169			if (signum == 9)
170	root	1.3	handle_asyncs ();
171	root	1.1	else
172			old_sighandler (signum);
173			}
174			#endif
175
176	root	1.4	static void
177	root	1.6	async_sigsend (int signum)
178			{
179			async_signal (sig_async [signum], 0);
180			}
181
182	root	1.7	#define block(async) ++(async)->blocked
183
184	root	1.6	static void
185	root	1.7	unblock (async_t *async)
186	root	1.4	{
187			--async->blocked;
188			if (async->pending && !async->blocked)
189			handle_async (async);
190	root	1.7	}
191	root	1.4
192	root	1.7	static void
193			scope_block_cb (pTHX_ void *async_sv)
194			{
195			async_t async = SvASYNC_nrv ((SV )async_sv);
196			unblock (async);
197	root	1.4	SvREFCNT_dec (async_sv);
198			}
199	root	1.1
200			MODULE = Async::Interrupt PACKAGE = Async::Interrupt
201
202			BOOT:
203			old_sighandler = PL_sighandlerp;
204			PL_sighandlerp = async_sighandler;
205			sig_pending = &PL_sig_pending;
206			psig_pend = PL_psig_pend;
207			asyncs = newAV ();
208	root	1.4	CvNODEBUG_on (get_cv ("Async::Interrupt::scope_block", 0)); /* otherwise calling scope can be the debugger */
209	root	1.1
210	root	1.3	PROTOTYPES: DISABLE
211
212	root	1.6	void
213			_alloc (SV cb, void c_cb, void c_arg, SV fh_r, SV fh_w, SV signl)
214			PPCODE:
215	root	1.1	{
216	root	1.8	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	root	1.6	async_t *async;
220	root	1.1
221	root	1.6	Newz (0, async, 1, async_t);
222	root	1.1
223	root	1.6	XPUSHs (sv_2mortal (newSViv (PTR2IV (async))));
224			av_push (asyncs, TOPs);
225	root	1.2
226	root	1.6	async->fh_r = fd_r >= 0 ? newSVsv (fh_r) : 0; async->fd_r = fd_r;
227			async->fh_w = fd_w >= 0 ? newSVsv (fh_w) : 0; async->fd_w = fd_w;
228			async->fd_wlen = 1;
229			async->fd_enable = 1;
230			async->cb = cv;
231			async->c_cb = c_cb;
232			async->c_arg = c_arg;
233	root	1.8	async->signum = SvOK (signl) ? s_signum_croak (signl) : 0;
234	root	1.6
235			if (async->signum)
236			{
237			if (async->signum < 0)
238			croak ("Async::Interrupt::new got passed illegal signal name or number: %s", SvPV_nolen (signl));
239
240			sig_async [async->signum] = async;
241			#if _WIN32
242			signal (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			}
252	root	1.1	}
253
254			void
255	root	1.6	signal_func (async_t *async)
256	root	1.1	PPCODE:
257			EXTEND (SP, 2);
258			PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
259	root	1.6	PUSHs (sv_2mortal (newSViv (PTR2IV (async))));
260	root	1.1
261			void
262	root	1.6	signal (async_t *async, int value = 0)
263	root	1.1	CODE:
264	root	1.6	async_signal (async, value);
265	root	1.2
266			void
267	root	1.6	block (async_t *async)
268	root	1.2	CODE:
269	root	1.7	block (async);
270	root	1.2
271			void
272	root	1.6	unblock (async_t *async)
273	root	1.2	CODE:
274	root	1.7	unblock (async);
275	root	1.1
276			void
277	root	1.4	scope_block (SV *self)
278			CODE:
279			{
280			SV *async_sv = SvRV (self);
281	root	1.6	async_t *async = SvASYNC_nrv (async_sv);
282	root	1.7	block (async);
283	root	1.4
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
289			void
290	root	1.6	pipe_enable (async_t *async)
291			ALIAS:
292			pipe_enable = 1
293			pipe_disable = 0
294			CODE:
295			async->fd_enable = ix;
296
297			void
298	root	1.1	DESTROY (SV *self)
299			CODE:
300			{
301			int i;
302			SV *async_sv = SvRV (self);
303	root	1.6	async_t *async = SvASYNC_nrv (async_sv);
304	root	1.1
305	root	1.2	for (i = AvFILLp (asyncs); i >= 0; --i)
306	root	1.1	if (AvARRAY (asyncs)[i] == async_sv)
307			{
308			if (i < AvFILLp (asyncs))
309	root	1.3	AvARRAY (asyncs)[i] = AvARRAY (asyncs)[AvFILLp (asyncs)];
310	root	1.1
311			assert (av_pop (asyncs) == async_sv);
312			goto found;
313			}
314
315			if (!PL_dirty)
316	root	1.2	warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report");
317	root	1.1
318			found:
319	root	1.6
320			if (async->signum)
321			{
322			#if _WIN32
323			signal (async->signum, SIG_DFL);
324			#else
325			{
326			struct sigaction sa = { };
327			sa.sa_handler = SIG_DFL;
328			sigaction (async->signum, &sa, 0);
329			}
330			#endif
331			}
332
333	root	1.2	SvREFCNT_dec (async->fh_r);
334			SvREFCNT_dec (async->fh_w);
335	root	1.1	SvREFCNT_dec (async->cb);
336
337			Safefree (async);
338			}
339