From 4299e7bc9840e5d23443ec4e5f7aa091868cbf6e Mon Sep 17 00:00:00 2001
From: David du Colombier <0intro@gmail.com>
Date: Wed, 20 Jun 1990 00:00:00 +0000
Subject: [PATCH] Plan 9 from Bell Labs 1990-06-20

---
 gnot/dat.h     |  4 +--
 gnot/devmnt.c  |  3 --
 gnot/devpipe.c | 51 +++++++++++++-----------------
 gnot/fns.h     |  2 ++
 gnot/stream.c  | 85 +++++++++++++++++++++++++++++++++++++++++++++-----
 port/devcons.c |  2 +-
 6 files changed, 104 insertions(+), 43 deletions(-)

diff --git a/gnot/dat.h b/gnot/dat.h
index 877b06ad7eda171fd1a25aa57a238e50809cc5bb..c33f5f950d4b3cd494303e775b2c10c5fd0e4843 100644
--- a/gnot/dat.h
+++ b/gnot/dat.h
@@ -421,7 +421,8 @@ struct Queue {
  */
 struct Stream {
 	Lock;			/* structure lock */
-	int	inuse;		/* use count */
+	int	inuse;		/* number of processes in stream */
+	int	opens;		/* number of processes with stream open */
 	int	hread;		/* number of reads after hangup */
 	int	type;		/* correclation with Chan */
 	int	dev;		/* ... */
@@ -430,7 +431,6 @@ struct Stream {
 	QLock	wrlock;		/* write lock */
 	Queue	*procq;		/* write queue at process end */
 	Queue	*devq;		/* read queue at device end */
-	char	tag[32];	/* when reading the tag qid */
 };
 #define	RD(q)		((q)->other < (q) ? (q->other) : q)
 #define	WR(q)		((q)->other > (q) ? (q->other) : q)
diff --git a/gnot/devmnt.c b/gnot/devmnt.c
index 18018b9c196ad1df3fcbb95c62f2bba97cf60557..4d9b2c6db0f23f838ce65c1cf9ff478b06d82ee6 100644
--- a/gnot/devmnt.c
+++ b/gnot/devmnt.c
@@ -771,7 +771,6 @@ mntxmit(Mnt *m, Mnthdr *mh)
 		qlocked = 0;
 		n = (*devtab[q->msg->type].read)(q->msg, mh->mbr->buf, BUFSIZE);
 		if(convM2S(mh->mbr->buf, &mh->rhdr, n) == 0){
-			print("format error in mntxmit\n");
 			mnterrdequeue(q, mh);
 			error(0, Ebadmsg);
 		}
@@ -795,7 +794,6 @@ mntxmit(Mnt *m, Mnthdr *mh)
 		/*
 		 * Hand response to correct recipient
 		 */
-		if(q->writer==0) print("response with empty queue\n");
 		for(ow=0,w=q->writer; w; ow=w,w=w->next)
 			if(mh->rhdr.fid == w->thdr.fid
 			&& mh->rhdr.type == w->thdr.type+1){
@@ -821,7 +819,6 @@ mntxmit(Mnt *m, Mnthdr *mh)
 		qunlock(q);
 		qlocked = 0;
 		if(waserror()){		/* interrupted sleep */
-			print("interrupted i/o\n");
 			mnterrdequeue(q, mh);
 			nexterror();
 		}
diff --git a/gnot/devpipe.c b/gnot/devpipe.c
index 2b960dedca066ece5ed18494ff9ec038a4a79cf2..e14f44c0e06dfa5cf93b85171a3c8b9a0c4c3fb1 100644
--- a/gnot/devpipe.c
+++ b/gnot/devpipe.c
@@ -32,6 +32,8 @@ Chan*
 pipeattach(char *spec)
 {
 	Chan *c;
+	int i;
+
 	/*
 	 *  make the first stream
 	 */
@@ -59,8 +61,17 @@ pipeclone(Chan *c, Chan *nc)
 	/*
 	 *  attach it to the first
 	 */
+	c->stream->devq->ptr = (Stream *)nc->stream;
+	nc->stream->devq->ptr = (Stream *)c->stream;
 	c->stream->devq->other->next = nc->stream->devq;
 	nc->stream->devq->other->next = c->stream->devq;
+
+	/*
+	 *  up the inuse count of each stream to reflect the
+	 *  pointer from the other stream.
+	 */
+	streamenter(c->stream);
+	streamenter(nc->stream);
 	return nc;
 }
 
@@ -110,7 +121,16 @@ pipewstat(Chan *c, char *db)
 void
 pipeclose(Chan *c)
 {
-	streamclose(c);
+	Stream *other;
+
+	other = (Stream *)c->stream->devq->ptr;
+
+	if(waserror()){
+		streamexit(other, 0);
+		nexterror();
+	}
+	streamclose(c);		/* close this stream */
+	streamexit(other, 0);	/* release stream for other half of pipe */
 }
 
 long
@@ -164,14 +184,7 @@ pipeiput(Queue *q, Block *bp)
 static void
 pipeoput(Queue *q, Block *bp)
 {
-	lock(q);
-	if(q->next)
-		pipeiput(q->next, bp);
-	else{
-		print("pipeoput losing block\n");
-		freeb(bp);
-	}
-	unlock(q);
+	PUTNEXT(q, bp);
 }
 
 /*
@@ -193,29 +206,9 @@ pipestclose(Queue *q)
 	 *  send a hangup
 	 */
 	q = q->other;
-	lock(q);
 	if(q->next){
 		bp = allocb(0);
 		bp->type = M_HANGUP;
 		pipeiput(q->next, bp);
 	}
-	unlock(q);
-
-	/*
-	 *  disconnect (possible livelock?)
-	 */
-	for(;;){
-		lock(q);
-		if(q->next){
-			if(!canlock(q->next->other)){
-				unlock(q);
-				continue;
-			}
-			q->next->other->next = 0;
-			unlock(q->next->other);
-			q->next = 0;
-		}
-		unlock(q);
-		break;
-	}
 }
diff --git a/gnot/fns.h b/gnot/fns.h
index a57c08fb76eb2b34b5b58e76a285f678ffcad0df..aa530dcaa06c06be00b7fb505fbf7331e905e156 100644
--- a/gnot/fns.h
+++ b/gnot/fns.h
@@ -135,6 +135,8 @@ int	spllo(void);
 void	splx(int);
 Devgen	streamgen;
 void	streamclose(Chan*);
+int	streamenter(Stream*);
+void	streamexit(Stream*, int);
 void	streaminit(void);
 long	streamread(Chan*, void*, long);
 long	streamwrite(Chan*, void*, long, int);
diff --git a/gnot/stream.c b/gnot/stream.c
index 16c8e83a0df887775e84c9bbceb5874854f9052d..c14e4a2e02269df5f9a59119b360e0954d29db3c 100644
--- a/gnot/stream.c
+++ b/gnot/stream.c
@@ -257,6 +257,22 @@ allocq(Qinfo *qi)
 	return q;
 }
 
+/*
+ *  flush a queue
+ */
+static void
+flushq(Queue *q)
+{
+	Block *bp;
+
+	q = RD(q);
+	while(bp = getq(q))
+		freeb(bp);
+	q = WR(q);
+	while(bp = getq(q))
+		freeb(bp);
+}
+
 /*
  *  free a queue
  */
@@ -658,8 +674,8 @@ streamnew(Chan *c, Qinfo *qi)
  	 *  hang a device and process q off the stream
 	 */
 	s->inuse = 1;
+	s->opens = 1;
 	s->hread = 0;
-	s->tag[0] = 0;
 	q = allocq(&procinfo);
 	s->procq = WR(q);
 	q = allocq(qi);
@@ -697,6 +713,7 @@ streamopen(Chan *c, Qinfo *qi)
 			&& s->dev == c->dev
 		 	&& s->id == STREAMID(c->qid)){
 				s->inuse++;
+				s->opens++;
 				c->stream = s;
 				unlock(s);
 				return;
@@ -711,6 +728,54 @@ streamopen(Chan *c, Qinfo *qi)
 	streamnew(c, qi);
 }
 
+/*
+ *  Enter a stream.  Increment the reference count so it can't disappear
+ *  under foot.
+ */
+int
+streamenter(Stream *s)
+{
+	lock(s);
+	if(s->opens == 0){
+		unlock(s);
+		return -1;
+	}
+	s->inuse++;
+	unlock(s);
+	return 0;
+}
+
+/*
+ *  Decrement the reference count on a stream.  If the count is
+ *  zero, free the stream.
+ */
+void
+streamexit(Stream *s, int locked)
+{
+	Queue *q;
+	Queue *nq;
+
+	if(!locked)
+		lock(s);
+	s->inuse--;
+	if(s->inuse != 0){
+		if(!locked)
+			unlock(s);
+		return;
+	}
+
+	/*
+	 *  ascend the stream freeing the queues
+	 */
+	for(q = s->devq; q; q = nq){
+		nq = q->next;
+		freeq(q);
+	}
+	s->id = s->dev = s->type = 0;
+	if(!locked)
+		unlock(s);
+}
+
 /*
  *  On the last close of a stream, for each queue on the
  *  stream release its blocks and call its close routine.
@@ -729,11 +794,11 @@ streamclose(Chan *c)
 		return;
 
 	/*
-	 *  decrement the reference cound
+	 *  decrement the reference count
 	 */
 	lock(s);
-	if(s->inuse != 1){
-		s->inuse--;
+	if(s->opens != 1){
+		s->opens--;
 		unlock(c->stream);
 		return;
 	}
@@ -747,15 +812,19 @@ streamclose(Chan *c)
 		if(q == s->devq->other)
 			break;
 	}
+
 	/*
-	 *  ascend the stream freeing the queues
+	 *  ascend the stream flushing the queues
 	 */
 	for(q = s->devq; q; q = nq){
 		nq = q->next;
-		freeq(q);
+		flushq(q);
 	}
-	s->id = s->dev = s->type = 0;
-	s->inuse--;
+
+	/*
+	 *  leave it and free it
+	 */
+	streamexit(s, 1);
 	unlock(s);
 }
 
diff --git a/port/devcons.c b/port/devcons.c
index c2575ff94c25ddfa2e9104937fbf4fde778b96a9..9f8673b48cec576916d7874ce813150e3543c64c 100644
--- a/port/devcons.c
+++ b/port/devcons.c
@@ -315,7 +315,7 @@ ulong	boottime;		/* seconds since epoch at boot */
 long
 seconds(void)
 {
-	return boottime + TK2MS(MACHP(0)->ticks);
+	return boottime + TK2SEC(MACHP(0)->ticks);
 }
 
 int