/*  * 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)			free(c->qentry);		free(c);	}} voidchanfree(Channel *c){	lock(&chanlock);	_chanfree(c);	unlock(&chanlock);} intchaninit(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 intisopenfor(Channel *c, int op){	return c->closed == 0 || (op == CHANRCV && c->n > 0);} intalt(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;} intchanclose(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;} intchanclosing(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 chanclosingintchanisclosed(Channel *c){	return chanisclosing(c) >= 0;} */ static intrunop(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;	}} intrecv(Channel *c, void *v){	return runop(CHANRCV, c, v, 0);} intnbrecv(Channel *c, void *v){	return runop(CHANRCV, c, v, 1);} intsend(Channel *c, void *v){	return runop(CHANSND, c, v, 0);} intnbsend(Channel *c, void *v){	return runop(CHANSND, c, v, 1);} static voidchannelsize(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();	}} intsendul(Channel *c, ulong v){	channelsize(c, sizeof(ulong));	return send(c, &v);} ulongrecvul(Channel *c){	ulong v; 	channelsize(c, sizeof(ulong));	if(recv(c, &v) < 0)		return ~0;	return v;} intsendp(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;} intnbsendul(Channel *c, ulong v){	channelsize(c, sizeof(ulong));	return nbsend(c, &v);} ulongnbrecvul(Channel *c){	ulong v; 	channelsize(c, sizeof(ulong));	if(nbrecv(c, &v) == 0)		return 0;	return v;} intnbsendp(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 intemptyentry(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 voidenqueue(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 voiddequeue(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 intcanexec(Alt *a){	int i, otherop;

Lines 1–500 of 622. Load more