From 2c89fa49d93b3af66f22725aab4e4e6a7d3f1dd1 Mon Sep 17 00:00:00 2001
From: David du Colombier <0intro@gmail.com>
Date: Thu, 5 Mar 1992 00:00:00 +0000
Subject: [PATCH] Plan 9 from Bell Labs 1992-03-05

---
 port/devdk.c   |  5 ++-
 port/devmnt.c  | 98 ++++++++++++++++++--------------------------------
 port/devpipe.c |  4 +--
 port/portdat.h |  2 +-
 port/portfns.h |  4 +--
 port/stasync.c |  3 +-
 port/stream.c  | 23 ++++++------
 7 files changed, 55 insertions(+), 84 deletions(-)

diff --git a/port/devdk.c b/port/devdk.c
index 9bb0a5cd8692959b215c3c4e9b63ec3f55887523..f3bb7ccb1cb7a864dabc066bf88cfb53cdb366e5 100644
--- a/port/devdk.c
+++ b/port/devdk.c
@@ -104,9 +104,9 @@ struct Line {
  */
 struct Dk {
 	QLock;
+	Lock;
 	Chan	*csc;
 
-	Lock;
 	int	ref;
 	int	opened;
 
@@ -588,8 +588,7 @@ dkoput(Queue *q, Block *bp)
 	bp->rptr[0] = line;
 	bp->rptr[1] = line>>8;
 
-	FLOWCTL(dp->wq);
-	PUTNEXT(dp->wq, bp);
+	FLOWCTL(dp->wq, bp);
 }
 
 /*
diff --git a/port/devmnt.c b/port/devmnt.c
index 9ecd15517e043a4fc452b3b073cda11acec981f5..86ba9d04aff794ec2aabde8d66fe6b81ae9fa678 100644
--- a/port/devmnt.c
+++ b/port/devmnt.c
@@ -19,10 +19,8 @@ struct Mntrpc
 	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 */
+	ushort	flushtag;	/* Tag flush sent on */
 	char	flush[MAXMSG];	/* Somewhere to build flush */
 };
 
@@ -36,6 +34,8 @@ struct Mnt
 	Mnt	*list;		/* Free list */
 	char	mux;		/* Set if the device aleady does the multiplexing */
 	int	blocksize;	/* read/write block size */
+	ushort	flushtag;	/* Tag to send flush on */
+	ushort	flushbase;	/* Base tag of flush window for this buffer */
 };
 
 struct Mntalloc
@@ -71,8 +71,7 @@ void	mntdoclunk(Mnt *, Mntrpc *);
 enum
 {
 	Tagspace = 1,
-	Flushspace = 64,
-	Flushtag = 512,
+	Tagend = 0xfffe,
 
 	ALIGN = 256,		/* Vme block mode alignment */
 };
@@ -82,24 +81,14 @@ mntreset(void)
 {
 	Mnt *me, *md;
 	Mntrpc *re, *rd;
-	ushort tag, ftag;
+	ushort tag;
 	ulong p;
 	int i;
 
 	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);
 	mntalloc.rpcarena = mntalloc.rpcfree;
 	re = &mntalloc.rpcfree[conf.nmntbuf];
@@ -112,15 +101,19 @@ mntreset(void)
 	i = MAXRPC+(ALIGN-1);
 	i &= ~(ALIGN-1);
 
+	tag = Tagspace;
 	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(i, 0);
 	}
 	re[-1].list = 0;
+	for(md = mntalloc.mntfree; md < me; md++){
+		md->list = md+1;
+		md->flushbase = tag;
+		md->flushtag = tag;
+	}
+	me[-1].list = 0;
 
 	mntalloc.id = 1;
 }
@@ -491,6 +484,7 @@ void
 mountrpc(Mnt *m, Mntrpc *r)
 {
 	r->reply.tag = 0;		/* safety check */
+	r->reply.type = 4;		/* safety check */
 	mountio(m, r);
 	if(r->reply.type == Rerror)
 		error(r->reply.ename);
@@ -500,6 +494,7 @@ mountrpc(Mnt *m, Mntrpc *r)
 	if(r->reply.type != r->request.type+1) {
 		print("devmnt: mismatched reply 0x%lux T%d R%d tags req %d fls %d rep %d\n",
 		r, r->request.type, r->reply.type, r->request.tag, r->flushtag, r->reply.tag);
+		mntdump();
 		error(Emountrpc);
 	}
 }
@@ -510,6 +505,7 @@ mountio(Mnt *m, Mntrpc *r)
 	int n;
 
 	lock(m);
+	r->flushed = 0;
 	r->m = m;
 	r->list = m->queue;
 	m->queue = r;
@@ -625,44 +621,29 @@ void
 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;
-		}
-		if(q->flushtag == r->reply.tag) {
+		if(q->request.tag == r->reply.tag
+		|| q->flushed && q->flushtag == r->reply.tag) {
 			*l = q->list;
-			q->flushed = 0;
-			done = q->done;
-			q->done = 1;
 			unlock(m);
-			if(done == 0)
-				goto dispatch;
-			if(q->bfree)
-				mntfree(q);
+			if(q != r) {		/* Completed someone else */
+				dp = q->rpc;
+				q->rpc = r->rpc;
+				r->rpc = dp;
+				memmove(&q->reply, &r->reply, sizeof(Fcall));
+				q->done = 1;
+				wakeup(&q->r);
+			}else
+				q->done = 1;
 			return;
 		}
 		l = &q->list;
 	}
 	unlock(m);
-	return;
-
-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);
-	}
 }
 
 int
@@ -671,9 +652,12 @@ mntflush(Mnt *m, Mntrpc *r)
 	Fcall flush;
 	int n;
 
-	r->flushtag++;
-	if((r->flushtag-r->flushbase) == Flushspace)
-		r->flushtag -= Flushspace;
+	lock(m);
+	r->flushtag = m->flushtag++;
+	if(m->flushtag == Tagend)
+		m->flushtag = m->flushbase;
+	r->flushed = 1;
+	unlock(m);
 
 	flush.type = Tflush;
 	flush.tag = r->flushtag;
@@ -696,10 +680,6 @@ mntflush(Mnt *m, Mntrpc *r)
 		qunlock(&m->c->wrl);
 	}
 	poperror();
-	lock(m);
-	if(!r->done)
-		r->flushed = 1;
-	unlock(m);
 	return 1;
 }
 
@@ -714,7 +694,7 @@ mntralloc(void)
 			mntalloc.rpcfree = new->list;
 			unlock(&mntalloc);
 			new->done = 0;
-			new->bfree = 0;
+			new->flushed = 0;
 			return new;
 		}
 		unlock(&mntalloc);
@@ -725,14 +705,6 @@ mntralloc(void)
 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;
@@ -803,10 +775,10 @@ mntdump(void)
 	print("rpcfree 0x%lux\n", mntalloc.rpcfree);
 	re = &mntalloc.rpcarena[conf.nmntbuf];
 	for(r = mntalloc.rpcarena; r < re; r++) 
-		print("%.8lux %.8lux T%d R%d tags req %d fls %d rep %d d %d b %d f %d\n",
+		print("%.8lux %.8lux T%d R%d tags req %d fls %d rep %d d %d f %d\n",
 			r, r->list, r->request.type, r->reply.type,
 			r->request.tag, r->flushtag, r->reply.tag, 
-			r->done, r->bfree, r->flushed);
+			r->done, r->flushed);
 
 }
 
diff --git a/port/devpipe.c b/port/devpipe.c
index dff38d9ca521e8683234ec5566b8176aaf243884..7aeef8e4387246b269db49e868716e94ef9c4648 100644
--- a/port/devpipe.c
+++ b/port/devpipe.c
@@ -276,9 +276,7 @@ pipewrite(Chan *c, void *va, long n, ulong offset)
 static void
 pipeiput(Queue *q, Block *bp)
 {
-	if(bp->type != M_HANGUP)
-		FLOWCTL(q);
-	PUTNEXT(q, bp);
+	FLOWCTL(q, bp);
 }
 
 /*
diff --git a/port/portdat.h b/port/portdat.h
index bcf63e2902ceaed03579eec342393880b0328fee..6f84dc56e8bccc50557fdd0e9578636a3aeac738 100644
--- a/port/portdat.h
+++ b/port/portdat.h
@@ -626,7 +626,7 @@ struct Stream {
 #define PUTNEXT(q,b)	(*(q)->next->put)((q)->next, b)
 #define BLEN(b)		((b)->wptr - (b)->rptr)
 #define QFULL(q)	((q)->flag & QHIWAT)
-#define FLOWCTL(q)	{ if(QFULL(q->next)) flowctl(q); }
+#define FLOWCTL(q,b)	{ if(QFULL(q->next)) flowctl(q,b); else PUTNEXT(q,b);}
 
 /*
  *  stream file qid's & high water mark
diff --git a/port/portfns.h b/port/portfns.h
index f45bc6dec17a3ad28d8856b7c0ef7d800f3f7aa1..9920f254fa2c114cc6ee6185b088baa6ffa3b2fb 100644
--- a/port/portfns.h
+++ b/port/portfns.h
@@ -62,13 +62,13 @@ void	error(char*);
 long	execregs(ulong, ulong, ulong);
 void	exhausted(char*);
 void	exit(void);
-Block*	expandb(Block *, int);
+Block*	expandb(Block*, int);
 int	fault(ulong, int);
 void	fdclose(int, int);
 Chan*	fdtochan(int, int, int);
 void	firmware(void);
 int	fixfault(Segment*, ulong, int, int);
-void	flowctl(Queue*);
+void	flowctl(Queue*, Block*);
 void	flushmmu(void);
 void	freealarm(Alarm*);
 void	freeb(Block*);
diff --git a/port/stasync.c b/port/stasync.c
index 818e03ff7ca59eafc467626bcd6786764933b411..c335e3b7fb8725157938c0528fc18486ddf7122a 100644
--- a/port/stasync.c
+++ b/port/stasync.c
@@ -173,8 +173,7 @@ aswrite(Async *ap)
 {
 	if(ap->bp->rptr == ap->bp->wptr)
 		return;
-	FLOWCTL(ap->wq);
-	PUTNEXT(ap->wq, ap->bp);
+	FLOWCTL(ap->wq, ap->bp);
 	ap->bp = 0;
 }
 
diff --git a/port/stream.c b/port/stream.c
index 37282a91964d23f23108d544147401928615a0f9..269c1a9b1d5c95c52a9a953ea1ddc228763a56ac 100644
--- a/port/stream.c
+++ b/port/stream.c
@@ -1286,17 +1286,21 @@ notfull(void *arg)
 	return !QFULL((Queue *)arg);
 }
 void
-flowctl(Queue *q)
+flowctl(Queue *q, Block *bp)
 {
-	qlock(&q->rlock);
-	if(waserror()){
+	if(bp->type != M_HANGUP){
+		qlock(&q->rlock);
+		if(waserror()){
+			qunlock(&q->rlock);
+			freeb(bp);
+			nexterror();
+		}
+		q->rp = &q->r;
+		sleep(q->rp, notfull, q->next);
 		qunlock(&q->rlock);
-		nexterror();
+		poperror();
 	}
-	q->rp = &q->r;
-	sleep(q->rp, notfull, q->next);
-	qunlock(&q->rlock);
-	poperror();
+	PUTNEXT(q, bp);
 }
 
 /*
@@ -1356,8 +1360,7 @@ streamwrite(Chan *c, void *a, long n, int docopy)
 	 *  send it down stream
 	 */
 	last->flags |= S_DELIM;
-	FLOWCTL(q);
-	PUTNEXT(q, first);
+	FLOWCTL(q, first);
 	return n;
 }