4 files changed, 54 insertions(+), 5 deletions(-)

M port/devmnt.c
M port/devnonet.c
M port/portdat.h
M port/stream.c

@@ 784,9 784,12 @@ mntxmit(Mnt *m, Mnthdr *mh)
 			nexterror();
 		}
 		mh->mbr = mballoc();
-		n = (*devtab[q->msg->type].read)(q->msg, mh->mbr->buf, BUFSIZE, 0);
+		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){
+print("POO bad len %d %ux!\n", n, mh->mbr->buf[0]);
 			mnterrdequeue(m, mh);
 			error(Ebadmsg);
 		}


@@ 868,6 871,9 @@ mntxmit(Mnt *m, Mnthdr *mh)
 		error(Eshutdown);
 	}else if(mh->rhdr.type != mh->thdr.type+1){
 		print("bad type %d not %d in mntxmit\n", mh->rhdr.type, mh->thdr.type+1);
+/*XXX*/		print("chan %c %d %lux %lux\n", devchar[m->q->msg->type],
+				m->q->msg->dev, m->q->msg->qid.path,
+				m->q->msg->stream);	
 		error(Ebadmsg);
 	}
 	/*

@@ 540,6 540,10 @@ noiput(Queue *q, Block *bp)
 /*
  *  queue a block
  */
+enum {
+	Window=	1,
+};
+
 static int
 windowopen(void *a)
 {


@@ 548,9 552,9 @@ windowopen(void *a)
 
 	cp = (Noconv *)a;
 	i = cp->next - cp->first;
-	if(i>=0 && i<32)	
+	if(i>=0 && i<Window)	
 		return 1;
-	if(i<0 && Nnomsg+i<32)
+	if(i<0 && Nnomsg+i<Window)
 		return 1;
 	return 0;	
 }


@@ 1355,6 1359,7 @@ nonetcksum(Block *bp, int offset)
 	int n;
 	ulong s;
 	Nohdr *hp;
+	Block *first;
 
 	s = 0;
 	p = bp->rptr + offset;


@@ 1390,7 1395,20 @@ nonetcksum(Block *bp, int offset)
 	s = (s&0xffff) + (s>>16);
 	hp->sum[1] = s>>8;
 	hp->sum[0] = s;
-	return s & 0xffff;
+	s &= 0xffff;
+	switch(s){
+	case 0xac9f:
+	case 0xc1a4:
+	case 0xc41c:
+	case 0xc46d:
+		{ int i;
+		print("%lux s,", s);
+		for(bp = first; bp; bp = bp->next)
+			for(i = 0; i < BLEN(bp); i++)
+				print(" %ux", bp->rptr[i]);
+		}
+	}
+	return s;
 }
 
 /*

@@ 409,6 409,7 @@ struct Stream {
 	Queue	*procq;		/* write queue at process end */
 	Queue	*devq;		/* read queue at device end */
 	Block	*err;		/* error message from down stream */
+	int	forcedelim;	/* force a delimiter before the next message */
 };
 
 /*

@@ 1010,6 1010,9 @@ stputq(Queue *q, Block *bp)
 		wakeup(q->other->rp);
 		delim = 1;
 	} else {
+		if(BLEN(bp) == 1 && *(bp->rptr) == 1)
+			print("stputq u->p->pid %d %s\n", u->p->pid, u->p->text);
+
 		lock(q);
 		if(q->first)
 			q->last->next = bp;


@@ 1268,6 1271,9 @@ streamwrite(Chan *c, void *a, long n, int docopy)
 
 	s = c->stream;
 
+	if(n == 1 && *((char*)a) == 1)
+		print("u->p->pid %d %s\n", u->p->pid, u->p->text);
+
 	/*
 	 *  decode the qid
 	 */


@@ 1285,7 1291,24 @@ streamwrite(Chan *c, void *a, long n, int docopy)
 			error(Ehungup);
 	}
 
-	if(!docopy && isphys(a)){
+	/*
+	 *  if an error occurs during write n,
+	 *  force a delim before write n+1
+	 */
+	if(waserror()){
+		s->forcedelim = 1;
+		nexterror();
+	}
+	if(s->forcedelim){
+		FLOWCTL(q);
+		bp = allocb(0);
+		bp->flags |= S_DELIM;
+		bp->type = M_DATA;
+		PUTNEXT(q, bp);
+		s->forcedelim = 0;
+	}
+
+	if(0 && !docopy && isphys(a)){
 		/*
 		 *  `a' is global to the whole system, just create a
 		 *  pointer to it and pass it on.


@@ 1322,6 1345,7 @@ streamwrite(Chan *c, void *a, long n, int docopy)
 			}
 		}
 	}
+	poperror();
 	return n;
 }