Coro-Multicore/Multicore.xs

/* most win32 perls are beyond fixing, requiring dTHX */
/* even for ISO-C functions such as malloc. avoid! avoid! avoid! */
/* and fail to define numerous symbols, but still overrwide them */
/* with non-working versions (e.g. setjmp). */
#ifdef _WIN32
/*# define PERL_CORE 1 fixes some, breaks others */
#else
# define PERL_NO_GET_CONTEXT
#endif

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

#define X_STACKSIZE 1024 * sizeof (void *)

#include "CoroAPI.h"
#include "perlmulticore.h"
#include "schmorp.h"
#include "xthread.h"

#ifdef _WIN32
  #ifndef sigset_t
    #define sigset_t int
  #endif
#endif

#ifndef SvREFCNT_dec_NN
  #define SvREFCNT_dec_NN(sv) SvREFCNT_dec (sv)
#endif

#ifndef SvREFCNT_dec_simple_void_NN
  #define SvREFCNT_dec_simple_void_NN(sv) SvREFCNT_dec (sv)
#endif

#ifndef SvREFCNT_inc_NN
  #define SvREFCNT_inc_NN(sv) SvREFCNT_inc (sv)
#endif

#ifndef RECURSION_CHECK
  #define RECURSION_CHECK 0
#endif

static X_TLS_DECLARE(current_key);
#if RECURSION_CHECK
static X_TLS_DECLARE(check_key);
#endif

static void
fatal (const char *msg)
{
  write (2, msg, strlen (msg));
  abort ();
}

static s_epipe ep;
static void *perl_thx;
static sigset_t cursigset, fullsigset;

static int global_enable = 0;
static int thread_enable; /* 0 undefined, 1 disabled, 2 enabled */

/* assigned to a thread for each release/acquire */
struct tctx
{
  void *coro;
  int wait_f;
  xcond_t acquire_c;
  int jeret;
};

static struct tctx *tctx_free;

static struct tctx *
tctx_get (void)
{
  struct tctx *ctx;

  if (!tctx_free)
    {
      ctx = malloc (sizeof (*tctx_free));
      X_COND_CREATE (ctx->acquire_c);
    }
  else
    {
      ctx = tctx_free;
      tctx_free = tctx_free->coro;
    }

  return ctx;
}

static void
tctx_put (struct tctx *ctx)
{
  ctx->coro = tctx_free;
  tctx_free = ctx;
}

/* a stack of tctxs */
struct tctxs
{
  struct tctx **ctxs;
  int cur, max;
};

static struct tctx *
tctxs_get (struct tctxs *ctxs)
{
  return ctxs->ctxs[--ctxs->cur];
}

static void
tctxs_put (struct tctxs *ctxs, struct tctx *ctx)
{
  if (ctxs->cur >= ctxs->max)
    {
      ctxs->max = ctxs->max ? ctxs->max * 2 : 16;
      ctxs->ctxs = realloc (ctxs->ctxs, ctxs->max * sizeof (ctxs->ctxs[0]));
    }

  ctxs->ctxs[ctxs->cur++] = ctx;
}

static xmutex_t release_m = X_MUTEX_INIT;
static xcond_t release_c = X_COND_INIT;
static struct tctxs releasers;
static int idle;
static int min_idle = 1;
static int curthreads, max_threads = 1; /* protected by release_m */

static xmutex_t acquire_m = X_MUTEX_INIT;
static struct tctxs acquirers;

X_THREAD_PROC(thread_proc)
{
  PERL_SET_CONTEXT (perl_thx);

  {
    dTHXa (perl_thx);
    dJMPENV;
    struct tctx *ctx;
    int catchret;

    X_LOCK (release_m);

    for (;;)
      {
        while (!releasers.cur)
          if (idle <= min_idle || 1)
            X_COND_WAIT (release_c, release_m);
          else
            {
              struct timespec ts = { time (0) + idle - min_idle, 0 };

              if (X_COND_TIMEDWAIT (release_c, release_m, ts) == ETIMEDOUT)
                if (idle > min_idle && !releasers.cur)
                  break;
            }

        ctx = tctxs_get (&releasers);
        --idle;
        X_UNLOCK (release_m);

        if (!ctx) /* timed out? */
          break;

        pthread_sigmask (SIG_SETMASK, &cursigset, 0);
        JMPENV_PUSH (ctx->jeret);

        if (!ctx->jeret)
          while (ctx->coro)
            CORO_SCHEDULE;

        JMPENV_POP;
        pthread_sigmask (SIG_SETMASK, &fullsigset, &cursigset);

        X_LOCK (acquire_m);
        ctx->wait_f = 1;
        X_COND_SIGNAL (ctx->acquire_c);
        X_UNLOCK (acquire_m);

        X_LOCK (release_m);
        ++idle;
      }
  }
}

static void
start_thread (void)
{
  xthread_t tid;

  if (curthreads >= max_threads && 0)
    return;

  ++curthreads;
  ++idle;
  xthread_create (&tid, thread_proc, 0);
}

static void
pmapi_release (void)
{
  if (! ((thread_enable ? thread_enable : global_enable) & 1))
    {
      X_TLS_SET (current_key, 0);
      return;
    }

  #if RECURSION_CHECK
  if (X_TLS_GET (check_key))
    fatal ("FATAL: perlinterp_release () called without valid perl context");

  X_TLS_SET (check_key, &check_key);
  #endif

  struct tctx *ctx = tctx_get ();
  ctx->coro = SvREFCNT_inc_simple_NN (CORO_CURRENT);
  ctx->wait_f = 0;

  X_TLS_SET (current_key, ctx);
  pthread_sigmask (SIG_SETMASK, &fullsigset, &cursigset);

  X_LOCK (release_m);

  if (idle <= min_idle)
    start_thread ();

  tctxs_put (&releasers, ctx);
  X_COND_SIGNAL (release_c);

  while (!idle && releasers.cur)
    {
      X_UNLOCK (release_m);
      X_LOCK (release_m);
    }

  X_UNLOCK (release_m);
}

static void
pmapi_acquire (void)
{
  int jeret;
  struct tctx *ctx = X_TLS_GET (current_key);

  if (!ctx)
    return;

  #if RECURSION_CHECK
  if (X_TLS_GET (check_key) != &check_key)
    fatal ("FATAL: perlinterp_acquire () called with valid perl context");

  X_TLS_SET (check_key, 0);
  #endif

  X_LOCK (acquire_m);

  tctxs_put (&acquirers, ctx);

  s_epipe_signal (&ep);
  while (!ctx->wait_f)
    X_COND_WAIT (ctx->acquire_c, acquire_m);
  X_UNLOCK (acquire_m);

  jeret = ctx->jeret;
  tctx_put (ctx);
  pthread_sigmask (SIG_SETMASK, &cursigset, 0);

  if (jeret)
    {
      dTHX;
      JMPENV_JUMP (jeret);
    }
}

static void
set_thread_enable (pTHX_ void *arg)
{
  thread_enable = PTR2IV (arg);
}

MODULE = Coro::Multicore                PACKAGE = Coro::Multicore

PROTOTYPES: DISABLE

BOOT:
{
#ifndef _WIN32
        sigfillset (&fullsigset);
#endif

        X_TLS_INIT (current_key);
#if RECURSION_CHECK
        X_TLS_INIT (check_key);
#endif

        if (s_epipe_new (&ep))
          croak ("Coro::Multicore: unable to initialise event pipe.\n");

        perl_thx = PERL_GET_CONTEXT;

        I_CORO_API ("Coro::Multicore");

        if (0) { /*D*/
        X_LOCK (release_m);
        while (idle < min_idle)
          start_thread ();
        X_UNLOCK (release_m);
        }

        /* not perfectly efficient to do it this way, but it is simple */
        perl_multicore_init (); /* calls release */
        perl_multicore_api->pmapi_release = pmapi_release;
        perl_multicore_api->pmapi_acquire = pmapi_acquire;
}

bool
enable (bool enable = NO_INIT)
        CODE:
        RETVAL = global_enable;
        if (items)
          global_enable = enable;
        OUTPUT:
        RETVAL

void
scoped_enable ()
        CODE:
        LEAVE; /* see Guard.xs */
        CORO_ENTERLEAVE_SCOPE_HOOK (set_thread_enable, (void *)1, set_thread_enable, (void *)0);
        ENTER; /* see Guard.xs */

void
scoped_disable ()
        CODE:
        LEAVE; /* see Guard.xs */
        CORO_ENTERLEAVE_SCOPE_HOOK (set_thread_enable, (void *)2, set_thread_enable, (void *)0);
        ENTER; /* see Guard.xs */

#if 0

U32
min_idle_threads (U32 min = NO_INIT)
        CODE:
        X_LOCK (acquire_m);
        RETVAL = min_idle;
        if (items)
          min_idle = min;
        X_UNLOCK (acquire_m);
        OUTPUT:
        RETVAL

#endif

int
fd ()
        CODE:
        RETVAL = s_epipe_fd (&ep);
        OUTPUT:
        RETVAL

void
poll (...)
        CODE:
        s_epipe_drain (&ep);
        X_LOCK (acquire_m);
        while (acquirers.cur)
          {
            struct tctx *ctx = tctxs_get (&acquirers);
            CORO_READY ((SV *)ctx->coro);
            SvREFCNT_dec_simple_void_NN ((SV *)ctx->coro);
            ctx->coro = 0;
          }
        X_UNLOCK (acquire_m);

void
sleep (NV seconds)
        CODE:
        perlinterp_release ();
        {
          int nsec = seconds;
          if (nsec) sleep (nsec);
          nsec = (seconds - nsec) * 1e9;
          if (nsec) usleep (nsec);
        }
        perlinterp_acquire ();

Revision:	1.25
Committed:	Wed Mar 6 16:39:45 2019 UTC (5 years, 2 months ago) by root
Branch:	MAIN
CVS Tags:	rel-1_03
Changes since 1.24:	+13 -10 lines
Log Message:	1.03
#	Content
1	/* most win32 perls are beyond fixing, requiring dTHX */
2	/* even for ISO-C functions such as malloc. avoid! avoid! avoid! */
3	/* and fail to define numerous symbols, but still overrwide them */
4	/* with non-working versions (e.g. setjmp). */
5	#ifdef _WIN32
6	/# define PERL_CORE 1 fixes some, breaks others /
7	#else
8	# define PERL_NO_GET_CONTEXT
9	#endif
10
11	#include "EXTERN.h"
12	#include "perl.h"
13	#include "XSUB.h"
14
15	#define X_STACKSIZE 1024 * sizeof (void *)
16
17	#include "CoroAPI.h"
18	#include "perlmulticore.h"
19	#include "schmorp.h"
20	#include "xthread.h"
21
22	#ifdef _WIN32
23	#ifndef sigset_t
24	#define sigset_t int
25	#endif
26	#endif
27
28	#ifndef SvREFCNT_dec_NN
29	#define SvREFCNT_dec_NN(sv) SvREFCNT_dec (sv)
30	#endif
31
32	#ifndef SvREFCNT_dec_simple_void_NN
33	#define SvREFCNT_dec_simple_void_NN(sv) SvREFCNT_dec (sv)
34	#endif
35
36	#ifndef SvREFCNT_inc_NN
37	#define SvREFCNT_inc_NN(sv) SvREFCNT_inc (sv)
38	#endif
39
40	#ifndef RECURSION_CHECK
41	#define RECURSION_CHECK 0
42	#endif
43
44	static X_TLS_DECLARE(current_key);
45	#if RECURSION_CHECK
46	static X_TLS_DECLARE(check_key);
47	#endif
48
49	static void
50	fatal (const char *msg)
51	{
52	write (2, msg, strlen (msg));
53	abort ();
54	}
55
56	static s_epipe ep;
57	static void *perl_thx;
58	static sigset_t cursigset, fullsigset;
59
60	static int global_enable = 0;
61	static int thread_enable; /* 0 undefined, 1 disabled, 2 enabled */
62
63	/* assigned to a thread for each release/acquire */
64	struct tctx
65	{
66	void *coro;
67	int wait_f;
68	xcond_t acquire_c;
69	int jeret;
70	};
71
72	static struct tctx *tctx_free;
73
74	static struct tctx *
75	tctx_get (void)
76	{
77	struct tctx *ctx;
78
79	if (!tctx_free)
80	{
81	ctx = malloc (sizeof (*tctx_free));
82	X_COND_CREATE (ctx->acquire_c);
83	}
84	else
85	{
86	ctx = tctx_free;
87	tctx_free = tctx_free->coro;
88	}
89
90	return ctx;
91	}
92
93	static void
94	tctx_put (struct tctx *ctx)
95	{
96	ctx->coro = tctx_free;
97	tctx_free = ctx;
98	}
99
100	/* a stack of tctxs */
101	struct tctxs
102	{
103	struct tctx **ctxs;
104	int cur, max;
105	};
106
107	static struct tctx *
108	tctxs_get (struct tctxs *ctxs)
109	{
110	return ctxs->ctxs[--ctxs->cur];
111	}
112
113	static void
114	tctxs_put (struct tctxs ctxs, struct tctx ctx)
115	{
116	if (ctxs->cur >= ctxs->max)
117	{
118	ctxs->max = ctxs->max ? ctxs->max * 2 : 16;
119	ctxs->ctxs = realloc (ctxs->ctxs, ctxs->max * sizeof (ctxs->ctxs[0]));
120	}
121
122	ctxs->ctxs[ctxs->cur++] = ctx;
123	}
124
125	static xmutex_t release_m = X_MUTEX_INIT;
126	static xcond_t release_c = X_COND_INIT;
127	static struct tctxs releasers;
128	static int idle;
129	static int min_idle = 1;
130	static int curthreads, max_threads = 1; /* protected by release_m */
131
132	static xmutex_t acquire_m = X_MUTEX_INIT;
133	static struct tctxs acquirers;
134
135	X_THREAD_PROC(thread_proc)
136	{
137	PERL_SET_CONTEXT (perl_thx);
138
139	{
140	dTHXa (perl_thx);
141	dJMPENV;
142	struct tctx *ctx;
143	int catchret;
144
145	X_LOCK (release_m);
146
147	for (;;)
148	{
149	while (!releasers.cur)
150	if (idle <= min_idle \|\| 1)
151	X_COND_WAIT (release_c, release_m);
152	else
153	{
154	struct timespec ts = { time (0) + idle - min_idle, 0 };
155
156	if (X_COND_TIMEDWAIT (release_c, release_m, ts) == ETIMEDOUT)
157	if (idle > min_idle && !releasers.cur)
158	break;
159	}
160
161	ctx = tctxs_get (&releasers);
162	--idle;
163	X_UNLOCK (release_m);
164
165	if (!ctx) /* timed out? */
166	break;
167
168	pthread_sigmask (SIG_SETMASK, &cursigset, 0);
169	JMPENV_PUSH (ctx->jeret);
170
171	if (!ctx->jeret)
172	while (ctx->coro)
173	CORO_SCHEDULE;
174
175	JMPENV_POP;
176	pthread_sigmask (SIG_SETMASK, &fullsigset, &cursigset);
177
178	X_LOCK (acquire_m);
179	ctx->wait_f = 1;
180	X_COND_SIGNAL (ctx->acquire_c);
181	X_UNLOCK (acquire_m);
182
183	X_LOCK (release_m);
184	++idle;
185	}
186	}
187	}
188
189	static void
190	start_thread (void)
191	{
192	xthread_t tid;
193
194	if (curthreads >= max_threads && 0)
195	return;
196
197	++curthreads;
198	++idle;
199	xthread_create (&tid, thread_proc, 0);
200	}
201
202	static void
203	pmapi_release (void)
204	{
205	if (! ((thread_enable ? thread_enable : global_enable) & 1))
206	{
207	X_TLS_SET (current_key, 0);
208	return;
209	}
210
211	#if RECURSION_CHECK
212	if (X_TLS_GET (check_key))
213	fatal ("FATAL: perlinterp_release () called without valid perl context");
214
215	X_TLS_SET (check_key, &check_key);
216	#endif
217
218	struct tctx *ctx = tctx_get ();
219	ctx->coro = SvREFCNT_inc_simple_NN (CORO_CURRENT);
220	ctx->wait_f = 0;
221
222	X_TLS_SET (current_key, ctx);
223	pthread_sigmask (SIG_SETMASK, &fullsigset, &cursigset);
224
225	X_LOCK (release_m);
226
227	if (idle <= min_idle)
228	start_thread ();
229
230	tctxs_put (&releasers, ctx);
231	X_COND_SIGNAL (release_c);
232
233	while (!idle && releasers.cur)
234	{
235	X_UNLOCK (release_m);
236	X_LOCK (release_m);
237	}
238
239	X_UNLOCK (release_m);
240	}
241
242	static void
243	pmapi_acquire (void)
244	{
245	int jeret;
246	struct tctx *ctx = X_TLS_GET (current_key);
247
248	if (!ctx)
249	return;
250
251	#if RECURSION_CHECK
252	if (X_TLS_GET (check_key) != &check_key)
253	fatal ("FATAL: perlinterp_acquire () called with valid perl context");
254
255	X_TLS_SET (check_key, 0);
256	#endif
257
258	X_LOCK (acquire_m);
259
260	tctxs_put (&acquirers, ctx);
261
262	s_epipe_signal (&ep);
263	while (!ctx->wait_f)
264	X_COND_WAIT (ctx->acquire_c, acquire_m);
265	X_UNLOCK (acquire_m);
266
267	jeret = ctx->jeret;
268	tctx_put (ctx);
269	pthread_sigmask (SIG_SETMASK, &cursigset, 0);
270
271	if (jeret)
272	{
273	dTHX;
274	JMPENV_JUMP (jeret);
275	}
276	}
277
278	static void
279	set_thread_enable (pTHX_ void *arg)
280	{
281	thread_enable = PTR2IV (arg);
282	}
283
284	MODULE = Coro::Multicore PACKAGE = Coro::Multicore
285
286	PROTOTYPES: DISABLE
287
288	BOOT:
289	{
290	#ifndef _WIN32
291	sigfillset (&fullsigset);
292	#endif
293
294	X_TLS_INIT (current_key);
295	#if RECURSION_CHECK
296	X_TLS_INIT (check_key);
297	#endif
298
299	if (s_epipe_new (&ep))
300	croak ("Coro::Multicore: unable to initialise event pipe.\n");
301
302	perl_thx = PERL_GET_CONTEXT;
303
304	I_CORO_API ("Coro::Multicore");
305
306	if (0) { /D/
307	X_LOCK (release_m);
308	while (idle < min_idle)
309	start_thread ();
310	X_UNLOCK (release_m);
311	}
312
313	/* not perfectly efficient to do it this way, but it is simple */
314	perl_multicore_init (); /* calls release */
315	perl_multicore_api->pmapi_release = pmapi_release;
316	perl_multicore_api->pmapi_acquire = pmapi_acquire;
317	}
318
319	bool
320	enable (bool enable = NO_INIT)
321	CODE:
322	RETVAL = global_enable;
323	if (items)
324	global_enable = enable;
325	OUTPUT:
326	RETVAL
327
328	void
329	scoped_enable ()
330	CODE:
331	LEAVE; /* see Guard.xs */
332	CORO_ENTERLEAVE_SCOPE_HOOK (set_thread_enable, (void )1, set_thread_enable, (void )0);
333	ENTER; /* see Guard.xs */
334
335	void
336	scoped_disable ()
337	CODE:
338	LEAVE; /* see Guard.xs */
339	CORO_ENTERLEAVE_SCOPE_HOOK (set_thread_enable, (void )2, set_thread_enable, (void )0);
340	ENTER; /* see Guard.xs */
341
342	#if 0
343
344	U32
345	min_idle_threads (U32 min = NO_INIT)
346	CODE:
347	X_LOCK (acquire_m);
348	RETVAL = min_idle;
349	if (items)
350	min_idle = min;
351	X_UNLOCK (acquire_m);
352	OUTPUT:
353	RETVAL
354
355	#endif
356
357	int
358	fd ()
359	CODE:
360	RETVAL = s_epipe_fd (&ep);
361	OUTPUT:
362	RETVAL
363
364	void
365	poll (...)
366	CODE:
367	s_epipe_drain (&ep);
368	X_LOCK (acquire_m);
369	while (acquirers.cur)
370	{
371	struct tctx *ctx = tctxs_get (&acquirers);
372	CORO_READY ((SV *)ctx->coro);
373	SvREFCNT_dec_simple_void_NN ((SV *)ctx->coro);
374	ctx->coro = 0;
375	}
376	X_UNLOCK (acquire_m);
377
378	void
379	sleep (NV seconds)
380	CODE:
381	perlinterp_release ();
382	{
383	int nsec = seconds;
384	if (nsec) sleep (nsec);
385	nsec = (seconds - nsec) * 1e9;
386	if (nsec) usleep (nsec);
387	}
388	perlinterp_acquire ();
389