cvsroot/EV-Loop-Async/Async.xs

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

#include <stddef.h>

#include "EVAPI.h"
#include "xthread.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;

  cond_t invoke_cv;
  mutex_t invoke_mutex;

  SV *interrupt;
} udat;

static void
c_func (pTHX_ void *loop_, int value)
{
  struct ev_loop *loop = (struct ev_loop *)loop_;
  udat *u = ev_userdata (EV_A);
  X_LOCK (u->lock);
  ev_invoke_pending (loop);
  X_UNLOCK (u->lock);
  X_COND_SIGNAL (u->invoke_cv);
}

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);
  X_UNLOCK (u->lock);
  u->signal_func (u->signal_arg, 1);
  X_COND_WAIT (u->invoke_cv, u->invoke_mutex);
  X_LOCK (u->lock);
}

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

  X_LOCK (u->invoke_mutex);
  X_LOCK (u->lock);
  ev_ref (loop); /* really? */
  for (;;) /* really? */
    ev_loop (loop, 0);
  X_UNLOCK (u->lock);
}

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

  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 (loop, &u->async_w);

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

        pthread_cond_init (&u->invoke_cv, 0);

        ev_set_userdata (loop, u);
        ev_set_invoke_pending_cb (loop, l_invoke);
        ev_set_loop_release_cb (loop, l_release, l_acquire);

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

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

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

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

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

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

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

        if (u)
          {
            X_LOCK (u->lock);
            /* now thread is either in l_invoke, or around the blocking syscall */
            pthread_cancel (u->tid);
            pthread_join (u->tid, 0);
            ev_async_send (loop, &u->async_w);
            ev_async_stop (loop, &u->async_w);
            pthread_mutex_destroy (&u->lock);
            pthread_cond_destroy (&u->invoke_cv);
            pthread_mutex_destroy (&u->invoke_mutex); /* TODO: locked by another thread... */
            SvREFCNT_dec (u->interrupt);
            Safefree (u);
          }
}


Revision:	1.3
Committed:	Tue Jul 14 15:09:44 2009 UTC (15 years ago) by root
Branch:	MAIN
Changes since 1.2:	+8 -5 lines
Log Message:	* empty log message *
#	Content
1	#include "EXTERN.h"
2	#include "perl.h"
3	#include "XSUB.h"
4
5	#include <stddef.h>
6
7	#include "EVAPI.h"
8	#include "xthread.h"
9
10	/* our userdata */
11	typedef struct {
12	mutex_t lock; /* global loop lock */
13	void (signal_func) (void signal_arg, int value);
14	void *signal_arg;
15	ev_async async_w;
16	thread_t tid;
17
18	cond_t invoke_cv;
19	mutex_t invoke_mutex;
20
21	SV *interrupt;
22	} udat;
23
24	static void
25	c_func (pTHX_ void *loop_, int value)
26	{
27	struct ev_loop loop = (struct ev_loop )loop_;
28	udat *u = ev_userdata (EV_A);
29	X_LOCK (u->lock);
30	ev_invoke_pending (loop);
31	X_UNLOCK (u->lock);
32	X_COND_SIGNAL (u->invoke_cv);
33	}
34
35	static void
36	async_cb (EV_P_ ev_async *w, int revents)
37	{
38	/* just used for the side effects */
39	}
40
41	static void
42	l_release (EV_P)
43	{
44	udat *u = ev_userdata (EV_A);
45	X_UNLOCK (u->lock);
46	}
47
48	static void
49	l_acquire (EV_P)
50	{
51	udat *u = ev_userdata (EV_A);
52	X_LOCK (u->lock);
53	}
54
55	static void
56	l_invoke (EV_P)
57	{
58	udat *u = ev_userdata (EV_A);
59	X_UNLOCK (u->lock);
60	u->signal_func (u->signal_arg, 1);
61	X_COND_WAIT (u->invoke_cv, u->invoke_mutex);
62	X_LOCK (u->lock);
63	}
64
65	X_THREAD_PROC(l_run)
66	{
67	struct ev_loop loop = (struct ev_loop )thr_arg;
68	udat *u = ev_userdata (loop);
69
70	X_LOCK (u->invoke_mutex);
71	X_LOCK (u->lock);
72	ev_ref (loop); /* really? */
73	for (;;) /* really? */
74	ev_loop (loop, 0);
75	X_UNLOCK (u->lock);
76	}
77
78	static void
79	scope_lock_cb (pTHX_ void *loop_)
80	{
81	struct ev_loop loop = (struct ev_loop )SvIVX ((SV *)loop_);
82	udat *u = ev_userdata (loop);
83
84	X_UNLOCK (u->lock);
85	SvREFCNT_dec ((SV *)loop_);
86	}
87
88	MODULE = EV::Loop::Async PACKAGE = EV::Loop::Async
89
90	PROTOTYPES: ENABLE
91
92	BOOT:
93	{
94	I_EV_API ("EV::Loop::Async");
95	CvNODEBUG_on (get_cv ("EV::Loop::Async::scope_lock", 0)); /* otherwise calling scope can be the debugger */
96	}
97
98	void
99	_c_func (SV *loop)
100	PPCODE:
101	EXTEND (SP, 2);
102	PUSHs (sv_2mortal (newSViv (PTR2IV (c_func))));
103	PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (loop)))));
104
105	void
106	_attach (SV loop_, SV interrupt, IV sig_func, void *sig_arg)
107	CODE:
108	{
109	pthread_mutexattr_t ma;
110	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
111	udat *u;
112
113	Newz (0, u, 1, udat);
114	u->interrupt = newSVsv (interrupt);
115	u->signal_func = (void ()(void , int))sig_func;
116	u->signal_arg = sig_arg;
117
118	ev_async_init (&u->async_w, async_cb);
119	ev_async_start (loop, &u->async_w);
120
121	pthread_mutexattr_init (&ma);
122	pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE);
123	pthread_mutex_init (&u->lock, &ma);
124	pthread_mutex_init (&u->invoke_mutex, &ma);
125	pthread_mutexattr_destroy (&ma);
126
127	pthread_cond_init (&u->invoke_cv, 0);
128
129	ev_set_userdata (loop, u);
130	ev_set_invoke_pending_cb (loop, l_invoke);
131	ev_set_loop_release_cb (loop, l_release, l_acquire);
132
133	pthread_create (&u->tid, 0, l_run, loop);
134	}
135
136	SV *
137	interrupt (SV *loop_)
138	CODE:
139	{
140	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
141	udat *u = ev_userdata (loop);
142
143	RETVAL = newSVsv (u->interrupt);
144	}
145	OUTPUT:
146	RETVAL
147
148	void
149	lock (SV *loop_)
150	ALIAS:
151	lock = 0
152	unlock = 1
153	notify = 2
154	CODE:
155	{
156	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
157	udat *u = ev_userdata (loop);
158
159	switch (ix)
160	{
161	case 0: X_LOCK (u->lock); break;
162	case 1: X_UNLOCK (u->lock); break;
163	case 2: ev_async_send (loop, &u->async_w); break;
164	}
165	}
166
167	void
168	scope_lock (SV *loop_)
169	CODE:
170	{
171	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
172	udat *u = ev_userdata (loop);
173
174	X_LOCK (u->lock);
175
176	LEAVE; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
177	SAVEDESTRUCTOR_X (scope_lock_cb, (void *)SvREFCNT_inc (SvRV (loop_)));
178	ENTER; /* unfortunately, perl sandwiches XS calls into ENTER/LEAVE */
179	}
180
181	void
182	DESTROY (SV *loop_)
183	CODE:
184	{
185	struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
186	udat *u = ev_userdata (loop);
187
188	if (u)
189	{
190	X_LOCK (u->lock);
191	/* now thread is either in l_invoke, or around the blocking syscall */
192	pthread_cancel (u->tid);
193	pthread_join (u->tid, 0);
194	ev_async_send (loop, &u->async_w);
195	ev_async_stop (loop, &u->async_w);
196	pthread_mutex_destroy (&u->lock);
197	pthread_cond_destroy (&u->invoke_cv);
198	pthread_mutex_destroy (&u->invoke_mutex); /* TODO: locked by another thread... */
199	SvREFCNT_dec (u->interrupt);
200	Safefree (u);
201	}
202	}
203
204
205