6 files changed, 104 insertions(+), 43 deletions(-)

M gnot/dat.h
M gnot/devmnt.c
M gnot/devpipe.c
M gnot/fns.h
M gnot/stream.c
M port/devcons.c

@@ 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)

@@ 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();
 		}

@@ 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;
-	}
 }

@@ 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);

@@ 258,6 258,22 @@ allocq(Qinfo *qi)
 }
 
 /*
+ *  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
  */
 static void


@@ 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;


@@ 712,6 729,54 @@ streamopen(Chan *c, Qinfo *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);
 }
 

@@ 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