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, newRV_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);
  else if (ev_is_active (&idler))
    ev_idle_stop (EV_A, &idler);

  --incede;
}

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

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

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_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);
  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_DEFAULT_UC, &dir->io);
  ev_timer_stop (EV_DEFAULT_UC, &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_DEFAULT);

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

        CORO_READYHOOK = readyhook;
}

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