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

  --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_DEFAULT, &dir->io);
  ev_timer_stop (EV_DEFAULT, &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);
}

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.13
Committed:	Thu Dec 27 21:48:27 2007 UTC (16 years, 5 months ago) by root
Branch:	MAIN
CVS Tags:	rel-4_50, rel-4_51, rel-4_4, rel-4_45, rel-4_49, rel-4_48, rel-4_47, rel-4_46, rel-4_35, rel-4_36, rel-4_37
Changes since 1.12:	+1 -1 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			#define ONCE_INIT AV *av = GvAV (PL_defgv);
23	root	1.13	#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.5
56			--incede;
57	root	1.2	}
58
59	root	1.10	/*****************************************************************************/
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	root	1.12	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	root	1.10	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	root	1.12	ev_io_stop (EV_DEFAULT, &dir->io);
95			ev_timer_stop (EV_DEFAULT, &dir->tw);
96	root	1.10
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	root	1.1	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	root	1.2
128	root	1.3	ev_prepare_init (&scheduler, prepare_cb);
129	root	1.10	ev_set_priority (&scheduler, EV_MINPRI);
130	root	1.12	ev_prepare_start (EV_DEFAULT, &scheduler);
131			ev_unref (EV_DEFAULT);
132	root	1.2
133	root	1.3	ev_idle_init (&idler, idle_cb);
134	root	1.10	ev_set_priority (&idler, EV_MINPRI);
135	root	1.1	}
136
137			void
138	root	1.9	_loop_oneshot ()
139			CODE:
140			++inhibit;
141	root	1.12	ev_loop (EV_DEFAULT, EVLOOP_ONESHOT);
142	root	1.9	--inhibit;
143
144			void
145	root	1.1	_timed_io_once (...)
146			CODE:
147			{
148			ONCE_INIT;
149			assert (AvFILLp (av) >= 1);
150	root	1.2	ev_once (
151	root	1.12	EV_DEFAULT,
152	root	1.2	sv_fileno (AvARRAY (av)[0]),
153			SvIV (AvARRAY (av)[1]),
154			AvFILLp (av) >= 2 && SvOK (AvARRAY (av)[2]) ? SvNV (AvARRAY (av)[2]) : -1.,
155			once_cb,
156			(void *)SvREFCNT_inc (av)
157	root	1.1	);
158			ONCE_DONE;
159			}
160
161			void
162			_timer_once (...)
163			CODE:
164			{
165			ONCE_INIT;
166			NV after = SvNV (AvARRAY (av)[0]);
167	root	1.2	ev_once (
168	root	1.12	EV_DEFAULT,
169	root	1.1	-1,
170	root	1.2	0,
171	root	1.4	after >= 0. ? after : 0.,
172	root	1.2	once_cb,
173			(void *)SvREFCNT_inc (av)
174	root	1.1	);
175			ONCE_DONE;
176			}
177
178	root	1.10	void
179			_readable_ev (SV handle_sv, SV done_sv)
180			ALIAS:
181			_writable_ev = 1
182			CODE:
183			{
184			AV handle = (AV )SvRV (handle_sv);
185			SV *data_sv = AvARRAY (handle)[5];
186			coro_handle *data;
187			coro_dir *dir;
188			assert (AvFILLp (handle) >= 7);
189
190			if (!SvOK (data_sv))
191			{
192			int fno = sv_fileno (AvARRAY (handle)[0]);
193			data_sv = AvARRAY (handle)[5] = NEWSV (0, sizeof (coro_handle));
194			SvPOK_only (data_sv);
195			SvREADONLY_on (data_sv);
196			data = (coro_handle *)SvPVX (data_sv);
197			memset (data, 0, sizeof (coro_handle));
198
199			ev_io_init (&data->r.io, handle_io_cb, fno, EV_READ);
200			ev_io_init (&data->w.io, handle_io_cb, fno, EV_WRITE);
201			ev_init (&data->r.tw, handle_timer_cb);
202			ev_init (&data->w.tw, handle_timer_cb);
203
204			sv_magicext (data_sv, 0, PERL_MAGIC_ext, &handle_vtbl, (char *)data, 0);
205			}
206			else
207			data = (coro_handle *)SvPVX (data_sv);
208
209			dir = ix ? &data->w : &data->r;
210
211			if (ev_is_active (&dir->io) \|\| ev_is_active (&dir->tw))
212			croak ("recursive invocation of readable_ev or writable_ev");
213
214			dir->done = SvREFCNT_inc (done_sv);
215			dir->current = SvREFCNT_inc (CORO_CURRENT);
216
217			{
218			SV *to = AvARRAY (handle)[2];
219
220			if (SvOK (to))
221			{
222			ev_timer_set (&dir->tw, 0., SvNV (to));
223	root	1.12	ev_timer_again (EV_DEFAULT, &dir->tw);
224	root	1.10	}
225			}
226	root	1.1
227	root	1.12	ev_io_start (EV_DEFAULT, &dir->io);
228	root	1.10	}
229	root	1.1