~kris/9p

9hist

4e0713cc0cd54b674654f89e4a6a2febb098ebc9 — David du Colombier 35 years ago c2979df
Plan 9 from Bell Labs 1991-09-01
2 files changed, 539 insertions(+), 777 deletions(-)

M port/devmnt.c
M port/stream.c
M port/devmnt.c => port/devmnt.c +539 -775
@@ 4,268 4,108 @@
#include	"dat.h"
#include	"fns.h"
#include	"errno.h"

#include	"devtab.h"

#include	"fcall.h"

#define		NTAG	65536	/*  1 <= tag < NTAG */

typedef struct Mnt	Mnt;
typedef struct Mnthdr	Mnthdr;
typedef struct MntQ	MntQ;

struct Mnt
{
	Ref;			/* for number of chans, incl. mntpt but not msg */
	ulong	mntid;		/* serial # */
	Chan	*mntpt;		/* channel in user's name space */
	MntQ	*q;
};

struct MntQ
{
	Ref;
	QLock;			/* for access */
	MntQ	*next;		/* for allocation */
	Chan	*msg;		/* for reading and writing messages */
	Proc	*reader;	/* process reading response */
	Mnthdr	*writer;	/* queue of headers of written messages */
typedef struct Mntrpc Mntrpc;
typedef struct Mnt Mnt;

struct Mntrpc
{
	Mntrpc	*list;		/* Free/pending list */
	Fcall	request;	/* Outgoing file system protocol message */
	Fcall	reply;		/* Incoming reply */
	Mnt	*m;		/* Mount device during rpc */
	Rendez	r;		/* Place to hang out */
	char	*rpc;		/* I/O Data buffer */
	char	done;		/* Rpc completed */
	char	bfree;		/* Buffer may be freed after flush */
	char	flushed;	/* Flush was sent */
	ushort	flushtag;	/* Tag to send flush on */
	ushort	flushbase;	/* Base tag of flush window for this buffer */
	char	flush[MAXMSG];	/* Somewhere to build flush */
};

#define	BITROUND 256
#define	BUFSIZE	(MAXFDATA+MAXMSG)
typedef struct Mntbuf Mntbuf;
struct Mntbuf
struct Mnt
{
	Mntbuf	*next;
	char	buf[BUFSIZE+BITROUND]; 	/* BUG */
	Ref;			/* Count of attached channels */
	Chan	*c;		/* Channel to file service */
	Proc	*rip;		/* Reader in progress */
	Mntrpc	*queue;		/* Queue of pending requests on this channel */
	int	id;		/* Multiplexor id for channel check */
	Mnt	*list;		/* Free list */
	char	mux;		/* Set if the device aleady does the multiplexing */
};

struct
struct Mntalloc
{
	Lock;
	Mntbuf	*free;
}mntbufalloc;

struct Mnthdr
{
	Mnthdr	*next;		/* in free list or writers list */
	Mnthdr	*prev;		/* in writers list only */
	char	active;
	char	flushing;	/* a Tflush has been sent */
	short	seq;
	Fcall	thdr;
	Fcall	rhdr;
	Rendez	r;
	Proc	*p;
	Mntbuf	*mbr;
	int	readreply;	/* true if we are reader or our reply has come */
	Mnt	*mntfree;
	Mnt	*mntarena;
	Mntrpc	*rpcfree;
	int	id;
}mntalloc;

#define BITBOTCH	256
#define MAXRPC		(MAXFDATA+MAXMSG+BITBOTCH)
#define limit(n, max)	(n > max ? max : n)

Chan 	*mattach(Mnt*, char*, char*);
Mntrpc	*mntralloc(void);
void	mntfree(Mntrpc*);
int	rpcattn(Mntrpc*);
void	mountrpc(Mnt*, Mntrpc*);
void	mountio(Mnt*, Mntrpc*);
Mnt	*mntchk(Chan*);
void	mountmux(Mnt*, Mntrpc*);
long	mntrdwr(int , Chan*, void*,long , ulong);
int	mntflush(Mnt*, Mntrpc*);
void	mntqrm(Mnt*, Mntrpc*);
void	mntdirfix(uchar*, Chan*);
void	mntgate(Mnt*);
void	mntrpcread(Mnt*, Mntrpc*);

enum
{
	Tagspace = 1,
	Flushspace = 64,
	Flushtag = 512,
};

struct
{
	Lock;
	Mnthdr	*arena;
	Mnthdr	*head;
	Mnthdr	*tail;
}mnthdralloc;

struct
{
	Lock;
	QLock;
	MntQ	*arena;
	MntQ	*free;
}mntqalloc;

struct
{
	Lock;
	long	id;
}mntid;

Mnt	*mnt;
void	mntxmit(Mnt*, Mnthdr*);

Mntbuf*
mballoc(void)
{
	Mntbuf *mb;

loop:
	lock(&mntbufalloc);
	if(mb = mntbufalloc.free){		/* assign = */
		mntbufalloc.free = mb->next;
		unlock(&mntbufalloc);
		return mb;
	}
	unlock(&mntbufalloc);
	print("no mntbufs\n");
	if(u == 0)
		panic("mballoc");
	u->p->state = Wakeme;
	alarm(1000, wakeme, u->p);
	sched();
	goto loop;
}

void
mbfree(Mntbuf *mb)
{
	lock(&mntbufalloc);
	mb->next = mntbufalloc.free;
	mntbufalloc.free = mb;
	unlock(&mntbufalloc);
}

Mnthdr*
mhalloc(Mnt *m)
{
	Mnthdr *mh;
	int seq;

loop:
	lock(&mnthdralloc);
	if(mh = mnthdralloc.head){		/* assign = */
		mnthdralloc.head = mh->next;
		if(mnthdralloc.head)
			mnthdralloc.head->prev = 0;
		else
			mnthdralloc.tail = 0;
		unlock(&mnthdralloc);
		mh->mbr = 0;
		seq = ++mh->seq;
		if(seq == (1<<7)){
			mh->seq = 1;
			seq = 1;
		}
		mh->thdr.tag = (((mh-mnthdralloc.arena)<<7)|seq) & (NTAG-1);
		return mh;
	}
	unlock(&mnthdralloc);
	print("no mnthdrs\n");
	if(u == 0)
		panic("mhalloc");
	u->p->state = Wakeme;
	alarm(1000, wakeme, u->p);
	sched();
	goto loop;
}

void
mhfree(Mnthdr *mh)
{
	if(mh->flushing)
		return;
	lock(&mnthdralloc);
	mh->active = 0;
	mh->thdr.tag = 0;
	mh->next = 0;
	mh->prev = mnthdralloc.tail;
	if(mnthdralloc.tail)
		mnthdralloc.tail->next = mh;
	else
		mnthdralloc.head = mh;
	mnthdralloc.tail = mh;
	unlock(&mnthdralloc);
}

MntQ*
mqalloc(Chan *msg)	/* mntqalloc is qlocked */
{
	MntQ *q;

	if(q = mntqalloc.free){		/* assign = */
		mntqalloc.free = q->next;
		lock(q);
		q->ref = 1;
		q->msg = msg;
		unlock(q);
		incref(msg);
		q->writer = 0;
		q->reader = 0;
		return q;
	}
	panic("no mntqs\n");			/* there MUST be enough */
}

void
mqfree(MntQ *mq)
{
	Chan *msg = 0;

	lock(mq);
	if(--mq->ref == 0){
		msg = mq->msg;
		mq->msg = 0;
		lock(&mntqalloc);
		mq->next = mntqalloc.free;
		mntqalloc.free = mq;
		unlock(&mntqalloc);
	}
	unlock(mq);
	if(msg)		/* after locks are down */
		close(msg);
}

Mnt*
mntdev(Chan *c, int noerr)
{
	Mnt *m;
	int i;

	m = &mnt[c->mntindex];
	if(m->mntid==c->dev && m->q!=0)
		return m;
	if(noerr)
		return 0;
	print("mntdev shutdown %d %d %d %lux\n", c->dev, c->mntindex,
			m->mntid, m->q);
	error(Eshutdown);
}

void
mntreset(void)
{
	int i;
	Mntbuf *mb;
	Mnthdr *mh;
	MntQ *mq;

	if(conf.nmnthdr > 512){
		print("conf.nmnthdr is %d set to 512\n", conf.nmnthdr);
		conf.nmnthdr = 512;
	}
	mnt = ialloc(conf.nmntdev*sizeof(Mnt), 0);

	mb = ialloc(conf.nmntbuf*sizeof(Mntbuf), 0);
	mntbufalloc.free = mb;
	for(i=0; i<conf.nmntbuf; i++,mb++)
		mb->next = mb+1;
	--mb;
	mb->next = 0;

	mh = ialloc(conf.nmnthdr*sizeof(Mnthdr), 0);
	mnthdralloc.arena = mh;
	mnthdralloc.head = mh;
	for(i=0; i<conf.nmnthdr; i++,mh++){
		mh->seq = 0;
		mh->next = mh+1;
		mh->prev = mh-1;
	}
	--mh;
	mnthdralloc.tail = mh;
	mh->next = 0;
	mnthdralloc.head->prev = 0;

	mq = ialloc(conf.nmntdev*sizeof(MntQ), 0);
	mntqalloc.arena = mq;
	mntqalloc.free = mq;
	for(i=0; i<conf.nmntdev; i++,mq++)
		mq->next = mq+1;
	--mq;
	mq->next = 0;
	Mnt *me, *md;
	Mntrpc *re, *rd;
	ushort tag, ftag;

	mntalloc.mntarena = ialloc(conf.nmntdev*sizeof(Mnt), 0);
	mntalloc.mntfree = mntalloc.mntarena;
	me = &mntalloc.mntfree[conf.nmntdev];
	for(md = mntalloc.mntfree; md < me; md++)
		md->list = md+1;
	me[-1].list = 0;

	if(conf.nmntbuf > Flushtag) {
		print("devmnt: buffers limited to %d\n", Flushtag);
		conf.nmntbuf = Flushtag;
	}

	tag = Tagspace;
	ftag = Flushtag;
	mntalloc.rpcfree = ialloc(conf.nmntbuf*sizeof(Mntrpc), 0);
	re = &mntalloc.rpcfree[conf.nmntbuf];
	for(rd = mntalloc.rpcfree; rd < re; rd++) {
		rd->list = rd+1;
		rd->request.tag = tag++;
		rd->flushbase = ftag;
		rd->flushtag = ftag;
		ftag += Flushspace;
		rd->rpc = ialloc(MAXRPC, 0);
	}
	re[-1].list = 0;

	mntalloc.id = 1;
}

void


@@ 274,79 114,86 @@ mntinit(void)
}

Chan*
mntattach(char *crud)
mntattach(char *muxattach)
{
	int i;
	Mnt *m, *mm;
	Mnthdr *mh;
	MntQ *q;
	Chan *c, *cm;
	Mnt *m, *e;
	struct bogus{
		Chan	*chan;
		char	*spec;
		char	*auth;
	}bogus;

	bogus = *((struct bogus *)crud);

	m = mnt;
	for(i=0; i<conf.nmntdev; i++,m++){
		lock(m);
		if(m->ref == 0)
			goto Found;
		unlock(m);
	bogus = *((struct bogus *)muxattach);
	e = &mntalloc.mntarena[conf.nmntdev];
	for(m = mntalloc.mntarena; m < e; m++) {
		if(m->c == bogus.chan && m->id) {
			lock(m);
			if(m->ref > 0 && m->id && m->c == bogus.chan) {
				m->ref++;
				unlock(m);
				return mattach(m, bogus.spec, bogus.auth);
			}
			unlock(m);	
		}
	}
	error(Enomntdev);

    Found:
	lock(&mntalloc);
	if(mntalloc.mntfree == 0) {
		unlock(&mntalloc);
		error(Enomntdev);
	}
	m = mntalloc.mntfree;
	mntalloc.mntfree = m->list;	
	m->id = mntalloc.id++;
	lock(m);
	unlock(&mntalloc);
	m->ref = 1;
	m->queue = 0;
	m->rip = 0;
	m->c = bogus.chan;

	switch(devchar[m->c->type]) {
	case 'H':			/* Hotrod */
	case '3':			/* BIT3 */
		m->mux = 1;
		break;
	default:
		m->mux = 0;
	}
	incref(m->c);
	unlock(m);
	lock(&mntid);
	m->mntid = ++mntid.id;
	unlock(&mntid);
	c = devattach('M', bogus.spec);
	c->dev = m->mntid;
	c->mntindex = m-mnt;
	m->mntpt = c;
	cm = bogus.chan;

	/*
	 * Look for queue to same msg channel
	 */
	q = mntqalloc.arena;
	qlock(&mntqalloc);
	for(i=0; i<conf.nmntdev; i++,q++)
		if(q->msg==cm){
			lock(q);
			if(q->ref && q->msg==cm){
				m->q = q;
				q->ref++;
				unlock(q);
				goto out;
			}
			unlock(q);
		}
	m->q = mqalloc(cm);

    out:
	qunlock(&mntqalloc);
	mh = mhalloc(m);
	return mattach(m, bogus.spec, bogus.auth);
}

Chan *
mattach(Mnt *m, char *spec, char *auth)
{
	Chan *c;
	Mntrpc *r;

	r = mntralloc();

	c = devattach('M', spec);
	c->dev = m->id;
	c->mntindex = m-mntalloc.mntarena;

	if(waserror()){
		mhfree(mh);
		mntfree(r);
		close(c);
		nexterror();
	}
	mh->thdr.type = Tattach;
	mh->thdr.fid = c->fid;
	memmove(mh->thdr.uname, u->p->pgrp->user, NAMELEN);
	strcpy(mh->thdr.aname, bogus.spec);
	strcpy(mh->thdr.auth, bogus.auth);
	mntxmit(m, mh);
	c->qid = mh->rhdr.qid;
	c->mchan = m->q->msg;
	r->request.type = Tattach;
	r->request.fid = c->fid;
	memmove(r->request.uname, u->p->pgrp->user, NAMELEN);
	strncpy(r->request.aname, spec, NAMELEN);
	strncpy(r->request.auth, auth, NAMELEN);
	mountrpc(m, r);

	c->qid = r->reply.qid;
	c->mchan = m->c;
	c->mqid = c->qid;
	mhfree(mh);
	poperror();
	mntfree(r);
	return c;
}



@@ 354,28 201,27 @@ Chan*
mntclone(Chan *c, Chan *nc)
{
	Mnt *m;
	Mnthdr *mh;
	int new;
	Mntrpc *r;
	int alloc = 0;

	new = 0;
	if(nc == 0){
	m = mntchk(c);
	r = mntralloc();
	if(nc == 0) {
		nc = newchan();
		new = 1;
		if(waserror()){
			close(nc);
			nexterror();
		}
		alloc = 1;
	}
	m = mntdev(c, 0);
	mh = mhalloc(m);
	if(waserror()){
		mhfree(mh);
		mntfree(r);
		if(alloc)
			close(nc);
		nexterror();
	}
	mh->thdr.type = Tclone;
	mh->thdr.fid = c->fid;
	mh->thdr.newfid = nc->fid;
	mntxmit(m, mh);

	r->request.type = Tclone;
	r->request.fid = c->fid;
	r->request.newfid = nc->fid;
	mountrpc(m, r);

	nc->type = c->type;
	nc->dev = c->dev;
	nc->qid = c->qid;


@@ 385,14 231,12 @@ mntclone(Chan *c, Chan *nc)
	nc->mnt = c->mnt;
	nc->mountid = c->mountid;
	nc->aux = c->aux;
	nc->mntindex = c->mntindex;
	nc->mchan = c->mchan;
	nc->mqid = c->qid;
	mhfree(mh);
	poperror();
	if(new)
		poperror();
	incref(m);

	poperror();
	mntfree(r);
	return nc;
}



@@ 400,73 244,71 @@ int
mntwalk(Chan *c, char *name)
{
	Mnt *m;
	Mnthdr *mh;
	int found;

	found = 1;
	m = mntdev(c, 0);
	mh = mhalloc(m);
	mh->thdr.type = Twalk;
	mh->thdr.fid = c->fid;
	strcpy(mh->thdr.name, name);
	if(waserror()){	/* BUG: can check type of error? */
		found = 0;
		goto Out;
	Mntrpc *r;

	m = mntchk(c);
	r = mntralloc();
	if(waserror()) {
		mntfree(r);
		return 0;
	}
	mntxmit(m, mh);
	c->qid = mh->rhdr.qid;
	r->request.type = Twalk;
	r->request.fid = c->fid;
	strncpy(r->request.name, name, NAMELEN);
	mountrpc(m, r);

	c->qid = r->reply.qid;

	poperror();
    Out:
	mhfree(mh);
	return found;
	mntfree(r);
	return 1;
}

void	 
mntstat(Chan *c, char *dp)
{
	Mnt *m;
	Mnthdr *mh;
	Mntrpc *r;

	m = mntdev(c, 0);
	mh = mhalloc(m);
	if(waserror()){
		mhfree(mh);
	m = mntchk(c);
	r = mntralloc();
	if(waserror()) {
		mntfree(r);
		nexterror();
	}
	mh->thdr.type = Tstat;
	mh->thdr.fid = c->fid;
	mntxmit(m, mh);
	memmove(dp, mh->rhdr.stat, DIRLEN);
	dp[DIRLEN-4] = devchar[c->type];
	dp[DIRLEN-3] = 0;
	dp[DIRLEN-2] = c->dev;
	dp[DIRLEN-1] = c->dev>>8;
	mhfree(mh);
	r->request.type = Tstat;
	r->request.fid = c->fid;
	mountrpc(m, r);

	memmove(dp, r->reply.stat, DIRLEN);
	mntdirfix((uchar*)dp, c);
	poperror();
	mntfree(r);
}

Chan*
mntopen(Chan *c, int omode)
{
	Mnt *m;
	Mnthdr *mh;
	Mntrpc *r;

	m = mntdev(c, 0);
	mh = mhalloc(m);
	if(waserror()){
		mhfree(mh);
	m = mntchk(c);
	r = mntralloc();
	if(waserror()) {
		mntfree(r);
		nexterror();
	}
	mh->thdr.type = Topen;
	mh->thdr.fid = c->fid;
	mh->thdr.mode = omode;
	mntxmit(m, mh);
	c->qid = mh->rhdr.qid;
	mhfree(mh);
	poperror();
	r->request.type = Topen;
	r->request.fid = c->fid;
	r->request.mode = omode;
	mountrpc(m, r);

	c->qid = r->reply.qid;
	c->offset = 0;
	c->mode = openmode(omode);
	c->flag |= COPEN;
	poperror();
	mntfree(r);
	return c;
}



@@ 474,60 316,58 @@ void
mntcreate(Chan *c, char *name, int omode, ulong perm)
{
	Mnt *m;
	Mnthdr *mh;
	Mntrpc *r;

	m = mntdev(c, 0);
	mh = mhalloc(m);
	if(waserror()){
		mhfree(mh);
	m = mntchk(c);
	r = mntralloc();
	if(waserror()) {
		mntfree(r);
		nexterror();
	}
	mh->thdr.type = Tcreate;
	mh->thdr.fid = c->fid;
	strcpy(mh->thdr.name, name);
	mh->thdr.mode = omode;
	mh->thdr.perm = perm;
	mntxmit(m, mh);
	c->qid = mh->rhdr.qid;
	mhfree(mh);
	poperror();
	r->request.type = Tcreate;
	r->request.fid = c->fid;
	r->request.mode = omode;
	r->request.perm = perm;
	strncpy(r->request.name, name, NAMELEN);
	mountrpc(m, r);

	c->qid = r->reply.qid;
	c->flag |= COPEN;
	c->mode = openmode(omode);
	c->qid = mh->rhdr.qid;
	poperror();
	mntfree(r);
}

void	 
mntclunk(Chan *c, int t)
{
	Mnt *m;
	Mnthdr *mh;
	MntQ *q;
	int waserr;

	m = mntdev(c, 0);
	mh = mhalloc(m);
	mh->thdr.type = t;
	mh->thdr.fid = c->fid;
	waserr = 0;
	if(waserror())		/* gotta clean up as if there wasn't */
		waserr = 1;
	else
		mntxmit(m, mh);
	mhfree(mh);
	if(c == m->mntpt)
		m->mntpt = 0;
	lock(m);
	if(--m->ref == 0){		/* BUG: need to hang up all pending i/o */
		q = m->q;
		m->q = 0;
		m->mntid = 0;
		unlock(m);		/* mqfree can take time */
		mqfree(q);
	}else
		unlock(m);
	if(waserr)
		nexterror();
	poperror();
	Mntrpc *r, *n, *q;
		
	m = mntchk(c);
	r = mntralloc();
	if(waserror()){
		mntfree(r);
		if(decref(m) == 0) {
			for(q = m->queue; q; q = r) {
				r = q->list;
				q->flushed = 0;
				mntfree(q);
			}
			m->id = 0;
			close(m->c);
			lock(&mntalloc);
			m->list = mntalloc.mntfree;
			mntalloc.mntfree = m;
			unlock(&mntalloc);
		}
		return;
	}

	r->request.type = t;
	r->request.fid = c->fid;
	mountrpc(m, r);
	nexterror();
}

void


@@ 536,433 376,357 @@ mntclose(Chan *c)
	mntclunk(c, Tclunk);
}

long
mntreadwrite(Chan *c, void *vbuf, long n, int type, ulong offset)
void	 
mntremove(Chan *c)
{
	mntclunk(c, Tremove);
}

void
mntwstat(Chan *c, char *dp)
{
	Mnt *m;
	Mnthdr *mh;
	long nt, nr, count;
	char *buf;

	buf = vbuf;
	count = 0;
	m = mntdev(c, 0);
	mh = mhalloc(m);
	if(waserror()){
		mhfree(mh);
	Mntrpc *r;

	m = mntchk(c);
	r = mntralloc();
	if(waserror()) {
		mntfree(r);
		nexterror();
	}
	mh->thdr.type = type;
	mh->thdr.fid = c->fid;
    Loop:
	nt = n;
	if(nt > MAXFDATA)
		nt = MAXFDATA;
	mh->thdr.offset = offset;
	mh->thdr.count = nt;
	mh->thdr.data = buf;
	mntxmit(m, mh);
	nr = mh->rhdr.count;
	offset += nr;
	count += nr;
	buf += nr;
	n -= nr;
	if(n && nr==nt)
		goto Loop;
	mhfree(mh);
	r->request.type = Twstat;
	r->request.fid = c->fid;
	memmove(r->request.stat, dp, DIRLEN);
	mountrpc(m, r);
	poperror();
	return count;
	mntfree(r);
}

long	 
mntread(Chan *c, void *buf, long n, ulong offset)
{
	long i;
	uchar *b;

	n = mntreadwrite(c, buf, n, Tread, offset);
	if(c->qid.path & CHDIR){
		b = (uchar*)buf;
		for(i=n-DIRLEN; i>=0; i-=DIRLEN){
			b[DIRLEN-4] = devchar[c->type];
			b[DIRLEN-3] = 0;
			b[DIRLEN-2] = c->dev;
			b[DIRLEN-1] = c->dev>>8;
			b += DIRLEN;
		}
	}
	uchar *p, *e;

	n = mntrdwr(Tread, c, buf, n, offset);
	if(c->qid.path & CHDIR) 
		for(p = (uchar*)buf, e = &p[n]; p < e; p += DIRLEN)
			mntdirfix(p, c);

	return n;
}

long	 
mntwrite(Chan *c, void *buf, long n, ulong offset)
{
	return mntreadwrite(c, buf, n, Twrite, offset);
}

void	 
mntremove(Chan *c)
{
	mntclunk(c, Tremove);
	return mntrdwr(Twrite, c, buf, n, offset);	
}

void
mntwstat(Chan *c, char *dp)
long
mntrdwr(int type, Chan *c, void *buf, long n, ulong offset)
{
	Mnt *m;
	Mnthdr *mh;

	m = mntdev(c, 0);
	mh = mhalloc(m);
	if(waserror()){
		mhfree(mh);
	Mntrpc *r;
	ulong cnt, nr;
	char *uba;

	r = mntralloc();
	m = mntchk(c);
	if(waserror()) {
		mntfree(r);
		nexterror();
	}
	mh->thdr.type = Twstat;
	mh->thdr.fid = c->fid;
	memmove(mh->thdr.stat, dp, DIRLEN);
	mntxmit(m, mh);
	mhfree(mh);
	r->request.type = type;
	r->request.fid = c->fid;
	r->request.offset = offset;
	uba = buf;
	for(cnt = 0; n; n -= nr) {
		r->request.data = uba;
		r->request.count = limit(n, MAXFDATA);
		mountrpc(m, r);
		nr = r->reply.count;
		if(type == Tread)
			memmove(uba, r->reply.data, nr);
		r->request.offset += nr;
		uba += nr;
		cnt += nr;
		if(nr != r->request.count)
			break;
	}
	poperror();
	mntfree(r);
	return cnt;
}

void
mntwunlink(MntQ *q, Mnthdr *w)		/* queue is locked and w is a writer */
{
	if(w->next)
		w->next->prev = w->prev;
	if(w->prev)
		w->prev->next = w->next;
	else{
		q->writer = w->next;
		if(q->writer)
			q->writer->prev = 0;
mountrpc(Mnt *m, Mntrpc *r)
{
	mountio(m, r);
	if(r->reply.type == Rerror)
		errors(r->reply.ename);
	if(r->reply.type != r->request.type+1) {
		print("devmnt: mismatched reply T%d R%d tags req %d fls %d rep %d\n",
		r->request.type, r->reply.type, r->request.tag, r->flushtag, r->reply.tag);
			errors("protocol error");
	}
}

/*
 * m->q is unlocked.  Send Tflush message to flush omh->tag.
 * Cut off all errors.   Caller will free omh
 */
void
mntflush(Mnt *m, Mnthdr *omh)	/* queue is unlocked */
mountio(Mnt *m, Mntrpc *r)
{
	Mnthdr *mh;
	int n;

	if(omh->thdr.type == Tflush){
		omh->flushing = 0;
	lock(m);
	r->m = m;
	r->list = m->queue;
	m->queue = r;
	unlock(m);

	/* Transmit a file system rpc */
	n = convS2M(&r->request, r->rpc);
	if(waserror()) {
		mntqrm(m, r);
		nexterror();
	}
	if((*devtab[m->c->type].write)(m->c, r->rpc, n, 0) != n)
		error(Eshortmsg);
	poperror();

	if(m->mux) {
		mntrpcread(m, r);
		return;
	}

	mh = mhalloc(m);
	if(waserror()){
		omh->flushing = 0;
		mhfree(mh);
		return;		/* no more errors please */
	/* Gate readers onto the mount point one at a time */
	for(;;) {
		lock(m);
		if(m->rip == 0)
			break;
		unlock(m);
		if(waserror()) {
			if(mntflush(m, r) == 0)
				nexterror();
			continue;
		}
		sleep(&r->r, rpcattn, r);
		poperror();
		if(r->done)
			return;
	}
	mh->thdr.type = Tflush;
	mh->thdr.oldtag = omh->thdr.tag;
	mntxmit(m, mh);
	omh->flushing = 0;
	mhfree(mh);
	poperror();
	m->rip = u->p;
	unlock(m);

	while(r->done == 0) {
		mntrpcread(m, r);
		mountmux(m, r);
	}
	mntgate(m);
}

void
mnterrdequeue(Mnt *m, Mnthdr *mh)	/* queue is unlocked */
{
	Mnthdr *w;
	MntQ *q;

	mh->flushing = 1;
	q = m->q;
	qlock(q);
	mh->readreply = 0;
	/* take self from queue if necessary */
	if(q->reader == u->p){	/* advance a writer to reader */
		w = q->writer;
		if(w){
			mntwunlink(q, w);
			q->reader = w->p;
			wakeup(&w->r);
		}else{
			q->reader = 0;
			q->writer = 0;
mntrpcread(Mnt *m, Mntrpc *r)
{
	int n;

	for(;;) {
		if(waserror()) {
			if(mntflush(m, r) == 0) {
				if(m->mux == 0)
					mntgate(m);
				nexterror();
			}
			continue;
		}
	}else
		mntwunlink(q, mh);
	qunlock(q);
	mntflush(m, mh);
		n = (*devtab[m->c->type].read)(m->c, r->rpc, MAXRPC, 0);
		poperror();
		if(n == 0)
			continue;
		if(convM2S(r->rpc, &r->reply, n) != 0)
			break;
	}
}

int
mntreadreply(void *a)
void
mntgate(Mnt *m)
{
	return ((Mnthdr *)a)->readreply;
	Mntrpc *q;

	lock(m);
	m->rip = 0;
	for(q = m->queue; q; q = q->list)
		if(q->done == 0) {
			lock(&q->r);
			if(q->r.p) {
				unlock(&q->r);
				unlock(m);
				wakeup(&q->r);
				return;
			}
			unlock(&q->r);
		}
	unlock(m);
}

void
mntxmit(Mnt *m, Mnthdr *mh)
{
	ulong n;
	Mntbuf *mbw;
	Mnthdr *w, *ow, *h;
	MntQ *q;
	int qlocked, tag, written;

	if(&qlocked);	/* force qlocked not to be registerized */
	mh->mbr = 0;
	mbw = mballoc();
	if(waserror()){			/* 1 */
		if(mh->mbr){
			mbfree(mh->mbr);
			mh->mbr = 0;
mountmux(Mnt *m, Mntrpc *r)
{
	Mntrpc **l, *q;
	int done;
	char *dp;

	lock(m);
	l = &m->queue;
	for(q = *l; q; q = q->list) {
		if(q->request.tag == r->reply.tag) {
			if(q->flushed == 0)
				*l = q->list;
			q->done = 1;
			unlock(m);
			goto dispatch;
		}
		mbfree(mbw);
		nexterror();
	}
	n = convS2M(&mh->thdr, mbw->buf);
	q = m->q;
	if(q == 0)
		error(Eshutdown);
#ifdef	BIT3
	/*
	 * Bit3 and Hotrod do their own multiplexing.  (Well, the file server does.)
	 * The code is different enough that it's broken out separately here.
	 */
	if(devchar[q->msg->type]!='3' && devchar[q->msg->type]!='H')
		goto Normal;

	incref(q);
	if(waserror()){		/* 2 */
		mqfree(q);
		nexterror();
	}
	if((*devtab[q->msg->type].write)(q->msg, mbw->buf, n, 0) != n){
		print("short write in mntxmit\n");
		error(Eshortmsg);
		if(q->flushtag == r->reply.tag) {
			*l = q->list;
			q->flushed = 0;
			done = q->done;
			q->done = 1;
			unlock(m);
			if(done == 0) {
				r->reply.type = Rerror;
				strcpy(r->reply.ename, errstrtab[Eintr]);
				goto dispatch;
			}
			if(q->bfree)
				mntfree(q);
			return;
		}
		l = &q->list;
	}
	unlock(m);
	return;

	/*
	 * Read response
	 */
	if(waserror()){		/* 3 */
		mntflush(m, mh);
		nexterror();
	}
	mh->mbr = mballoc();
	n = (*devtab[q->msg->type].read)(q->msg, mh->mbr->buf, BUFSIZE, 0);
	poperror();		/* 3 */
	mqfree(q);
	poperror();		/* 2 */

	if(convM2S(mh->mbr->buf, &mh->rhdr, n) == 0){
		print("format error in mntxmit\n");
		error(Ebadmsg);
dispatch:
	if(q != r) {		/* Completed someone else */
		dp = q->rpc;
		q->rpc = r->rpc;
		r->rpc = dp;
		memmove(&q->reply, &r->reply, sizeof(Fcall));
		wakeup(&q->r);					
	}

	/*
	 * Various checks
	 */
	if(mh->rhdr.tag != mh->thdr.tag){
		print("tag mismatch %d %d\n", mh->rhdr.tag, mh->thdr.tag);
		error(Ebadmsg);
	}
	if(mh->rhdr.type == Rerror){
		if(m->mntpt)
			errors(mh->rhdr.ename);
		error(Eshutdown);
	}
	if(mh->rhdr.type != mh->thdr.type+1){
		print("type mismatch %d %d\n", mh->rhdr.type, mh->thdr.type+1);
		error(Ebadmsg);
	}
	if(mh->rhdr.fid != mh->thdr.fid){
		print("fid mismatch %d %d type %d\n", mh->rhdr.fid, mh->thdr.fid, mh->rhdr.type);
		error(Ebadmsg);
	}
}

	/*
	 * Copy out on read
	 */
	if(mh->thdr.type == Tread)
		memmove(mh->thdr.data, mh->rhdr.data, mh->rhdr.count);
	mbfree(mh->mbr);
	mh->mbr = 0;
	mbfree(mbw);
	poperror();		/* 1 */
	return;
int
mntflush(Mnt *m, Mntrpc *r)
{
	Fcall flush;
	int n;

    Normal:
#endif
	incref(q);
	qlock(q);
	qlocked = 1;
	if(waserror()){		/* 2 */
		if(qlocked)
			qunlock(q);
		mqfree(q);
		nexterror();
	}
	mh->readreply = 0;
	mh->active = 1;
	if((*devtab[q->msg->type].write)(q->msg, mbw->buf, n, 0) != n){
		print("short write in mntxmit\n");
		error(Eshortmsg);
	r->flushtag++;
	if((r->flushtag-r->flushbase) == Flushspace)
		r->flushtag -= Flushspace;

	flush.type = Tflush;
	flush.tag = r->flushtag;
	flush.oldtag = r->request.tag;
	n = convS2M(&flush, r->flush);

	if(waserror()) {
		if(strcmp(u->error, errstrtab[Eintr]) == 0)
			return 1;
		mntqrm(m, r);
		return 0;
	}
	if(q->reader == 0){		/* i will read */
		q->reader = u->p;
    Read:
		USED(qlocked);
		qunlock(q);
		qlocked = 0;
		if(waserror()){		/* 3 */
			mnterrdequeue(m, mh);
			nexterror();
		}
		mh->mbr = mballoc();
		do{
			n = (*devtab[q->msg->type].read)(q->msg, mh->mbr->buf, BUFSIZE, 0);
		}while(n == 0);
		poperror();		/* 3 */
		if(convM2S(mh->mbr->buf, &mh->rhdr, n) == 0){
			/* BUG? IS THIS RIGHT? IGNORE AND RETRY */
			print(" MR ");
			qlock(q);
			qlocked = 1;
			goto FreeRead;
		}
		/*
		 * Response might not be mine
		 */
		USED(qlocked);
		qlock(q);
		qlocked = 1;
		tag = mh->rhdr.tag;
		if(tag == mh->thdr.tag){	/* it's mine */
			if(mh->rhdr.type != Rerror)
			if(mh->rhdr.type != mh->thdr.type+1){
				print("mail rob: '%s xxT(%d)%c %d %d'\n", u->p->text,
					tag, devchar[m->q->msg->type],
					mh->rhdr.type, mh->thdr.type+1);
				goto FreeRead;
			}
			q->reader = 0;
			if(w = q->writer){	/* advance a writer to reader */
				mntwunlink(q, w);
				q->reader = w->p;
				w->readreply = 1;
				wakeup(&w->r);
			}
			mh->active = 0;
			USED(qlocked);
			qunlock(q);
			qlocked = 0;
			goto Respond;
		}
		/*
		 * Hand response to correct recipient
		 */
		if(tag==0 || tag>=NTAG){
			print("unknown tag %d\n", tag);
	FreeRead:
			mbfree(mh->mbr);
			mh->mbr = 0;
			goto Read;
		}
		/*
		 * Find writer in queue
		 */
		for(w=q->writer; w; w=w->next)
			if(w->thdr.tag == tag)
				goto Inqueue;
		goto FreeRead;
	Inqueue:
		if(w->flushing || !w->active)	/* nothing to do; mntflush will clean up */
			goto FreeRead;
		if(mh->rhdr.type != Rerror)
		if(mh->rhdr.type != w->thdr.type+1){
			print("mail rob: '%s xxw(%d)%c %d %d'\n",
				u->p->text, tag, devchar[m->q->msg->type],
				mh->rhdr.type, w->thdr.type+1);
			goto FreeRead;
	(*devtab[m->c->type].write)(m->c, r->flush, n, 0);
	poperror();
	lock(m);
	if(!r->done)
		r->flushed = 1;
	unlock(m);
	return 1;
}

Mntrpc *
mntralloc(void)
{
	Mntrpc *new;

	for(;;) {
		lock(&mntalloc);
		if(new = mntalloc.rpcfree) {
			mntalloc.rpcfree = new->list;
			unlock(&mntalloc);
			new->done = 0;
			new->bfree = 0;
			return new;
		}
		w->mbr = mh->mbr;
		mh->mbr = 0;
		memmove(&w->rhdr, &mh->rhdr, sizeof mh->rhdr);
		mntwunlink(q, w);
		w->readreply = 1;
		wakeup(&w->r);
		goto Read;
	}else{
		mh->p = u->p;
		/* put self in queue */
		mh->next = q->writer;
		mh->prev = 0;
		if(q->writer)
			q->writer->prev = mh;
		q->writer = mh;
		qunlock(q);
		qlocked = 0;
		if(waserror()){		/* interrupted sleep */
			mnterrdequeue(m, mh);
			nexterror();
		unlock(&mntalloc);
		resrcwait("no mount buffers");
	}
}

void
mntfree(Mntrpc *r)
{
	Mntrpc *q;
	Mnt *m, *e;
	int i;

	r->bfree = 1;
	if(r->flushed)
		return;

	lock(&mntalloc);
	r->list = mntalloc.rpcfree;
	mntalloc.rpcfree = r;
	unlock(&mntalloc);
}

void
mntqrm(Mnt *m, Mntrpc *r)
{
	Mntrpc **l, *f;

	lock(m);
	r->done = 1;
	r->flushed = 0;

	l = &m->queue;
	for(f = *l; f; f = f->list) {
		if(f == r) {
			*l = r->list;
			break;
		}
		sleep(&mh->r, mntreadreply, mh);
		poperror();
		USED(qlocked);
		qlock(q);
		qlocked = 1;
		mh->readreply = 0;
		if(q->reader == u->p)	/* i got promoted */
			goto Read;
		mh->active = 0;
		USED(qlocked);
		qunlock(q);
		qlocked = 0;
		goto Respond;
		l = &f->list;
	}
	unlock(m);
}

Mnt *
mntchk(Chan *c)
{
	Mnt *m;

    Respond:
	mqfree(q);
	poperror();		/* 2 */
	if(mh->rhdr.type == Rerror){
		if(m->mntpt)
			errors(mh->rhdr.ename);
	m = &mntalloc.mntarena[c->mntindex];
	if(m->id != c->dev)
		error(Eshutdown);
	}
	/*
	 * Copy out on read
	 */
	if(mh->thdr.type == Tread){
		if(mh->rhdr.count > mh->thdr.count)
			error(Ebadcnt);
		memmove(mh->thdr.data, mh->rhdr.data, mh->rhdr.count);
	}
	mbfree(mh->mbr);
	mh->mbr = 0;
	mbfree(mbw);
	USED(qlocked);
	poperror();		/* 1 */
	return m;
}

void
mntdump(void)
mntdirfix(uchar *dirbuf, Chan *c)
{
	int i;
	MntQ *q;
	Mnthdr *h;
	Proc *p;
	dirbuf[DIRLEN-4] = devchar[c->type];
	dirbuf[DIRLEN-3] = 0;
	dirbuf[DIRLEN-2] = c->dev;
	dirbuf[DIRLEN-1] = c->dev>>8;
}

	for(i=0; i<conf.nmntdev; i++){
		q = &mntqalloc.arena[i];
		if(!q->msg)
			continue;
		p = q->reader;
		print("q rdr %d wrtr ", p? p->pid : 0);
		for(h=q->writer; h; h=h->next)
			print("(%lux %lux %d %d)", h, &h->r, h->thdr.tag,
				(p=h->p)? p->pid : 0);
		print("\n");
	}
int
rpcattn(Mntrpc *r)
{
	return r->done || r->m->rip == 0;
}

void
mntdump(void)
{
}


M port/stream.c => port/stream.c +0 -2
@@ 247,8 247,6 @@ freeb(Block *bp)
		nbp = bp->next;
		bp->next = 0;
		unlock(bcp);
		if(bcp->r.p)
			wakeup(&bcp->r);
	}
}