Async-Interrupt/Interrupt.xs

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.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

static int
extract_fd (SV *fh, int wr)
{
  int fd = PerlIO_fileno (wr ? IoOFP (sv_2io (fh)) : IoIFP (sv_2io (fh)));

  if (fd < 0)
    croak ("illegal fh argument, either not an OS file or read/write mode mismatch");

  return fd;
}

static SV *
get_cb (SV *cb_sv)
{
  HV *st;
  GV *gvp;
  CV *cv;

  if (!SvOK (cb_sv))
    return 0;

  cv = sv_2cv (cb_sv, &st, &gvp, 0);

  if (!cv)
    croak ("Async::Interrupt callback must be undef or a CODE reference");

  return (SV *)cv;
}

static AV *asyncs;

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

  int fd_r, fd_w;
  atomic_t value;
  atomic_t pending;
};

/* the main workhorse to signal */
static void
async_signal (void *signal_arg, int value)
{
  struct async *async = (struct async *)signal_arg;
  int pending = async->pending;

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

  if (!pending && async->fd_w >= 0)
    write (async->fd_w, async, 1);
}

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

  async->pending = 0;

  /* drain pipe */
  if (async->fd_r >= 0)
    {
      char dummy [4];

      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)
    {
      struct async *async = INT2PTR (struct async *, SvIVX (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
scope_block_cb (pTHX_ void *async_sv)
{
  struct async *async = INT2PTR (struct async *, SvIVX ((SV *)async_sv));

  --async->blocked;
  if (async->pending && !async->blocked)
    handle_async (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

SV *
_alloc (SV *cb, void *c_cb, void *c_arg, SV *fh_r, SV *fh_w)
        CODE:
{
        SV *cv   = SvOK (cb) ? SvREFCNT_inc (get_cb (cb)) : 0;
        int fd_r = SvOK (fh_r) ? extract_fd (fh_r, 0) : -1;
        int fd_w = SvOK (fh_w) ? extract_fd (fh_w, 1) : -1;
        struct async *async;

        Newz (0, async, 1, struct async);

        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->cb     = cv;
        async->c_cb   = c_cb;
        async->c_arg  = c_arg;

        printf ("r,w %d,%d\n", fd_r, fd_w);//D

        RETVAL = newSViv (PTR2IV (async));
        av_push (asyncs, RETVAL);
}
        OUTPUT:
        RETVAL

void
signal_func (SV *self)
        PPCODE:
        EXTEND (SP, 2);
        PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
        PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (self)))));

void
signal (SV *self, int value = 0)
        CODE:
        async_signal (INT2PTR (void *, SvIVX (SvRV (self))), value);

void
block (SV *self)
        CODE:
{
        struct async *async = INT2PTR (struct async *, SvIVX (SvRV (self)));
        ++async->blocked;
}

void
unblock (SV *self)
        CODE:
{
        struct async *async = INT2PTR (struct async *, SvIVX (SvRV (self)));
        --async->blocked;
        if (async->pending && !async->blocked)
          handle_async (async);
}

void
scope_block (SV *self)
        CODE:
{
        SV *async_sv = SvRV (self);
        struct async *async = INT2PTR (struct async *, SvIVX (async_sv));
        ++async->blocked;

        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
DESTROY (SV *self)
        CODE:
{
        int i;
        SV *async_sv = SvRV (self);
        struct async *async = INT2PTR (struct async *, SvIVX (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:
        SvREFCNT_dec (async->fh_r);
        SvREFCNT_dec (async->fh_w);
        SvREFCNT_dec (async->cb);

        Safefree (async);
}

Revision:	1.5
Committed:	Fri Jul 3 21:11:22 2009 UTC (14 years, 11 months ago) by root
Branch:	MAIN
CVS Tags:	rel-0_03
Changes since 1.4:	+16 -2 lines
Log Message:	0.03
#	Content
1	#include "EXTERN.h"
2	#include "perl.h"
3	#include "XSUB.h"
4
5	typedef volatile sig_atomic_t atomic_t;
6
7	static int sig_pending, psig_pend; /* make local copies because of missing THX */
8	static Sighandler_t old_sighandler;
9	static atomic_t async_pending;
10
11	#define PERL_VERSION_ATLEAST(a,b,c) \
12	(PERL_REVISION > (a) \
13	\|\| (PERL_REVISION == (a) \
14	&& (PERL_VERSION > (b) \
15	\|\| (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))
16
17	#if defined(HAS_SIGACTION) && defined(SA_SIGINFO)
18	# define HAS_SA_SIGINFO 1
19	#endif
20
21	#if !PERL_VERSION_ATLEAST(5,10,0)
22	# undef HAS_SA_SIGINFO
23	#endif
24
25	static int
26	extract_fd (SV *fh, int wr)
27	{
28	int fd = PerlIO_fileno (wr ? IoOFP (sv_2io (fh)) : IoIFP (sv_2io (fh)));
29
30	if (fd < 0)
31	croak ("illegal fh argument, either not an OS file or read/write mode mismatch");
32
33	return fd;
34	}
35
36	static SV *
37	get_cb (SV *cb_sv)
38	{
39	HV *st;
40	GV *gvp;
41	CV *cv;
42
43	if (!SvOK (cb_sv))
44	return 0;
45
46	cv = sv_2cv (cb_sv, &st, &gvp, 0);
47
48	if (!cv)
49	croak ("Async::Interrupt callback must be undef or a CODE reference");
50
51	return (SV *)cv;
52	}
53
54	static AV *asyncs;
55
56	struct async {
57	SV *cb;
58	void (c_cb)(pTHX_ void c_arg, int value);
59	void *c_arg;
60	SV fh_r, fh_w;
61	int blocked;
62
63	int fd_r, fd_w;
64	atomic_t value;
65	atomic_t pending;
66	};
67
68	/* the main workhorse to signal */
69	static void
70	async_signal (void *signal_arg, int value)
71	{
72	struct async async = (struct async )signal_arg;
73	int pending = async->pending;
74
75	async->value = value;
76	async->pending = 1;
77	async_pending = 1;
78	psig_pend [9] = 1;
79	*sig_pending = 1;
80
81	if (!pending && async->fd_w >= 0)
82	write (async->fd_w, async, 1);
83	}
84
85	static void
86	handle_async (struct async *async)
87	{
88	int old_errno = errno;
89	int value = async->value;
90
91	async->pending = 0;
92
93	/* drain pipe */
94	if (async->fd_r >= 0)
95	{
96	char dummy [4];
97
98	while (read (async->fd_r, dummy, sizeof (dummy)) == sizeof (dummy))
99	;
100	}
101
102	if (async->c_cb)
103	{
104	dTHX;
105	async->c_cb (aTHX_ async->c_arg, value);
106	}
107
108	if (async->cb)
109	{
110	dSP;
111
112	SV *saveerr = SvOK (ERRSV) ? sv_mortalcopy (ERRSV) : 0;
113	SV *savedie = PL_diehook;
114
115	PL_diehook = 0;
116
117	PUSHSTACKi (PERLSI_SIGNAL);
118
119	PUSHMARK (SP);
120	XPUSHs (sv_2mortal (newSViv (value)));
121	PUTBACK;
122	call_sv (async->cb, G_VOID \| G_DISCARD \| G_EVAL);
123
124	if (SvTRUE (ERRSV))
125	{
126	SPAGAIN;
127
128	PUSHMARK (SP);
129	PUTBACK;
130	call_sv (get_sv ("Async::Interrupt::DIED", 1), G_VOID \| G_DISCARD \| G_EVAL \| G_KEEPERR);
131
132	sv_setpvn (ERRSV, "", 0);
133	}
134
135	if (saveerr)
136	sv_setsv (ERRSV, saveerr);
137
138	{
139	SV *oldhook = PL_diehook;
140	PL_diehook = savedie;
141	SvREFCNT_dec (oldhook);
142	}
143
144	POPSTACK;
145	}
146
147	errno = old_errno;
148	}
149
150	static void
151	handle_asyncs (void)
152	{
153	int i;
154
155	async_pending = 0;
156
157	for (i = AvFILLp (asyncs); i >= 0; --i)
158	{
159	struct async async = INT2PTR (struct async , SvIVX (AvARRAY (asyncs)[i]));
160
161	if (async->pending && !async->blocked)
162	handle_async (async);
163	}
164	}
165
166	#if HAS_SA_SIGINFO
167	static Signal_t async_sighandler (int signum, siginfo_t si, void sarg)
168	{
169	if (signum == 9)
170	handle_asyncs ();
171	else
172	old_sighandler (signum, si, sarg);
173	}
174	#else
175	static Signal_t async_sighandler (int signum)
176	{
177	if (signum == 9)
178	handle_asyncs ();
179	else
180	old_sighandler (signum);
181	}
182	#endif
183
184	static void
185	scope_block_cb (pTHX_ void *async_sv)
186	{
187	struct async async = INT2PTR (struct async , SvIVX ((SV *)async_sv));
188
189	--async->blocked;
190	if (async->pending && !async->blocked)
191	handle_async (async);
192
193	SvREFCNT_dec (async_sv);
194	}
195
196	MODULE = Async::Interrupt PACKAGE = Async::Interrupt
197
198	BOOT:
199	old_sighandler = PL_sighandlerp;
200	PL_sighandlerp = async_sighandler;
201	sig_pending = &PL_sig_pending;
202	psig_pend = PL_psig_pend;
203	asyncs = newAV ();
204	CvNODEBUG_on (get_cv ("Async::Interrupt::scope_block", 0)); /* otherwise calling scope can be the debugger */
205
206	PROTOTYPES: DISABLE
207
208	SV *
209	_alloc (SV cb, void c_cb, void c_arg, SV fh_r, SV *fh_w)
210	CODE:
211	{
212	SV *cv = SvOK (cb) ? SvREFCNT_inc (get_cb (cb)) : 0;
213	int fd_r = SvOK (fh_r) ? extract_fd (fh_r, 0) : -1;
214	int fd_w = SvOK (fh_w) ? extract_fd (fh_w, 1) : -1;
215	struct async *async;
216
217	Newz (0, async, 1, struct async);
218
219	async->fh_r = fd_r >= 0 ? newSVsv (fh_r) : 0; async->fd_r = fd_r;
220	async->fh_w = fd_w >= 0 ? newSVsv (fh_w) : 0; async->fd_w = fd_w;
221	async->cb = cv;
222	async->c_cb = c_cb;
223	async->c_arg = c_arg;
224
225	printf ("r,w %d,%d\n", fd_r, fd_w);//D
226
227	RETVAL = newSViv (PTR2IV (async));
228	av_push (asyncs, RETVAL);
229	}
230	OUTPUT:
231	RETVAL
232
233	void
234	signal_func (SV *self)
235	PPCODE:
236	EXTEND (SP, 2);
237	PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
238	PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (self)))));
239
240	void
241	signal (SV *self, int value = 0)
242	CODE:
243	async_signal (INT2PTR (void *, SvIVX (SvRV (self))), value);
244
245	void
246	block (SV *self)
247	CODE:
248	{
249	struct async async = INT2PTR (struct async , SvIVX (SvRV (self)));
250	++async->blocked;
251	}
252
253	void
254	unblock (SV *self)
255	CODE:
256	{
257	struct async async = INT2PTR (struct async , SvIVX (SvRV (self)));
258	--async->blocked;
259	if (async->pending && !async->blocked)
260	handle_async (async);
261	}
262
263	void
264	scope_block (SV *self)
265	CODE:
266	{
267	SV *async_sv = SvRV (self);
268	struct async async = INT2PTR (struct async , SvIVX (async_sv));
269	++async->blocked;
270
271	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
272	SAVEDESTRUCTOR_X (scope_block_cb, (void *)SvREFCNT_inc (async_sv));
273	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
274	}
275
276	void
277	DESTROY (SV *self)
278	CODE:
279	{
280	int i;
281	SV *async_sv = SvRV (self);
282	struct async async = INT2PTR (struct async , SvIVX (async_sv));
283
284	for (i = AvFILLp (asyncs); i >= 0; --i)
285	if (AvARRAY (asyncs)[i] == async_sv)
286	{
287	if (i < AvFILLp (asyncs))
288	AvARRAY (asyncs)[i] = AvARRAY (asyncs)[AvFILLp (asyncs)];
289
290	assert (av_pop (asyncs) == async_sv);
291	goto found;
292	}
293
294	if (!PL_dirty)
295	warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report");
296
297	found:
298	SvREFCNT_dec (async->fh_r);
299	SvREFCNT_dec (async->fh_w);
300	SvREFCNT_dec (async->cb);
301
302	Safefree (async);
303	}
304