Subversion Repositories planix.SVN

#include <u.h>

#include <libc.h>

#include <thread.h>

#include "threadimpl.h"

/* Value to indicate the channel is closed */

enum {

	CHANCLOSD = 0xc105ed,

};

static char errcl[] = "channel was closed";

static Lock chanlock;		/* central channel access lock */

static void enqueue(Alt*, Channel**);

static void dequeue(Alt*);

static int canexec(Alt*);

static int altexec(Alt*, int);

#define Closed	((void*)CHANCLOSD)

#define Intred	((void*)~0)		/* interrupted */

static void

_chanfree(Channel *c)

{

	int i, inuse;

	if(c->closed == 1)			/* chanclose is ongoing */

		inuse = 1;

	else{

		inuse = 0;

		for(i = 0; i < c->nentry; i++)	/* alt ongoing */

			if(c->qentry[i])

				inuse = 1;

	}

	if(inuse)

		c->freed = 1;

	else{

		if(c->qentry)

			free(c->qentry);

		free(c);

	}

}

void

chanfree(Channel *c)

{

	lock(&chanlock);

	_chanfree(c);

	unlock(&chanlock);

}

int

chaninit(Channel *c, int elemsize, int elemcnt)

{

	if(elemcnt < 0 || elemsize <= 0 || c == nil)

		return -1;

	c->f = 0;

	c->n = 0;

	c->closed = 0;

	c->freed = 0;

	c->e = elemsize;

	c->s = elemcnt;

	_threaddebug(DBGCHAN, "chaninit %p", c);

	return 1;

}

Channel*

chancreate(int elemsize, int elemcnt)

{

	Channel *c;

	if(elemcnt < 0 || elemsize <= 0)

		return nil;

	c = _threadmalloc(sizeof(Channel)+elemsize*elemcnt, 1);

	c->e = elemsize;

	c->s = elemcnt;

	_threaddebug(DBGCHAN, "chancreate %p", c);

	return c;

}

static int

isopenfor(Channel *c, int op)

{

	return c->closed == 0 || (op == CHANRCV && c->n > 0);

}

int

alt(Alt *alts)

{

	Alt *a, *xa, *ca;

	Channel volatile *c;

	int n, s, waiting, allreadycl;

	void* r;

	Thread *t;

	/*

	 * The point of going splhi here is that note handlers

	 * might reasonably want to use channel operations,

	 * but that will hang if the note comes while we hold the

	 * chanlock.  Instead, we delay the note until we've dropped

	 * the lock.

	 */

	t = _threadgetproc()->thread;

	if(t->moribund || _threadexitsallstatus)

		yield();	/* won't return */

	s = _procsplhi();

	lock(&chanlock);

	t->alt = alts;

	t->chan = Chanalt;

	/* test whether any channels can proceed */

	n = 0;

	a = nil;

	for(xa=alts; xa->op!=CHANEND && xa->op!=CHANNOBLK; xa++){

		xa->entryno = -1;

		if(xa->op == CHANNOP)

			continue;

		c = xa->c;

		if(c==nil){

			unlock(&chanlock);

			_procsplx(s);

			t->chan = Channone;

			return -1;

		}

		if(isopenfor(c, xa->op) && canexec(xa))

			if(nrand(++n) == 0)

				a = xa;

	}

	if(a==nil){

		/* nothing can proceed */

		if(xa->op == CHANNOBLK){

			unlock(&chanlock);

			_procsplx(s);

			t->chan = Channone;

			if(xa->op == CHANNOBLK)

				return xa - alts;

		}

		/* enqueue on all channels open for us. */

		c = nil;

		ca = nil;

		waiting = 0;

		allreadycl = 0;

		for(xa=alts; xa->op!=CHANEND; xa++)

			if(xa->op==CHANNOP)

				continue;

			else if(isopenfor(xa->c, xa->op)){

				waiting = 1;

				enqueue(xa, &c);

			} else if(xa->err != errcl)

				ca = xa;

			else

				allreadycl = 1;

		if(waiting == 0)

			if(ca != nil){

				/* everything was closed, select last channel */

				ca->err = errcl;

				unlock(&chanlock);

				_procsplx(s);

				t->chan = Channone;

				return ca - alts;

			} else if(allreadycl){

				/* everything was already closed */

				unlock(&chanlock);

				_procsplx(s);

				t->chan = Channone;

				return -1;

			}

		/*

		 * wait for successful rendezvous.

		 * we can't just give up if the rendezvous

		 * is interrupted -- someone else might come

		 * along and try to rendezvous with us, so

		 * we need to be here.

		 * if the channel was closed, the op is done

		 * and we flag an error for the entry.

		 */

	    Again:

		unlock(&chanlock);

		_procsplx(s);

		r = _threadrendezvous(&c, 0);

		s = _procsplhi();

		lock(&chanlock);

		if(r==Intred){		/* 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 */

		a = nil;

		for(xa=alts; xa->op!=CHANEND; xa++){

			if(xa->op==CHANNOP)

				continue;

			if(xa->c == c){

				a = xa;

				a->err = nil;

				if(r == Closed)

					a->err = errcl;

			}

			dequeue(xa);

		}

		unlock(&chanlock);

		_procsplx(s);

		if(a == nil){	/* we were interrupted */

			assert(c==(Channel*)~0);

			return -1;

		}

	}else

		altexec(a, s);	/* unlocks chanlock, does splx */

	_sched();

	t->chan = Channone;

	return a - alts;

}

int

chanclose(Channel *c)

{

	Alt *a;

	int i, s;

	s = _procsplhi();	/* note handlers; see :/^alt */

	lock(&chanlock);

	if(c->closed){

		/* Already close; we fail but it's ok. don't print */

		unlock(&chanlock);

		_procsplx(s);

		return -1;

	}

	c->closed = 1;		/* Being closed */

	/*

	 * Locate entries that will fail due to close

	 * (send, and receive if nothing buffered) and wake them up.

	 * the situation cannot change because all queries

	 * should be committed by now and new ones will find the channel

	 * closed.  We still need to take the lock during the iteration

	 * because we can wake threads on qentrys we have not seen yet

	 * as in alt and there would be a race in the access to *a.

	 */

	for(i = 0; i < c->nentry; i++){

		if((a = c->qentry[i]) == nil || *a->tag != nil)

			continue;

		if(a->op != CHANSND && (a->op != CHANRCV || c->n != 0))

			continue;

		*a->tag = c;

		unlock(&chanlock);

		_procsplx(s);

		while(_threadrendezvous(a->tag, Closed) == Intred)

			;

		s = _procsplhi();

		lock(&chanlock);

	}

	c->closed = 2;		/* Fully closed */

	if(c->freed)

		_chanfree(c);

	unlock(&chanlock);

	_procsplx(s);

	return 0;

}

int

chanclosing(Channel *c)

{

	int n, s;

	s = _procsplhi();	/* note handlers; see :/^alt */

	lock(&chanlock);

	if(c->closed == 0)

		n = -1;

	else

		n = c->n;

	unlock(&chanlock);

	_procsplx(s);

	return n;

}

/*

 * superseded by chanclosing

int

chanisclosed(Channel *c)

{

	return chanisclosing(c) >= 0;

}

 */

static int

runop(int op, Channel *c, void *v, int nb)

{

	int r;

	Alt a[2];

	/*

	 * we could do this without calling alt,

	 * but the only reason would be performance,

	 * and i'm not convinced it matters.

	 */

	a[0].op = op;

	a[0].c = c;

	a[0].v = v;

	a[0].err = nil;

	a[1].op = CHANEND;

	if(nb)

		a[1].op = CHANNOBLK;

	switch(r=alt(a)){

	case -1:	/* interrupted */

		return -1;

	case 1:	/* nonblocking, didn't accomplish anything */

		assert(nb);

		return 0;

	case 0:

		/*

		 * Okay, but return -1 if the op is done because of a close.

		 */

		if(a[0].err != nil)

			return -1;

		return 1;

	default:

		fprint(2, "ERROR: channel alt returned %d\n", r);

		abort();

		return -1;

	}

}

int

recv(Channel *c, void *v)

{

	return runop(CHANRCV, c, v, 0);

}

int

nbrecv(Channel *c, void *v)

{

	return runop(CHANRCV, c, v, 1);

}

int

send(Channel *c, void *v)

{

	return runop(CHANSND, c, v, 0);

}

int

nbsend(Channel *c, void *v)

{

	return runop(CHANSND, c, v, 1);

}

static void

channelsize(Channel *c, int sz)

{

	if(c->e != sz){

		fprint(2, "expected channel with elements of size %d, got size %d\n",

			sz, c->e);

		abort();

	}

}

int

sendul(Channel *c, ulong v)

{

	channelsize(c, sizeof(ulong));

	return send(c, &v);

}

ulong

recvul(Channel *c)

{

	ulong v;

	channelsize(c, sizeof(ulong));

	if(recv(c, &v) < 0)

		return ~0;

	return v;

}

int

sendp(Channel *c, void *v)

{

	channelsize(c, sizeof(void*));

	return send(c, &v);

}

void*

recvp(Channel *c)

{

	void *v;

	channelsize(c, sizeof(void*));

	if(recv(c, &v) < 0)

		return nil;

	return v;

}

int

nbsendul(Channel *c, ulong v)

{

	channelsize(c, sizeof(ulong));

	return nbsend(c, &v);

}

ulong

nbrecvul(Channel *c)

{

	ulong v;

	channelsize(c, sizeof(ulong));

	if(nbrecv(c, &v) == 0)

		return 0;

	return v;

}

int

nbsendp(Channel *c, void *v)

{

	channelsize(c, sizeof(void*));

	return nbsend(c, &v);

}

void*

nbrecvp(Channel *c)

{

	void *v;

	channelsize(c, sizeof(void*));

	if(nbrecv(c, &v) == 0)

		return nil;

	return v;

}

static int

emptyentry(Channel *c)

{

	int i, extra;

	assert((c->nentry==0 && c->qentry==nil) || (c->nentry && c->qentry));

	for(i=0; i<c->nentry; i++)

		if(c->qentry[i]==nil)

			return i;

	extra = 16;

	c->nentry += extra;

	c->qentry = realloc((void*)c->qentry, c->nentry*sizeof(c->qentry[0]));

	if(c->qentry == nil)

		sysfatal("realloc channel entries: %r");

	memset(&c->qentry[i], 0, extra*sizeof(c->qentry[0]));

	return i;

}

static void

enqueue(Alt *a, Channel **c)

{

	int i;

	_threaddebug(DBGCHAN, "Queuing alt %p on channel %p", a, a->c);

	a->tag = c;

	i = emptyentry(a->c);

	a->c->qentry[i] = a;

}

static void

dequeue(Alt *a)

{

	int i;

	Channel *c;

	c = a->c;

	for(i=0; i<c->nentry; i++)

		if(c->qentry[i]==a){

			_threaddebug(DBGCHAN, "Dequeuing alt %p from channel %p", a, a->c);

			c->qentry[i] = nil;

			/* release if freed and not closing */

			if(c->freed && c->closed != 1)

				_chanfree(c);

			return;

		}

}

static int

canexec(Alt *a)

{

	int i, otherop;

	Channel *c;

	c = a->c;

	/* are there senders or receivers blocked? */

	otherop = (CHANSND+CHANRCV) - a->op;

	for(i=0; i<c->nentry; i++)

		if(c->qentry[i] && c->qentry[i]->op==otherop && *c->qentry[i]->tag==nil){

			_threaddebug(DBGCHAN, "can rendez alt %p chan %p", a, c);

			return 1;

		}

	/* is there room in the channel? */

	if((a->op==CHANSND && c->n < c->s)

	|| (a->op==CHANRCV && c->n > 0)){

		_threaddebug(DBGCHAN, "can buffer alt %p chan %p", a, c);

		return 1;

	}

	return 0;

}

static void*

altexecbuffered(Alt *a, int willreplace)

{

	uchar *v;

	Channel *c;

	c = a->c;

	/* use buffered channel queue */

	if(a->op==CHANRCV && c->n > 0){

		_threaddebug(DBGCHAN, "buffer recv alt %p chan %p", a, c);

		v = c->v + c->e*(c->f%c->s);

		if(!willreplace)

			c->n--;

		c->f++;

		return v;

	}

	if(a->op==CHANSND && c->n < c->s){

		_threaddebug(DBGCHAN, "buffer send alt %p chan %p", a, c);

		v = c->v + c->e*((c->f+c->n)%c->s);

		if(!willreplace)

			c->n++;

		return v;

	}

	abort();

	return nil;

}

static void

altcopy(void *dst, void *src, int sz)

{

	if(dst){

		if(src)

			memmove(dst, src, sz);

		else

			memset(dst, 0, sz);

	}

}

static int

altexec(Alt *a, int spl)

{

	volatile Alt *b;

	int i, n, otherop;

	Channel *c;

	void *me, *waiter, *buf;

	c = a->c;

	/* rendezvous with others */

	otherop = (CHANSND+CHANRCV) - a->op;

	n = 0;

	b = nil;

	me = a->v;

	for(i=0; i<c->nentry; i++)

		if(c->qentry[i] && c->qentry[i]->op==otherop && *c->qentry[i]->tag==nil)

			if(nrand(++n) == 0)

				b = c->qentry[i];

	if(b != nil){

		_threaddebug(DBGCHAN, "rendez %s alt %p chan %p alt %p", a->op==CHANRCV?"recv":"send", a, c, b);

		waiter = b->v;

		if(c->s && c->n){

			/*

			 * if buffer is full and there are waiters

			 * and we're meeting a waiter,

			 * we must be receiving.

			 *

			 * we use the value in the channel buffer,

			 * copy the waiter's value into the channel buffer

			 * on behalf of the waiter, and then wake the waiter.

			 */

			if(a->op!=CHANRCV)

				abort();

			buf = altexecbuffered(a, 1);

			altcopy(me, buf, c->e);

			altcopy(buf, waiter, c->e);

		}else{

			if(a->op==CHANRCV)

				altcopy(me, waiter, c->e);

			else

				altcopy(waiter, me, c->e);

		}

		*b->tag = c;	/* commits us to rendezvous */

		_threaddebug(DBGCHAN, "unlocking the chanlock");

		unlock(&chanlock);

		_procsplx(spl);

		_threaddebug(DBGCHAN, "chanlock is %lud", *(ulong*)&chanlock);

		while(_threadrendezvous(b->tag, 0) == Intred)

			;

		return 1;

	}

	buf = altexecbuffered(a, 0);

	if(a->op==CHANRCV)

		altcopy(me, buf, c->e);

	else

		altcopy(buf, me, c->e);

	unlock(&chanlock);

	_procsplx(spl);

	return 1;

}