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