o5r/plan9port
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Rewrite to remove dependence on rendezvous and its bizarre

Browse source 
data structures.  Makes it easier to use pthreads too.
Still need to add code for non-pthreads systems.

Just a checkpoint to switch work to another machine.
This commit is contained in:
rsc 2004-09-17 00:38:29 +00:00
parent 984e353160
commit 06bb4ed20d
15 changed files with 205 additions and 218 deletions
Download patch file Download diff file Expand all files Collapse all files

2
bin/9l
View file

@ -10,7 +10,7 @@ case "$tag" in
extralibs="$extralibs -lutil" extralibs="$extralibs -lutil"
;; ;;
*Linux*) ld=gcc *Linux*) ld=gcc
extralibs="$extralibs -lutil" extralibs="$extralibs -lutil -lpthread"
;; ;;
*Darwin*) ld=gcc ;; *Darwin*) ld=gcc ;;
*SunOS*) ld="${CC9:-cc} -g" *SunOS*) ld="${CC9:-cc} -g"

43
include/libc.h
View file

@ -425,22 +425,49 @@ extern void needstack(int);
/* /*
* synchronization * synchronization
*/ */
typedef typedef struct Lock Lock;
struct Lock { struct Lock
int val; {
} Lock; #ifdef PLAN9_PTHREADS
int init;
extern int _tas(int*); pthread_mutex_t mutex;
#else
int val;
#endif
};
extern void lock(Lock*); extern void lock(Lock*);
extern void unlock(Lock*); extern void unlock(Lock*);
extern int canlock(Lock*); extern int canlock(Lock*);
/*
* Used to implement process sleep and wakeup,
* either in terms of pthreads or our own primitives.
* This will be more portable than writing our own
* per-system implementations, and on some systems
* non-pthreads threading implementations break libc
* (cough, Linux, cough).
*/
typedef struct _Procrend _Procrend;
struct _Procrend
{
int asleep;
Lock *l;
void *arg;
#ifdef PLAN9_PTHREADS
pthread_cond_t cond;
#endif
};
extern void _procsleep(_Procrend*);
extern void _procwakeup(_Procrend*);
typedef struct QLp QLp; typedef struct QLp QLp;
struct QLp struct QLp
{ {
int inuse; Lock inuse;
QLp *next; QLp *next;
_Procrend rend;
char state; char state;
}; };
@ -456,7 +483,7 @@ struct QLock
extern void qlock(QLock*); extern void qlock(QLock*);
extern void qunlock(QLock*); extern void qunlock(QLock*);
extern int canqlock(QLock*); extern int canqlock(QLock*);
extern void _qlockinit(ulong (*)(ulong, ulong)); /* called only by the thread library */ extern void _qlockinit(void(*)(_Procrend*), void(*)(_Procrend*)); /* called only by the thread library */
typedef typedef
struct RWLock struct RWLock

2
include/thread.h
View file

@ -55,7 +55,7 @@ struct Alt {
/* the next variables are used internally to alt /* the next variables are used internally to alt
* they need not be initialized * they need not be initialized
*/ */
Channel **tag; /* pointer to rendez-vous tag */ struct Thread *thread; /* thread waiting on this alt */
int entryno; /* entry number */ int entryno; /* entry number */
}; };

7
include/u.h
View file

@ -41,6 +41,8 @@ typedef long p9jmp_buf[sizeof(sigjmp_buf)/sizeof(long)];
# undef _NEEDUSHORT # undef _NEEDUSHORT
# undef _NEEDUINT # undef _NEEDUINT
# undef _NEEDULONG # undef _NEEDULONG
# include <pthread.h>
# define PLAN9_PTHREADS
# endif # endif
#endif #endif
#if defined(__sun__) #if defined(__sun__)
@ -48,6 +50,8 @@ typedef long p9jmp_buf[sizeof(sigjmp_buf)/sizeof(long)];
# undef _NEEDUSHORT # undef _NEEDUSHORT
# undef _NEEDUINT # undef _NEEDUINT
# undef _NEEDULONG # undef _NEEDULONG
# include <pthread.h>
# define PLAN9_PTHREADS
#endif #endif
#if defined(__FreeBSD__) #if defined(__FreeBSD__)
# include <sys/types.h> # include <sys/types.h>
@ -61,8 +65,11 @@ typedef long p9jmp_buf[sizeof(sigjmp_buf)/sizeof(long)];
# undef _NEEDUSHORT # undef _NEEDUSHORT
# undef _NEEDUINT # undef _NEEDUINT
# define _NEEDLL 1 # define _NEEDLL 1
# include <pthread.h>
# define PLAN9_PTHREADS
#endif #endif
typedef signed char schar; typedef signed char schar;
typedef unsigned int u32int; typedef unsigned int u32int;
typedef int s32int; typedef int s32int;

4
src/lib9/ffork-Linux.c
View file

@ -1,3 +1,6 @@
#include "ffork-pthread.c"
#ifdef OLD
/* /*
* Is nothing simple? * Is nothing simple?
* *
@ -191,3 +194,4 @@ getfforkid(void)
return getpid(); return getpid();
} }
#endif

78
src/lib9/lock.c
View file

@ -2,52 +2,54 @@
#include <unistd.h> #include <unistd.h>
#include <sys/time.h> #include <sys/time.h>
#include <sched.h> #include <sched.h>
#include <errno.h>
#include <libc.h> #include <libc.h>
int _ntas; static pthread_mutex_t initmutex = PTHREAD_MUTEX_INITIALIZER;
static int
_xtas(void *v)
{
int x;
_ntas++; static void
x = _tas(v); lockinit(Lock *lk)
return x;
}
int
canlock(Lock *l)
{ {
return !_xtas(&l->val); pthread_mutexattr_t attr;
}
void pthread_mutex_lock(&initmutex);
unlock(Lock *l) if(lk->init == 0){
{ pthread_mutexattr_init(&attr);
l->val = 0; pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL);
pthread_mutex_init(&lk->mutex, &attr);
pthread_mutexattr_destroy(&attr);
lk->init = 1;
}
pthread_mutex_unlock(&initmutex);
} }
void void
lock(Lock *lk) lock(Lock *lk)
{ {
int i; if(!lk->init)
lockinit(lk);
/* once fast */ if(pthread_mutex_lock(&lk->mutex) != 0)
if(!_xtas(&lk->val)) abort();
return; }
/* a thousand times pretty fast */
for(i=0; i<1000; i++){ int
if(!_xtas(&lk->val)) canlock(Lock *lk)
return; {
sched_yield(); int r;
}
/* now nice and slow */ if(!lk->init)
for(i=0; i<1000; i++){ lockinit(lk);
if(!_xtas(&lk->val)) r = pthread_mutex_trylock(&lk->mutex);
return; if(r == 0)
usleep(100*1000); return 1;
} if(r == EBUSY)
/* take your time */ return 0;
while(_xtas(&lk->val)) abort();
usleep(1000*1000); }
void
unlock(Lock *lk)
{
if(pthread_mutex_unlock(&lk->mutex) != 0)
abort();
} }

87
src/lib9/qlock.c
View file

@ -16,13 +16,15 @@ enum
Waking, Waking,
}; };
static ulong (*_rendezvousp)(ulong, ulong) = rendezvous; static void (*procsleep)(_Procrend*) = _procsleep;
static void (*procwakeup)(_Procrend*) = _procwakeup;
/* this gets called by the thread library ONLY to get us to use its rendezvous */ /* this gets called by the thread library ONLY to get us to use its rendezvous */
void void
_qlockinit(ulong (*r)(ulong, ulong)) _qlockinit(void (*sleep)(_Procrend*), void (*wakeup)(_Procrend*))
{ {
_rendezvousp = r; procsleep = sleep;
procwakeup = wakeup;
} }
/* find a free shared memory location to queue ourselves in */ /* find a free shared memory location to queue ourselves in */
@ -39,7 +41,7 @@ getqlp(void)
fprint(2, "qlock: out of qlp\n"); fprint(2, "qlock: out of qlp\n");
abort(); abort();
} }
if(_tas(&(p->inuse)) == 0){ if(canlock(&p->inuse)){
ql.p = p; ql.p = p;
p->next = nil; p->next = nil;
break; break;
@ -70,13 +72,11 @@ qlock(QLock *q)
p->next = mp; p->next = mp;
q->tail = mp; q->tail = mp;
mp->state = Queuing; mp->state = Queuing;
mp->rend.l = &q->lock;
_procsleep(&mp->rend);
unlock(&q->lock); unlock(&q->lock);
/* wait */
while((*_rendezvousp)((ulong)mp, 1) == ~0)
;
assert(mp->state == Waking); assert(mp->state == Waking);
mp->inuse = 0; unlock(&mp->inuse);
} }
void void
@ -91,10 +91,9 @@ qunlock(QLock *q)
q->head = p->next; q->head = p->next;
if(q->head == nil) if(q->head == nil)
q->tail = nil; q->tail = nil;
unlock(&q->lock);
p->state = Waking; p->state = Waking;
while((*_rendezvousp)((ulong)p, 0x12345) == ~0) _procwakeup(&p->rend);
; unlock(&q->lock);
return; return;
} }
q->locked = 0; q->locked = 0;
@ -137,13 +136,11 @@ rlock(RWLock *q)
q->tail = mp; q->tail = mp;
mp->next = nil; mp->next = nil;
mp->state = QueuingR; mp->state = QueuingR;
mp->rend.l = &q->lock;
_procsleep(&mp->rend);
unlock(&q->lock); unlock(&q->lock);
/* wait in kernel */
while((*_rendezvousp)((ulong)mp, 1) == ~0)
;
assert(mp->state == Waking); assert(mp->state == Waking);
mp->inuse = 0; unlock(&mp->inuse);
} }
int int
@ -181,12 +178,11 @@ runlock(RWLock *q)
if(q->head == 0) if(q->head == 0)
q->tail = 0; q->tail = 0;
q->writer = 1; q->writer = 1;
unlock(&q->lock);
/* wakeup waiter */ /* wakeup waiter */
p->state = Waking; p->state = Waking;
while((*_rendezvousp)((ulong)p, 0) == ~0) _procwakeup(&p->rend);
; unlock(&q->lock);
} }
void void
@ -212,13 +208,13 @@ wlock(RWLock *q)
q->tail = mp; q->tail = mp;
mp->next = nil; mp->next = nil;
mp->state = QueuingW; mp->state = QueuingW;
unlock(&q->lock);
/* wait in kernel */ /* wait in kernel */
while((*_rendezvousp)((ulong)mp, 1) == ~0) mp->rend.l = &q->lock;
; _procsleep(&mp->rend);
unlock(&q->lock);
assert(mp->state == Waking); assert(mp->state == Waking);
mp->inuse = 0; unlock(&mp->inuse);
} }
int int
@ -256,10 +252,9 @@ wunlock(RWLock *q)
q->head = p->next; q->head = p->next;
if(q->head == nil) if(q->head == nil)
q->tail = nil; q->tail = nil;
unlock(&q->lock);
p->state = Waking; p->state = Waking;
while((*_rendezvousp)((ulong)p, 0) == ~0) _procwakeup(&p->rend);
; unlock(&q->lock);
return; return;
} }
@ -274,8 +269,7 @@ wunlock(RWLock *q)
q->head = p->next; q->head = p->next;
q->readers++; q->readers++;
p->state = Waking; p->state = Waking;
while((*_rendezvousp)((ulong)p, 0) == ~0) _procwakeup(&p->rend);
;
} }
if(q->head == nil) if(q->head == nil)
q->tail = nil; q->tail = nil;
@ -315,20 +309,17 @@ rsleep(Rendez *r)
r->l->head = t->next; r->l->head = t->next;
if(r->l->head == nil) if(r->l->head == nil)
r->l->tail = nil; r->l->tail = nil;
unlock(&r->l->lock);
t->state = Waking; t->state = Waking;
while((*_rendezvousp)((ulong)t, 0x12345) == ~0) _procwakeup(&t->rend);
; }else
}else{
r->l->locked = 0; r->l->locked = 0;
unlock(&r->l->lock);
}
/* wait for a wakeup */ /* wait for a wakeup */
while((*_rendezvousp)((ulong)me, 0x23456) == ~0) me->rend.l = &r->l->lock;
; _procsleep(&me->rend);
assert(me->state == Waking); assert(me->state == Waking);
me->inuse = 0; unlock(&me->inuse);
if(!r->l->locked){ if(!r->l->locked){
fprint(2, "rsleep: not locked after wakeup\n"); fprint(2, "rsleep: not locked after wakeup\n");
abort(); abort();
@ -384,3 +375,23 @@ rwakeupall(Rendez *r)
; ;
return i; return i;
} }
void
_procsleep(_Procrend *rend)
{
//print("sleep %p %d\n", rend, getpid());
pthread_cond_init(&rend->cond, 0);
rend->asleep = 1;
while(rend->asleep)
pthread_cond_wait(&rend->cond, &rend->l->mutex);
pthread_cond_destroy(&rend->cond);
}
void
_procwakeup(_Procrend *rend)
{
//print("wakeup %p\n", rend);
rend->asleep = 0;
pthread_cond_signal(&rend->cond);
}

43
src/libthread/channel.c
View file

@ -2,7 +2,7 @@
static Lock chanlock; /* central channel access lock */ static Lock chanlock; /* central channel access lock */
static void enqueue(Alt*, Channel**); static void enqueue(Alt*, Thread*);
static void dequeue(Alt*); static void dequeue(Alt*);
static int altexec(Alt*, int); static int altexec(Alt*, int);
@ -29,7 +29,7 @@ canexec(Alt *a)
/* are there senders or receivers blocked? */ /* are there senders or receivers blocked? */
otherop = (CHANSND+CHANRCV) - a->op; otherop = (CHANSND+CHANRCV) - a->op;
for(i=0; i<c->nentry; i++) for(i=0; i<c->nentry; i++)
if(c->qentry[i] && c->qentry[i]->op==otherop && *c->qentry[i]->tag==nil){ if(c->qentry[i] && c->qentry[i]->op==otherop && c->qentry[i]->thread==nil){
_threaddebug(DBGCHAN, "can rendez alt %p chan %p", a, c); _threaddebug(DBGCHAN, "can rendez alt %p chan %p", a, c);
return 1; return 1;
} }
@ -100,9 +100,8 @@ static int
_alt(Alt *alts) _alt(Alt *alts)
{ {
Alt *a, *xa; Alt *a, *xa;
Channel *volatile c; Channel *c;
int n, s; int n, s;
ulong r;
Thread *t; Thread *t;
/* /*
@ -131,7 +130,7 @@ _alt(Alt *alts)
xa->entryno = -1; xa->entryno = -1;
if(xa->op == CHANNOP) if(xa->op == CHANNOP)
continue; continue;
c = xa->c; c = xa->c;
if(c==nil){ if(c==nil){
unlock(&chanlock); unlock(&chanlock);
@ -153,11 +152,11 @@ _threadnalt++;
} }
/* enqueue on all channels. */ /* enqueue on all channels. */
c = nil; t->altc = nil;
for(xa=alts; xa->op!=CHANEND; xa++){ for(xa=alts; xa->op!=CHANEND; xa++){
if(xa->op==CHANNOP) if(xa->op==CHANNOP)
continue; continue;
enqueue(xa, (Channel**)&c); enqueue(xa, t);
} }
/* /*
@ -166,25 +165,20 @@ _threadnalt++;
* is interrupted -- someone else might come * is interrupted -- someone else might come
* along and try to rendezvous with us, so * along and try to rendezvous with us, so
* we need to be here. * we need to be here.
*
* actually, now we're assuming no interrupts.
*/ */
Again: /*Again:*/
t->alt = alts; t->alt = alts;
t->chan = Chanalt; t->chan = Chanalt;
t->altrend.l = &chanlock;
unlock(&chanlock);
_procsplx(s); _procsplx(s);
r = _threadrendezvous((ulong)&c, 0); _threadsleep(&t->altrend);
s = _procsplhi(); s = _procsplhi();
lock(&chanlock);
if(r==~0){ /* interrupted */
if(c!=nil) /* someone will meet us; go back */
goto Again;
c = (Channel*)~0; /* so no one tries to meet us */
}
/* dequeue from channels, find selected one */ /* dequeue from channels, find selected one */
a = nil; a = nil;
c = t->altc;
for(xa=alts; xa->op!=CHANEND; xa++){ for(xa=alts; xa->op!=CHANEND; xa++){
if(xa->op==CHANNOP) if(xa->op==CHANNOP)
continue; continue;
@ -385,12 +379,12 @@ if(c->nentry > _threadhighnentry) _threadhighnentry = c->nentry;
} }
static void static void
enqueue(Alt *a, Channel **c) enqueue(Alt *a, Thread *t)
{ {
int i; int i;
_threaddebug(DBGCHAN, "Queueing alt %p on channel %p", a, a->c); _threaddebug(DBGCHAN, "Queueing alt %p on channel %p", a, a->c);
a->tag = c; a->thread = t;
i = emptyentry(a->c); i = emptyentry(a->c);
a->c->qentry[i] = a; a->c->qentry[i] = a;
} }
@ -466,7 +460,7 @@ altexec(Alt *a, int spl)
b = nil; b = nil;
me = a->v; me = a->v;
for(i=0; i<c->nentry; i++) for(i=0; i<c->nentry; i++)
if(c->qentry[i] && c->qentry[i]->op==otherop && *c->qentry[i]->tag==nil) if(c->qentry[i] && c->qentry[i]->op==otherop && c->qentry[i]->thread==nil)
if(nrand(++n) == 0) if(nrand(++n) == 0)
b = c->qentry[i]; b = c->qentry[i];
if(b != nil){ if(b != nil){
@ -493,13 +487,12 @@ altexec(Alt *a, int spl)
else else
altcopy(waiter, me, c->e); altcopy(waiter, me, c->e);
} }
*b->tag = c; /* commits us to rendezvous */ b->thread->altc = c;
_procwakeup(&b->thread->altrend);
_threaddebug(DBGCHAN, "chanlock is %lud", *(ulong*)(void*)&chanlock);
_threaddebug(DBGCHAN, "unlocking the chanlock"); _threaddebug(DBGCHAN, "unlocking the chanlock");
unlock(&chanlock); unlock(&chanlock);
_procsplx(spl); _procsplx(spl);
_threaddebug(DBGCHAN, "chanlock is %lud", *(ulong*)(void*)&chanlock);
while(_threadrendezvous((ulong)b->tag, 0) == ~0)
;
return 1; return 1;
} }

2
src/libthread/create.c
View file

@ -2,6 +2,7 @@
Pqueue _threadpq; Pqueue _threadpq;
int _threadprocs; int _threadprocs;
int __pthread_nonstandard_stacks;
static int nextID(void); static int nextID(void);
@ -21,6 +22,7 @@ newthread(Proc *p, void (*f)(void *arg), void *arg, uint stacksize, char *name,
Thread *t; Thread *t;
char *s; char *s;
__pthread_nonstandard_stacks = 1;
if(stacksize < 32) if(stacksize < 32)
sysfatal("bad stacksize %d", stacksize); sysfatal("bad stacksize %d", stacksize);
t = _threadmalloc(sizeof(Thread), 1); t = _threadmalloc(sizeof(Thread), 1);

7
src/libthread/main.c
View file

@ -40,7 +40,7 @@ main(int argc, char **argv)
//_threaddebuglevel = (DBGSCHED|DBGCHAN|DBGREND)^~0; //_threaddebuglevel = (DBGSCHED|DBGCHAN|DBGREND)^~0;
_systhreadinit(); _systhreadinit();
_qlockinit(_threadrendezvous); _qlockinit(_threadsleep, _threadwakeup);
_sysfatal = _threadsysfatal; _sysfatal = _threadsysfatal;
notify(_threadnote); notify(_threadnote);
if(mainstacksize == 0) if(mainstacksize == 0)
@ -49,8 +49,9 @@ main(int argc, char **argv)
a = _threadmalloc(sizeof *a, 1); a = _threadmalloc(sizeof *a, 1);
a->argc = argc; a->argc = argc;
a->argv = argv; a->argv = argv;
malloc(10);
p = _newproc(mainlauncher, a, mainstacksize, "threadmain", 0, 0); p = _newproc(mainlauncher, a, mainstacksize, "threadmain", 0, 0);
malloc(10);
_schedinit(p); _schedinit(p);
abort(); /* not reached */ abort(); /* not reached */
return 0; return 0;
@ -61,7 +62,9 @@ mainlauncher(void *arg)
{ {
Mainarg *a; Mainarg *a;
malloc(10);
a = arg; a = arg;
malloc(10);
threadmain(a->argc, a->argv); threadmain(a->argc, a->argv);
threadexits("threadmain"); threadexits("threadmain");
} }

7
src/libthread/mkfile
View file

@ -23,7 +23,6 @@ OFILES=\
ioreadn.$O\ ioreadn.$O\
iosleep.$O\ iosleep.$O\
iowrite.$O\ iowrite.$O\
kill.$O\
lib.$O\ lib.$O\
main.$O\ main.$O\
memset.$O\ memset.$O\
@ -43,13 +42,13 @@ HFILES=\
<$PLAN9/src/mksyslib <$PLAN9/src/mksyslib
tprimes: tprimes.$O $PLAN9/lib/$LIB tprimes: tprimes.$O $PLAN9/lib/$LIB
$LD -o tprimes tprimes.$O $LDFLAGS -lthread -l9 -lfmt -lutf $LD -o tprimes tprimes.$O $LDFLAGS -lthread -l9
texec: texec.$O $PLAN9/lib/$LIB texec: texec.$O $PLAN9/lib/$LIB
$LD -o texec texec.$O $LDFLAGS -lthread -l9 -lfmt -lutf $LD -o texec texec.$O $LDFLAGS -lthread -l9
trend: trend.$O $PLAN9/lib/$LIB trend: trend.$O $PLAN9/lib/$LIB
$LD -o trend trend.$O $LDFLAGS -lthread -l9 -lfmt -lutf $LD -o trend trend.$O $LDFLAGS -lthread -l9
CLEANFILES=$CLEANFILES tprimes texec CLEANFILES=$CLEANFILES tprimes texec

116
src/libthread/rendez.c
View file

@ -1,104 +1,38 @@
#include "threadimpl.h" #include "threadimpl.h"
Rgrp _threadrgrp;
static int isdirty;
int _threadhighnrendez; int _threadhighnrendez;
int _threadnrendez; int _threadnrendez;
static int nrendez;
static ulong void
finish(Thread *t, ulong val) _threadsleep(_Procrend *r)
{ {
ulong ret; Thread *t;
ret = t->rendval; t = _threadgetproc()->thread;
t->rendval = val; r->arg = t;
t->nextstate = Rendezvous;
t->inrendez = 1;
unlock(r->l);
_sched();
t->inrendez = 0;
lock(r->l);
}
void
_threadwakeup(_Procrend *r)
{
Thread *t;
t = r->arg;
while(t->state == Running) while(t->state == Running)
sleep(0); sleep(0);
lock(&t->proc->lock); lock(&t->proc->lock);
if(t->state == Rendezvous){ /* not always true: might be Dead */ if(t->state == Dead){
t->state = Ready; unlock(&t->proc->lock);
_threadready(t); return;
} }
assert(t->state == Rendezvous && t->inrendez);
t->state = Ready;
_threadready(t);
unlock(&t->proc->lock); unlock(&t->proc->lock);
return ret;
}
ulong
_threadrendezvous(ulong tag, ulong val)
{
ulong ret;
Thread *t, **l;
lock(&_threadrgrp.lock);
_threadnrendez++;
l = &_threadrgrp.hash[tag%nelem(_threadrgrp.hash)];
for(t=*l; t; l=&t->rendhash, t=*l){
if(t->rendtag==tag){
_threaddebug(DBGREND, "Rendezvous with thread %d.%d", t->proc->pid, t->id);
*l = t->rendhash;
ret = finish(t, val);
--nrendez;
unlock(&_threadrgrp.lock);
return ret;
}
}
/* Going to sleep here. */
t = _threadgetproc()->thread;
t->rendbreak = 0;
t->inrendez = 1;
t->rendtag = tag;
t->rendval = val;
t->rendhash = *l;
*l = t;
++nrendez;
if(nrendez > _threadhighnrendez)
_threadhighnrendez = nrendez;
_threaddebug(DBGREND, "Rendezvous for tag %lud (m=%d)", t->rendtag, t->moribund);
unlock(&_threadrgrp.lock);
t->nextstate = Rendezvous;
_sched();
t->inrendez = 0;
_threaddebug(DBGREND, "Woke after rendezvous; val is %lud", t->rendval);
return t->rendval;
}
/*
* This is called while holding _threadpq.lock and p->lock,
* so we can't lock _threadrgrp.lock. Instead our caller has
* to call _threadbreakrendez after dropping those locks.
*/
void
_threadflagrendez(Thread *t)
{
t->rendbreak = 1;
isdirty = 1;
}
void
_threadbreakrendez(void)
{
int i;
Thread *t, **l;
if(isdirty == 0)
return;
lock(&_threadrgrp.lock);
if(isdirty == 0){
unlock(&_threadrgrp.lock);
return;
}
isdirty = 0;
for(i=0; i<nelem(_threadrgrp.hash); i++){
l = &_threadrgrp.hash[i];
for(t=*l; t; t=*l){
if(t->rendbreak){
*l = t->rendhash;
finish(t, ~0);
}else
l=&t->rendhash;
}
}
unlock(&_threadrgrp.lock);
} }

11
src/libthread/sched.c
View file

@ -36,6 +36,7 @@ _schedinit(void *arg)
unlock(&p->lock); unlock(&p->lock);
while(_setlabel(&p->sched)) while(_setlabel(&p->sched))
; ;
malloc(10);
_threaddebug(DBGSCHED, "top of schedinit, _threadexitsallstatus=%p", _threadexitsallstatus); _threaddebug(DBGSCHED, "top of schedinit, _threadexitsallstatus=%p", _threadexitsallstatus);
if(_threadexitsallstatus) if(_threadexitsallstatus)
_exits(_threadexitsallstatus); _exits(_threadexitsallstatus);
@ -57,8 +58,9 @@ _schedinit(void *arg)
p->threads.tail = t->prevt; p->threads.tail = t->prevt;
unlock(&p->lock); unlock(&p->lock);
if(t->inrendez){ if(t->inrendez){
_threadflagrendez(t); abort();
_threadbreakrendez(); // _threadflagrendez(t);
// _threadbreakrendez();
} }
_stackfree(t->stk); _stackfree(t->stk);
free(t->cmdname); free(t->cmdname);
@ -183,15 +185,18 @@ _sched(void)
Resched: Resched:
p = _threadgetproc(); p = _threadgetproc();
//fprint(2, "p %p\n", p); //fprint(2, "p %p\n", p);
malloc(10);
if((t = p->thread) != nil){ if((t = p->thread) != nil){
needstack(512); needstack(512);
// _threaddebug(DBGSCHED, "pausing, state=%s set %p goto %p", // _threaddebug(DBGSCHED, "pausing, state=%s set %p goto %p",
// psstate(t->state), &t->sched, &p->sched); // psstate(t->state), &t->sched, &p->sched);
print("swap\n");
if(_setlabel(&t->sched)==0) if(_setlabel(&t->sched)==0)
_gotolabel(&p->sched); _gotolabel(&p->sched);
_threadstacklimit(t->stk, t->stk+t->stksize); _threadstacklimit(t->stk, t->stk+t->stksize);
return p->nsched++; return p->nsched++;
}else{ }else{
malloc(10);
t = runthread(p); t = runthread(p);
if(t == nil){ if(t == nil){
_threaddebug(DBGSCHED, "all threads gone; exiting"); _threaddebug(DBGSCHED, "all threads gone; exiting");
@ -206,6 +211,8 @@ Resched:
} }
t->state = Running; t->state = Running;
t->nextstate = Ready; t->nextstate = Ready;
malloc(10);
print("gotolabel\n");
_gotolabel(&t->sched); _gotolabel(&t->sched);
for(;;); for(;;);
} }

13
src/libthread/threadimpl.h
View file

@ -88,11 +88,9 @@ struct Thread
char *cmdname; /* ptr to name of thread */ char *cmdname; /* ptr to name of thread */
int inrendez; int inrendez;
Thread *rendhash; /* Trgrp linked list */ Channel *altc;
ulong rendtag; /* rendezvous tag */ _Procrend altrend;
ulong rendval; /* rendezvous value */
int rendbreak; /* rendezvous has been taken */
Chanstate chan; /* which channel operation is current */ Chanstate chan; /* which channel operation is current */
Alt *alt; /* pointer to current alt structure (debugging) */ Alt *alt; /* pointer to current alt structure (debugging) */
@ -179,11 +177,9 @@ int _schedfork(Proc*);
void _schedinit(void*); void _schedinit(void*);
void _systhreadinit(void); void _systhreadinit(void);
void _threadassert(char*); void _threadassert(char*);
void _threadbreakrendez(void);
void __threaddebug(ulong, char*, ...); void __threaddebug(ulong, char*, ...);
#define _threaddebug if(!_threaddebuglevel){}else __threaddebug #define _threaddebug if(!_threaddebuglevel){}else __threaddebug
void _threadexitsall(char*); void _threadexitsall(char*);
void _threadflagrendez(Thread*);
Proc* _threadgetproc(void); Proc* _threadgetproc(void);
extern void _threadmultiproc(void); extern void _threadmultiproc(void);
Proc* _threaddelproc(void); Proc* _threaddelproc(void);
@ -193,7 +189,8 @@ void* _threadmalloc(long, int);
void _threadnote(void*, char*); void _threadnote(void*, char*);
void _threadready(Thread*); void _threadready(Thread*);
void _threadidle(void); void _threadidle(void);
ulong _threadrendezvous(ulong, ulong); void _threadsleep(_Procrend*);
void _threadwakeup(_Procrend*);
void _threadsignal(void); void _threadsignal(void);
void _threadsysfatal(char*, va_list); void _threadsysfatal(char*, va_list);
long _xdec(long*); long _xdec(long*);

1
src/libthread/tprimes.c
View file

@ -41,6 +41,7 @@ threadmain(int argc, char **argv)
int i; int i;
Channel *c; Channel *c;
malloc(10);
ARGBEGIN{ ARGBEGIN{
case 'D': case 'D':
_threaddebuglevel = atoi(ARGF()); _threaddebuglevel = atoi(ARGF());