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;

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