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_data, int value);
  void *c_data;
  SV *fh;

  int fd;
  atomic_t value;
  atomic_t signalled;
};

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

  async->value = value;

  if (!async->signalled)
    {
      async->signalled = 1;

      if (async->fd >= 0)
        write (async->fd, async, 1);
    }

  async_pending = 1;
  psig_pend [9] = 1;
  *sig_pending  = 1;
}

static void
async_handle (void)
{
  async_pending = 0;
}

#if defined(HAS_SIGACTION) && defined(SA_SIGINFO)
static Signal_t async_sighandler (int signum, siginfo_t *si, void *sarg)
{
  if (signum == 9)
    async_handle ();
  else
    old_sighandler (signum, si, sarg);
}
#else
static Signal_t async_sighandler (int)
{
  if (signum == 9)
    async_handle ();
  else
    old_sighandler (signum);
}
#endif


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 ();

SV *
_alloc (SV *cb, void *c_cb, void *c_data, SV *fh)
        CODE:
{
        SV *cv = SvOK (cb) ? SvREFCNT_inc_NN (get_cb (cb)) : 0;
        int fd = SvOK (fh) ? extract_fd (fh, 1) : -1;

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

        async->fh     = fd >= 0 ? newSVsv (fh) : 0;
        async->fd     = fd;
        async->cb     = cb;
        async->c_cb   = c_cb;
        async->c_data = c_data;

        RETVAL = newSViv (PTR2IV (async));
        av_push (asyncs, RETVAL);

        RETVAL = newRV_noinc (RETVAL);
}
        OUTPUT:
        RETVAL

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

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

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

        Safefree (async);
}

Revision:	1.1
Committed:	Thu Jul 2 13:41:44 2009 UTC (14 years, 11 months ago) by root
Branch:	MAIN
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			void (c_cb)(pTHX_ void c_data, int value);
45			void *c_data;
46			SV *fh;
47
48			int fd;
49			atomic_t value;
50			atomic_t signalled;
51			};
52
53			/* the main workhorse to signal */
54			static void
55			async_signal (void *signal_arg, int value)
56			{
57			struct async async = (struct async )signal_arg;
58
59			async->value = value;
60
61			if (!async->signalled)
62			{
63			async->signalled = 1;
64
65			if (async->fd >= 0)
66			write (async->fd, async, 1);
67			}
68
69			async_pending = 1;
70			psig_pend [9] = 1;
71			*sig_pending = 1;
72			}
73
74			static void
75			async_handle (void)
76			{
77			async_pending = 0;
78			}
79
80			#if defined(HAS_SIGACTION) && defined(SA_SIGINFO)
81			static Signal_t async_sighandler (int signum, siginfo_t si, void sarg)
82			{
83			if (signum == 9)
84			async_handle ();
85			else
86			old_sighandler (signum, si, sarg);
87			}
88			#else
89			static Signal_t async_sighandler (int)
90			{
91			if (signum == 9)
92			async_handle ();
93			else
94			old_sighandler (signum);
95			}
96			#endif
97
98
99			MODULE = Async::Interrupt PACKAGE = Async::Interrupt
100
101			BOOT:
102			old_sighandler = PL_sighandlerp;
103			PL_sighandlerp = async_sighandler;
104			sig_pending = &PL_sig_pending;
105			psig_pend = PL_psig_pend;
106			asyncs = newAV ();
107
108			SV *
109			_alloc (SV cb, void c_cb, void c_data, SV fh)
110			CODE:
111			{
112			SV *cv = SvOK (cb) ? SvREFCNT_inc_NN (get_cb (cb)) : 0;
113			int fd = SvOK (fh) ? extract_fd (fh, 1) : -1;
114
115			struct async *async;
116			Newz (0, async, 1, struct async);
117
118			async->fh = fd >= 0 ? newSVsv (fh) : 0;
119			async->fd = fd;
120			async->cb = cb;
121			async->c_cb = c_cb;
122			async->c_data = c_data;
123
124			RETVAL = newSViv (PTR2IV (async));
125			av_push (asyncs, RETVAL);
126
127			RETVAL = newRV_noinc (RETVAL);
128			}
129			OUTPUT:
130			RETVAL
131
132			void
133			signal_cb (SV *self)
134			PPCODE:
135			EXTEND (SP, 2);
136			PUSHs (sv_2mortal (newSViv (PTR2IV (async_signal))));
137			PUSHs (sv_2mortal (newSViv (SvIV (SvRV (self)))));
138
139			void
140			signal (SV *self, int value = 0)
141			CODE:
142			async_signal (INT2PTR (void *, SvIV (SvRV (self))), value);
143
144			void
145			DESTROY (SV *self)
146			CODE:
147			{
148			int i;
149			SV *async_sv = SvRV (self);
150			struct async async = INT2PTR (struct async , SvIV (async_sv));
151
152			for (i = AvFILLp (asyncs); i > 0; --i)
153			if (AvARRAY (asyncs)[i] == async_sv)
154			{
155			if (i < AvFILLp (asyncs))
156			AvARRAY (asyncs)[i] == AvARRAY (asyncs)[AvFILLp (asyncs)];
157
158			assert (av_pop (asyncs) == async_sv);
159			goto found;
160			}
161
162			if (!PL_dirty)
163			warn ("Async::Interrupt::DESTROY could not find async object in list of asyncs, please report!");
164
165			found:
166			SvREFCNT_dec (async->fh);
167			SvREFCNT_dec (async->cb);
168
169			Safefree (async);
170			}
171