cvsroot/EV-Loop-Async/Async.xs

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

  cond_t invoke_cv;

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

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

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