/*  * 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 "stdinc.h" #include "9.h"#include "dat.h"#include "fns.h" enum {	NConInit	= 128,	NMsgInit	= 384,	NMsgProcInit	= 64,	NMsizeInit	= 8192+IOHDRSZ,}; static struct {	VtLock*	alock;			/* alloc */	Msg*	ahead;	VtRendez* arendez; 	int	maxmsg;	int	nmsg;	int	nmsgstarve; 	VtLock*	rlock;			/* read */	Msg*	rhead;	Msg*	rtail;	VtRendez* rrendez; 	int	maxproc;	int	nproc;	int	nprocstarve; 	u32int	msize;			/* immutable */} mbox; static struct {	VtLock*	alock;			/* alloc */	Con*	ahead;	VtRendez* arendez; 	VtLock*	clock;	Con*	chead;	Con*	ctail; 	int	maxcon;	int	ncon;	int	nconstarve; 	u32int	msize;} cbox; static voidconFree(Con* con){	assert(con->version == nil);	assert(con->mhead == nil);	assert(con->whead == nil);	assert(con->nfid == 0);	assert(con->state == ConMoribund); 	if(con->fd >= 0){		close(con->fd);		con->fd = -1;	}	con->state = ConDead;	con->aok = 0;	con->flags = 0;	con->isconsole = 0; 	vtLock(cbox.alock);	if(con->cprev != nil)		con->cprev->cnext = con->cnext;	else		cbox.chead = con->cnext;	if(con->cnext != nil)		con->cnext->cprev = con->cprev;	else		cbox.ctail = con->cprev;	con->cprev = con->cnext = nil; 	if(cbox.ncon > cbox.maxcon){		if(con->name != nil)			vtMemFree(con->name);		vtLockFree(con->fidlock);		vtMemFree(con->data);		vtRendezFree(con->wrendez);		vtLockFree(con->wlock);		vtRendezFree(con->mrendez);		vtLockFree(con->mlock);		vtRendezFree(con->rendez);		vtLockFree(con->lock);		vtMemFree(con);		cbox.ncon--;		vtUnlock(cbox.alock);		return;	}	con->anext = cbox.ahead;	cbox.ahead = con;	if(con->anext == nil)		vtWakeup(cbox.arendez);	vtUnlock(cbox.alock);} static voidmsgFree(Msg* m){	assert(m->rwnext == nil);	assert(m->flush == nil); 	vtLock(mbox.alock);	if(mbox.nmsg > mbox.maxmsg){		vtMemFree(m->data);		vtMemFree(m);		mbox.nmsg--;		vtUnlock(mbox.alock);		return;	}	m->anext = mbox.ahead;	mbox.ahead = m;	if(m->anext == nil)		vtWakeup(mbox.arendez);	vtUnlock(mbox.alock);} static Msg*msgAlloc(Con* con){	Msg *m; 	vtLock(mbox.alock);	while(mbox.ahead == nil){		if(mbox.nmsg >= mbox.maxmsg){			mbox.nmsgstarve++;			vtSleep(mbox.arendez);			continue;		}		m = vtMemAllocZ(sizeof(Msg));		m->data = vtMemAlloc(mbox.msize);		m->msize = mbox.msize;		mbox.nmsg++;		mbox.ahead = m;		break;	}	m = mbox.ahead;	mbox.ahead = m->anext;	m->anext = nil;	vtUnlock(mbox.alock); 	m->con = con;	m->state = MsgR;	m->nowq = 0; 	return m;} static voidmsgMunlink(Msg* m){	Con *con; 	con = m->con; 	if(m->mprev != nil)		m->mprev->mnext = m->mnext;	else		con->mhead = m->mnext;	if(m->mnext != nil)		m->mnext->mprev = m->mprev;	else		con->mtail = m->mprev;	m->mprev = m->mnext = nil;} voidmsgFlush(Msg* m){	Con *con;	Msg *flush, *old; 	con = m->con; 	if(Dflag)		fprint(2, "msgFlush %F\n", &m->t); 	/*	 * If this Tflush has been flushed, nothing to do.	 * Look for the message to be flushed in the	 * queue of all messages still on this connection.	 * If it's not found must assume Elvis has already	 * left the building and reply normally.	 */	vtLock(con->mlock);	if(m->state == MsgF){		vtUnlock(con->mlock);		return;	}	for(old = con->mhead; old != nil; old = old->mnext)		if(old->t.tag == m->t.oldtag)			break;	if(old == nil){		if(Dflag)			fprint(2, "msgFlush: cannot find %d\n", m->t.oldtag);		vtUnlock(con->mlock);		return;	} 	if(Dflag)		fprint(2, "\tmsgFlush found %F\n", &old->t); 	/*	 * Found it.	 * There are two cases where the old message can be	 * truly flushed and no reply to the original message given.	 * The first is when the old message is in MsgR state; no	 * processing has been done yet and it is still on the read	 * queue. The second is if old is a Tflush, which doesn't	 * affect the server state. In both cases, put the old	 * message into MsgF state and let MsgWrite toss it after	 * pulling it off the queue.	 */	if(old->state == MsgR || old->t.type == Tflush){		old->state = MsgF;		if(Dflag)			fprint(2, "msgFlush: change %d from MsgR to MsgF\n",				m->t.oldtag);	} 	/*	 * Link this flush message and the old message	 * so multiple flushes can be coalesced (if there are	 * multiple Tflush messages for a particular pending	 * request, it is only necessary to respond to the last	 * one, so any previous can be removed) and to be	 * sure flushes wait for their corresponding old	 * message to go out first.	 * Waiting flush messages do not go on the write queue,	 * they are processed after the old message is dealt	 * with. There's no real need to protect the setting of	 * Msg.nowq, the only code to check it runs in this	 * process after this routine returns.	 */	if((flush = old->flush) != nil){		if(Dflag)			fprint(2, "msgFlush: remove %d from %d list\n",				old->flush->t.tag, old->t.tag);		m->flush = flush->flush;		flush->flush = nil;		msgMunlink(flush);		msgFree(flush);	}	old->flush = m;	m->nowq = 1; 	if(Dflag)		fprint(2, "msgFlush: add %d to %d queue\n",			m->t.tag, old->t.tag);	vtUnlock(con->mlock);} static voidmsgProc(void*){	Msg *m;	char *e;	Con *con; 	vtThreadSetName("msgProc"); 	for(;;){		/*		 * If surplus to requirements, exit.		 * If not, wait for and pull a message off		 * the read queue.		 */		vtLock(mbox.rlock);		if(mbox.nproc > mbox.maxproc){			mbox.nproc--;			vtUnlock(mbox.rlock);			break;		}		while(mbox.rhead == nil)			vtSleep(mbox.rrendez);		m = mbox.rhead;		mbox.rhead = m->rwnext;		m->rwnext = nil;		vtUnlock(mbox.rlock); 		con = m->con;		e = nil; 		/*		 * If the message has been flushed before		 * any 9P processing has started, mark it so		 * none will be attempted.		 */		vtLock(con->mlock);		if(m->state == MsgF)			e = "flushed";		else			m->state = Msg9;		vtUnlock(con->mlock); 		if(e == nil){			/*			 * explain this			 */			vtLock(con->lock);			if(m->t.type == Tversion){				con->version = m;				con->state = ConDown;				while(con->mhead != m)					vtSleep(con->rendez);				assert(con->state == ConDown);				if(con->version == m){					con->version = nil;					con->state = ConInit;				}				else					e = "Tversion aborted";			}			else if(con->state != ConUp)				e = "connection not ready";			vtUnlock(con->lock);		} 		/*		 * Dispatch if not error already.		 */		m->r.tag = m->t.tag;		if(e == nil && !(*rFcall[m->t.type])(m))			e = vtGetError();		if(e != nil){			m->r.type = Rerror;			m->r.ename = e;		}		else			m->r.type = m->t.type+1; 		/*		 * Put the message (with reply) on the		 * write queue and wakeup the write process.		 */		if(!m->nowq){			vtLock(con->wlock);			if(con->whead == nil)				con->whead = m;			else				con->wtail->rwnext = m;			con->wtail = m;			vtWakeup(con->wrendez);			vtUnlock(con->wlock);		}	}} static voidmsgRead(void* v){	Msg *m;	Con *con;	int eof, fd, n; 	vtThreadSetName("msgRead"); 	con = v;	fd = con->fd;	eof = 0; 	while(!eof){		m = msgAlloc(con); 		while((n = read9pmsg(fd, m->data, con->msize)) == 0)			;		if(n < 0){			m->t.type = Tversion;			m->t.fid = NOFID;			m->t.tag = NOTAG;			m->t.msize = con->msize;			m->t.version = "9PEoF";			eof = 1;		}		else if(convM2S(m->data, n, &m->t) != n){			if(Dflag)				fprint(2, "msgRead: convM2S error: %s\n",					con->name);			msgFree(m);			continue;		}		if(Dflag)			fprint(2, "msgRead %p: t %F\n", con, &m->t); 		vtLock(con->mlock);		if(con->mtail != nil){			m->mprev = con->mtail;			con->mtail->mnext = m;		}		else{			con->mhead = m;			m->mprev = nil;		}		con->mtail = m;		vtUnlock(con->mlock); 		vtLock(mbox.rlock);		if(mbox.rhead == nil){			mbox.rhead = m;			if(!vtWakeup(mbox.rrendez)){				if(mbox.nproc < mbox.maxproc){					if(vtThread(msgProc, nil) > 0)						mbox.nproc++;				}				else					mbox.nprocstarve++;			}			/*			 * don't need this surely?			vtWakeup(mbox.rrendez);			 */		}		else			mbox.rtail->rwnext = m;		mbox.rtail = m;		vtUnlock(mbox.rlock);	}} static voidmsgWrite(void* v){	Con *con;	int eof, n;	Msg *flush, *m; 	vtThreadSetName("msgWrite"); 	con = v;	if(vtThread(msgRead, con) < 0){		conFree(con);		return;	} 	for(;;){		/*		 * Wait for and pull a message off the write queue.		 */		vtLock(con->wlock);		while(con->whead == nil)			vtSleep(con->wrendez);		m = con->whead;		con->whead = m->rwnext;		m->rwnext = nil;		assert(!m->nowq);		vtUnlock(con->wlock); 		eof = 0; 		/*		 * Write each message (if it hasn't been flushed)		 * followed by any messages waiting for it to complete.		 */		vtLock(con->mlock);		while(m != nil){			msgMunlink(m); 			if(Dflag)				fprint(2, "msgWrite %d: r %F\n",					m->state, &m->r); 			if(m->state != MsgF){				m->state = MsgW;				vtUnlock(con->mlock); 				n = convS2M(&m->r, con->data, con->msize);				if(write(con->fd, con->data, n) != n)					eof = 1; 				vtLock(con->mlock);			} 			if((flush = m->flush) != nil){				assert(flush->nowq);				m->flush = nil;			}			msgFree(m);			m = flush;		}		vtUnlock(con->mlock); 		vtLock(con->lock);		if(eof && con->fd >= 0){			close(con->fd);			con->fd = -1;

Lines 1–500 of 834. Load more