Coro/EV/EV.xs

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

#include <stddef.h>
#include <assert.h>
#include <string.h>

#include "EVAPI.h"
#include "../Coro/CoroAPI.h"

static struct ev_prepare scheduler;
static struct ev_idle idler;
static int inhibit;

static void
idle_cb (EV_P_ ev_idle *w, int revents)
{
  ev_idle_stop (EV_A, w);
}

static void
prepare_cb (EV_P_ ev_prepare *w, int revents)
{
  static int incede;

  if (inhibit)
    return;

  ++incede;

  CORO_CEDE_NOTSELF;

  while (CORO_NREADY >= incede && CORO_CEDE)
    ;

  /* if still ready, then we have lower-priority coroutines.
   * poll anyways, but do not block.
   */
  if (CORO_NREADY >= incede)
    {
      if (!ev_is_active (&idler))
        ev_idle_start (EV_A, &idler);
    }
  else
    {
      if (ev_is_active (&idler))
        ev_idle_stop (EV_A, &idler);
    }

  --incede;
}

static void
readyhook (void)
{
  if (!ev_is_active (&idler))
    ev_idle_start (EV_DEFAULT_UC, &idler);
}

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

static void
once_cb (int revents, void *arg)
{
  SV *data = (SV *)arg;

  CORO_READY (data);
  sv_setiv (data, revents);
  SvREFCNT_dec (data);
}

static int
slf_check_once (pTHX_ struct CoroSLF *frame)
{
  SV *data = (SV *)frame->data;

  /* return early when an exception is pending */
  if (CORO_THROW)
    return 0;

  if (SvROK (data))
    return 1; /* repeat until we have been signalled */
  else
    {
      dSP;

      XPUSHs (data);

      PUTBACK;
      return 0;
    }
}

static void
slf_init_timed_io (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
{
  SV *data;
  int fd;

  if (items < 2 || items > 3)
    croak ("Coro::EV::timed_io_once requires exactly two or three parameters, not %d.\n", items);

  SvGETMAGIC (arg [0]);
  SvGETMAGIC (arg [1]);

  if (items >= 3)
    SvGETMAGIC (arg [2]);

  fd = sv_fileno (arg [0]);

  data = sv_2mortal (newRV_inc (CORO_CURRENT));
  frame->data    = (void *)data;
  frame->prepare = GCoroAPI->prepare_schedule;
  frame->check   = slf_check_once;

  return;
  ev_once (
    EV_DEFAULT_UC,
    fd,
    SvIV (arg [1]),
    items >= 3 && SvOK (arg [2]) ? SvNV (arg [2]) : -1.,
    once_cb,
    SvREFCNT_inc (data)
  );
}

static void
slf_init_timer (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
{
  SV *data;
  NV after;

  if (items > 1)
    croak ("Coro::EV::timer_once requires at most one parameter, not %d.\n", items);

  data = sv_2mortal (newRV_inc (CORO_CURRENT));
  frame->data    = (void *)data;
  frame->prepare = GCoroAPI->prepare_schedule;
  frame->check   = slf_check_once;

  after = items ? SvNV (arg [0]) : 0;

  ev_once (
    EV_DEFAULT_UC,
    -1,
    0,
    after >= 0. ? after : 0.,
    once_cb,
    SvREFCNT_inc (data)
  );
}

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

typedef struct
{
  ev_io io;
  ev_timer tw;
  SV *data;
} coro_dir;

typedef struct
{
  coro_dir r, w;
} coro_handle;

static int
handle_free (pTHX_ SV *sv, MAGIC *mg)
{
  coro_handle *data = (coro_handle *)mg->mg_ptr;
  mg->mg_ptr = 0;

  ev_io_stop    (EV_DEFAULT_UC, &data->r.io); ev_io_stop    (EV_DEFAULT_UC, &data->w.io);
  ev_timer_stop (EV_DEFAULT_UC, &data->r.tw); ev_timer_stop (EV_DEFAULT_UC, &data->w.tw);

  return 0;
}

static MGVTBL handle_vtbl = { 0,  0,  0,  0, handle_free };

static void
handle_cb (coro_dir *dir, int success)
{
  ev_io_stop    (EV_DEFAULT_UC, &dir->io);
  ev_timer_stop (EV_DEFAULT_UC, &dir->tw);

  CORO_READY (dir->data);
  sv_setiv (dir->data, success);
}

static void
handle_io_cb (EV_P_ ev_io *w, int revents)
{
  handle_cb ((coro_dir *)(((char *)w) - offsetof (coro_dir, io)), 1);
}

static void
handle_timer_cb (EV_P_ ev_timer *w, int revents)
{
  handle_cb ((coro_dir *)(((char *)w) - offsetof (coro_dir, tw)), 0);
}

static int
slf_check_rw (pTHX_ struct CoroSLF *frame)
{
  SV *data = (SV *)frame->data;

  /* return early when an exception is pending */
  if (CORO_THROW)
    return 0;

  if (SvROK (data))
    return 1;
  else
    {
      dSP;

      XPUSHs (data);

      PUTBACK;
      return 0;
    }
}

static void
slf_init_rw (pTHX_ struct CoroSLF *frame, SV *arg, int wr)
{
  AV *handle = (AV *)SvRV (arg);
  SV *data_sv = AvARRAY (handle)[5];
  coro_handle *data;
  coro_dir *dir;
  assert (AvFILLp (handle) >= 7);

  if (!SvOK (data_sv))
    {
      int fno = sv_fileno (AvARRAY (handle)[0]);
      data_sv = AvARRAY (handle)[5] = NEWSV (0, sizeof (coro_handle));
      SvPOK_only (data_sv);
      SvREADONLY_on (data_sv);
      data = (coro_handle *)SvPVX (data_sv);
      memset (data, 0, sizeof (coro_handle));

      ev_io_init (&data->r.io, handle_io_cb, fno, EV_READ);
      ev_io_init (&data->w.io, handle_io_cb, fno, EV_WRITE);
      ev_init    (&data->r.tw, handle_timer_cb);
      ev_init    (&data->w.tw, handle_timer_cb);

      sv_magicext (data_sv, 0, PERL_MAGIC_ext, &handle_vtbl, (char *)data, 0);
    }
  else
    data = (coro_handle *)SvPVX (data_sv);

  dir = wr ? &data->w : &data->r;

  if (ev_is_active (&dir->io) || ev_is_active (&dir->tw))
    croak ("recursive invocation of readable_ev or writable_ev (concurrent Coro::Handle calls on same handle?), detected");

  dir->data = sv_2mortal (newRV_inc (CORO_CURRENT));

  {
    SV *to = AvARRAY (handle)[2];

    if (SvOK (to))
      {
        ev_timer_set (&dir->tw, 0., SvNV (to));
        ev_timer_again (EV_DEFAULT_UC, &dir->tw);
      }
  }

  ev_io_start (EV_DEFAULT_UC, &dir->io);

  frame->data    = (void *)dir->data;
  frame->prepare = GCoroAPI->prepare_schedule;
  frame->check   = slf_check_rw;
}

static void
slf_init_readable (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
{
  slf_init_rw (aTHX_ frame, arg [0], 0);
}

static void
slf_init_writable (pTHX_ struct CoroSLF *frame, CV *cv, SV **arg, int items)
{
  slf_init_rw (aTHX_ frame, arg [0], 1);
}

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

MODULE = Coro::EV                PACKAGE = Coro::EV

PROTOTYPES: ENABLE

BOOT:
{
        I_EV_API   ("Coro::EV");
        I_CORO_API ("Coro::EV");

        EV_DEFAULT; /* make sure it is initialised */

        ev_prepare_init (&scheduler, prepare_cb);
        ev_set_priority (&scheduler, EV_MINPRI);
        ev_prepare_start (EV_DEFAULT_UC, &scheduler);
        ev_unref (EV_DEFAULT_UC);

        ev_idle_init (&idler, idle_cb);
        ev_set_priority (&idler, EV_MINPRI);

        if (!CORO_READYHOOK) /* do not override if Coro::AnyEvent already did */
          {
            CORO_READYHOOK = readyhook;
            CORO_READYHOOK (); /* make sure we don't miss previous ready's */
          }
}

void
_set_readyhook ()
        CODE:
        CORO_READYHOOK = readyhook;
        CORO_READYHOOK ();

void
_loop_oneshot ()
        CODE:
{
        /* inhibit the prepare watcher, as we know we are the only
         * ready coroutine and we don't want it to start an idle watcher
         * just because of the fallback idle coro being of lower priority.
         */
        ++inhibit;

        /* same reasoning as above, make sure it is stopped */
        if (ev_is_active (&idler))
          ev_idle_stop (EV_DEFAULT_UC, &idler);
        ev_run (EV_DEFAULT_UC, EVRUN_ONCE);
        --inhibit;
}

void
timed_io_once (...)
        PROTOTYPE: $$;$
        CODE:
        CORO_EXECUTE_SLF_XS (slf_init_timed_io);

void
timer_once (...)
        PROTOTYPE: $
        CODE:
        CORO_EXECUTE_SLF_XS (slf_init_timer);

void
_poll (...)
        PROTOTYPE:
        CODE:
        CORO_EXECUTE_SLF_XS (slf_init_timer);

PROTOTYPES: DISABLE

void
_readable_ev (...)
        CODE:
        items = 1; /* ignore the remaining args for speed inside Coro::Handle */
        CORO_EXECUTE_SLF_XS (slf_init_readable);

void
_writable_ev (...)
        CODE:
        items = 1; /* ignore the remaining args for speed inside Coro::Handle */
        CORO_EXECUTE_SLF_XS (slf_init_writable);

Revision:	1.38
Committed:	Wed Oct 31 17:03:30 2012 UTC (11 years, 7 months ago) by root
Branch:	MAIN
Changes since 1.37:	+1 -0 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.10	#include <stddef.h>
6	root	1.1	#include <assert.h>
7			#include <string.h>
8
9			#include "EVAPI.h"
10			#include "../Coro/CoroAPI.h"
11
12	root	1.2	static struct ev_prepare scheduler;
13			static struct ev_idle idler;
14	root	1.9	static int inhibit;
15	root	1.2
16			static void
17	root	1.10	idle_cb (EV_P_ ev_idle *w, int revents)
18	root	1.2	{
19	root	1.12	ev_idle_stop (EV_A, w);
20	root	1.2	}
21
22			static void
23	root	1.10	prepare_cb (EV_P_ ev_prepare *w, int revents)
24	root	1.2	{
25	root	1.5	static int incede;
26
27	root	1.9	if (inhibit)
28			return;
29
30	root	1.5	++incede;
31
32			CORO_CEDE_NOTSELF;
33
34	root	1.6	while (CORO_NREADY >= incede && CORO_CEDE)
35	root	1.2	;
36
37			/* if still ready, then we have lower-priority coroutines.
38	root	1.5	* poll anyways, but do not block.
39	root	1.2	*/
40	root	1.16	if (CORO_NREADY >= incede)
41			{
42			if (!ev_is_active (&idler))
43			ev_idle_start (EV_A, &idler);
44			}
45			else
46			{
47			if (ev_is_active (&idler))
48			ev_idle_stop (EV_A, &idler);
49			}
50	root	1.5
51			--incede;
52	root	1.2	}
53
54	root	1.15	static void
55			readyhook (void)
56			{
57	root	1.17	if (!ev_is_active (&idler))
58			ev_idle_start (EV_DEFAULT_UC, &idler);
59	root	1.15	}
60
61	root	1.10	/*****************************************************************************/
62
63	root	1.18	static void
64			once_cb (int revents, void *arg)
65			{
66			SV data = (SV )arg;
67
68			CORO_READY (data);
69			sv_setiv (data, revents);
70	root	1.22	SvREFCNT_dec (data);
71	root	1.18	}
72
73			static int
74			slf_check_once (pTHX_ struct CoroSLF *frame)
75			{
76			SV data = (SV )frame->data;
77
78	root	1.22	/* return early when an exception is pending */
79			if (CORO_THROW)
80			return 0;
81
82	root	1.18	if (SvROK (data))
83			return 1; /* repeat until we have been signalled */
84			else
85			{
86			dSP;
87
88			XPUSHs (data);
89
90			PUTBACK;
91			return 0;
92			}
93			}
94
95			static void
96	root	1.21	slf_init_timed_io (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
97	root	1.18	{
98			SV *data;
99	root	1.37	int fd;
100	root	1.18
101			if (items < 2 \|\| items > 3)
102			croak ("Coro::EV::timed_io_once requires exactly two or three parameters, not %d.\n", items);
103
104	root	1.36	SvGETMAGIC (arg [0]);
105			SvGETMAGIC (arg [1]);
106
107			if (items >= 3)
108			SvGETMAGIC (arg [2]);
109
110	root	1.37	fd = sv_fileno (arg [0]);
111
112	root	1.18	data = sv_2mortal (newRV_inc (CORO_CURRENT));
113			frame->data = (void *)data;
114			frame->prepare = GCoroAPI->prepare_schedule;
115			frame->check = slf_check_once;
116
117	root	1.38	return;
118	root	1.18	ev_once (
119			EV_DEFAULT_UC,
120	root	1.37	fd,
121	root	1.18	SvIV (arg [1]),
122			items >= 3 && SvOK (arg [2]) ? SvNV (arg [2]) : -1.,
123			once_cb,
124	root	1.22	SvREFCNT_inc (data)
125	root	1.18	);
126			}
127
128			static void
129	root	1.21	slf_init_timer (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
130	root	1.18	{
131			SV *data;
132			NV after;
133
134	root	1.24	if (items > 1)
135			croak ("Coro::EV::timer_once requires at most one parameter, not %d.\n", items);
136	root	1.18
137			data = sv_2mortal (newRV_inc (CORO_CURRENT));
138	root	1.19	frame->data = (void *)data;
139			frame->prepare = GCoroAPI->prepare_schedule;
140			frame->check = slf_check_once;
141	root	1.18
142	root	1.24	after = items ? SvNV (arg [0]) : 0;
143	root	1.18
144			ev_once (
145			EV_DEFAULT_UC,
146			-1,
147			0,
148			after >= 0. ? after : 0.,
149			once_cb,
150	root	1.22	SvREFCNT_inc (data)
151	root	1.18	);
152			}
153
154			/*****************************************************************************/
155
156	root	1.10	typedef struct
157			{
158			ev_io io;
159			ev_timer tw;
160	root	1.18	SV *data;
161	root	1.10	} coro_dir;
162
163			typedef struct
164			{
165			coro_dir r, w;
166			} coro_handle;
167
168			static int
169			handle_free (pTHX_ SV sv, MAGIC mg)
170			{
171			coro_handle data = (coro_handle )mg->mg_ptr;
172			mg->mg_ptr = 0;
173
174	root	1.15	ev_io_stop (EV_DEFAULT_UC, &data->r.io); ev_io_stop (EV_DEFAULT_UC, &data->w.io);
175			ev_timer_stop (EV_DEFAULT_UC, &data->r.tw); ev_timer_stop (EV_DEFAULT_UC, &data->w.tw);
176	root	1.10
177			return 0;
178			}
179
180			static MGVTBL handle_vtbl = { 0, 0, 0, 0, handle_free };
181
182			static void
183	root	1.18	handle_cb (coro_dir *dir, int success)
184	root	1.10	{
185	root	1.15	ev_io_stop (EV_DEFAULT_UC, &dir->io);
186			ev_timer_stop (EV_DEFAULT_UC, &dir->tw);
187	root	1.10
188	root	1.18	CORO_READY (dir->data);
189			sv_setiv (dir->data, success);
190	root	1.10	}
191
192			static void
193			handle_io_cb (EV_P_ ev_io *w, int revents)
194			{
195			handle_cb ((coro_dir )(((char )w) - offsetof (coro_dir, io)), 1);
196			}
197
198			static void
199			handle_timer_cb (EV_P_ ev_timer *w, int revents)
200			{
201			handle_cb ((coro_dir )(((char )w) - offsetof (coro_dir, tw)), 0);
202			}
203
204	root	1.18	static int
205			slf_check_rw (pTHX_ struct CoroSLF *frame)
206			{
207			SV data = (SV )frame->data;
208
209	root	1.34	/* return early when an exception is pending */
210			if (CORO_THROW)
211			return 0;
212
213	root	1.18	if (SvROK (data))
214			return 1;
215			else
216			{
217			dSP;
218
219			XPUSHs (data);
220
221			PUTBACK;
222			return 0;
223			}
224			}
225
226			static void
227	root	1.20	slf_init_rw (pTHX_ struct CoroSLF frame, SV arg, int wr)
228	root	1.18	{
229			AV handle = (AV )SvRV (arg);
230			SV *data_sv = AvARRAY (handle)[5];
231			coro_handle *data;
232			coro_dir *dir;
233			assert (AvFILLp (handle) >= 7);
234
235			if (!SvOK (data_sv))
236			{
237			int fno = sv_fileno (AvARRAY (handle)[0]);
238			data_sv = AvARRAY (handle)[5] = NEWSV (0, sizeof (coro_handle));
239			SvPOK_only (data_sv);
240			SvREADONLY_on (data_sv);
241			data = (coro_handle *)SvPVX (data_sv);
242			memset (data, 0, sizeof (coro_handle));
243
244			ev_io_init (&data->r.io, handle_io_cb, fno, EV_READ);
245			ev_io_init (&data->w.io, handle_io_cb, fno, EV_WRITE);
246			ev_init (&data->r.tw, handle_timer_cb);
247			ev_init (&data->w.tw, handle_timer_cb);
248
249			sv_magicext (data_sv, 0, PERL_MAGIC_ext, &handle_vtbl, (char *)data, 0);
250			}
251			else
252			data = (coro_handle *)SvPVX (data_sv);
253
254			dir = wr ? &data->w : &data->r;
255
256			if (ev_is_active (&dir->io) \|\| ev_is_active (&dir->tw))
257	root	1.33	croak ("recursive invocation of readable_ev or writable_ev (concurrent Coro::Handle calls on same handle?), detected");
258	root	1.18
259			dir->data = sv_2mortal (newRV_inc (CORO_CURRENT));
260
261			{
262			SV *to = AvARRAY (handle)[2];
263
264			if (SvOK (to))
265			{
266			ev_timer_set (&dir->tw, 0., SvNV (to));
267			ev_timer_again (EV_DEFAULT_UC, &dir->tw);
268			}
269			}
270
271			ev_io_start (EV_DEFAULT_UC, &dir->io);
272
273			frame->data = (void *)dir->data;
274			frame->prepare = GCoroAPI->prepare_schedule;
275			frame->check = slf_check_rw;
276			}
277
278			static void
279	root	1.21	slf_init_readable (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
280	root	1.18	{
281			slf_init_rw (aTHX_ frame, arg [0], 0);
282			}
283
284			static void
285	root	1.21	slf_init_writable (pTHX_ struct CoroSLF frame, CV cv, SV **arg, int items)
286	root	1.18	{
287			slf_init_rw (aTHX_ frame, arg [0], 1);
288			}
289
290	root	1.10	/*****************************************************************************/
291
292	root	1.1	MODULE = Coro::EV PACKAGE = Coro::EV
293
294			PROTOTYPES: ENABLE
295
296			BOOT:
297			{
298	root	1.23	I_EV_API ("Coro::EV");
299			I_CORO_API ("Coro::EV");
300	root	1.2
301	root	1.17	EV_DEFAULT; /* make sure it is initialised */
302
303	root	1.3	ev_prepare_init (&scheduler, prepare_cb);
304	root	1.10	ev_set_priority (&scheduler, EV_MINPRI);
305	root	1.17	ev_prepare_start (EV_DEFAULT_UC, &scheduler);
306			ev_unref (EV_DEFAULT_UC);
307	root	1.2
308	root	1.3	ev_idle_init (&idler, idle_cb);
309	root	1.10	ev_set_priority (&idler, EV_MINPRI);
310	root	1.15
311	root	1.29	if (!CORO_READYHOOK) /* do not override if Coro::AnyEvent already did */
312			{
313			CORO_READYHOOK = readyhook;
314			CORO_READYHOOK (); /* make sure we don't miss previous ready's */
315			}
316			}
317
318			void
319			_set_readyhook ()
320			CODE:
321	root	1.15	CORO_READYHOOK = readyhook;
322	root	1.29	CORO_READYHOOK ();
323	root	1.1
324			void
325	root	1.9	_loop_oneshot ()
326			CODE:
327	root	1.17	{
328			/* inhibit the prepare watcher, as we know we are the only
329			* ready coroutine and we don't want it to start an idle watcher
330			* just because of the fallback idle coro being of lower priority.
331			*/
332	root	1.9	++inhibit;
333	root	1.17
334			/* same reasoning as above, make sure it is stopped */
335			if (ev_is_active (&idler))
336			ev_idle_stop (EV_DEFAULT_UC, &idler);
337	root	1.30	ev_run (EV_DEFAULT_UC, EVRUN_ONCE);
338	root	1.9	--inhibit;
339	root	1.17	}
340	root	1.9
341			void
342	root	1.18	timed_io_once (...)
343	root	1.24	PROTOTYPE: $$;$
344	root	1.1	CODE:
345	root	1.18	CORO_EXECUTE_SLF_XS (slf_init_timed_io);
346	root	1.1
347			void
348	root	1.18	timer_once (...)
349	root	1.24	PROTOTYPE: $
350	root	1.1	CODE:
351	root	1.18	CORO_EXECUTE_SLF_XS (slf_init_timer);
352	root	1.1
353	root	1.10	void
354	root	1.27	_poll (...)
355			PROTOTYPE:
356			CODE:
357			CORO_EXECUTE_SLF_XS (slf_init_timer);
358
359	root	1.28	PROTOTYPES: DISABLE
360
361	root	1.27	void
362	root	1.24	_readable_ev (...)
363	root	1.10	CODE:
364	root	1.23	items = 1; /* ignore the remaining args for speed inside Coro::Handle */
365	root	1.18	CORO_EXECUTE_SLF_XS (slf_init_readable);
366	root	1.1
367	root	1.18	void
368	root	1.24	_writable_ev (...)
369	root	1.18	CODE:
370	root	1.23	items = 1; /* ignore the remaining args for speed inside Coro::Handle */
371	root	1.18	CORO_EXECUTE_SLF_XS (slf_init_writable);
372	root	1.1