16 files changed, 525 insertions(+), 125 deletions(-)

M gnot/devcons.c
M gnot/devincon.c
M gnot/fns.h
M gnot/proc.c
M gnot/screen.c
M gnot/sturp.c
M port/devbit.c
M port/devcons.c
M port/devmnt.c
M port/devpipe.c
M port/proc.c
M port/stream.c
M port/sturp.c
M power/conf.h
M power/dat.h
M power/fns.h

@@ 17,7 17,7 @@ static struct
 
 typedef struct IOQ	IOQ;
 
-#define	NQ	4096
+#define	NQ	1024
 struct IOQ{
 	union{
 		Lock;


@@ 31,6 31,8 @@ struct IOQ{
 };
 
 IOQ	lineq;
+IOQ	rs232iq;
+IOQ	rs232oq;
 
 struct{
 	IOQ;		/* qlock to getc; interrupt putc's */


@@ 50,12 52,20 @@ printinit(void)
 
 	kbdq.in = kbdq.buf;
 	kbdq.out = kbdq.buf;
+	rs232iq.in = rs232iq.buf;
+	rs232iq.out = rs232iq.buf;
+	rs232oq.in = rs232oq.buf;
+	rs232oq.out = rs232oq.buf;
 	lineq.in = lineq.buf;
 	lineq.out = lineq.buf;
 	qlock(&kbdq);		/* allocate qlock */
 	qunlock(&kbdq);
 	lock(&lineq);		/* allocate lock */
 	unlock(&lineq);
+	lock(&rs232iq);		/* allocate lock */
+	unlock(&rs232iq);
+	lock(&rs232oq);		/* allocate lock */
+	unlock(&rs232oq);
 
 	screeninit();
 }


@@ 114,6 124,14 @@ getc(IOQ *q)
 	return c;
 }
 
+void
+putc(IOQ *q, int c)
+{
+	*q->in++ = c;
+	if(q->in == q->buf+sizeof(q->buf))
+		q->in = q->buf;
+}
+
 int
 sprint(char *s, char *fmt, ...)
 {


@@ 184,12 202,15 @@ echo(int c)
 	 */
 	if(c == 0x14)
 		DEBUG();
+	if(c == 0x16){
+		incontoggle();
+		urpdump();
+		dumpqueues();
+	}
 	if(raw.ref)
 		return;
 	if(c == 0x15)
 		putstrn("^U\n", 3);
-	if(c == 0x16)
-		dumpqueues();
 	else{
 		ch = c;
 		putstrn(&ch, 1);


@@ 235,6 256,26 @@ kbdclock(void)
 		kbdchar(kbdq.c);
 }
 
+void
+rs232ichar(int c)
+{
+	*rs232iq.in++ = c;
+	if(rs232iq.in == rs232iq.buf+sizeof(rs232iq.buf))
+		rs232iq.in = rs232iq.buf;
+	wakeup(&rs232iq.r);
+}
+
+int
+getrs232o(void)
+{
+	int c;
+
+	c = getc(&rs232oq);
+	if(c == -1)
+		rs232oq.state = 0;
+	return c;
+}
+
 int
 consactive(void)
 {


@@ 253,6 294,7 @@ enum{
 	Qpid,
 	Qppid,
 	Qrcons,
+	Qrs232,
 	Qtime,
 	Quser,
 };


@@ 265,6 307,7 @@ Dirtab consdir[]={
 	"pid",		Qpid,		12,	0600,
 	"ppid",		Qppid,		12,	0600,
 	"rcons",	Qrcons,		0,	0600,
+	"rs232",	Qrs232,		0,	0600,
 	"time",		Qtime,		12,	0600,
 	"user",		Quser,		0,	0600,
 };


@@ 356,7 399,7 @@ consopen(Chan *c, int omode)
 			error(0, Eperm);
 	}
 	if(c->qid == Qrcons)
-		if(incref(&raw) == 0){
+		if(incref(&raw) == 1){
 			lock(&lineq);
 			while((ch=getc(&kbdq)) != -1){
 				*lineq.in++ = ch;


@@ 446,6 489,22 @@ consread(Chan *c, void *buf, long n)
 		qunlock(&kbdq);
 		return i;
 
+	case Qrs232:
+		qlock(&rs232iq);
+		if(waserror()){
+			qunlock(&rs232iq);
+			nexterror();
+		}
+		while(!cangetc(&rs232iq))
+			sleep(&rs232iq.r, (int(*)(void*))cangetc, &rs232iq);
+		for(i=0; i<n; i++){
+			if((ch=getc(&rs232iq)) == -1)
+				break;
+			*cbuf++ = ch;
+		}
+		qunlock(&rs232iq);
+		return i;
+
 	case Qcputime:
 		k = c->offset;
 		if(k >= sizeof tmp)


@@ 513,6 572,20 @@ conswrite(Chan *c, void *va, long n)
 		}
 		break;
 
+	case Qrs232:
+		qlock(&rs232oq);
+		l = n;
+		while(--l >= 0)
+			putc(&rs232oq, *a++);
+		splhi();
+		if(rs232oq.state == 0){
+			rs232oq.state = 1;
+			duartstartrs232o();
+		}
+		spllo();
+		qunlock(&rs232oq);
+		break;
+
 	case Qtime:
 		if(n<=0 || boottime!=0)	/* only one write please */
 			return 0;


@@ 571,22 644,3 @@ consuserstr(Error *e, char *buf)
 {
 	strcpy(buf, u->p->pgrp->user);
 }
-
-typedef struct Incon{
-	unsigned char	cdata;		unsigned char u0;
-	unsigned char	cstatus;	unsigned char u1;
-	unsigned char	creset;		unsigned char u2;
-	unsigned char	csend;		unsigned char u3;
-	unsigned short	data_cntl;	/* data is high byte, cntl is low byte */
-	unsigned char	status;		unsigned char u5;
-	unsigned char	reset;		unsigned char u6;
-	unsigned char	send;		unsigned char u7;
-}Incon;
-
-/*
-inconintr(Ureg *ur)
-{
-	int x;
-	x = ((Incon*)0x40700000)->status;
-}
-*/

@@ 133,6 133,8 @@ static void inconstopen(Queue*, Stream*);
 static void inconstclose(Queue*);
 Qinfo inconinfo = { nullput, inconoput, inconstopen, inconstclose, "incon" };
 
+int incondebug;
+
 /*
  *  set the incon parameters
  */


@@ 492,6 494,9 @@ inconoput(Queue *q, Block *bp)
 	ctl = bp->rptr[2];
 	bp->rptr += 3;
 
+	if(incondebug)
+		print("->(%d)%uo %d\n", chan, ctl, bp->wptr - bp->rptr);
+
 	/*
 	 *  make sure there's an incon out there
 	 */


@@ 677,6 682,8 @@ nextin(Incon *ip, unsigned int c)
 	bp->base[0] = ip->chan;
 	bp->base[1] = ip->chan>>8;
 	bp->base[2] = c;
+	if(incondebug)
+		print("<-(%d)%uo %d\n", ip->chan, c, bp->wptr-bp->rptr)-3;
 
 	next = (ip->wi+3)%Nin;
 	if(next == ip->ri){


@@ 808,3 815,8 @@ inconintr(Ureg *ur)
 		ip->state = Dead;
 	}
 }
+
+incontoggle()
+{
+	incondebug ^= 1;
+}

@@ 31,6 31,7 @@ int	devno(int, int);
 Chan*	devopen(Chan*, int, Dirtab*, int, Devgen*);
 void	devstat(Chan*, char*, Dirtab*, int, Devgen*);
 int	devwalk(Chan*, char*, Dirtab*, int, Devgen*);
+void	duartstartrs232o(void);
 void	duartstarttimer(void);
 void	duartstoptimer(void);
 void	dumpregs(Ureg*);


@@ 54,6 55,7 @@ void	freealarm(Alarm*);
 Block	*getb(Blist*);
 int	getfields(char*, char**, int, char);
 Block	*getq(Queue*);
+int	getrs232o(void);
 void	gotolabel(Label*);
 void	growpte(Orig*, ulong);
 void	*ialloc(ulong, int);


@@ 114,6 116,7 @@ int	readnum(ulong, char*, ulong, ulong, int);
 void	ready(Proc*);
 int	return0(void*);
 void	rooterrstr(Error*, char*);
+void	rs232ichar(int);
 void	qlock(QLock*);
 void	qunlock(QLock*);
 void	restartprint(Alarm*);

@@ 559,9 559,9 @@ DEBUG()
 	for(i=0; i<conf.nproc; i++){
 		p = procalloc.arena+i;
 		if(p->state != Dead)
-			print("%d:%s upc %lux %s ut %ld st %ld %lux\n",
+			print("%d:%s upc %lux %s ut %ld st %ld q %lux r %lux\n",
 				p->pid, p->text, p->pc, statename[p->state],
-				p->time[0], p->time[1], p->qlock);
+				p->time[0], p->time[1], p->qlock, p->r);
 	}
 }
 

@@ 183,7 183,7 @@ duartinit(void)
 	duart[0].sr_csr = BD4800;
 
 	/*
-	 * Pen
+	 * RS232
 	 */
 	duart[1].cmnd = RESET_RCV|DIS_TX|DIS_RX;
 	duart[1].cmnd = RESET_TRANS;


@@ 191,7 191,7 @@ duartinit(void)
 	duart[1].cmnd = RESET_MR;
 	duart[1].mr1_2 = CHAR_ERR|NO_PAR|CBITS8;
 	duart[1].mr1_2 = NORM_OP|ONESTOPB;
-	duart[1].sr_csr = BD2400;
+	duart[1].sr_csr = BD9600;
 
 	/*
 	 * Output port


@@ 240,6 240,27 @@ duartstoptimer(void)
 }
 
 void
+duartrs232intr(void)
+{
+	int c;
+	Duart *duart;
+
+	duart = DUARTREG;
+	c = getrs232o();
+	if(c == -1)
+		duart[1].cmnd = DIS_TX;
+	else
+		duart[1].data = c;
+}
+
+void
+duartstartrs232o(void)
+{
+	DUARTREG[1].cmnd = ENB_TX;
+	duartrs232intr();
+}
+
+void
 duartintr(Ureg *ur)
 {
 	int cause, status, c;


@@ 282,14 303,19 @@ duartintr(Ureg *ur)
 	/*
 	 * Is it 2?
 	 */
-	if(cause & IM_RRDYB)		/* pen input */
+	if(cause & IM_RRDYB){		/* rs232 input */
+		status = duart[1].sr_csr;
 		c = duart[1].data;
+		if(status & (FRM_ERR|OVR_ERR|PAR_ERR))
+			duart[1].cmnd = RESET_ERR;
+		else
+			rs232ichar(c);
+	}
 	/*
 	 * Is it 3?
 	 */
-	if(cause & IM_XRDYB){
-		duart[1].cmnd = DIS_TX;
-	}
+	if(cause & IM_XRDYB)		/* rs232 output */
+		duartrs232intr();
 	/*
 	 * Is it 4?
 	 */

@@ 65,7 65,7 @@ struct Urp {
 };
 #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 IN(x, f, n) (f<=n ? (x>=f && x<n) : (x<n || x>=f))
 #define NEXT(x) (((x)+1)&Nmask)
 
 /*


@@ 117,6 117,7 @@ static void	sendrej(Urp*);
 static void	initoutput(Urp*, int);
 static void	initinput(Urp*, int);
 static void	urpkproc(void *arg);
+static void	urpvomit(char*, Urp*);
 
 Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
 


@@ 584,11 585,14 @@ output(Urp *up)
 	}
 
 	/*
-	 *  fill the transmit buffers
+	 *  fill the transmit buffers, `nxb' can never overtake `unechoed'
 	 */
 	q = up->wq;
-	if(up->xb[up->nxb]==0) {
-		for(bp=getq(q); bp && up->xb[up->nxb]==0; up->nxb=NEXT(up->nxb)){
+	i = NEXT(up->nxb);
+	if(i != up->unechoed) {
+		for(bp = getq(q); bp && i!=up->unechoed; i = NEXT(i)){
+			if(up->xb[up->nxb] != 0)
+				urpvomit("output", up);
 			if(BLEN(bp) > up->maxblock){
 				nbp = up->xb[up->nxb] = allocb(0);
 				nbp->rptr = bp->rptr;


@@ 597,17 601,27 @@ output(Urp *up)
 				up->xb[up->nxb] = bp;
 				bp = getq(q);
 			}
+			up->nxb = i;
 		}
 		if(bp)
 			putbq(q, bp);
 	}
-/*	print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n",
-		WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state);
-/**/
+
 	/*
-	 *  if a retransmit time has elapsed since a transmit, send an ENQ
+	 *  retransmit cruft
 	 */
-	if(up->unechoed!=up->next && NOW>up->timer){
+	if(up->rexmit){
+		/*
+		 *  if a retransmit is requested, move next back to
+		 *  the unacked blocks
+		 */
+		up->rexmit = 0;
+		up->next = up->unacked;
+	} else if(up->unacked!=up->next && NOW>up->timer){
+		/*
+		 *  if a retransmit time has elapsed since a transmit,
+		 *  send an ENQ
+		 */
 		up->timer = NOW + MSrexmit;
 		up->state &= ~REJECTING;
 		sendctl(up, ENQ);


@@ 618,12 632,11 @@ output(Urp *up)
 
 	/*
 	 *  if there's a window open, push some blocks out
+	 *
+	 *  the lock is to synchronize with acknowledges that free
+	 *  blocks.
 	 */
-	if(up->rexmit){
-		up->rexmit = 0;
-		up->next = up->unechoed;
-	}
-	while(WINDOW(up)>0 && up->xb[up->next]!=0){
+	while(WINDOW(up)>0 && up->next!=up->nxb){
 		i = up->next;
 		qlock(&up->xl[i]);
 		if(waserror()){


@@ 722,6 735,10 @@ sendblock(Urp *up, int bn)
 	 *  message 1, the BOT and the data
 	 */
 	bp = up->xb[bn];
+	if(bp == 0){
+		urpvomit("sendblock", up);
+		return;
+	}
 	m = allocb(1);
 	m->rptr = m->lim - 1;
 	m->wptr = m->lim;


@@ 768,6 785,8 @@ rcvack(Urp *up, int msg)
 			qlock(&up->xl[i]);
 			if(up->xb[i])
 				freeb(up->xb[i]);
+			else
+				urpvomit("rcvack", up);
 			up->xb[i] = 0;
 			qunlock(&up->xl[i]);
 		}


@@ 924,3 943,31 @@ urpkproc(void *arg)
 	up->kstarted = 0;
 	DPRINT("urpkproc %ux\n", up);
 }
+
+/*
+ *  urp got very confused, complain
+ */
+static void
+urpvomit(char *msg, Urp* up)
+{
+	print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n",
+		msg, up, up->next, up->unechoed, up->unacked, up->nxb);
+	print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n",
+		up->xb[0], up->xb[1], up->xb[2], up->xb[3], up->xb[4], 
+		up->xb[5], up->xb[6], up->xb[7]);
+	print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n",
+		up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1],
+		up->trbuf[2]);
+	print("\tupq: %ux %d %d\n", up->rq->next->first,  up->rq->next->nb,
+		up->rq->next->len);
+}
+
+int
+urpdump(void)
+{
+	Urp *up;
+
+	for(up = urp; up < &urp[Nurp]; up++)
+		if(up->rq)
+			urpvomit("", up);
+}

@@ 128,7 128,7 @@ Dirtab bitdir[]={
 };
 
 #define	NBIT	(sizeof bitdir/sizeof(Dirtab))
-#define	NINFO	150
+#define	NINFO	257
 
 void
 bitreset(void)


@@ 277,7 277,8 @@ bitread(Chan *c, void *va, long n)
 	uchar *p, *q;
 	long miny, maxy, t, x, y;
 	ulong l, nw, ws;
-	int off;
+	int off, j;
+	Fontchar *i;
 	Bitmap *src;
 
 	if(c->qid & CHDIR)


@@ 322,6 323,9 @@ bitread(Chan *c, void *va, long n)
 			 *	'I'		1
 			 *	ldepth		1
 			 * 	rectangle	16
+			 * if count great enough, also
+			 *	font info	3*12
+			 *	fontchars	6*(defont->n+1)
 			 */
 			if(n < 18)
 				error(0, Ebadblt);


@@ 331,8 335,22 @@ bitread(Chan *c, void *va, long n)
 			PLONG(p+6, screen.r.min.y);
 			PLONG(p+10, screen.r.max.x);
 			PLONG(p+14, screen.r.max.y);
+			if(n >= 18+3*12+6*(defont->n+1)){
+				p += 18;
+				sprint((char*)p, "%11d %11d %11d ", defont->n,
+					defont->height, defont->ascent);
+				p += 3*12;
+				for(i=defont->info,j=0; j<=defont->n; j++,i++,p+=6){
+					PSHORT(p, i->x);
+					p[2] = i->top;
+					p[3] = i->bottom;
+					p[4] = i->left;
+					p[5] = i->width;
+				}
+				n = 18+3*12+6*(defont->n+1);
+			}else
+				n = 18;
 			bit.init = 0;
-			n = 18;
 			break;
 		}
 		if(bit.lastid > 0){

@@ 240,7 240,7 @@ echo(int c)
 	if(c == 0x15)
 		putstrn("^U\n", 3);
 	if(c == 0x16)
-		dumpqueues();
+		urpdump();
 	else{
 		ch = c;
 		putstrn(&ch, 1);

@@ 810,6 810,10 @@ mntxmit(Mnt *m, Mnthdr *mh)
 				wakeup(&w->r);
 				goto Read;
 			}
+print("devmnt: undelivered response fid %d type %d\n", mh->rhdr.fid, mh->rhdr.type);
+print("reader pid %d fid %d type %d\n", u->p->pid, mh->thdr.fid, mh->thdr.type);
+for(w=q->writer; w; w=w->next)print("writer pid %d fid %d type %d\n",w->p->pid,w->thdr.fid,w->thdr.type);
+
 		goto Read;
 	}else{
 		mh->p = u->p;

@@ 32,6 32,8 @@ Chan*
 pipeattach(char *spec)
 {
 	Chan *c;
+	int i;
+
 	/*
 	 *  make the first stream
 	 */


@@ 59,8 61,19 @@ 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.
+	 */
+	if(streamenter(c->stream)<0)
+		panic("pipeattach");
+	if(streamenter(nc->stream)<0)
+		panic("pipeattach");
 	return nc;
 }
 


@@ 110,7 123,17 @@ 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 */
+	poperror();
 }
 
 long


@@ 164,14 187,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 209,7 @@ 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;
-	}
+	bp = allocb(0);
+	bp->type = M_HANGUP;
+	PUTNEXT(q, bp);
 }

@@ 561,9 561,9 @@ DEBUG()
 	for(i=0; i<conf.nproc; i++){
 		p = procalloc.arena+i;
 		if(p->state != Dead){
-			print("%d:%s upc %lux %s ut %ld st %ld\n",
+			print("%d:%s upc %lux %s ut %ld st %ld r %lux\n",
 				p->pid, p->text, p->pc, statename[p->state],
-				p->time[0], p->time[1]);
+				p->time[0], p->time[1], p->r);
 		}
 	}
 }

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


@@ 515,8 531,12 @@ prepend(Block *bp, int n)
 void
 nullput(Queue *q, Block *bp)
 {
-	freeb(bp);
-	error(0, Ehungup);
+	if(bp->type == M_HANGUP)
+		freeb(bp);
+	else {
+		freeb(bp);
+		error(0, Ehungup);
+	}
 }
 
 /*


@@ 658,8 678,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 717,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 733,50 @@ 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);
+	if(s->inuse == 1){
+		/*
+		 *  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;
+	}
+	s->inuse--;
+	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,33 794,35 @@ streamclose(Chan *c)
 		return;
 
 	/*
-	 *  decrement the reference cound
+	 *  decrement the reference count
 	 */
 	lock(s);
-	if(s->inuse != 1){
-		s->inuse--;
-		unlock(c->stream);
-		return;
+	if(s->opens == 1){
+		/*
+		 *  descend the stream closing the queues
+		 */
+		for(q = s->procq; q; q = q->next){
+			if(q->info->close)
+				(*q->info->close)(q->other);
+			/* this may be 2 streams joined device end to device end */
+			if(q == s->devq->other)
+				break;
+		}
+	
+		/*
+		 *  ascend the stream flushing the queues
+		 */
+		for(q = s->devq; q; q = nq){
+			nq = q->next;
+			flushq(q);
+		}
 	}
+	s->opens--;
 
 	/*
-	 *  descend the stream closing the queues
+	 *  leave it and free it
 	 */
-	for(q = s->procq; q; q = q->next){
-		if(q->info->close)
-			(*q->info->close)(q->other);
-		if(q == s->devq->other)
-			break;
-	}
-	/*
-	 *  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;
-	s->inuse--;
+	streamexit(s, 1);
 	unlock(s);
 }
 

@@ 65,7 65,7 @@ struct Urp {
 };
 #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 IN(x, f, n) (f<=n ? (x>=f && x<n) : (x<n || x>=f))
 #define NEXT(x) (((x)+1)&Nmask)
 
 /*


@@ 117,6 117,7 @@ static void	sendrej(Urp*);
 static void	initoutput(Urp*, int);
 static void	initinput(Urp*, int);
 static void	urpkproc(void *arg);
+static void	urpvomit(char*, Urp*);
 
 Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
 


@@ 584,11 585,14 @@ output(Urp *up)
 	}
 
 	/*
-	 *  fill the transmit buffers
+	 *  fill the transmit buffers, `nxb' can never overtake `unechoed'
 	 */
 	q = up->wq;
-	if(up->xb[up->nxb]==0) {
-		for(bp=getq(q); bp && up->xb[up->nxb]==0; up->nxb=NEXT(up->nxb)){
+	i = NEXT(up->nxb);
+	if(i != up->unechoed) {
+		for(bp = getq(q); bp && i!=up->unechoed; i = NEXT(i)){
+			if(up->xb[up->nxb] != 0)
+				urpvomit("output", up);
 			if(BLEN(bp) > up->maxblock){
 				nbp = up->xb[up->nxb] = allocb(0);
 				nbp->rptr = bp->rptr;


@@ 597,17 601,27 @@ output(Urp *up)
 				up->xb[up->nxb] = bp;
 				bp = getq(q);
 			}
+			up->nxb = i;
 		}
 		if(bp)
 			putbq(q, bp);
 	}
-/*	print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n",
-		WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state);
-/**/
+
 	/*
-	 *  if a retransmit time has elapsed since a transmit, send an ENQ
+	 *  retransmit cruft
 	 */
-	if(up->unechoed!=up->next && NOW>up->timer){
+	if(up->rexmit){
+		/*
+		 *  if a retransmit is requested, move next back to
+		 *  the unacked blocks
+		 */
+		up->rexmit = 0;
+		up->next = up->unacked;
+	} else if(up->unacked!=up->next && NOW>up->timer){
+		/*
+		 *  if a retransmit time has elapsed since a transmit,
+		 *  send an ENQ
+		 */
 		up->timer = NOW + MSrexmit;
 		up->state &= ~REJECTING;
 		sendctl(up, ENQ);


@@ 618,12 632,11 @@ output(Urp *up)
 
 	/*
 	 *  if there's a window open, push some blocks out
+	 *
+	 *  the lock is to synchronize with acknowledges that free
+	 *  blocks.
 	 */
-	if(up->rexmit){
-		up->rexmit = 0;
-		up->next = up->unechoed;
-	}
-	while(WINDOW(up)>0 && up->xb[up->next]!=0){
+	while(WINDOW(up)>0 && up->next!=up->nxb){
 		i = up->next;
 		qlock(&up->xl[i]);
 		if(waserror()){


@@ 722,6 735,10 @@ sendblock(Urp *up, int bn)
 	 *  message 1, the BOT and the data
 	 */
 	bp = up->xb[bn];
+	if(bp == 0){
+		urpvomit("sendblock", up);
+		return;
+	}
 	m = allocb(1);
 	m->rptr = m->lim - 1;
 	m->wptr = m->lim;


@@ 768,6 785,8 @@ rcvack(Urp *up, int msg)
 			qlock(&up->xl[i]);
 			if(up->xb[i])
 				freeb(up->xb[i]);
+			else
+				urpvomit("rcvack", up);
 			up->xb[i] = 0;
 			qunlock(&up->xl[i]);
 		}


@@ 924,3 943,31 @@ urpkproc(void *arg)
 	up->kstarted = 0;
 	DPRINT("urpkproc %ux\n", up);
 }
+
+/*
+ *  urp got very confused, complain
+ */
+static void
+urpvomit(char *msg, Urp* up)
+{
+	print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n",
+		msg, up, up->next, up->unechoed, up->unacked, up->nxb);
+	print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n",
+		up->xb[0], up->xb[1], up->xb[2], up->xb[3], up->xb[4], 
+		up->xb[5], up->xb[6], up->xb[7]);
+	print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n",
+		up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1],
+		up->trbuf[2]);
+	print("\tupq: %ux %d %d\n", up->rq->next->first,  up->rq->next->nb,
+		up->rq->next->len);
+}
+
+int
+urpdump(void)
+{
+	Urp *up;
+
+	for(up = urp; up < &urp[Nurp]; up++)
+		if(up->rq)
+			urpvomit("", up);
+}

@@ 21,6 21,8 @@ Conftab conftab[] = {
 	{"nqueue", &conf.nqueue },
 	{"nblock", &conf.nblock },
 	{"nsrv", &conf.nsrv },
+	{"nnoifc", &conf.nnoifc },
+	{"nnoconv", &conf.nnoconv },
 	{ 0, 0 },
 };
 

@@ 17,7 17,12 @@ typedef struct Lock	Lock;
 typedef struct Mach	Mach;
 typedef struct Mount	Mount;
 typedef struct Mtab	Mtab;
+typedef struct Noconv	Noconv;
+typedef struct Nohdr	Nohdr;
+typedef struct Noifc	Noifc;
 typedef struct Note	Note;
+typedef struct Nomsg	Nomsg;
+typedef struct Nocall	Nocall;
 typedef struct Orig	Orig;
 typedef struct PTE	PTE;
 typedef struct Page	Page;


@@ 137,6 142,8 @@ struct Conf
 	ulong	nqueue;		/* stream queues */
 	ulong	nblock;		/* stream blocks */
 	ulong	nsrv;		/* public servers (devsrv.c) */
+	ulong	nnoifc;		/* number of nonet interfaces */
+	ulong	nnoconv;	/* number of nonet conversations/ifc */
 };
 
 struct Dev


@@ 435,7 442,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;		/* ... */


@@ 444,7 452,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)


@@ 469,6 476,119 @@ enum {
 	Streambhi= 16,		/* block count high water mark */
 };
 
+/*
+ *  nonet constants
+ */
+enum {
+	Nnomsg = 128,		/* max number of outstanding messages */
+	Nnocalls = 5,		/* maximum queued incoming calls */
+};
+
+/*
+ *  generic nonet header
+ */
+struct Nohdr {
+	uchar	circuit[3];	/* circuit number */
+	uchar	flag;
+	uchar	mid;		/* message id */
+	uchar	ack;		/* piggy back ack */
+	uchar	remain[2];	/* count of remaing bytes of data */
+	uchar	sum[2];		/* checksum (0 means none) */
+};
+#define HDRSIZE 10
+#define NEWCALL	0x1		/* flag bit marking a new circuit */
+#define HANGUP 0x2		/* flag bit requesting hangup */
+#define ACKME 0x4		/* acknowledge this message */
+
+/*
+ *  a buffer describing a nonet message
+ */
+struct Nomsg {
+	QLock;
+	Blist;
+	int	mid;		/* sequence number */
+	int	rem;		/* remaining */
+	long	time;
+	int	acked;
+};
+
+/*
+ *  one exists for each Nonet conversation.
+ */
+struct Noconv {
+	QLock;
+
+	Queue	*rq;		/* input queue */
+	int	version;	/* incremented each time struct is changed */
+	int	state;		/* true if listening */
+
+	Nomsg	in[Nnomsg];	/* messages being received */
+	int	rcvcircuit;	/* circuit number of incoming packets */
+
+	uchar	ack[Nnomsg];	/* acknowledgements waiting to be sent */
+	long	atime[Nnomsg];
+	int	afirst;
+	int	anext;
+
+	QLock	xlock;		/* one trasmitter at a time */
+	Rendez	r;		/* process waiting for an output mid */
+	Nomsg	ctl;		/* for control messages */
+	Nomsg	out[Nnomsg];	/* messages being sent */
+	int	first;		/* first unacknowledged message */
+	int	next;		/* next message buffer to use */
+	int	lastacked;	/* last message acked */		
+	Block	*media;		/* prototype media output header */
+	Nohdr	*hdr;		/* nonet header inside of media header */
+
+	Noifc	*ifc;
+	int	kstarted;
+	char	raddr[64];	/* remote address */
+	int	rexmit;		/* statistics */
+	int	retry;
+	int	bad;
+	int	sent;
+	int	rcvd;
+};
+
+/*
+ *  an incoming call
+ */
+struct Nocall {
+	Block	*msg;
+	char	raddr[64];
+	long	circuit;
+};
+
+/*
+ *  a nonet interface.  one exists for every stream that a 
+ *  nonet multiplexor is pushed onto.
+ */
+struct Noifc {
+	Lock;
+	int	ref;
+	char	name[64];	/* interface name */		
+	Queue	*wq;		/* interface output queue */
+	Noconv	*conv;
+
+	/*
+	 *  media dependent
+	 */
+	int	maxtu;		/* maximum transfer unit */
+	int	mintu;		/* minimum transfer unit */
+	int	hsize;		/* media header size */
+	void	(*connect)(Noconv *, char *);
+
+	/*
+	 *  calls and listeners
+	 */
+	QLock	listenl;
+	Rendez	listenr;
+	Lock	lock;
+	Nocall	call[Nnocalls];
+	int	rptr;
+	int	wptr;
+};
+
 #define	PRINTSIZE	256
 struct
 {

@@ 108,6 108,10 @@ void	newqinfo(Qinfo*);
 char	*nextelem(char*, char*);
 void	newstart(void);
 int	newtlbpid(Proc*);
+int	nonetcksum(Block*, int);
+void	nonetfreeifc(Noifc*);
+Noifc*	nonetnewifc(Queue*, Stream*, int, int, int, void (*)(Noconv*, char*));
+void	nonetrcvmsg(Noconv*, Block*);
 void	novme(int);
 void	nullput(Queue*, Block*);
 void	online(void);


@@ 158,6 162,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);