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 *asy;
} 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);
  fprintf (stderr, "invoke2\n");//D
  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);
}

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

PROTOTYPES: ENABLE

BOOT:
{
        I_EV_API ("EV::Loop::Async");
}

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 *asy, IV sig_func, IV sig_arg)
        CODE:
{
        pthread_mutexattr_t ma;
        struct ev_loop *loop = (struct ev_loop *)SvIVX (SvRV (loop_));
        udat *u;
        printf ("f,a %p,%p\n", sig_func, sig_arg);//D

        Newz (0, u, 1, udat);
        u->asy         = newSVsv (asy);
        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);

        X_LOCK (u->lock);
        thread_create (&u->tid, l_run, (void *)loop);
}

void
lock (SV *loop_)
        ALIAS:
        lock   = 0
        unlock = 1
        poke   = 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
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_send (loop, &u->async_w);
            pthread_cancel (u->tid);
            ev_async_stop (loop, &u->async_w);
            pthread_mutex_destroy (&u->lock);
            pthread_cond_destroy (&u->invoke_cv);
            pthread_mutex_destroy (&u->invoke_mutex);
            Safefree (u);
            SvREFCNT_dec (u->asy);
          }
}


Revision:	1.1
Committed:	Tue Jul 14 02:59:55 2009 UTC (14 years, 10 months ago) by root
Branch:	MAIN
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			#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 *asy;
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			fprintf (stderr, "invoke2\n");//D
63			X_LOCK (u->lock);
64			}
65
66			X_THREAD_PROC(l_run)
67			{
68			struct ev_loop loop = (struct ev_loop )thr_arg;
69			udat *u = ev_userdata (loop);
70
71			X_LOCK (u->invoke_mutex);
72			X_LOCK (u->lock);
73			ev_ref (loop); /* really? */
74			for (;;) /* really? */
75			ev_loop (loop, 0);
76			X_UNLOCK (u->lock);
77			}
78
79			MODULE = EV::Loop::Async PACKAGE = EV::Loop::Async
80
81			PROTOTYPES: ENABLE
82
83			BOOT:
84			{
85			I_EV_API ("EV::Loop::Async");
86			}
87
88			void
89			_c_func (SV *loop)
90			PPCODE:
91			EXTEND (SP, 2);
92			PUSHs (sv_2mortal (newSViv (PTR2IV (c_func))));
93			PUSHs (sv_2mortal (newSViv (SvIVX (SvRV (loop)))));
94
95			void
96			_attach (SV loop_, SV asy, IV sig_func, IV sig_arg)
97			CODE:
98			{
99			pthread_mutexattr_t ma;
100			struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
101			udat *u;
102			printf ("f,a %p,%p\n", sig_func, sig_arg);//D
103
104			Newz (0, u, 1, udat);
105			u->asy = newSVsv (asy);
106			u->signal_func = (void ()(void , int))sig_func;
107			u->signal_arg = sig_arg;
108
109			ev_async_init (&u->async_w, async_cb);
110			ev_async_start (loop, &u->async_w);
111
112			pthread_mutexattr_init (&ma);
113			pthread_mutexattr_settype (&ma, PTHREAD_MUTEX_RECURSIVE);
114			pthread_mutex_init (&u->lock, &ma);
115			pthread_mutex_init (&u->invoke_mutex, &ma);
116			pthread_mutexattr_destroy (&ma);
117
118			pthread_cond_init (&u->invoke_cv, 0);
119
120			ev_set_userdata (loop, u);
121			ev_set_invoke_pending_cb (loop, l_invoke);
122			ev_set_loop_release_cb (loop, l_release, l_acquire);
123
124			X_LOCK (u->lock);
125			thread_create (&u->tid, l_run, (void *)loop);
126			}
127
128			void
129			lock (SV *loop_)
130			ALIAS:
131			lock = 0
132			unlock = 1
133			poke = 2
134			CODE:
135			{
136			struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
137			udat *u = ev_userdata (loop);
138
139			switch (ix)
140			{
141			case 0: X_LOCK (u->lock); break;
142			case 1: X_UNLOCK (u->lock); break;
143			case 2: ev_async_send (loop, &u->async_w); break;
144			}
145			}
146
147			void
148			DESTROY (SV *loop_)
149			CODE:
150			{
151			struct ev_loop loop = (struct ev_loop )SvIVX (SvRV (loop_));
152			udat *u = ev_userdata (loop);
153
154			if (u)
155			{
156			X_LOCK (u->lock);
157			ev_async_send (loop, &u->async_w);
158			pthread_cancel (u->tid);
159			ev_async_stop (loop, &u->async_w);
160			pthread_mutex_destroy (&u->lock);
161			pthread_cond_destroy (&u->invoke_cv);
162			pthread_mutex_destroy (&u->invoke_mutex);
163			Safefree (u);
164			SvREFCNT_dec (u->asy);
165			}
166			}
167
168
169