10 files changed, 90 insertions(+), 24 deletions(-)

M gnot/dat.h
M gnot/devdk.c
M gnot/devincon.c
M gnot/devpipe.c
M gnot/stream.c
M gnot/sturp.c
M port/devkprof.c
M port/stream.c
M port/sturp.c
M power/dat.h

@@ 408,6 408,7 @@ struct Queue {
 struct Stream {
 	Lock;			/* structure lock */
 	int	inuse;		/* use count */
+	int	hread;		/* number of reads after hangup */
 	int	type;		/* correclation with Chan */
 	int	dev;		/* ... */
 	int	id;		/* ... */

@@ 417,7 417,11 @@ dkoput(Queue *q, Block *bp)
 	bp->rptr[0] = line;
 	bp->rptr[1] = line>>8;
 
-	PUTNEXT(dp->wq, bp);
+	if(dp->wq->len >= Streamhi){
+		print("dkoput free\n");
+		freeb(bp);
+	} else
+		PUTNEXT(dp->wq, bp);
 }
 
 /*

@@ 243,7 243,7 @@ inconrestart(Incon *ip)
 		break;
 	case Selected:
 		print("incon[%d] station %d\n", ip-incon, ip->station);
-		inconset(ip, 8, 9);
+		inconset(ip, 3, 15);
 		break;
 	default:
 		print("incon[%d] bollixed\n", ip-incon);


@@ 264,7 264,7 @@ inconreset(void)
 	incon[0].dev = INCON;
 	incon[0].state = Selected;
 	incon[0].ri = incon[0].wi = 0;
-/*	inconset(&incon[0], 8, 9); /**/
+/*	inconset(&incon[0], 3, 15); /**/
 	for(i=1; i<Nincon; i++){
 		incon[i].dev = INCON+i;
 		incon[i].state = Dead;


@@ 625,15 625,7 @@ inconkproc(void *arg)
 		while(ip->ri != ip->wi){
 			bp = ip->inb[ip->ri];
 			n = BLEN(bp);
-			if(n <= 64){
-				nbp = allocb(n);
-				memcpy(nbp->wptr, bp->rptr, n);
-				nbp->wptr += n;
-				freeb(bp);
-				PUTNEXT(ip->rq, nbp);
-			} else {
-				PUTNEXT(ip->rq, bp);
-			}
+			PUTNEXT(ip->rq, bp);
 			bp = ip->inb[ip->ri] = allocb(Bsize);
 			bp->wptr += 3;
 			ip->ri = (ip->ri+1)%Nin;

@@ 11,7 11,7 @@
 static void pipeiput(Queue*, Block*);
 static void pipeoput(Queue*, Block*);
 static void pipestclose(Queue *);
-Qinfo pipeinfo = { pipeiput, pipeoput, 0, pipestclose, "process" };
+Qinfo pipeinfo = { pipeiput, pipeoput, 0, pipestclose, "pipe" };
 
 void
 pipeinit(void)


@@ 154,7 154,8 @@ pipeerrstr(Error *e, char *buf)
 static void
 pipeiput(Queue *q, Block *bp)
 {
-	flowctl(q);
+	if(q->next->len >= Streamhi)
+		flowctl(q);
 	PUTNEXT(q, bp);
 }
 


@@ 168,6 169,10 @@ pipeoput(Queue *q, Block *bp)
 	lock(q);
 	if(q->next)
 		pipeiput(q->next, bp);
+	else{
+		print("pipeoput losing block\n");
+		freeb(bp);
+	}
 	unlock(q);
 }
 

@@ 58,10 58,32 @@ Bclass bclass[Nclass]={
 	{ 0 },
 	{ 68 },
 	{ 260 },
-	{ 4096 },
+	{ 1024 },
 };
 
 /*
+ *  Dump all block information of how many blocks are in which queues
+ */
+void
+dumpqueues(void)
+{
+	Queue *q;
+	int count;
+	Block *bp;
+
+	for(q = qlist; q < qlist + conf.nqueue; q++, q++){
+		if(!(q->flag & QINUSE))
+			continue;
+		for(count = 0, bp = q->first; bp; bp = bp->next)
+			count++;
+		print("%s %ux  RD count %d len %d", q->info->name, q, count, q->len);
+		for(count = 0, bp = WR(q)->first; bp; bp = bp->next)
+			count++;
+		print("  WR count %d len %d\n", count, WR(q)->len);
+	}
+}
+
+/*
  *  Allocate streams, queues, and blocks.  Allocate n block classes with
  *	1/2(m+1) to class m < n-1
  *	1/2(n-1) to class n-1


@@ 123,8 145,11 @@ allocb(ulong size)
 	lock(bcp);
 	while(bcp->first == 0){
 		unlock(bcp);
-		if(loop++ > 10)
+		if(loop++ > 10){
+			dumpqueues();
+			dumpstack();
 			panic("waiting for blocks\n");
+		}
 		qlock(bcp);
 		tsleep(&bcp->r, isblock, (void *)bcp, 250);
 		qunlock(bcp);


@@ 614,6 639,7 @@ streamnew(Chan *c, Qinfo *qi)
  	 *  hang a device and process q off the stream
 	 */
 	s->inuse = 1;
+	s->hread = 0;
 	s->tag[0] = 0;
 	q = allocq(&procinfo);
 	s->procq = WR(q);


@@ 825,8 851,12 @@ streamread(Chan *c, void *vbuf, long n)
 	while(left){
 		bp = getq(q);
 		if(bp == 0){
-			if(q->flag & QHUNGUP)
-				break;
+			if(q->flag & QHUNGUP){
+				if(s->hread++ < 3)
+					break;
+				else
+					error(0, Ehungup);
+			}
 			sleep(&q->r, &isinput, (void *)q);
 			continue;
 		}

@@ 63,7 63,8 @@ struct Urp {
 
 	int	kstarted;
 };
-#define WINDOW(u) ((u->unechoed + u->maxout - u->next)%8)
+#define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\
+			(u)->unechoed+(u)->maxout-(u)->next)
 #define IN(x, f, n) (f<=n ? x>=f && x<n : x<n || x>=f)
 #define NEXT(x) (((x)+1)&Nmask)
 

@@ 162,5 162,6 @@ kproftimer(ulong pc)
 		pc -= KTZERO;
 		pc >>= LRES;
 		timerbuf[pc]++;
-	}
+	} else
+		timerbuf[1]++;
 }

@@ 62,6 62,28 @@ Bclass bclass[Nclass]={
 };
 
 /*
+ *  Dump all block information of how many blocks are in which queues
+ */
+void
+dumpqueues(void)
+{
+	Queue *q;
+	int count;
+	Block *bp;
+
+	for(q = qlist; q < qlist + conf.nqueue; q++, q++){
+		if(!(q->flag & QINUSE))
+			continue;
+		for(count = 0, bp = q->first; bp; bp = bp->next)
+			count++;
+		print("%s %ux  RD count %d len %d", q->info->name, q, count, q->len);
+		for(count = 0, bp = WR(q)->first; bp; bp = bp->next)
+			count++;
+		print("  WR count %d len %d\n", count, WR(q)->len);
+	}
+}
+
+/*
  *  Allocate streams, queues, and blocks.  Allocate n block classes with
  *	1/2(m+1) to class m < n-1
  *	1/2(n-1) to class n-1


@@ 123,8 145,11 @@ allocb(ulong size)
 	lock(bcp);
 	while(bcp->first == 0){
 		unlock(bcp);
-		if(loop++ > 10)
+		if(loop++ > 10){
+			dumpqueues();
+			dumpstack();
 			panic("waiting for blocks\n");
+		}
 		qlock(bcp);
 		tsleep(&bcp->r, isblock, (void *)bcp, 250);
 		qunlock(bcp);


@@ 614,6 639,7 @@ streamnew(Chan *c, Qinfo *qi)
  	 *  hang a device and process q off the stream
 	 */
 	s->inuse = 1;
+	s->hread = 0;
 	s->tag[0] = 0;
 	q = allocq(&procinfo);
 	s->procq = WR(q);


@@ 825,8 851,12 @@ streamread(Chan *c, void *vbuf, long n)
 	while(left){
 		bp = getq(q);
 		if(bp == 0){
-			if(q->flag & QHUNGUP)
-				break;
+			if(q->flag & QHUNGUP){
+				if(s->hread++ < 3)
+					break;
+				else
+					error(0, Ehungup);
+			}
 			sleep(&q->r, &isinput, (void *)q);
 			continue;
 		}

@@ 63,7 63,8 @@ struct Urp {
 
 	int	kstarted;
 };
-#define WINDOW(u) ((u->unechoed + u->maxout - u->next)%8)
+#define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\
+			(u)->unechoed+(u)->maxout-(u)->next)
 #define IN(x, f, n) (f<=n ? x>=f && x<n : x<n || x>=f)
 #define NEXT(x) (((x)+1)&Nmask)
 

@@ 425,6 425,7 @@ struct Queue {
 struct Stream {
 	Lock;			/* structure lock */
 	int	inuse;		/* use count */
+	int	hread;		/* number of reads after hangup */
 	int	type;		/* correclation with Chan */
 	int	dev;		/* ... */
 	int	id;		/* ... */