[p5sagit/p5-mst-13.2.git] / ext / Thread / Thread.xs

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

/* Magic signature for Thread's mg_private is "Th" */ 
#define Thread_MAGIC_SIGNATURE 0x5468

#ifdef __cplusplus
#ifdef I_UNISTD
#include <unistd.h>
#endif
#endif
#include <fcntl.h>
                        
static U32 threadnum = 0;
static int sig_pipe[2];
            
#ifndef THREAD_RET_TYPE
typedef struct perl_thread *Thread;
#define THREAD_RET_TYPE void *
#define THREAD_RET_CAST(x) ((THREAD_RET_TYPE) x)
#endif

static void
remove_thread(struct perl_thread *t)
{
#ifdef USE_THREADS
    DEBUG_L(WITH_THR(PerlIO_printf(PerlIO_stderr(),
				   "%p: remove_thread %p\n", thr, t)));
    MUTEX_LOCK(&threads_mutex);
    MUTEX_DESTROY(&t->mutex);
    nthreads--;
    t->prev->next = t->next;
    t->next->prev = t->prev;
    COND_BROADCAST(&nthreads_cond);
    MUTEX_UNLOCK(&threads_mutex);
#endif
}

static THREAD_RET_TYPE
threadstart(void *arg)
{
#ifdef USE_THREADS
#ifdef FAKE_THREADS
    Thread savethread = thr;
    LOGOP myop;
    dSP;
    I32 oldscope = scopestack_ix;
    I32 retval;
    AV *av;
    int i;

    DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p starting at %s\n",
			  thr, SvPEEK(TOPs)));
    thr = (Thread) arg;
    savemark = TOPMARK;
    thr->prev = thr->prev_run = savethread;
    thr->next = savethread->next;
    thr->next_run = savethread->next_run;
    savethread->next = savethread->next_run = thr;
    thr->wait_queue = 0;
    thr->private = 0;

    /* Now duplicate most of perl_call_sv but with a few twists */
    op = (OP*)&myop;
    Zero(op, 1, LOGOP);
    myop.op_flags = OPf_STACKED;
    myop.op_next = Nullop;
    myop.op_flags |= OPf_KNOW;
    myop.op_flags |= OPf_WANT_LIST;
    op = pp_entersub(ARGS);
    DEBUG_L(if (!op)
	    PerlIO_printf(PerlIO_stderr(), "thread starts at Nullop\n"));
    /*
     * When this thread is next scheduled, we start in the right
     * place. When the thread runs off the end of the sub, perl.c
     * handles things, using savemark to figure out how much of the
     * stack is the return value for any join.
     */
    thr = savethread;		/* back to the old thread */
    return 0;
#else
    Thread thr = (Thread) arg;
    LOGOP myop;
    djSP;
    I32 oldmark = TOPMARK;
    I32 oldscope = scopestack_ix;
    I32 retval;
    SV *sv;
    AV *av = newAV();
    int i, ret;
    dJMPENV;
    DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p waiting to start\n",
			  thr));

    /* Don't call *anything* requiring dTHR until after SET_THR() */
    /*
     * Wait until our creator releases us. If we didn't do this, then
     * it would be potentially possible for out thread to carry on and
     * do stuff before our creator fills in our "self" field. For example,
     * if we went and created another thread which tried to JOIN with us,
     * then we'd be in a mess.
     */
    MUTEX_LOCK(&thr->mutex);
    MUTEX_UNLOCK(&thr->mutex);

    /*
     * It's safe to wait until now to set the thread-specific pointer
     * from our pthread_t structure to our struct perl_thread, since
     * we're the only thread who can get at it anyway.
     */
    SET_THR(thr);

    /* Only now can we use SvPEEK (which calls sv_newmortal which does dTHR) */
    DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p starting at %s\n",
			  thr, SvPEEK(TOPs)));

    sv = POPs;
    PUTBACK;
    perl_call_sv(sv, G_ARRAY|G_EVAL);
    SPAGAIN;
    retval = sp - (stack_base + oldmark);
    sp = stack_base + oldmark + 1;
    if (SvCUR(thr->errsv)) {
	MUTEX_LOCK(&thr->mutex);
	thr->flags |= THRf_DID_DIE;
	MUTEX_UNLOCK(&thr->mutex);
	av_store(av, 0, &sv_no);
	av_store(av, 1, newSVsv(thr->errsv));
	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p died: %s\n",
			      SvPV(thr->errsv, na));
    } else {
	DEBUG_L(STMT_START {
	    for (i = 1; i <= retval; i++) {
		PerlIO_printf(PerlIO_stderr(), "%p return[%d] = %s\n",
				thr, i, SvPEEK(sp[i - 1]));)
	    }
	} STMT_END);
	av_store(av, 0, &sv_yes);
	for (i = 1; i <= retval; i++, sp++)
	    sv_setsv(*av_fetch(av, i, TRUE), SvREFCNT_inc(*sp));
    }

  finishoff:
#if 0    
    /* removed for debug */
    SvREFCNT_dec(curstack);
#endif
    SvREFCNT_dec(thr->cvcache);
    SvREFCNT_dec(thr->threadsv);
    SvREFCNT_dec(thr->specific);
    SvREFCNT_dec(thr->errsv);
    SvREFCNT_dec(thr->errhv);
    Safefree(markstack);
    Safefree(scopestack);
    Safefree(savestack);
    Safefree(retstack);
    Safefree(cxstack);
    Safefree(tmps_stack);
    Safefree(ofs);

    MUTEX_LOCK(&thr->mutex);
    DEBUG_L(PerlIO_printf(PerlIO_stderr(),
			  "%p: threadstart finishing: state is %u\n",
			  thr, ThrSTATE(thr)));
    switch (ThrSTATE(thr)) {
    case THRf_R_JOINABLE:
	ThrSETSTATE(thr, THRf_ZOMBIE);
	MUTEX_UNLOCK(&thr->mutex);
	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
			      "%p: R_JOINABLE thread finished\n", thr));
	break;
    case THRf_R_JOINED:
	ThrSETSTATE(thr, THRf_DEAD);
	MUTEX_UNLOCK(&thr->mutex);
	remove_thread(thr);
	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
			      "%p: R_JOINED thread finished\n", thr));
	break;
    case THRf_R_DETACHED:
	ThrSETSTATE(thr, THRf_DEAD);
	MUTEX_UNLOCK(&thr->mutex);
	SvREFCNT_dec(av);
	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
			      "%p: DETACHED thread finished\n", thr));
	remove_thread(thr);	/* This might trigger main thread to finish */
	break;
    default:
	MUTEX_UNLOCK(&thr->mutex);
	croak("panic: illegal state %u at end of threadstart", ThrSTATE(thr));
	/* NOTREACHED */
    }
    return THREAD_RET_CAST(av);	/* Available for anyone to join with */
					/* us unless we're detached, in which */
					/* case noone sees the value anyway. */
#endif    
#else
    return THREAD_RET_CAST(NULL);
#endif
}

static SV *
newthread (SV *startsv, AV *initargs, char *classname)
{
#ifdef USE_THREADS
    dSP;
    Thread savethread;
    int i;
    SV *sv;
    int err;
#ifndef THREAD_CREATE
    sigset_t fullmask, oldmask;
#endif
    
    savethread = thr;
    thr = new_struct_thread(thr);
    SPAGAIN;
    DEBUG_L(PerlIO_printf(PerlIO_stderr(),
			  "%p: newthread (%p), tid is %u, preparing stack\n",
			  savethread, thr, thr->tid));
    /* The following pushes the arg list and startsv onto the *new* stack */
    PUSHMARK(sp);
    /* Could easily speed up the following greatly */
    for (i = 0; i <= AvFILL(initargs); i++)
	XPUSHs(SvREFCNT_inc(*av_fetch(initargs, i, FALSE)));
    XPUSHs(SvREFCNT_inc(startsv));
    PUTBACK;
#ifdef THREAD_CREATE
    err = THREAD_CREATE(thr, threadstart);
#else    
    /* On your marks... */
    MUTEX_LOCK(&thr->mutex);
    /* Get set...  */
    sigfillset(&fullmask);
    if (sigprocmask(SIG_SETMASK, &fullmask, &oldmask) == -1)
	croak("panic: sigprocmask");
#ifdef PTHREADS_CREATED_JOINABLE
    err = pthread_create(&thr->self, pthread_attr_default,
			 threadstart, (void*) thr);
#else
    {
	pthread_attr_t attr;

	err = pthread_attr_init(&attr);
	if (err == 0) {
#ifdef PTHREAD_CREATE_UNDETACHED
	  err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_UNDETACHED);
#else
	  croak("panic: pthread_attr_setdetachstate");
#endif
	  if (err == 0) 
	    err = pthread_create(&thr->self, &attr,
				 threadstart, (void*) thr);
	}
	pthread_attr_destroy(&attr);
    }
#endif
    /* Go */
    MUTEX_UNLOCK(&thr->mutex);
#endif
    if (err) {
        DEBUG_L(PerlIO_printf(PerlIO_stderr(),
			  "%p: create of %p failed %d\n", savethread, thr, err));
	/* Thread creation failed--clean up */
	SvREFCNT_dec(thr->cvcache);
	remove_thread(thr);
	MUTEX_DESTROY(&thr->mutex);
	for (i = 0; i <= AvFILL(initargs); i++)
	    SvREFCNT_dec(*av_fetch(initargs, i, FALSE));
	SvREFCNT_dec(startsv);
	return NULL;
    }
#ifdef THREAD_POST_CREATE
    THREAD_POST_CREATE(thr);
#else
    if (sigprocmask(SIG_SETMASK, &oldmask, 0))
	croak("panic: sigprocmask");
#endif
    sv = newSViv(thr->tid);
    sv_magic(sv, thr->oursv, '~', 0, 0);
    SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
    return sv_bless(newRV_noinc(sv), gv_stashpv(classname, TRUE));
#else
    croak("No threads in this perl");
    return &sv_undef;
#endif
}

static Signal_t handle_thread_signal _((int sig));

static Signal_t
handle_thread_signal(int sig)
{
    char c = (char) sig;
    write(sig_pipe[0], &c, 1);
}

MODULE = Thread		PACKAGE = Thread
PROTOTYPES: DISABLE

void
new(classname, startsv, ...)
	char *		classname
	SV *		startsv
	AV *		av = av_make(items - 2, &ST(2));
    PPCODE:
	XPUSHs(sv_2mortal(newthread(startsv, av, classname)));

void
join(t)
	Thread	t
	AV *	av = NO_INIT
	int	i = NO_INIT
    PPCODE:
#ifdef USE_THREADS
	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: joining %p (state %u)\n",
			      thr, t, ThrSTATE(t)););
    	MUTEX_LOCK(&t->mutex);
	switch (ThrSTATE(t)) {
	case THRf_R_JOINABLE:
	case THRf_R_JOINED:
	    ThrSETSTATE(t, THRf_R_JOINED);
	    MUTEX_UNLOCK(&t->mutex);
	    break;
	case THRf_ZOMBIE:
	    ThrSETSTATE(t, THRf_DEAD);
	    MUTEX_UNLOCK(&t->mutex);
	    remove_thread(t);
	    break;
	default:
	    MUTEX_UNLOCK(&t->mutex);
	    croak("can't join with thread");
	    /* NOTREACHED */
	}
	JOIN(t, &av);

	if (SvTRUE(*av_fetch(av, 0, FALSE))) {
	    /* Could easily speed up the following if necessary */
	    for (i = 1; i <= AvFILL(av); i++)
		XPUSHs(sv_2mortal(*av_fetch(av, i, FALSE)));
	} else {
	    char *mess = SvPV(*av_fetch(av, 1, FALSE), na);
	    DEBUG_L(PerlIO_printf(PerlIO_stderr(),
				  "%p: join propagating die message: %s\n",
				  thr, mess));
	    croak(mess);
	}
#endif

void
detach(t)
	Thread	t
    CODE:
#ifdef USE_THREADS
	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: detaching %p (state %u)\n",
			      thr, t, ThrSTATE(t)););
    	MUTEX_LOCK(&t->mutex);
	switch (ThrSTATE(t)) {
	case THRf_R_JOINABLE:
	    ThrSETSTATE(t, THRf_R_DETACHED);
	    /* fall through */
	case THRf_R_DETACHED:
	    DETACH(t);
	    MUTEX_UNLOCK(&t->mutex);
	    break;
	case THRf_ZOMBIE:
	    ThrSETSTATE(t, THRf_DEAD);
	    DETACH(t);
	    MUTEX_UNLOCK(&t->mutex);
	    remove_thread(t);
	    break;
	default:
	    MUTEX_UNLOCK(&t->mutex);
	    croak("can't detach thread");
	    /* NOTREACHED */
	}
#endif

void
equal(t1, t2)
	Thread	t1
	Thread	t2
    PPCODE:
	PUSHs((t1 == t2) ? &sv_yes : &sv_no);

void
flags(t)
	Thread	t
    PPCODE:
#ifdef USE_THREADS
	PUSHs(sv_2mortal(newSViv(t->flags)));
#endif

void
self(classname)
	char *	classname
    PREINIT:
	SV *sv;
    PPCODE:        
#ifdef USE_THREADS
	sv = newSViv(thr->tid);
	sv_magic(sv, thr->oursv, '~', 0, 0);
	SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
	PUSHs(sv_2mortal(sv_bless(newRV_noinc(sv),
				  gv_stashpv(classname, TRUE))));
#endif

U32
tid(t)
	Thread	t
    CODE:
#ifdef USE_THREADS
    	MUTEX_LOCK(&t->mutex);
	RETVAL = t->tid;
    	MUTEX_UNLOCK(&t->mutex);
#else 
	RETVAL = 0;
#endif
    OUTPUT:
	RETVAL

void
DESTROY(t)
	SV *	t
    PPCODE:
	PUSHs(&sv_yes);

void
yield()
    CODE:
{
#ifdef USE_THREADS
	YIELD;
#endif
}

void
cond_wait(sv)
	SV *	sv
	MAGIC *	mg = NO_INIT
CODE:                       
#ifdef USE_THREADS
	if (SvROK(sv))
	    sv = SvRV(sv);

	mg = condpair_magic(sv);
	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_wait %p\n", thr, sv));
	MUTEX_LOCK(MgMUTEXP(mg));
	if (MgOWNER(mg) != thr) {
	    MUTEX_UNLOCK(MgMUTEXP(mg));
	    croak("cond_wait for lock that we don't own\n");
	}
	MgOWNER(mg) = 0;
	COND_WAIT(MgCONDP(mg), MgMUTEXP(mg));
	while (MgOWNER(mg))
	    COND_WAIT(MgOWNERCONDP(mg), MgMUTEXP(mg));
	MgOWNER(mg) = thr;
	MUTEX_UNLOCK(MgMUTEXP(mg));
#endif

void
cond_signal(sv)
	SV *	sv
	MAGIC *	mg = NO_INIT
CODE:
#ifdef USE_THREADS
	if (SvROK(sv))
	    sv = SvRV(sv);

	mg = condpair_magic(sv);
	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_signal %p\n",thr,sv));
	MUTEX_LOCK(MgMUTEXP(mg));
	if (MgOWNER(mg) != thr) {
	    MUTEX_UNLOCK(MgMUTEXP(mg));
	    croak("cond_signal for lock that we don't own\n");
	}
	COND_SIGNAL(MgCONDP(mg));
	MUTEX_UNLOCK(MgMUTEXP(mg));
#endif

void
cond_broadcast(sv)
	SV *	sv
	MAGIC *	mg = NO_INIT
CODE: 
#ifdef USE_THREADS
	if (SvROK(sv))
	    sv = SvRV(sv);

	mg = condpair_magic(sv);
	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_broadcast %p\n",
			      thr, sv));
	MUTEX_LOCK(MgMUTEXP(mg));
	if (MgOWNER(mg) != thr) {
	    MUTEX_UNLOCK(MgMUTEXP(mg));
	    croak("cond_broadcast for lock that we don't own\n");
	}
	COND_BROADCAST(MgCONDP(mg));
	MUTEX_UNLOCK(MgMUTEXP(mg));
#endif

void
list(classname)
	char *	classname
    PREINIT:
	Thread	t;
	AV *	av;
	SV **	svp;
	int	n = 0;
    PPCODE:
#ifdef USE_THREADS
	av = newAV();
	/*
	 * Iterate until we have enough dynamic storage for all threads.
	 * We mustn't do any allocation while holding threads_mutex though.
	 */
	MUTEX_LOCK(&threads_mutex);
	do {
	    n = nthreads;
	    MUTEX_UNLOCK(&threads_mutex);
	    if (AvFILL(av) < n - 1) {
		int i = AvFILL(av);
		for (i = AvFILL(av); i < n - 1; i++) {
		    SV *sv = newSViv(0);	/* fill in tid later */
		    sv_magic(sv, 0, '~', 0, 0);	/* fill in other magic later */
		    av_push(av, sv_bless(newRV_noinc(sv),
					 gv_stashpv(classname, TRUE)));
	
		}
	    }
	    MUTEX_LOCK(&threads_mutex);
	} while (n < nthreads);
	n = nthreads;	/* Get the final correct value */

	/*
	 * At this point, there's enough room to fill in av.
	 * Note that we are holding threads_mutex so the list
	 * won't change out from under us but all the remaining
	 * processing is "fast" (no blocking, malloc etc.)
	 */
	t = thr;
	svp = AvARRAY(av);
	do {
	    SV *sv = (SV*)SvRV(*svp);
	    sv_setiv(sv, t->tid);
	    SvMAGIC(sv)->mg_obj = SvREFCNT_inc(t->oursv);
	    SvMAGIC(sv)->mg_flags |= MGf_REFCOUNTED;
	    SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
	    t = t->next;
	    svp++;
	} while (t != thr);
	/*  */
	MUTEX_UNLOCK(&threads_mutex);
	/* Truncate any unneeded slots in av */
	av_fill(av, n - 1);
	/* Finally, push all the new objects onto the stack and drop av */
	EXTEND(sp, n);
	for (svp = AvARRAY(av); n > 0; n--, svp++)
	    PUSHs(*svp);
	(void)sv_2mortal((SV*)av);
#endif


MODULE = Thread		PACKAGE = Thread::Signal

void
kill_sighandler_thread()
    PPCODE:
	write(sig_pipe[0], "\0", 1);
	PUSHs(&sv_yes);

void
init_thread_signals()
    PPCODE:
	sighandlerp = handle_thread_signal;
	if (pipe(sig_pipe) == -1)
	    XSRETURN_UNDEF;
	PUSHs(&sv_yes);

SV *
await_signal()
    PREINIT:
	char c;
	SSize_t ret;
    CODE:
	do {
	    ret = read(sig_pipe[1], &c, 1);
	} while (ret == -1 && errno == EINTR);
	if (ret == -1)
	    croak("panic: await_signal");
	if (ret == 0)
	    XSRETURN_UNDEF;
	RETVAL = c ? psig_ptr[c] : &sv_no;
    OUTPUT:
	RETVAL

MODULE = Thread		PACKAGE = Thread::Specific

void
data(classname = "Thread::Specific")
	char *	classname
    PPCODE:
#ifdef USE_THREADS
	if (AvFILL(thr->specific) == -1) {
	    GV *gv = gv_fetchpv("Thread::Specific::FIELDS", TRUE, SVt_PVHV);
	    av_store(thr->specific, 0, newRV((SV*)GvHV(gv)));
	}
	XPUSHs(sv_bless(newRV((SV*)thr->specific),gv_stashpv(classname,TRUE)));
#endif
Commit	Line	Data
d9bb3666	1	#include "EXTERN.h"
	2	#include "perl.h"
	3	#include "XSUB.h"
	4
7d901afa	5	/* Magic signature for Thread's mg_private is "Th" */
	6	#define Thread_MAGIC_SIGNATURE 0x5468
	7
f0f333f4	8	#ifdef __cplusplus
	9	#ifdef I_UNISTD
	10	#include <unistd.h>
	11	#endif
	12	#endif
	13	#include <fcntl.h>
	14
7d901afa	15	static U32 threadnum = 0;
85ced67f	16	static int sig_pipe[2];
f0f333f4	17
f0f333f4	18	#ifndef THREAD_RET_TYPE
52e1cb5e	19	typedef struct perl_thread *Thread;
f0f333f4	20	#define THREAD_RET_TYPE void *
f0f333f4	21	#define THREAD_RET_CAST(x) ((THREAD_RET_TYPE) x)
458fb581	22	#endif
683929b4	23
7d901afa	24	static void
52e1cb5e	25	remove_thread(struct perl_thread *t)
7d901afa	26	{
f0f333f4	27	#ifdef USE_THREADS
7d901afa	28	DEBUG_L(WITH_THR(PerlIO_printf(PerlIO_stderr(),
	29	"%p: remove_thread %p\n", thr, t)));
	30	MUTEX_LOCK(&threads_mutex);
0a00ffdb	31	MUTEX_DESTROY(&t->mutex);
7d901afa	32	nthreads--;
	33	t->prev->next = t->next;
	34	t->next->prev = t->prev;
	35	COND_BROADCAST(&nthreads_cond);
	36	MUTEX_UNLOCK(&threads_mutex);
f0f333f4	37	#endif
7d901afa	38	}
7d901afa	39
ea0efc06	40	static THREAD_RET_TYPE
f0f333f4	41	threadstart(void *arg)
d9bb3666	42	{
f0f333f4	43	#ifdef USE_THREADS
783070da	44	#ifdef FAKE_THREADS
	45	Thread savethread = thr;
	46	LOGOP myop;
	47	dSP;
	48	I32 oldscope = scopestack_ix;
	49	I32 retval;
458fb581	50	AV *av;
783070da	51	int i;
783070da	52
683929b4	53	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p starting at %s\n",
683929b4	54	thr, SvPEEK(TOPs)));
783070da	55	thr = (Thread) arg;
	56	savemark = TOPMARK;
	57	thr->prev = thr->prev_run = savethread;
	58	thr->next = savethread->next;
	59	thr->next_run = savethread->next_run;
	60	savethread->next = savethread->next_run = thr;
	61	thr->wait_queue = 0;
	62	thr->private = 0;
	63
	64	/* Now duplicate most of perl_call_sv but with a few twists */
	65	op = (OP*)&myop;
	66	Zero(op, 1, LOGOP);
	67	myop.op_flags = OPf_STACKED;
	68	myop.op_next = Nullop;
	69	myop.op_flags \|= OPf_KNOW;
	70	myop.op_flags \|= OPf_WANT_LIST;
	71	op = pp_entersub(ARGS);
	72	DEBUG_L(if (!op)
	73	PerlIO_printf(PerlIO_stderr(), "thread starts at Nullop\n"));
	74	/*
	75	* When this thread is next scheduled, we start in the right
	76	* place. When the thread runs off the end of the sub, perl.c
	77	* handles things, using savemark to figure out how much of the
	78	* stack is the return value for any join.
	79	*/
	80	thr = savethread; /* back to the old thread */
	81	return 0;
	82	#else
d9bb3666	83	Thread thr = (Thread) arg;
d9bb3666	84	LOGOP myop;
4e35701f	85	djSP;
d9bb3666	86	I32 oldmark = TOPMARK;
	87	I32 oldscope = scopestack_ix;
	88	I32 retval;
458fb581	89	SV *sv;
458fb581	90	AV *av = newAV();
14fcddff	91	int i, ret;
783070da	92	dJMPENV;
0b9678a8	93	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p waiting to start\n",
0b9678a8	94	thr));
783070da	95
0b9678a8	96	/* Don't call anything requiring dTHR until after SET_THR() */
d9bb3666	97	/*
	98	* Wait until our creator releases us. If we didn't do this, then
	99	* it would be potentially possible for out thread to carry on and
	100	* do stuff before our creator fills in our "self" field. For example,
ea0efc06	101	* if we went and created another thread which tried to JOIN with us,
ea0efc06	102	* then we'd be in a mess.
d9bb3666	103	*/
50112d62	104	MUTEX_LOCK(&thr->mutex);
50112d62	105	MUTEX_UNLOCK(&thr->mutex);
d9bb3666	106
d9bb3666	107	/*
d9bb3666	108	* It's safe to wait until now to set the thread-specific pointer
52e1cb5e	109	* from our pthread_t structure to our struct perl_thread, since
52e1cb5e	110	* we're the only thread who can get at it anyway.
d9bb3666	111	*/
ea0efc06	112	SET_THR(thr);
d9bb3666	113
783070da	114	/* Only now can we use SvPEEK (which calls sv_newmortal which does dTHR) */
683929b4	115	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "new thread %p starting at %s\n",
683929b4	116	thr, SvPEEK(TOPs)));
783070da	117
458fb581	118	sv = POPs;
	119	PUTBACK;
	120	perl_call_sv(sv, G_ARRAY\|G_EVAL);
734689b1	121	SPAGAIN;
	122	retval = sp - (stack_base + oldmark);
	123	sp = stack_base + oldmark + 1;
458fb581	124	if (SvCUR(thr->errsv)) {
	125	MUTEX_LOCK(&thr->mutex);
	126	thr->flags \|= THRf_DID_DIE;
	127	MUTEX_UNLOCK(&thr->mutex);
	128	av_store(av, 0, &sv_no);
	129	av_store(av, 1, newSVsv(thr->errsv));
	130	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p died: %s\n",
	131	SvPV(thr->errsv, na));
	132	} else {
	133	DEBUG_L(STMT_START {
	134	for (i = 1; i <= retval; i++) {
	135	PerlIO_printf(PerlIO_stderr(), "%p return[%d] = %s\n",
	136	thr, i, SvPEEK(sp[i - 1]));)
	137	}
	138	} STMT_END);
	139	av_store(av, 0, &sv_yes);
	140	for (i = 1; i <= retval; i++, sp++)
	141	sv_setsv(av_fetch(av, i, TRUE), SvREFCNT_inc(sp));
	142	}
	143
d9bb3666	144	finishoff:
783070da	145	#if 0
	146	/* removed for debug */
	147	SvREFCNT_dec(curstack);
	148	#endif
199100c8	149	SvREFCNT_dec(thr->cvcache);
54b9620d	150	SvREFCNT_dec(thr->threadsv);
554b3eca	151	SvREFCNT_dec(thr->specific);
38a03e6e	152	SvREFCNT_dec(thr->errsv);
38a03e6e	153	SvREFCNT_dec(thr->errhv);
d9bb3666	154	Safefree(markstack);
	155	Safefree(scopestack);
	156	Safefree(savestack);
	157	Safefree(retstack);
	158	Safefree(cxstack);
	159	Safefree(tmps_stack);
199100c8	160	Safefree(ofs);
d9bb3666	161
14fcddff	162	MUTEX_LOCK(&thr->mutex);
50112d62	163	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
	164	"%p: threadstart finishing: state is %u\n",
	165	thr, ThrSTATE(thr)));
14fcddff	166	switch (ThrSTATE(thr)) {
	167	case THRf_R_JOINABLE:
	168	ThrSETSTATE(thr, THRf_ZOMBIE);
	169	MUTEX_UNLOCK(&thr->mutex);
783070da	170	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
14fcddff	171	"%p: R_JOINABLE thread finished\n", thr));
	172	break;
	173	case THRf_R_JOINED:
	174	ThrSETSTATE(thr, THRf_DEAD);
	175	MUTEX_UNLOCK(&thr->mutex);
50112d62	176	remove_thread(thr);
14fcddff	177	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
	178	"%p: R_JOINED thread finished\n", thr));
	179	break;
50112d62	180	case THRf_R_DETACHED:
683929b4	181	ThrSETSTATE(thr, THRf_DEAD);
14fcddff	182	MUTEX_UNLOCK(&thr->mutex);
458fb581	183	SvREFCNT_dec(av);
14fcddff	184	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
14fcddff	185	"%p: DETACHED thread finished\n", thr));
50112d62	186	remove_thread(thr); /* This might trigger main thread to finish */
14fcddff	187	break;
	188	default:
	189	MUTEX_UNLOCK(&thr->mutex);
	190	croak("panic: illegal state %u at end of threadstart", ThrSTATE(thr));
	191	/* NOTREACHED */
734689b1	192	}
458fb581	193	return THREAD_RET_CAST(av); /* Available for anyone to join with */
ea0efc06	194	/* us unless we're detached, in which */
ea0efc06	195	/* case noone sees the value anyway. */
783070da	196	#endif
f0f333f4	197	#else
	198	return THREAD_RET_CAST(NULL);
	199	#endif
d9bb3666	200	}
d9bb3666	201
683929b4	202	static SV *
458fb581	203	newthread (SV startsv, AV initargs, char *classname)
d9bb3666	204	{
f0f333f4	205	#ifdef USE_THREADS
d9bb3666	206	dSP;
	207	Thread savethread;
	208	int i;
683929b4	209	SV *sv;
ea0efc06	210	int err;
ea0efc06	211	#ifndef THREAD_CREATE
f152979c	212	sigset_t fullmask, oldmask;
ea0efc06	213	#endif
d9bb3666	214
d9bb3666	215	savethread = thr;
a863c7d1	216	thr = new_struct_thread(thr);
d9bb3666	217	SPAGAIN;
50112d62	218	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
0b9678a8	219	"%p: newthread (%p), tid is %u, preparing stack\n",
0b9678a8	220	savethread, thr, thr->tid));
d9bb3666	221	/* The following pushes the arg list and startsv onto the new stack */
	222	PUSHMARK(sp);
	223	/* Could easily speed up the following greatly */
734689b1	224	for (i = 0; i <= AvFILL(initargs); i++)
d9bb3666	225	XPUSHs(SvREFCNT_inc(*av_fetch(initargs, i, FALSE)));
	226	XPUSHs(SvREFCNT_inc(startsv));
	227	PUTBACK;
ea0efc06	228	#ifdef THREAD_CREATE
f0f333f4	229	err = THREAD_CREATE(thr, threadstart);
783070da	230	#else
d9bb3666	231	/* On your marks... */
14fcddff	232	MUTEX_LOCK(&thr->mutex);
ea0efc06	233	/* Get set... */
f152979c	234	sigfillset(&fullmask);
	235	if (sigprocmask(SIG_SETMASK, &fullmask, &oldmask) == -1)
	236	croak("panic: sigprocmask");
52e1cb5e	237	#ifdef PTHREADS_CREATED_JOINABLE
46930d8f	238	err = pthread_create(&thr->self, pthread_attr_default,
46930d8f	239	threadstart, (void*) thr);
52e1cb5e	240	#else
	241	{
	242	pthread_attr_t attr;
	243
	244	err = pthread_attr_init(&attr);
	245	if (err == 0) {
	246	#ifdef PTHREAD_CREATE_UNDETACHED
	247	err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_UNDETACHED);
	248	#else
	249	croak("panic: pthread_attr_setdetachstate");
	250	#endif
	251	if (err == 0)
	252	err = pthread_create(&thr->self, &attr,
	253	threadstart, (void*) thr);
	254	}
	255	pthread_attr_destroy(&attr);
	256	}
	257	#endif
d9bb3666	258	/* Go */
14fcddff	259	MUTEX_UNLOCK(&thr->mutex);
ea0efc06	260	#endif
ea0efc06	261	if (err) {
0b9678a8	262	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
0b9678a8	263	"%p: create of %p failed %d\n", savethread, thr, err));
ea0efc06	264	/* Thread creation failed--clean up */
199100c8	265	SvREFCNT_dec(thr->cvcache);
ea0efc06	266	remove_thread(thr);
	267	MUTEX_DESTROY(&thr->mutex);
	268	for (i = 0; i <= AvFILL(initargs); i++)
	269	SvREFCNT_dec(*av_fetch(initargs, i, FALSE));
	270	SvREFCNT_dec(startsv);
	271	return NULL;
	272	}
	273	#ifdef THREAD_POST_CREATE
	274	THREAD_POST_CREATE(thr);
	275	#else
f152979c	276	if (sigprocmask(SIG_SETMASK, &oldmask, 0))
f152979c	277	croak("panic: sigprocmask");
783070da	278	#endif
7d901afa	279	sv = newSViv(thr->tid);
199100c8	280	sv_magic(sv, thr->oursv, '~', 0, 0);
7d901afa	281	SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
458fb581	282	return sv_bless(newRV_noinc(sv), gv_stashpv(classname, TRUE));
f0f333f4	283	#else
	284	croak("No threads in this perl");
	285	return &sv_undef;
	286	#endif
d9bb3666	287	}
d9bb3666	288
f0f333f4	289	static Signal_t handle_thread_signal _((int sig));
f0f333f4	290
f152979c	291	static Signal_t
f0f333f4	292	handle_thread_signal(int sig)
f152979c	293	{
	294	char c = (char) sig;
	295	write(sig_pipe[0], &c, 1);
	296	}
	297
d9bb3666	298	MODULE = Thread PACKAGE = Thread
0b9678a8	299	PROTOTYPES: DISABLE
d9bb3666	300
683929b4	301	void
458fb581	302	new(classname, startsv, ...)
458fb581	303	char * classname
d9bb3666	304	SV * startsv
734689b1	305	AV * av = av_make(items - 2, &ST(2));
683929b4	306	PPCODE:
458fb581	307	XPUSHs(sv_2mortal(newthread(startsv, av, classname)));
d9bb3666	308
d9bb3666	309	void
d9bb3666	310	join(t)
	311	Thread t
	312	AV * av = NO_INIT
	313	int i = NO_INIT
	314	PPCODE:
f0f333f4	315	#ifdef USE_THREADS
7d901afa	316	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: joining %p (state %u)\n",
7d901afa	317	thr, t, ThrSTATE(t)););
50112d62	318	MUTEX_LOCK(&t->mutex);
50112d62	319	switch (ThrSTATE(t)) {
14fcddff	320	case THRf_R_JOINABLE:
14fcddff	321	case THRf_R_JOINED:
50112d62	322	ThrSETSTATE(t, THRf_R_JOINED);
50112d62	323	MUTEX_UNLOCK(&t->mutex);
14fcddff	324	break;
14fcddff	325	case THRf_ZOMBIE:
50112d62	326	ThrSETSTATE(t, THRf_DEAD);
	327	MUTEX_UNLOCK(&t->mutex);
	328	remove_thread(t);
14fcddff	329	break;
14fcddff	330	default:
50112d62	331	MUTEX_UNLOCK(&t->mutex);
14fcddff	332	croak("can't join with thread");
	333	/* NOTREACHED */
	334	}
ea0efc06	335	JOIN(t, &av);
7d901afa	336
458fb581	337	if (SvTRUE(*av_fetch(av, 0, FALSE))) {
	338	/* Could easily speed up the following if necessary */
	339	for (i = 1; i <= AvFILL(av); i++)
	340	XPUSHs(sv_2mortal(*av_fetch(av, i, FALSE)));
	341	} else {
	342	char mess = SvPV(av_fetch(av, 1, FALSE), na);
	343	DEBUG_L(PerlIO_printf(PerlIO_stderr(),
	344	"%p: join propagating die message: %s\n",
	345	thr, mess));
	346	croak(mess);
	347	}
f0f333f4	348	#endif
d9bb3666	349
d9bb3666	350	void
734689b1	351	detach(t)
d9bb3666	352	Thread t
d9bb3666	353	CODE:
f0f333f4	354	#ifdef USE_THREADS
7d901afa	355	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: detaching %p (state %u)\n",
7d901afa	356	thr, t, ThrSTATE(t)););
50112d62	357	MUTEX_LOCK(&t->mutex);
50112d62	358	switch (ThrSTATE(t)) {
14fcddff	359	case THRf_R_JOINABLE:
50112d62	360	ThrSETSTATE(t, THRf_R_DETACHED);
14fcddff	361	/* fall through */
50112d62	362	case THRf_R_DETACHED:
14fcddff	363	DETACH(t);
50112d62	364	MUTEX_UNLOCK(&t->mutex);
14fcddff	365	break;
14fcddff	366	case THRf_ZOMBIE:
50112d62	367	ThrSETSTATE(t, THRf_DEAD);
7d901afa	368	DETACH(t);
50112d62	369	MUTEX_UNLOCK(&t->mutex);
50112d62	370	remove_thread(t);
14fcddff	371	break;
14fcddff	372	default:
50112d62	373	MUTEX_UNLOCK(&t->mutex);
14fcddff	374	croak("can't detach thread");
14fcddff	375	/* NOTREACHED */
734689b1	376	}
f0f333f4	377	#endif
d9bb3666	378
d9bb3666	379	void
7d901afa	380	equal(t1, t2)
	381	Thread t1
	382	Thread t2
	383	PPCODE:
	384	PUSHs((t1 == t2) ? &sv_yes : &sv_no);
	385
	386	void
	387	flags(t)
	388	Thread t
	389	PPCODE:
f0f333f4	390	#ifdef USE_THREADS
7d901afa	391	PUSHs(sv_2mortal(newSViv(t->flags)));
f0f333f4	392	#endif
7d901afa	393
7d901afa	394	void
458fb581	395	self(classname)
458fb581	396	char * classname
7d901afa	397	PREINIT:
7d901afa	398	SV *sv;
f0f333f4	399	PPCODE:
f0f333f4	400	#ifdef USE_THREADS
7d901afa	401	sv = newSViv(thr->tid);
199100c8	402	sv_magic(sv, thr->oursv, '~', 0, 0);
7d901afa	403	SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
458fb581	404	PUSHs(sv_2mortal(sv_bless(newRV_noinc(sv),
458fb581	405	gv_stashpv(classname, TRUE))));
f0f333f4	406	#endif
7d901afa	407
50112d62	408	U32
	409	tid(t)
	410	Thread t
	411	CODE:
f0f333f4	412	#ifdef USE_THREADS
50112d62	413	MUTEX_LOCK(&t->mutex);
	414	RETVAL = t->tid;
	415	MUTEX_UNLOCK(&t->mutex);
f0f333f4	416	#else
	417	RETVAL = 0;
	418	#endif
50112d62	419	OUTPUT:
	420	RETVAL
	421
	422	void
	423	DESTROY(t)
	424	SV * t
	425	PPCODE:
	426	PUSHs(&sv_yes);
	427
7d901afa	428	void
734689b1	429	yield()
d9bb3666	430	CODE:
f0f333f4	431	{
f0f333f4	432	#ifdef USE_THREADS
ea0efc06	433	YIELD;
f0f333f4	434	#endif
f0f333f4	435	}
d9bb3666	436
d9bb3666	437	void
734689b1	438	cond_wait(sv)
	439	SV * sv
	440	MAGIC * mg = NO_INIT
f0f333f4	441	CODE:
f0f333f4	442	#ifdef USE_THREADS
2c127b02	443	if (SvROK(sv))
734689b1	444	sv = SvRV(sv);
2c127b02	445
734689b1	446	mg = condpair_magic(sv);
683929b4	447	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_wait %p\n", thr, sv));
734689b1	448	MUTEX_LOCK(MgMUTEXP(mg));
	449	if (MgOWNER(mg) != thr) {
	450	MUTEX_UNLOCK(MgMUTEXP(mg));
	451	croak("cond_wait for lock that we don't own\n");
	452	}
	453	MgOWNER(mg) = 0;
	454	COND_WAIT(MgCONDP(mg), MgMUTEXP(mg));
50112d62	455	while (MgOWNER(mg))
50112d62	456	COND_WAIT(MgOWNERCONDP(mg), MgMUTEXP(mg));
734689b1	457	MgOWNER(mg) = thr;
734689b1	458	MUTEX_UNLOCK(MgMUTEXP(mg));
f0f333f4	459	#endif
f0f333f4	460
734689b1	461	void
	462	cond_signal(sv)
	463	SV * sv
	464	MAGIC * mg = NO_INIT
	465	CODE:
f0f333f4	466	#ifdef USE_THREADS
50112d62	467	if (SvROK(sv))
734689b1	468	sv = SvRV(sv);
50112d62	469
734689b1	470	mg = condpair_magic(sv);
683929b4	471	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_signal %p\n",thr,sv));
734689b1	472	MUTEX_LOCK(MgMUTEXP(mg));
	473	if (MgOWNER(mg) != thr) {
	474	MUTEX_UNLOCK(MgMUTEXP(mg));
	475	croak("cond_signal for lock that we don't own\n");
	476	}
	477	COND_SIGNAL(MgCONDP(mg));
	478	MUTEX_UNLOCK(MgMUTEXP(mg));
f0f333f4	479	#endif
d9bb3666	480
734689b1	481	void
	482	cond_broadcast(sv)
	483	SV * sv
	484	MAGIC * mg = NO_INIT
f0f333f4	485	CODE:
f0f333f4	486	#ifdef USE_THREADS
783070da	487	if (SvROK(sv))
734689b1	488	sv = SvRV(sv);
783070da	489
734689b1	490	mg = condpair_magic(sv);
683929b4	491	DEBUG_L(PerlIO_printf(PerlIO_stderr(), "%p: cond_broadcast %p\n",
683929b4	492	thr, sv));
734689b1	493	MUTEX_LOCK(MgMUTEXP(mg));
	494	if (MgOWNER(mg) != thr) {
	495	MUTEX_UNLOCK(MgMUTEXP(mg));
	496	croak("cond_broadcast for lock that we don't own\n");
	497	}
	498	COND_BROADCAST(MgCONDP(mg));
	499	MUTEX_UNLOCK(MgMUTEXP(mg));
f0f333f4	500	#endif
f152979c	501
7d901afa	502	void
458fb581	503	list(classname)
458fb581	504	char * classname
7d901afa	505	PREINIT:
	506	Thread t;
	507	AV * av;
	508	SV ** svp;
	509	int n = 0;
	510	PPCODE:
f0f333f4	511	#ifdef USE_THREADS
7d901afa	512	av = newAV();
	513	/*
	514	* Iterate until we have enough dynamic storage for all threads.
	515	* We mustn't do any allocation while holding threads_mutex though.
	516	*/
	517	MUTEX_LOCK(&threads_mutex);
	518	do {
	519	n = nthreads;
	520	MUTEX_UNLOCK(&threads_mutex);
	521	if (AvFILL(av) < n - 1) {
	522	int i = AvFILL(av);
	523	for (i = AvFILL(av); i < n - 1; i++) {
	524	SV sv = newSViv(0); / fill in tid later */
	525	sv_magic(sv, 0, '~', 0, 0); /* fill in other magic later */
	526	av_push(av, sv_bless(newRV_noinc(sv),
458fb581	527	gv_stashpv(classname, TRUE)));
50112d62	528
7d901afa	529	}
	530	}
	531	MUTEX_LOCK(&threads_mutex);
	532	} while (n < nthreads);
50112d62	533	n = nthreads; /* Get the final correct value */
7d901afa	534
	535	/*
	536	* At this point, there's enough room to fill in av.
	537	* Note that we are holding threads_mutex so the list
	538	* won't change out from under us but all the remaining
	539	* processing is "fast" (no blocking, malloc etc.)
	540	*/
	541	t = thr;
	542	svp = AvARRAY(av);
	543	do {
0a00ffdb	544	SV sv = (SV)SvRV(*svp);
7d901afa	545	sv_setiv(sv, t->tid);
199100c8	546	SvMAGIC(sv)->mg_obj = SvREFCNT_inc(t->oursv);
7d901afa	547	SvMAGIC(sv)->mg_flags \|= MGf_REFCOUNTED;
	548	SvMAGIC(sv)->mg_private = Thread_MAGIC_SIGNATURE;
	549	t = t->next;
0a00ffdb	550	svp++;
7d901afa	551	} while (t != thr);
50112d62	552	/* */
7d901afa	553	MUTEX_UNLOCK(&threads_mutex);
7d901afa	554	/* Truncate any unneeded slots in av */
50112d62	555	av_fill(av, n - 1);
7d901afa	556	/* Finally, push all the new objects onto the stack and drop av */
	557	EXTEND(sp, n);
	558	for (svp = AvARRAY(av); n > 0; n--, svp++)
	559	PUSHs(*svp);
	560	(void)sv_2mortal((SV*)av);
f0f333f4	561	#endif
7d901afa	562
7d901afa	563
f152979c	564	MODULE = Thread PACKAGE = Thread::Signal
	565
	566	void
	567	kill_sighandler_thread()
	568	PPCODE:
	569	write(sig_pipe[0], "\0", 1);
	570	PUSHs(&sv_yes);
	571
	572	void
	573	init_thread_signals()
	574	PPCODE:
	575	sighandlerp = handle_thread_signal;
	576	if (pipe(sig_pipe) == -1)
	577	XSRETURN_UNDEF;
	578	PUSHs(&sv_yes);
	579
	580	SV *
	581	await_signal()
	582	PREINIT:
	583	char c;
ea0efc06	584	SSize_t ret;
f152979c	585	CODE:
	586	do {
	587	ret = read(sig_pipe[1], &c, 1);
	588	} while (ret == -1 && errno == EINTR);
	589	if (ret == -1)
	590	croak("panic: await_signal");
	591	if (ret == 0)
	592	XSRETURN_UNDEF;
	593	RETVAL = c ? psig_ptr[c] : &sv_no;
	594	OUTPUT:
	595	RETVAL
4e35701f	596
458fb581	597	MODULE = Thread PACKAGE = Thread::Specific
	598
	599	void
	600	data(classname = "Thread::Specific")
	601	char * classname
	602	PPCODE:
fb223100	603	#ifdef USE_THREADS
458fb581	604	if (AvFILL(thr->specific) == -1) {
	605	GV *gv = gv_fetchpv("Thread::Specific::FIELDS", TRUE, SVt_PVHV);
	606	av_store(thr->specific, 0, newRV((SV*)GvHV(gv)));
	607	}
	608	XPUSHs(sv_bless(newRV((SV*)thr->specific),gv_stashpv(classname,TRUE)));
fb223100	609	#endif