EV-Loop-Async/Async.xs

#include "xthread.h"

#include <errno.h>

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

#include <stddef.h>

#include "EVAPI.h"

/* our userdata */
typedef struct {
  mutex_t lock; /* global loop lock */
  void (*signal_func) (void *signal_arg, int value);
  void *signal_arg;
  ev_async async_w;
  thread_t tid;
  unsigned int max_loops;
  unsigned int count;

  cond_t invoke_cv;

  SV *interrupt;
} udat;

static void loop_set_cb (EV_P);

static void
fg_invoke_pending (EV_P)
{
  udat *u = ev_userdata (EV_A);

  u->count = ev_pending_count (EV_A);

  if (u->count)
    ev_invoke_pending (EV_A);
}

static void
c_func (pTHX_ void *loop_, int value)
{
  struct ev_loop *loop = (struct ev_loop *)loop_;
  udat *u = ev_userdata (EV_A);
  int i;

  X_LOCK (u->lock);
  ev_invoke_pending (EV_A);

  /* do any additional foreground loop runs */
  for (i = u->max_loops; i--; )
    {
      /* this is a bit tricky, but we can manage... */
      u->count = 0;

      ev_set_invoke_pending_cb (EV_A, fg_invoke_pending);
      ev_set_loop_release_cb (EV_A, 0, 0);
      ev_loop (EV_A, EVLOOP_NONBLOCK);
      loop_set_cb (EV_A);

      if (!u->count)
        break;
    }

  X_COND_SIGNAL (u->invoke_cv);
  X_UNLOCK (u->lock);
}

static void
async_cb (EV_P_ ev_async *w, int revents)
{
  /* just used for the side effects */
}

static void
l_release (EV_P)
{
  udat *u = ev_userdata (EV_A);
  X_UNLOCK (u->lock);
}

static void
l_acquire (EV_P)
{
  udat *u = ev_userdata (EV_A);
  X_LOCK (u->lock);
}

static void
l_invoke (EV_P)
{
  udat *u = ev_userdata (EV_A);

  while (ev_pending_count (EV_A))
    {
      u->signal_func (u->signal_arg, 1);
      X_COND_WAIT (u->invoke_cv, u->lock);
    }
}

static void
loop_set_cb (EV_P)
{
  ev_set_invoke_pending_cb (EV_A, l_invoke);
  ev_set_loop_release_cb (EV_A, l_release, l_acquire);
}

X_THREAD_PROC(l_run)
{
  struct ev_loop *loop = (struct ev_loop *)thr_arg;

  l_acquire (EV_A);

  /* yeah */
  pthread_setcanceltype (PTHREAD_CANCEL_ASYNCHRONOUS, 0);

  ev_ref (EV_A);
  ev_loop (EV_A, 0);
  ev_unref (EV_A);

  l_release (EV_A);

  return 0;
}

static void
scope_lock_cb (pTHX_ void *loop_)
{
  struct ev_loop *loop = (struct ev_loop *)SvIVX ((SV *)loop_);
  udat *u = ev_userdata (EV_A);

  X_UNLOCK (u->lock);
  SvREFCNT_dec ((SV *)loop_);
}

MODULE = EV::Loop::Async                PACKAGE = EV::Loop::Async

PROTOTYPES: ENABLE

BOOT:
{
        I_EV_API ("EV::Loop::Async");
        CvNODEBUG_on (get_cv ("EV::Loop::Async::scope_lock", 0)); /* otherwise calling scope can be the debugger */
}

void
_c_func (SV *loop)
        PPCODE:
        EXTEND (SP, 2);
        PUSHs (sv_2mortal (newSViv (PTR2IV (c_func))));
        PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (loop)))));

void
_attach (SV *loop_, SV *interrupt, IV sig_func, void *sig_arg)
        CODE:
{
        pthread_mutexattr_t ma;
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u;

        Newz (0, u, 1, udat);
        u->interrupt   = newSVsv (interrupt);
        u->signal_func = (void (*)(void *, int))sig_func;
        u->signal_arg  = sig_arg;

        ev_async_init (&u->async_w, async_cb);
        ev_async_start (EV_A, &u->async_w);

        pthread_mutexattr_init (&ma);
        pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE);
        pthread_mutex_init (&u->lock, &ma);
        pthread_mutexattr_destroy (&ma);

        pthread_cond_init (&u->invoke_cv, 0);

        ev_set_userdata (EV_A, u);
        loop_set_cb (EV_A);

        thread_create (&u->tid, l_run, loop);
}

SV *
interrupt (SV *loop_)
        CODE:
{
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u = ev_userdata (EV_A);

        RETVAL = newSVsv (u->interrupt);
}
        OUTPUT:
        RETVAL

void
set_max_foreground_loops (SV *loop_, UV max_loops)
        CODE:
{
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u = ev_userdata (EV_A);

        u->max_loops = max_loops;
}

void
lock (SV *loop_)
        ALIAS:
        lock   = 0
        unlock = 1
        notify = 2
        CODE:
{
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u = ev_userdata (EV_A);

        switch (ix)
          {
            case 0: X_LOCK   (u->lock); break;
            case 1: X_UNLOCK (u->lock); break;
            case 2: ev_async_send (EV_A, &u->async_w); break;
          }
}

void
scope_lock (SV *loop_)
        CODE:
{
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u = ev_userdata (EV_A);

        X_LOCK (u->lock);

        LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
        SAVEDESTRUCTOR_X (scope_lock_cb, (void *)SvREFCNT_inc (SvRV (loop_)));
        ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
}

void
DESTROY (SV *loop_)
        CODE:
{
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u = ev_userdata (EV_A);

        if (u)
          {
            X_LOCK (u->lock);
            ev_async_stop (EV_A, &u->async_w);
            /* now thread is around blocking call, or in pthread_cond_wait */
            pthread_cancel (u->tid);
            pthread_mutex_destroy (&u->lock);
            pthread_cond_destroy (&u->invoke_cv);
            SvREFCNT_dec (u->interrupt);
            Safefree (u);
          }
}


Revision:	1.9
Committed:	Fri Jul 17 03:28:57 2009 UTC (17 years, 2 months ago) by root
Branch:	MAIN
CVS Tags:	rel-0_02
Changes since 1.8:	+2 -0 lines
Log Message:	* empty log message *
#	Content
1	#include "xthread.h"
2
3	#include <errno.h>
4
5	#include "EXTERN.h"
6	#include "perl.h"
7	#include "XSUB.h"
8
9	#include <stddef.h>
10
11	#include "EVAPI.h"
12
13	/* our userdata */
14	typedef struct {
15	mutex_t lock; /* global loop lock */
16	void (signal_func) (void signal_arg, int value);
17	void *signal_arg;
18	ev_async async_w;
19	thread_t tid;
20	unsigned int max_loops;
21	unsigned int count;
22
23	cond_t invoke_cv;
24
25	SV *interrupt;
26	} udat;
27
28	static void loop_set_cb (EV_P);
29
30	static void
31	fg_invoke_pending (EV_P)
32	{
33	udat *u = ev_userdata (EV_A);
34
35	u->count = ev_pending_count (EV_A);
36
37	if (u->count)
38	ev_invoke_pending (EV_A);
39	}
40
41	static void
42	c_func (pTHX_ void *loop_, int value)
43	{
44	struct ev_loop loop = (struct ev_loop )loop_;
45	udat *u = ev_userdata (EV_A);
46	int i;
47
48	X_LOCK (u->lock);
49	ev_invoke_pending (EV_A);
50
51	/* do any additional foreground loop runs */
52	for (i = u->max_loops; i--; )
53	{
54	/* this is a bit tricky, but we can manage... */
55	u->count = 0;
56
57	ev_set_invoke_pending_cb (EV_A, fg_invoke_pending);
58	ev_set_loop_release_cb (EV_A, 0, 0);
59	ev_loop (EV_A, EVLOOP_NONBLOCK);
60	loop_set_cb (EV_A);
61
62	if (!u->count)
63	break;
64	}
65
66	X_COND_SIGNAL (u->invoke_cv);
67	X_UNLOCK (u->lock);
68	}
69
70	static void
71	async_cb (EV_P_ ev_async *w, int revents)
72	{
73	/* just used for the side effects */
74	}
75
76	static void
77	l_release (EV_P)
78	{
79	udat *u = ev_userdata (EV_A);
80	X_UNLOCK (u->lock);
81	}
82
83	static void
84	l_acquire (EV_P)
85	{
86	udat *u = ev_userdata (EV_A);
87	X_LOCK (u->lock);
88	}
89
90	static void
91	l_invoke (EV_P)
92	{
93	udat *u = ev_userdata (EV_A);
94
95	while (ev_pending_count (EV_A))
96	{
97	u->signal_func (u->signal_arg, 1);
98	X_COND_WAIT (u->invoke_cv, u->lock);
99	}
100	}
101
102	static void
103	loop_set_cb (EV_P)
104	{
105	ev_set_invoke_pending_cb (EV_A, l_invoke);
106	ev_set_loop_release_cb (EV_A, l_release, l_acquire);
107	}
108
109	X_THREAD_PROC(l_run)
110	{
111	struct ev_loop loop = (struct ev_loop )thr_arg;
112
113	l_acquire (EV_A);
114
115	/* yeah */
116	pthread_setcanceltype (PTHREAD_CANCEL_ASYNCHRONOUS, 0);
117
118	ev_ref (EV_A);
119	ev_loop (EV_A, 0);
120	ev_unref (EV_A);
121
122	l_release (EV_A);
123
124	return 0;
125	}
126
127	static void
128	scope_lock_cb (pTHX_ void *loop_)
129	{
130	struct ev_loop loop = (struct ev_loop )SvIVX ((SV *)loop_);
131	udat *u = ev_userdata (EV_A);
132
133	X_UNLOCK (u->lock);
134	SvREFCNT_dec ((SV *)loop_);
135	}
136
137	MODULE = EV::Loop::Async PACKAGE = EV::Loop::Async
138
139	PROTOTYPES: ENABLE
140
141	BOOT:
142	{
143	I_EV_API ("EV::Loop::Async");
144	CvNODEBUG_on (get_cv ("EV::Loop::Async::scope_lock", 0)); /* otherwise calling scope can be the debugger */
145	}
146
147	void
148	_c_func (SV *loop)
149	PPCODE:
150	EXTEND (SP, 2);
151	PUSHs (sv_2mortal (newSViv (PTR2IV (c_func))));
152	PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (loop)))));
153
154	void
155	_attach (SV loop_, SV interrupt, IV sig_func, void *sig_arg)
156	CODE:
157	{
158	pthread_mutexattr_t ma;
159	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
160	udat *u;
161
162	Newz (0, u, 1, udat);
163	u->interrupt = newSVsv (interrupt);
164	u->signal_func = (void ()(void , int))sig_func;
165	u->signal_arg = sig_arg;
166
167	ev_async_init (&u->async_w, async_cb);
168	ev_async_start (EV_A, &u->async_w);
169
170	pthread_mutexattr_init (&ma);
171	pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE);
172	pthread_mutex_init (&u->lock, &ma);
173	pthread_mutexattr_destroy (&ma);
174
175	pthread_cond_init (&u->invoke_cv, 0);
176
177	ev_set_userdata (EV_A, u);
178	loop_set_cb (EV_A);
179
180	thread_create (&u->tid, l_run, loop);
181	}
182
183	SV *
184	interrupt (SV *loop_)
185	CODE:
186	{
187	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
188	udat *u = ev_userdata (EV_A);
189
190	RETVAL = newSVsv (u->interrupt);
191	}
192	OUTPUT:
193	RETVAL
194
195	void
196	set_max_foreground_loops (SV *loop_, UV max_loops)
197	CODE:
198	{
199	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
200	udat *u = ev_userdata (EV_A);
201
202	u->max_loops = max_loops;
203	}
204
205	void
206	lock (SV *loop_)
207	ALIAS:
208	lock = 0
209	unlock = 1
210	notify = 2
211	CODE:
212	{
213	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
214	udat *u = ev_userdata (EV_A);
215
216	switch (ix)
217	{
218	case 0: X_LOCK (u->lock); break;
219	case 1: X_UNLOCK (u->lock); break;
220	case 2: ev_async_send (EV_A, &u->async_w); break;
221	}
222	}
223
224	void
225	scope_lock (SV *loop_)
226	CODE:
227	{
228	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
229	udat *u = ev_userdata (EV_A);
230
231	X_LOCK (u->lock);
232
233	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
234	SAVEDESTRUCTOR_X (scope_lock_cb, (void *)SvREFCNT_inc (SvRV (loop_)));
235	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
236	}
237
238	void
239	DESTROY (SV *loop_)
240	CODE:
241	{
242	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
243	udat *u = ev_userdata (EV_A);
244
245	if (u)
246	{
247	X_LOCK (u->lock);
248	ev_async_stop (EV_A, &u->async_w);
249	/* now thread is around blocking call, or in pthread_cond_wait */
250	pthread_cancel (u->tid);
251	pthread_mutex_destroy (&u->lock);
252	pthread_cond_destroy (&u->invoke_cv);
253	SvREFCNT_dec (u->interrupt);
254	Safefree (u);
255	}
256	}
257
258
259