Coro/EV/EV.xs

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

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

#define EV_PROTOTYPES 0
#include "EVAPI.h"
#include "../Coro/CoroAPI.h"

static void
once_cb (int revents, void *arg)
{
  AV *av = (AV *)arg; /* @_ */
  av_push (av, newSViv (revents));
  CORO_READY (AvARRAY (av)[0]);
  SvREFCNT_dec (av);
}

#define ONCE_INIT  AV *av = GvAV (PL_defgv);
#define ONCE_DONE  av_clear (av); av_push (av, SvREFCNT_inc (CORO_CURRENT));

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 && !ev_is_active (&idler))
    ev_idle_start (EV_A_ &idler);

  --incede;
}

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

typedef struct
{
  ev_io io;
  ev_timer tw;
  SV *done;
  SV *current;
} 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_ &data->r.io); ev_io_stop    (EV_DEFAULT_ &data->w.io);
  ev_timer_stop (EV_DEFAULT_ &data->r.tw); ev_timer_stop (EV_DEFAULT_ &data->w.tw);
  SvREFCNT_dec (data->r.done);             SvREFCNT_dec (data->w.done);
  SvREFCNT_dec (data->r.current);          SvREFCNT_dec (data->w.current);

  return 0;
}

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

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

  sv_setiv (dir->done, done);
  SvREFCNT_dec (dir->done);
  dir->done = 0;
  CORO_READY (dir->current);
  SvREFCNT_dec (dir->current);
  dir->current = 0;
}

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

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

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

PROTOTYPES: ENABLE

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

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

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

void
_loop_oneshot ()
        CODE:
        ++inhibit;
        ev_loop (EV_DEFAULT_ EVLOOP_ONESHOT);
        --inhibit;

void
_timed_io_once (...)
        CODE:
{
        ONCE_INIT;
        assert (AvFILLp (av) >= 1);
        ev_once (
          sv_fileno (AvARRAY (av)[0]),
          SvIV (AvARRAY (av)[1]),
          AvFILLp (av) >= 2 && SvOK (AvARRAY (av)[2]) ? SvNV (AvARRAY (av)[2]) : -1.,
          once_cb,
          (void *)SvREFCNT_inc (av)
        );
        ONCE_DONE;
}

void
_timer_once (...)
        CODE:
{
        ONCE_INIT;
        NV after = SvNV (AvARRAY (av)[0]);
        ev_once (
          -1,
          0,
          after >= 0. ? after : 0.,
          once_cb,
          (void *)SvREFCNT_inc (av)
        );
        ONCE_DONE;
}

void
_readable_ev (SV *handle_sv, SV *done_sv)
        ALIAS:
        _writable_ev = 1
        CODE:
{
        AV *handle = (AV *)SvRV (handle_sv);
        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 = ix ? &data->w : &data->r;

        if (ev_is_active (&dir->io) || ev_is_active (&dir->tw))
          croak ("recursive invocation of readable_ev or writable_ev");

        dir->done = SvREFCNT_inc (done_sv);
        dir->current = SvREFCNT_inc (CORO_CURRENT);

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

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

        ev_io_start (EV_DEFAULT_ &dir->io);
}

Revision:	1.11
Committed:	Thu Dec 13 16:39:01 2007 UTC (16 years, 6 months ago) by root
Branch:	MAIN
CVS Tags:	rel-4_32, rel-4_33
Changes since 1.10:	+1 -1 lines
Log Message:	* empty log message *
#	Content
1	#include "EXTERN.h"
2	#include "perl.h"
3	#include "XSUB.h"
4
5	#include <stddef.h>
6	#include <assert.h>
7	#include <string.h>
8
9	#define EV_PROTOTYPES 0
10	#include "EVAPI.h"
11	#include "../Coro/CoroAPI.h"
12
13	static void
14	once_cb (int revents, void *arg)
15	{
16	AV av = (AV )arg; /* @_ */
17	av_push (av, newSViv (revents));
18	CORO_READY (AvARRAY (av)[0]);
19	SvREFCNT_dec (av);
20	}
21
22	#define ONCE_INIT AV *av = GvAV (PL_defgv);
23	#define ONCE_DONE av_clear (av); av_push (av, SvREFCNT_inc (CORO_CURRENT));
24
25	static struct ev_prepare scheduler;
26	static struct ev_idle idler;
27	static int inhibit;
28
29	static void
30	idle_cb (EV_P_ ev_idle *w, int revents)
31	{
32	ev_idle_stop (EV_A_ w);
33	}
34
35	static void
36	prepare_cb (EV_P_ ev_prepare *w, int revents)
37	{
38	static int incede;
39
40	if (inhibit)
41	return;
42
43	++incede;
44
45	CORO_CEDE_NOTSELF;
46
47	while (CORO_NREADY >= incede && CORO_CEDE)
48	;
49
50	/* if still ready, then we have lower-priority coroutines.
51	* poll anyways, but do not block.
52	*/
53	if (CORO_NREADY >= incede && !ev_is_active (&idler))
54	ev_idle_start (EV_A_ &idler);
55
56	--incede;
57	}
58
59	/*****************************************************************************/
60
61	typedef struct
62	{
63	ev_io io;
64	ev_timer tw;
65	SV *done;
66	SV *current;
67	} coro_dir;
68
69	typedef struct
70	{
71	coro_dir r, w;
72	} coro_handle;
73
74
75	static int
76	handle_free (pTHX_ SV sv, MAGIC mg)
77	{
78	coro_handle data = (coro_handle )mg->mg_ptr;
79	mg->mg_ptr = 0;
80
81	ev_io_stop (EV_DEFAULT_ &data->r.io); ev_io_stop (EV_DEFAULT_ &data->w.io);
82	ev_timer_stop (EV_DEFAULT_ &data->r.tw); ev_timer_stop (EV_DEFAULT_ &data->w.tw);
83	SvREFCNT_dec (data->r.done); SvREFCNT_dec (data->w.done);
84	SvREFCNT_dec (data->r.current); SvREFCNT_dec (data->w.current);
85
86	return 0;
87	}
88
89	static MGVTBL handle_vtbl = { 0, 0, 0, 0, handle_free };
90
91	static void
92	handle_cb (coro_dir *dir, int done)
93	{
94	ev_io_stop (EV_A_ &dir->io);
95	ev_timer_stop (EV_A_ &dir->tw);
96
97	sv_setiv (dir->done, done);
98	SvREFCNT_dec (dir->done);
99	dir->done = 0;
100	CORO_READY (dir->current);
101	SvREFCNT_dec (dir->current);
102	dir->current = 0;
103	}
104
105	static void
106	handle_io_cb (EV_P_ ev_io *w, int revents)
107	{
108	handle_cb ((coro_dir )(((char )w) - offsetof (coro_dir, io)), 1);
109	}
110
111	static void
112	handle_timer_cb (EV_P_ ev_timer *w, int revents)
113	{
114	handle_cb ((coro_dir )(((char )w) - offsetof (coro_dir, tw)), 0);
115	}
116
117	/*****************************************************************************/
118
119	MODULE = Coro::EV PACKAGE = Coro::EV
120
121	PROTOTYPES: ENABLE
122
123	BOOT:
124	{
125	I_EV_API ("Coro::EV");
126	I_CORO_API ("Coro::Event");
127
128	ev_prepare_init (&scheduler, prepare_cb);
129	ev_set_priority (&scheduler, EV_MINPRI);
130	ev_prepare_start (EV_DEFAULT_ &scheduler);
131	ev_unref ();
132
133	ev_idle_init (&idler, idle_cb);
134	ev_set_priority (&idler, EV_MINPRI);
135	}
136
137	void
138	_loop_oneshot ()
139	CODE:
140	++inhibit;
141	ev_loop (EV_DEFAULT_ EVLOOP_ONESHOT);
142	--inhibit;
143
144	void
145	_timed_io_once (...)
146	CODE:
147	{
148	ONCE_INIT;
149	assert (AvFILLp (av) >= 1);
150	ev_once (
151	sv_fileno (AvARRAY (av)[0]),
152	SvIV (AvARRAY (av)[1]),
153	AvFILLp (av) >= 2 && SvOK (AvARRAY (av)[2]) ? SvNV (AvARRAY (av)[2]) : -1.,
154	once_cb,
155	(void *)SvREFCNT_inc (av)
156	);
157	ONCE_DONE;
158	}
159
160	void
161	_timer_once (...)
162	CODE:
163	{
164	ONCE_INIT;
165	NV after = SvNV (AvARRAY (av)[0]);
166	ev_once (
167	-1,
168	0,
169	after >= 0. ? after : 0.,
170	once_cb,
171	(void *)SvREFCNT_inc (av)
172	);
173	ONCE_DONE;
174	}
175
176	void
177	_readable_ev (SV handle_sv, SV done_sv)
178	ALIAS:
179	_writable_ev = 1
180	CODE:
181	{
182	AV handle = (AV )SvRV (handle_sv);
183	SV *data_sv = AvARRAY (handle)[5];
184	coro_handle *data;
185	coro_dir *dir;
186	assert (AvFILLp (handle) >= 7);
187
188	if (!SvOK (data_sv))
189	{
190	int fno = sv_fileno (AvARRAY (handle)[0]);
191	data_sv = AvARRAY (handle)[5] = NEWSV (0, sizeof (coro_handle));
192	SvPOK_only (data_sv);
193	SvREADONLY_on (data_sv);
194	data = (coro_handle *)SvPVX (data_sv);
195	memset (data, 0, sizeof (coro_handle));
196
197	ev_io_init (&data->r.io, handle_io_cb, fno, EV_READ);
198	ev_io_init (&data->w.io, handle_io_cb, fno, EV_WRITE);
199	ev_init (&data->r.tw, handle_timer_cb);
200	ev_init (&data->w.tw, handle_timer_cb);
201
202	sv_magicext (data_sv, 0, PERL_MAGIC_ext, &handle_vtbl, (char *)data, 0);
203	}
204	else
205	data = (coro_handle *)SvPVX (data_sv);
206
207	dir = ix ? &data->w : &data->r;
208
209	if (ev_is_active (&dir->io) \|\| ev_is_active (&dir->tw))
210	croak ("recursive invocation of readable_ev or writable_ev");
211
212	dir->done = SvREFCNT_inc (done_sv);
213	dir->current = SvREFCNT_inc (CORO_CURRENT);
214
215	{
216	SV *to = AvARRAY (handle)[2];
217
218	if (SvOK (to))
219	{
220	ev_timer_set (&dir->tw, 0., SvNV (to));
221	ev_timer_again (EV_DEFAULT_ &dir->tw);
222	}
223	}
224
225	ev_io_start (EV_DEFAULT_ &dir->io);
226	}
227