jehanne/sys/src/lib/thread/channel.c

/*
 * This file is part of the UCB release of Plan 9. It is subject to the license
 * terms in the LICENSE file found in the top-level directory of this
 * distribution and at http://akaros.cs.berkeley.edu/files/Plan9License. No
 * part of the UCB release of Plan 9, including this file, may be copied,
 * modified, propagated, or distributed except according to the terms contained
 * in the LICENSE file.
 */

#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)
			jehanne_free(c->qentry);
		jehanne_free(c);
	}
}

void
chanfree(Channel *c)
{
	jehanne_lock(&chanlock);
	_chanfree(c);
	jehanne_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;
}

/* isopenfor is meant to be called under the chanlock. */
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();
	jehanne_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){
			jehanne_unlock(&chanlock);
			_procsplx(s);
			t->chan = Channone;
			return -1;
		}

		if(isopenfor((Channel*)c, xa->op) && canexec(xa))
			if(jehanne_nrand(++n) == 0)
				a = xa;
	}


	if(a==nil){
		/* nothing can proceed */
		if(xa->op == CHANNOBLK){
			jehanne_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, (struct Channel **)&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;
				jehanne_unlock(&chanlock);
				_procsplx(s);
				t->chan = Channone;
				return ca - alts;
			} else if(allreadycl){
				/* everything was already closed */
				jehanne_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:
		jehanne_unlock(&chanlock);
		_procsplx(s);
		r = _threadrendezvous(&c, 0);
		s = _procsplhi();
		jehanne_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);
		}
		jehanne_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)
{
	volatile Alt *a;
	int i, s;

	s = _procsplhi();	/* note handlers; see :/^alt */
	jehanne_lock(&chanlock);
	if(c->closed){
		/* Already close; we fail but it's ok. don't print */
		jehanne_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;
		jehanne_unlock(&chanlock);
		_procsplx(s);
		while(_threadrendezvous(a->tag, Closed) == Intred)
			;
		s = _procsplhi();
		jehanne_lock(&chanlock);
	}

	c->closed = 2;		/* Fully closed */
	if(c->freed)
		_chanfree(c);
	jehanne_unlock(&chanlock);
	_procsplx(s);
	return 0;
}

int
chanclosing(Channel *c)
{
	int n, s;

	s = _procsplhi();	/* note handlers; see :/^alt */
	jehanne_lock(&chanlock);
	if(c->closed == 0)
		n = -1;
	else
		n = c->n;
	jehanne_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:
		jehanne_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){
		jehanne_fprint(2, "expected channel with elements of size %d, got size %d\n",
			sz, c->e);
		abort();
	}
}

int
sendul(Channel *c, uint32_t v)
{
	channelsize(c, sizeof(uint32_t));
	return send(c, &v);
}

uint32_t
recvul(Channel *c)
{
	uint32_t v;

	channelsize(c, sizeof(uint32_t));
	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, uint32_t v)
{
	channelsize(c, sizeof(uint32_t));
	return nbsend(c, &v);
}

uint32_t
nbrecvul(Channel *c)
{
	uint32_t v;

	channelsize(c, sizeof(uint32_t));
	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 = jehanne_realloc((void*)c->qentry, c->nentry*sizeof(c->qentry[0]));
	if(c->qentry == nil)
		jehanne_sysfatal("realloc channel entries: %r");
	jehanne_memset(&c->qentry[i], 0, extra*sizeof(c->qentry[0]));
	return i;
}

/* enqueue is called under the chanlock, the Channel can be treated as non-volatile. */
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)
{
	uint8_t *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)
			jehanne_memmove(dst, src, sz);
		else
			jehanne_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(jehanne_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");
		jehanne_unlock(&chanlock);
		_procsplx(spl);
		_threaddebug(DBGCHAN, "chanlock is %lud",
			     *(uint32_t*)&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);

	jehanne_unlock(&chanlock);
	_procsplx(spl);
	return 1;
}