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 {
  xmutex_t lock; /* global loop lock */
  void (*signal_func) (void *signal_arg, int value);
  void *signal_arg;
  ev_async async_w;
  xthread_t tid;
  unsigned int max_loops;
  unsigned int count;

  xcond_t invoke_cv;

  SV *interrupt;
#if defined(_WIN32) && defined(USE_ITHREADS)
  void *thx;
#endif
} 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_run (EV_A, EVRUN_NOWAIT);
      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;
#if defined(_WIN32) && defined(USE_ITHREADS)
  udat *u = ev_userdata (EV_A);

  /* just setting the same context pointer as the other thread is */
  /* probably fatal, yet, I have no clue what makes libev crash (malloc?) */
  /* as visual c also crashes when it tries to debug the crash */
  /* the loser platform is indeed a crashy OS */
  PERL_SET_CONTEXT (u->thx);
#endif

  l_acquire (EV_A);

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

  ev_ref (EV_A);
  ev_run (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)
        PROTOTYPE: @
        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;
#if defined(_WIN32) && defined(USE_ITHREADS)
        u->thx         = PERL_GET_CONTEXT;
#endif

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

        pthread_mutexattr_init (&ma);
#ifdef PTHREAD_MUTEX_RECURSIVE
        pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE);
#else
        pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE_NP);
#endif
        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);

        xthread_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);
            X_UNLOCK (u->lock);
            pthread_mutex_destroy (&u->lock);
            pthread_cond_destroy (&u->invoke_cv);
            SvREFCNT_dec (u->interrupt);
            Safefree (u);
          }
}


Revision:	1.13
Committed:	Thu Oct 11 05:02:54 2012 UTC (13 years, 6 months ago) by root
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.12:	+1 -1 lines
Log Message:	xthread_create
#	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	xmutex_t lock; /* global loop lock */
16	void (signal_func) (void signal_arg, int value);
17	void *signal_arg;
18	ev_async async_w;
19	xthread_t tid;
20	unsigned int max_loops;
21	unsigned int count;
22
23	xcond_t invoke_cv;
24
25	SV *interrupt;
26	#if defined(_WIN32) && defined(USE_ITHREADS)
27	void *thx;
28	#endif
29	} udat;
30
31	static void loop_set_cb (EV_P);
32
33	static void
34	fg_invoke_pending (EV_P)
35	{
36	udat *u = ev_userdata (EV_A);
37
38	u->count = ev_pending_count (EV_A);
39
40	if (u->count)
41	ev_invoke_pending (EV_A);
42	}
43
44	static void
45	c_func (pTHX_ void *loop_, int value)
46	{
47	struct ev_loop loop = (struct ev_loop )loop_;
48	udat *u = ev_userdata (EV_A);
49	int i;
50
51	X_LOCK (u->lock);
52	ev_invoke_pending (EV_A);
53
54	/* do any additional foreground loop runs */
55	for (i = u->max_loops; i--; )
56	{
57	/* this is a bit tricky, but we can manage... */
58	u->count = 0;
59
60	ev_set_invoke_pending_cb (EV_A, fg_invoke_pending);
61	ev_set_loop_release_cb (EV_A, 0, 0);
62	ev_run (EV_A, EVRUN_NOWAIT);
63	loop_set_cb (EV_A);
64
65	if (!u->count)
66	break;
67	}
68
69	X_COND_SIGNAL (u->invoke_cv);
70	X_UNLOCK (u->lock);
71	}
72
73	static void
74	async_cb (EV_P_ ev_async *w, int revents)
75	{
76	/* just used for the side effects */
77	}
78
79	static void
80	l_release (EV_P)
81	{
82	udat *u = ev_userdata (EV_A);
83	X_UNLOCK (u->lock);
84	}
85
86	static void
87	l_acquire (EV_P)
88	{
89	udat *u = ev_userdata (EV_A);
90	X_LOCK (u->lock);
91	}
92
93	static void
94	l_invoke (EV_P)
95	{
96	udat *u = ev_userdata (EV_A);
97
98	while (ev_pending_count (EV_A))
99	{
100	u->signal_func (u->signal_arg, 1);
101	X_COND_WAIT (u->invoke_cv, u->lock);
102	}
103	}
104
105	static void
106	loop_set_cb (EV_P)
107	{
108	ev_set_invoke_pending_cb (EV_A, l_invoke);
109	ev_set_loop_release_cb (EV_A, l_release, l_acquire);
110	}
111
112	X_THREAD_PROC(l_run)
113	{
114	struct ev_loop loop = (struct ev_loop )thr_arg;
115	#if defined(_WIN32) && defined(USE_ITHREADS)
116	udat *u = ev_userdata (EV_A);
117
118	/* just setting the same context pointer as the other thread is */
119	/* probably fatal, yet, I have no clue what makes libev crash (malloc?) */
120	/* as visual c also crashes when it tries to debug the crash */
121	/* the loser platform is indeed a crashy OS */
122	PERL_SET_CONTEXT (u->thx);
123	#endif
124
125	l_acquire (EV_A);
126
127	/* yeah */
128	pthread_setcanceltype (PTHREAD_CANCEL_ASYNCHRONOUS, 0);
129
130	ev_ref (EV_A);
131	ev_run (EV_A, 0);
132	ev_unref (EV_A);
133
134	l_release (EV_A);
135
136	return 0;
137	}
138
139	static void
140	scope_lock_cb (pTHX_ void *loop_)
141	{
142	struct ev_loop loop = (struct ev_loop )SvIVX ((SV *)loop_);
143	udat *u = ev_userdata (EV_A);
144
145	X_UNLOCK (u->lock);
146	SvREFCNT_dec ((SV *)loop_);
147	}
148
149	MODULE = EV::Loop::Async PACKAGE = EV::Loop::Async
150
151	PROTOTYPES: ENABLE
152
153	BOOT:
154	{
155	I_EV_API ("EV::Loop::Async");
156	CvNODEBUG_on (get_cv ("EV::Loop::Async::scope_lock", 0)); /* otherwise calling scope can be the debugger */
157	}
158
159	void
160	_c_func (SV *loop)
161	PPCODE:
162	EXTEND (SP, 2);
163	PUSHs (sv_2mortal (newSViv (PTR2IV (c_func))));
164	PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (loop)))));
165
166	void
167	_attach (SV loop_, SV interrupt, IV sig_func, void *sig_arg)
168	PROTOTYPE: @
169	CODE:
170	{
171	pthread_mutexattr_t ma;
172	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
173	udat *u;
174
175	Newz (0, u, 1, udat);
176	u->interrupt = newSVsv (interrupt);
177	u->signal_func = (void ()(void , int))sig_func;
178	u->signal_arg = sig_arg;
179	#if defined(_WIN32) && defined(USE_ITHREADS)
180	u->thx = PERL_GET_CONTEXT;
181	#endif
182
183	ev_async_init (&u->async_w, async_cb);
184	ev_async_start (EV_A, &u->async_w);
185
186	pthread_mutexattr_init (&ma);
187	#ifdef PTHREAD_MUTEX_RECURSIVE
188	pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE);
189	#else
190	pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE_NP);
191	#endif
192	pthread_mutex_init (&u->lock, &ma);
193	pthread_mutexattr_destroy (&ma);
194
195	pthread_cond_init (&u->invoke_cv, 0);
196
197	ev_set_userdata (EV_A, u);
198	loop_set_cb (EV_A);
199
200	xthread_create (&u->tid, l_run, loop);
201	}
202
203	SV *
204	interrupt (SV *loop_)
205	CODE:
206	{
207	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
208	udat *u = ev_userdata (EV_A);
209
210	RETVAL = newSVsv (u->interrupt);
211	}
212	OUTPUT:
213	RETVAL
214
215	void
216	set_max_foreground_loops (SV *loop_, UV max_loops)
217	CODE:
218	{
219	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
220	udat *u = ev_userdata (EV_A);
221
222	u->max_loops = max_loops;
223	}
224
225	void
226	lock (SV *loop_)
227	ALIAS:
228	lock = 0
229	unlock = 1
230	notify = 2
231	CODE:
232	{
233	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
234	udat *u = ev_userdata (EV_A);
235
236	switch (ix)
237	{
238	case 0: X_LOCK (u->lock); break;
239	case 1: X_UNLOCK (u->lock); break;
240	case 2: ev_async_send (EV_A, &u->async_w); break;
241	}
242	}
243
244	void
245	scope_lock (SV *loop_)
246	CODE:
247	{
248	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
249	udat *u = ev_userdata (EV_A);
250
251	X_LOCK (u->lock);
252
253	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
254	SAVEDESTRUCTOR_X (scope_lock_cb, (void *)SvREFCNT_inc (SvRV (loop_)));
255	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
256	}
257
258	void
259	DESTROY (SV *loop_)
260	CODE:
261	{
262	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
263	udat *u = ev_userdata (EV_A);
264
265	if (u)
266	{
267	X_LOCK (u->lock);
268	ev_async_stop (EV_A, &u->async_w);
269	/* now thread is around blocking call, or in pthread_cond_wait */
270	pthread_cancel (u->tid);
271	X_UNLOCK (u->lock);
272	pthread_mutex_destroy (&u->lock);
273	pthread_cond_destroy (&u->invoke_cv);
274	SvREFCNT_dec (u->interrupt);
275	Safefree (u);
276	}
277	}
278
279
280