text: /*
   * 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;
  }
  