#include "threadimpl.h" /* * One can go through a lot of effort to avoid this global lock. * You have to put locks in all the channels and all the Alt * structures. At the beginning of an alt you have to lock all * the channels, but then to try to actually exec an op you * have to lock the other guy's alt structure, so that other * people aren't trying to use him in some other op at the * same time. * * For Plan 9 apps, it's just not worth the extra effort. */static QLock chanlock; Channel*chancreate(int elemsize, int bufsize){ Channel *c; c = malloc(sizeof *c+bufsize*elemsize); if(c == nil) sysfatal("chancreate malloc: %r"); memset(c, 0, sizeof *c); c->elemsize = elemsize; c->bufsize = bufsize; c->nbuf = 0; c->buf = (uchar*)(c+1); return c;} voidchansetname(Channel *c, char *fmt, ...){ char *name; va_list arg; va_start(arg, fmt); name = vsmprint(fmt, arg); va_end(arg); free(c->name); c->name = name;} /* bug - work out races */voidchanfree(Channel *c){ if(c == nil) return; free(c->name); free(c->arecv.a); free(c->asend.a); free(c);} static voidaddarray(_Altarray *a, Alt *alt){ if(a->n == a->m){ a->m += 16; a->a = realloc(a->a, a->m*sizeof a->a[0]); } a->a[a->n++] = alt;} static voiddelarray(_Altarray *a, int i){ --a->n; a->a[i] = a->a[a->n];} /* * doesn't really work for things other than CHANSND and CHANRCV * but is only used as arg to chanarray, which can handle it */#define otherop(op) (CHANSND+CHANRCV-(op)) static _Altarray*chanarray(Channel *c, uint op){ switch(op){ default: return nil; case CHANSND: return &c->asend; case CHANRCV: return &c->arecv; }} static intaltcanexec(Alt *a){ _Altarray *ar; Channel *c; if(a->op == CHANNOP || (c=a->c) == nil) return 0; if(c->bufsize == 0){ ar = chanarray(c, otherop(a->op)); return ar && ar->n; }else{ switch(a->op){ default: return 0; case CHANSND: return c->nbuf < c->bufsize; case CHANRCV: return c->nbuf > 0; } }} static voidaltqueue(Alt *a){ _Altarray *ar; if(a->c == nil) return; ar = chanarray(a->c, a->op); addarray(ar, a);} static voidaltdequeue(Alt *a){ int i; _Altarray *ar; ar = chanarray(a->c, a->op); if(ar == nil){ fprint(2, "bad use of altdequeue op=%d\n", a->op); abort(); } for(i=0; i<ar->n; i++) if(ar->a[i] == a){ delarray(ar, i); return; } fprint(2, "cannot find self in altdq\n"); abort();} static voidaltalldequeue(Alt *a){ int i; for(i=0; a[i].op!=CHANEND && a[i].op!=CHANNOBLK; i++) if(a[i].op != CHANNOP) altdequeue(&a[i]);} static voidamove(void *dst, void *src, uint n){ if(dst){ if(src == nil) memset(dst, 0, n); else memmove(dst, src, n); }} /* * Actually move the data around. There are up to three * players: the sender, the receiver, and the channel itself. * If the channel is unbuffered or the buffer is empty, * data goes from sender to receiver. If the channel is full, * the receiver removes some from the channel and the sender * gets to put some in. */static voidaltcopy(Alt *s, Alt *r){ Alt *t; Channel *c; uchar *cp; /* * Work out who is sender and who is receiver */ if(s == nil && r == nil) return; assert(s != nil); c = s->c; if(s->op == CHANRCV){ t = s; s = r; r = t; } assert(s==nil || s->op == CHANSND); assert(r==nil || r->op == CHANRCV); /* * Channel is empty (or unbuffered) - copy directly. */ if(s && r && c->nbuf == 0){ amove(r->v, s->v, c->elemsize); return; } /* * Otherwise it's always okay to receive and then send. */ if(r){ cp = c->buf + c->off*c->elemsize; amove(r->v, cp, c->elemsize); --c->nbuf; if(++c->off == c->bufsize) c->off = 0; } if(s){ cp = c->buf + (c->off+c->nbuf)%c->bufsize*c->elemsize; amove(cp, s->v, c->elemsize); ++c->nbuf; }} static voidaltexec(Alt *a){ int i; _Altarray *ar; Alt *other; Channel *c; c = a->c; ar = chanarray(c, otherop(a->op)); if(ar && ar->n){ i = rand()%ar->n; other = ar->a[i]; altcopy(a, other); altalldequeue(other->thread->alt); other->thread->alt = other; _threadready(other->thread); }else altcopy(a, nil);} #define dbgalt 0intchanalt(Alt *a){ int i, j, ncan, n, canblock; Channel *c; _Thread *t; needstack(512); for(i=0; a[i].op != CHANEND && a[i].op != CHANNOBLK; i++) ; n = i; canblock = a[i].op == CHANEND; t = proc()->thread; for(i=0; i<n; i++) a[i].thread = t; t->alt = a; qlock(&chanlock);if(dbgalt) print("alt "); ncan = 0; for(i=0; i<n; i++){ c = a[i].c;if(dbgalt) print(" %c:", "esrnb"[a[i].op]);if(dbgalt) if(c->name) print("%s", c->name); else print("%p", c); if(altcanexec(&a[i])){if(dbgalt) print("*"); ncan++; } } if(ncan){ j = rand()%ncan; for(i=0; i<n; i++){ if(altcanexec(&a[i])){ if(j-- == 0){if(dbgalt){c = a[i].c;print(" => %c:", "esrnb"[a[i].op]);if(c->name) print("%s", c->name); else print("%p", c);print("\n");} altexec(&a[i]); qunlock(&chanlock); return i; } } } }if(dbgalt)print("\n"); if(!canblock){ qunlock(&chanlock); return -1; } for(i=0; i<n; i++){ if(a[i].op != CHANNOP) altqueue(&a[i]); } qunlock(&chanlock); _threadswitch(); /* * the guy who ran the op took care of dequeueing us * and then set t->alt to the one that was executed. */ if(t->alt < a || t->alt >= a+n) sysfatal("channel bad alt"); return t->alt - a;} static int_chanop(Channel *c, int op, void *p, int canblock){ Alt a[2]; a[0].c = c; a[0].op = op; a[0].v = p; a[1].op = canblock ? CHANEND : CHANNOBLK; if(chanalt(a) < 0) return -1; return 1;} intchansend(Channel *c, void *v){ return _chanop(c, CHANSND, v, 1);} intchannbsend(Channel *c, void *v){ return _chanop(c, CHANSND, v, 0);} intchanrecv(Channel *c, void *v){ return _chanop(c, CHANRCV, v, 1);} intchannbrecv(Channel *c, void *v){ return _chanop(c, CHANRCV, v, 0);} intchansendp(Channel *c, void *v){ return _chanop(c, CHANSND, (void*)&v, 1);} void*chanrecvp(Channel *c){ void *v; if(_chanop(c, CHANRCV, (void*)&v, 1) > 0) return v; return nil;} intchannbsendp(Channel *c, void *v){ return _chanop(c, CHANSND, (void*)&v, 0);} void*channbrecvp(Channel *c){ void *v; if(_chanop(c, CHANRCV, (void*)&v, 0) > 0) return v; return nil;} intchansendul(Channel *c, ulong val){ return _chanop(c, CHANSND, &val, 1);} ulongchanrecvul(Channel *c){ ulong val; if(_chanop(c, CHANRCV, &val, 1) > 0) return val; return -1;} intchannbsendul(Channel *c, ulong val){ return _chanop(c, CHANSND, &val, 0);} ulongchannbrecvul(Channel *c){ ulong val; if(_chanop(c, CHANRCV, &val, 0) > 0) return val; return -1;}
#include "threadimpl.h"
/*
* One can go through a lot of effort to avoid this global lock.
* You have to put locks in all the channels and all the Alt
* structures. At the beginning of an alt you have to lock all
* the channels, but then to try to actually exec an op you
* have to lock the other guy's alt structure, so that other
* people aren't trying to use him in some other op at the
* same time.
*
* For Plan 9 apps, it's just not worth the extra effort.
*/
static QLock chanlock;
Channel*
chancreate(int elemsize, int bufsize)
{
Channel *c;
c = malloc(sizeof *c+bufsize*elemsize);
if(c == nil)
sysfatal("chancreate malloc: %r");
memset(c, 0, sizeof *c);
c->elemsize = elemsize;
c->bufsize = bufsize;
c->nbuf = 0;
c->buf = (uchar*)(c+1);
return c;
}
void
chansetname(Channel *c, char *fmt, ...)
{
char *name;
va_list arg;
va_start(arg, fmt);
name = vsmprint(fmt, arg);
va_end(arg);
free(c->name);
c->name = name;
}
/* bug - work out races */
void
chanfree(Channel *c)
{
if(c == nil)
return;
free(c->name);
free(c->arecv.a);
free(c->asend.a);
free(c);
}
static void
addarray(_Altarray *a, Alt *alt)
{
if(a->n == a->m){
a->m += 16;
a->a = realloc(a->a, a->m*sizeof a->a[0]);
}
a->a[a->n++] = alt;
}
static void
delarray(_Altarray *a, int i)
{
--a->n;
a->a[i] = a->a[a->n];
}
/*
* doesn't really work for things other than CHANSND and CHANRCV
* but is only used as arg to chanarray, which can handle it
*/
#define otherop(op) (CHANSND+CHANRCV-(op))
static _Altarray*
chanarray(Channel *c, uint op)
{
switch(op){
default:
return nil;
case CHANSND:
return &c->asend;
case CHANRCV:
return &c->arecv;
}
}
static int
altcanexec(Alt *a)
{
_Altarray *ar;
Channel *c;
if(a->op == CHANNOP || (c=a->c) == nil)
return 0;
if(c->bufsize == 0){
ar = chanarray(c, otherop(a->op));
return ar && ar->n;
}else{
switch(a->op){
default:
return 0;
case CHANSND:
return c->nbuf < c->bufsize;
case CHANRCV:
return c->nbuf > 0;
}
}
}
static void
altqueue(Alt *a)
{
_Altarray *ar;
if(a->c == nil)
return;
ar = chanarray(a->c, a->op);
addarray(ar, a);
}
static void
altdequeue(Alt *a)
{
int i;
_Altarray *ar;
ar = chanarray(a->c, a->op);
if(ar == nil){
fprint(2, "bad use of altdequeue op=%d\n", a->op);
abort();
}
for(i=0; i<ar->n; i++)
if(ar->a[i] == a){
delarray(ar, i);
return;
}
fprint(2, "cannot find self in altdq\n");
abort();
}
static void
altalldequeue(Alt *a)
{
int i;
for(i=0; a[i].op!=CHANEND && a[i].op!=CHANNOBLK; i++)
if(a[i].op != CHANNOP)
altdequeue(&a[i]);
}
static void
amove(void *dst, void *src, uint n)
{
if(dst){
if(src == nil)
memset(dst, 0, n);
else
memmove(dst, src, n);
}
}
/*
* Actually move the data around. There are up to three
* players: the sender, the receiver, and the channel itself.
* If the channel is unbuffered or the buffer is empty,
* data goes from sender to receiver. If the channel is full,
* the receiver removes some from the channel and the sender
* gets to put some in.
*/
static void
altcopy(Alt *s, Alt *r)
{
Alt *t;
Channel *c;
uchar *cp;
/*
* Work out who is sender and who is receiver
*/
if(s == nil && r == nil)
return;
assert(s != nil);
c = s->c;
if(s->op == CHANRCV){
t = s;
s = r;
r = t;
}
assert(s==nil || s->op == CHANSND);
assert(r==nil || r->op == CHANRCV);
/*
* Channel is empty (or unbuffered) - copy directly.
*/
if(s && r && c->nbuf == 0){
amove(r->v, s->v, c->elemsize);
return;
}
/*
* Otherwise it's always okay to receive and then send.
*/
if(r){
cp = c->buf + c->off*c->elemsize;
amove(r->v, cp, c->elemsize);
--c->nbuf;
if(++c->off == c->bufsize)
c->off = 0;
}
if(s){
cp = c->buf + (c->off+c->nbuf)%c->bufsize*c->elemsize;
amove(cp, s->v, c->elemsize);
++c->nbuf;
}
}
static void
altexec(Alt *a)
{
int i;
_Altarray *ar;
Alt *other;
Channel *c;
c = a->c;
ar = chanarray(c, otherop(a->op));
if(ar && ar->n){
i = rand()%ar->n;
other = ar->a[i];
altcopy(a, other);
altalldequeue(other->thread->alt);
other->thread->alt = other;
_threadready(other->thread);
}else
altcopy(a, nil);
}
#define dbgalt 0
int
chanalt(Alt *a)
{
int i, j, ncan, n, canblock;
Channel *c;
_Thread *t;
needstack(512);
for(i=0; a[i].op != CHANEND && a[i].op != CHANNOBLK; i++)
;
n = i;
canblock = a[i].op == CHANEND;
t = proc()->thread;
for(i=0; i<n; i++)
a[i].thread = t;
t->alt = a;
qlock(&chanlock);
if(dbgalt) print("alt ");
ncan = 0;
for(i=0; i<n; i++){
c = a[i].c;
if(dbgalt) print(" %c:", "esrnb"[a[i].op]);
if(dbgalt) if(c->name) print("%s", c->name); else print("%p", c);
if(altcanexec(&a[i])){
if(dbgalt) print("*");
ncan++;
}
}
if(ncan){
j = rand()%ncan;
for(i=0; i<n; i++){
if(altcanexec(&a[i])){
if(j-- == 0){
if(dbgalt){
c = a[i].c;
print(" => %c:", "esrnb"[a[i].op]);
if(c->name) print("%s", c->name); else print("%p", c);
print("\n");
}
altexec(&a[i]);
qunlock(&chanlock);
return i;
}
}
}
}
if(dbgalt)print("\n");
if(!canblock){
qunlock(&chanlock);
return -1;
}
for(i=0; i<n; i++){
if(a[i].op != CHANNOP)
altqueue(&a[i]);
}
qunlock(&chanlock);
_threadswitch();
/*
* the guy who ran the op took care of dequeueing us
* and then set t->alt to the one that was executed.
*/
if(t->alt < a || t->alt >= a+n)
sysfatal("channel bad alt");
return t->alt - a;
}
static int
_chanop(Channel *c, int op, void *p, int canblock)
{
Alt a[2];
a[0].c = c;
a[0].op = op;
a[0].v = p;
a[1].op = canblock ? CHANEND : CHANNOBLK;
if(chanalt(a) < 0)
return -1;
return 1;
}
int
chansend(Channel *c, void *v)
{
return _chanop(c, CHANSND, v, 1);
}
int
channbsend(Channel *c, void *v)
{
return _chanop(c, CHANSND, v, 0);
}
int
chanrecv(Channel *c, void *v)
{
return _chanop(c, CHANRCV, v, 1);
}
int
channbrecv(Channel *c, void *v)
{
return _chanop(c, CHANRCV, v, 0);
}
int
chansendp(Channel *c, void *v)
{
return _chanop(c, CHANSND, (void*)&v, 1);
}
void*
chanrecvp(Channel *c)
{
void *v;
if(_chanop(c, CHANRCV, (void*)&v, 1) > 0)
return v;
return nil;
}
int
channbsendp(Channel *c, void *v)
{
return _chanop(c, CHANSND, (void*)&v, 0);
}
void*
channbrecvp(Channel *c)
{
void *v;
if(_chanop(c, CHANRCV, (void*)&v, 0) > 0)
return v;
return nil;
}
int
chansendul(Channel *c, ulong val)
{
return _chanop(c, CHANSND, &val, 1);
}
ulong
chanrecvul(Channel *c)
{
ulong val;
if(_chanop(c, CHANRCV, &val, 1) > 0)
return val;
return -1;
}
int
channbsendul(Channel *c, ulong val)
{
return _chanop(c, CHANSND, &val, 0);
}
ulong
channbrecvul(Channel *c)
{
ulong val;
if(_chanop(c, CHANRCV, &val, 0) > 0)
return val;
return -1;
}