thread.xs

UNIX下perl实现代码

static Signal_t
handle_thread_signal(int sig)
{
    dTHXo;
    unsigned char c = (unsigned char) sig;
    /*
     * We're not really allowed to call fprintf in a signal handler
     * so don't be surprised if this isn't robust while debugging
     * with -DL.
     */
    DEBUG_S(PerlIO_printf(Perl_debug_log,
	    "handle_thread_signal: got signal %d\n", sig););
    write(sig_pipe[1], &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(aTHX_ startsv, av, classname)));

void
join(t)
	Thread	t
	AV *	av = NO_INIT
	int	i = NO_INIT
    PPCODE:
#ifdef USE_THREADS
	if (t == thr)
	    croak("Attempt to join self");
	DEBUG_S(PerlIO_printf(Perl_debug_log, "%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(aTHX_ t);
	    break;
	default:
	    MUTEX_UNLOCK(&t->mutex);
	    croak("can't join with thread");
	    /* NOTREACHED */
	}
	JOIN(t, &av);

	sv_2mortal((SV*)av);

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

void
detach(t)
	Thread	t
    CODE:
#ifdef USE_THREADS
	DEBUG_S(PerlIO_printf(Perl_debug_log, "%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(aTHX_ 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) ? &PL_sv_yes : &PL_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(&PL_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_S(PerlIO_printf(Perl_debug_log, "%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_SIGNAL(MgOWNERCONDP(mg));
	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_S(PerlIO_printf(Perl_debug_log, "%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_S(PerlIO_printf(Perl_debug_log, "%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(&PL_threads_mutex);
	do {
	    n = PL_nthreads;
	    MUTEX_UNLOCK(&PL_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(&PL_threads_mutex);
	} while (n < PL_nthreads);
	n = PL_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(&PL_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[1], "\0", 1);
	PUSHs(&PL_sv_yes);

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

void
await_signal()
    PREINIT:
	unsigned char c;
	SSize_t ret;
    CODE:
	do {
	    ret = read(sig_pipe[0], &c, 1);
	} while (ret == -1 && errno == EINTR);
	if (ret == -1)
	    croak("panic: await_signal");
	ST(0) = sv_newmortal();
	if (ret)
	    sv_setsv(ST(0), c ? PL_psig_ptr[c] : &PL_sv_no);
	DEBUG_S(PerlIO_printf(Perl_debug_log,
			      "await_signal returning %s\n", SvPEEK(ST(0))););

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