8 files changed, 396 insertions(+), 366 deletions(-)

M boot/boot.c
M port/devpipe.c
M port/netif.c
M port/portdat.h
M port/portfns.h
M port/proc.c
M port/qio.c
M port/taslock.c

@@ 25,9 25,96 @@ static void	recover(Method*);
 static Method	*rootserver(char*);
 
 void
+pipetest(void)
+{
+	int i, n, p[2];
+	ulong buf[1];
+
+	if(bind("#c", "/dev", MREPL) < 0)
+		print("can't bind #c\n");
+
+	if(pipe(p) < 0){
+		print("pipe fails\n");
+		exits(0);
+	}
+	switch(fork()){
+	case -1:
+		print("fork fails\n");
+		exits(0);
+	case 0:
+		close(p[0]);
+		for(;;){
+			n = read(p[1], buf, 1);
+			if(n <= 0){
+				print("1 exiting %d\n", time(0));
+				exits(0);
+			}
+			write(p[1], buf, 1);
+		}
+	default:
+		close(p[1]);
+		print("start %d\n", time(0));
+		for(i = 0; i < 100000; i++){
+			if(write(p[0], buf, 1) != 1){
+				print("sndr broke\n");
+				exits(0);
+			}
+			n = read(p[0], buf, 1);
+			if(n <= 0){
+				print("0 exiting %d\n", time(0));
+				exits(0);
+			}
+		}
+		close(p[0]);
+		sleep(1000);
+	}
+}
+
+void
+pipetest2(void)
+{
+	int i, n, p[2];
+	uchar buf[4096];
+
+	if(bind("#c", "/dev", MREPL) < 0)
+		print("can't bind #c\n");
+
+	if(pipe(p) < 0){
+		print("pipe fails\n");
+		exits(0);
+	}
+	switch(fork()){
+	case -1:
+		print("fork fails\n");
+		exits(0);
+	case 0:
+		close(p[0]);
+		for(;;){
+			n = read(p[1], buf, sizeof(buf));
+			if(n <= 0){
+				print("1 exiting %d\n", time(0));
+				exits(0);
+			}
+		}
+	default:
+		close(p[1]);
+		print("start %d\n", time(0));
+		for(i = 0; i < 40000; i++){
+			if(write(p[0], buf, sizeof(buf)) != sizeof(buf)){
+				print("sndr broke\n");
+				exits(0);
+			}
+		}
+		close(p[0]);
+		sleep(1000);
+	}
+
+}
+
+void
 ethertest(void)
 {
-	int cf, df, n, t;
+	int cf, df, n, t, i;
 	char buf[64];
 	struct Etherpkt
 	{


@@ 58,7 145,7 @@ ethertest(void)
 		exits(0);
 	}
 	close(cf);
-	for(;;){
+	for(i=0;i<20;i++){
 		n = read(df, &p, sizeof(p));
 		if(n <= 0){
 			print("read returns %d: %r\n", n);


@@ 67,6 154,7 @@ ethertest(void)
 		t = (p.type[0]<<8) | p.type[1];
 		print("%d %2.2ux%2.2ux%2.2ux%2.2ux%2.2ux%2.2ux -> %2.2ux%2.2ux%2.2ux%2.2ux%2.2ux%2.2ux %ux\n", n, p.s[0], p.s[1], p.s[2], p.s[3], p.s[4], p.s[5], p.d[0], p.d[1], p.d[2], p.d[3], p.d[4], p.d[5], t);
 	}
+	close(df);
 }
 
 void


@@ 83,11 171,15 @@ boot(int argc, char *argv[])
 	open("#c/cons", OREAD);
 	open("#c/cons", OWRITE);
 	open("#c/cons", OWRITE);
+#ifdef DEBUG
 	print("argc=%d\n", argc);
 	for(fd = 0; fd < argc; fd++)
 		print("%s ", argv[fd]);
-	print("\n");/**/
+	print("\n");
+#endif DEBUG
 
+	pipetest();
+	pipetest2();
 	ethertest();
 
 	if(argc <= 1)

@@ 6,14 6,17 @@
 #include	"../port/error.h"
 
 #include	"devtab.h"
+#include	"netif.h"
 
 typedef struct Pipe	Pipe;
 struct Pipe
 {
-	Ref;
 	QLock;
 	Pipe	*next;
+	int	ref;
 	ulong	path;
+	Queue	*q[2];
+	int	qref[2];
 };
 
 struct


@@ 23,26 26,17 @@ struct
 	ulong	path;
 } pipealloc;
 
-static Pipe *getpipe(ulong);
-static void pipeiput(Queue*, Block*);
-static void pipeoput(Queue*, Block*);
-static void pipestclose(Queue *);
-
-Qinfo pipeinfo =
+enum
 {
-	pipeiput,
-	pipeoput,
-	0,
-	pipestclose,
-	"pipe"
+	Qdir,
+	Qdata0,
+	Qdata1,
 };
 
 Dirtab pipedir[] =
 {
-	"data",		{Sdataqid},	0,			0600,
-	"ctl",		{Sctlqid},	0,			0600,
-	"data1",	{Sdataqid},	0,			0600,
-	"ctl1",		{Sctlqid},	0,			0600,
+	"data",		{Qdata0},	0,			0600,
+	"data1",	{Qdata1},	0,			0600,
 };
 #define NPIPEDIR 4
 


@@ 66,16 60,33 @@ pipeattach(char *spec)
 	Chan *c;
 
 	c = devattach('|', spec);
-	p = smalloc(sizeof(Pipe));
+	p = malloc(sizeof(Pipe));
+	if(p == 0)
+		exhausted("memory");
 	p->ref = 1;
 
+	p->q[0] = qopen(64*1024, 0, 0);
+	if(p->q[0] == 0){
+		free(p);
+		exhausted("memory");
+	}
+	p->q[0]->state &= ~Qmsg;
+	p->q[1] = qopen(32*1024, 0, 0);
+	if(p->q[1] == 0){
+		free(p->q[0]);
+		free(p);
+		exhausted("memory");
+	}
+	p->q[1]->state &= ~Qmsg;
+
 	lock(&pipealloc);
 	p->path = ++pipealloc.path;
 	p->next = pipealloc.pipe;
 	pipealloc.pipe = p;
 	unlock(&pipealloc);
 
-	c->qid = (Qid){CHDIR|STREAMQID(2*p->path, 0), 0};
+	c->qid = (Qid){CHDIR|NETQID(2*p->path, Qdir), 0};
+	c->aux = p;
 	c->dev = 0;
 	return c;
 }


@@ 85,10 96,11 @@ pipeclone(Chan *c, Chan *nc)
 {
 	Pipe *p;
 
-	p = getpipe(STREAMID(c->qid.path)/2);
+	p = c->aux;
 	nc = devclone(c, nc);
-	if(incref(p) <= 1)
-		panic("pipeclone");
+	qlock(p);
+	p->ref++;
+	qunlock(p);
 	return nc;
 }
 


@@ 97,13 109,13 @@ pipegen(Chan *c, Dirtab *tab, int ntab, int i, Dir *dp)
 {
 	int id;
 
-	id = STREAMID(c->qid.path);
+	id = NETID(c->qid.path);
 	if(i > 1)
 		id++;
 	if(tab==0 || i>=ntab)
 		return -1;
 	tab += i;
-	devdir(c, (Qid){STREAMQID(id, tab->qid.path),0}, tab->name, tab->length, eve, tab->perm, dp);
+	devdir(c, (Qid){NETQID(id, tab->qid.path),0}, tab->name, tab->length, eve, tab->perm, dp);
 	return 1;
 }
 


@@ 117,7 129,25 @@ pipewalk(Chan *c, char *name)
 void
 pipestat(Chan *c, char *db)
 {
-	streamstat(c, db, "pipe", 0666);
+	Pipe *p;
+	Dir dir;
+
+	p = c->aux;
+
+	switch(NETTYPE(c->qid.path)){
+	case Qdir:
+		devdir(c, c->qid, ".", 2*DIRLEN, eve, CHDIR|0555, &dir);
+		break;
+	case Qdata0:
+		devdir(c, c->qid, "data", p->q[0]->len, eve, 0660, &dir);
+		break;
+	case Qdata1:
+		devdir(c, c->qid, "data1", p->q[1]->len, eve, 0660, &dir);
+		break;
+	default:
+		panic("pipestat");
+	}
+	convD2M(&dir, db);
 }
 
 /*


@@ 127,8 157,6 @@ Chan *
 pipeopen(Chan *c, int omode)
 {
 	Pipe *p;
-	int other;
-	Stream *local, *remote;
 
 	if(c->qid.path & CHDIR){
 		if(omode != OREAD)


@@ 139,36 167,17 @@ pipeopen(Chan *c, int omode)
 		return c;
 	}
 
-	p = getpipe(STREAMID(c->qid.path)/2);
-	if(waserror()){
-		qunlock(p);
-		nexterror();
-	}
+	p = c->aux;
 	qlock(p);
-	streamopen(c, &pipeinfo);
-	local = c->stream;
-	if(local->devq->ptr == 0){
-		/*
-		 *  first open, create the other end also
-		 */
-		other = STREAMID(c->qid.path)^1;
-		remote = streamnew(c->type, c->dev, other, &pipeinfo,1);
-
-		/*
-		 *  connect the device ends of both streams
-		 */
-		local->devq->ptr = remote;
-		remote->devq->ptr = local;
-		local->devq->other->next = remote->devq;
-		remote->devq->other->next = local->devq;
-	} else if(local->opens == 1){
-		/*
-		 *  keep other side around till last close of this side
-		 */
-		streamenter(local->devq->ptr);
+	switch(NETTYPE(c->qid.path)){
+	case Qdata0:
+		p->qref[0]++;
+		break;
+	case Qdata1:
+		p->qref[1]++;
+		break;
 	}
 	qunlock(p);
-	poperror();
 
 	c->mode = omode&~OTRUNC;
 	c->flag |= COPEN;


@@ 201,27 210,44 @@ void
 pipeclose(Chan *c)
 {
 	Pipe *p, *f, **l;
-	Stream *remote;
 
-	p = getpipe(STREAMID(c->qid.path)/2);
+	p = c->aux;
+	qlock(p);
 
 	/*
-	 *  take care of local and remote streams
+	 *  closing either side hangs up the stream
 	 */
-	if(c->stream){
-		qlock(p);
-		remote = c->stream->devq->ptr;
-		if(streamclose(c) == 0){
-			if(remote)
-				streamexit(remote);
+	switch(NETTYPE(c->qid.path)){
+	case Qdata0:
+		p->qref[0]--;
+		if(p->qref[0] == 0){
+			qclose(p->q[0]);
+			qhangup(p->q[1]);
 		}
-		qunlock(p);
+		break;
+	case Qdata1:
+		p->qref[1]--;
+		if(p->qref[1] == 0){
+			qclose(p->q[1]);
+			qhangup(p->q[0]);
+		}
+		break;
 	}
 
 	/*
-	 *  free the structure
+	 *  if both sides are closed, they are reusable
 	 */
-	if(decref(p) == 0){
+	if(p->qref[0] == 0 && p->qref[1] == 0){
+		qreopen(p->q[0]);
+		qreopen(p->q[1]);
+	}
+
+	/*
+	 *  free the structure on last close
+	 */
+	p->ref--;
+	if(p->ref == 0){
+		qunlock(p);
 		lock(&pipealloc);
 		l = &pipealloc.pipe;
 		for(f = *l; f; f = f->next) {


@@ 232,18 258,34 @@ pipeclose(Chan *c)
 			l = &f->next;
 		}
 		unlock(&pipealloc);
+		free(p->q[0]);
+		free(p->q[1]);
 		free(p);
 	}
+
+	qunlock(p);
 }
 
 long
 piperead(Chan *c, void *va, long n, ulong offset)
 {
+	Pipe *p;
+
 	USED(offset);
-	if(c->qid.path & CHDIR)
-		return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen);
 
-	return streamread(c, va, n);
+	p = c->aux;
+
+	switch(NETTYPE(c->qid.path)){
+	case Qdir:
+		return devdirread(c, va, n, pipedir, NPIPEDIR, pipegen);
+	case Qdata0:
+		return qread(p->q[0], va, n);
+	case Qdata1:
+		return qread(p->q[1], va, n);
+	default:
+		panic("piperead");
+	}
+	return -1;	/* not reached */
 }
 
 /*


@@ 253,79 295,19 @@ piperead(Chan *c, void *va, long n, ulong offset)
 long
 pipewrite(Chan *c, void *va, long n, ulong offset)
 {
+	Pipe *p;
+
 	USED(offset);
 
-	/* avoid notes when pipe is a mounted stream */
-	if(c->flag & CMSG)
-		return streamwrite(c, va, n, 0);
+	p = c->aux;
 
-	if(waserror()) {
-		postnote(up, 1, "sys: write on closed pipe", NUser);
-		error(Egreg);
+	switch(NETTYPE(c->qid.path)){
+	case Qdata0:
+		return qwrite(p->q[1], va, n);
+	case Qdata1:
+		return qwrite(p->q[0], va, n);
+	default:
+		panic("piperead");
 	}
-	n = streamwrite(c, va, n, 0);
-	poperror();
 	return n;
 }
-
-/*
- *  send a block upstream to the process.
- *  sleep until there's room upstream.
- */
-static void
-pipeiput(Queue *q, Block *bp)
-{
-	FLOWCTL(q, bp);
-}
-
-/*
- *  send the block to the other side
- */
-static void
-pipeoput(Queue *q, Block *bp)
-{
-	PUTNEXT(q, bp);
-}
-
-/*
- *  send a hangup and disconnect the streams
- */
-static void
-pipestclose(Queue *q)
-{
-	Block *bp;
-
-	/*
-	 *  point to the bit-bucket and let any in-progress
-	 *  write's finish.
-	 */
-	q->put = nullput;
-	wakeup(&q->r);
-
-	/*
-	 *  send a hangup
-	 */
-	q = q->other;
-	if(q->next == 0)
-		return;
-	bp = allocb(0);
-	bp->type = M_HANGUP;
-	PUTNEXT(q, bp);
-}
-
-Pipe*
-getpipe(ulong path)
-{
-	Pipe *p;
-
-	lock(&pipealloc);
-	for(p = pipealloc.pipe; p; p = p->next) {
-		if(path == p->path) {
-			unlock(&pipealloc);
-			return p;
-		}
-	}
-	unlock(&pipealloc);
-	panic("getpipe");
-	return 0;		/* not reached */
-}

@@ 269,6 269,8 @@ netifclose(Netif *nif, Chan *c)
 			qunlock(nif);
 		}
 		f->owner[0] = 0;
+		f->type = 0;
+		qclose(f->in);
 	}
 	qunlock(f);
 }


@@ 320,6 322,7 @@ openfile(Netif *nif, int id)
 		if(f == 0)
 			error(Enodev);
 		qlock(f);
+		qreopen(f->in);
 		f->inuse++;
 		qunlock(f);
 		return id;


@@ 346,6 349,7 @@ openfile(Netif *nif, int id)
 			}
 		}
 		f->inuse = 1;
+		qreopen(f->in);
 		netown(f, up->user, 0);
 		qunlock(f);
 		qunlock(nif);

@@ 656,11 656,6 @@ struct Block
 struct Queue
 {
 	Lock;
-	QLock	rlock;		/* mutex for reading processes */
-	QLock	wlock;		/* mutex for writing processes */
-
-	Block	*rfirst;	/* waiting readers */
-	Block	*rlast;
 
 	Block	*bfirst;	/* buffer */
 	Block	*blast;


@@ 672,7 667,10 @@ struct Queue
 	void	(*kick)(void*);	/* restart output */
 	void	*arg;		/* argument to kick */
 
-	Rendez	r;
+	QLock	rlock;		/* mutex for reading processes */
+	Rendez	rr;		/* process waiting to read */
+	QLock	wlock;		/* mutex for writing processes */
+	Rendez	wr;		/* process waiting to write */
 };
 
 enum


@@ 681,8 679,10 @@ enum
 	Bfilled=1,		/* block filled */
 
 	/* Queue.state */	
-	Qstarve=1,		/* consumer starved */
-	Qmsg=2,			/* message stream */
+	Qstarve=	(1<<0),		/* consumer starved */
+	Qmsg=		(1<<1),		/* message stream */
+	Qclosed=	(1<<2),
+	Qflow=		(1<<3),
 };
 
 

@@ 170,11 170,14 @@ void		putstr(char*);
 void		putstrn(char*, long);
 void		putswap(Page*);
 ulong		pwait(Waitmsg*);
+void		qclose(Queue*);
 int		qconsume(Queue*, uchar*, int);
+void		qhangup(Queue*);
 void		qlock(QLock*);
 Queue*		qopen(int, void (*)(void*), void*);
 int		qproduce(Queue*, uchar*, int);
 long		qread(Queue*, char*, int);
+void		qreopen(Queue*);
 void		qunlock(QLock*);
 long		qwrite(Queue*, char*, int);
 int		readnum(ulong, char*, ulong, ulong, int);

@@ 160,6 160,8 @@ loop:
 	/* p->mach==0 only when process state is saved */
 	if(p == 0 || p->mach){	
 		unlock(&runhiq);
+		if(conf.nmach > 1)
+			delay(7);	/* keep off the bus (tuned for everest) */
 		goto loop;
 	}
 	if(p->rnext == 0)

@@ 150,10 150,10 @@ iallocb(int size)
 			cl->first = p->next;
 			unlock(cl);
 			b = (Block *)p;
+			memset(b, 0, sizeof(Block));
 			b->base = (uchar*)(b+1);
 			b->wp = b->rp = b->base;
 			b->lim = b->base + (1<<pow) - sizeof(Block);
-			b->flag = 0;
 			return b;
 		}
 	panic("iallocb %d\n", size);


@@ 186,6 186,7 @@ allocb(int size)
 	if(b == 0)
 		exhausted("Blocks");
 
+	memset(b, 0, sizeof(Block));
 	b->base = (uchar*)(b+1);
 	b->rp = b->wp = b->base;
 	b->lim = b->base + size;


@@ 202,7 203,7 @@ int
 qconsume(Queue *q, uchar *p, int len)
 {
 	Block *b;
-	int n;
+	int n, dowakeup;
 
 	/* sync with qwrite */
 	lock(q);


@@ 213,6 214,7 @@ qconsume(Queue *q, uchar *p, int len)
 		unlock(q);
 		return -1;
 	}
+
 	n = BLEN(b);
 	if(n < len)
 		len = n;


@@ 223,42 225,34 @@ qconsume(Queue *q, uchar *p, int len)
 		b->rp += len;
 	q->len -= len;
 
-	/* wakeup flow controlled writers (with a bit of histeresis) */
-	if(q->len+len >= q->limit && q->len < q->limit/2)
-		wakeup(&q->r);
+	/* if writer flow controlled, restart */
+	if((q->state & Qflow) && q->len < q->limit/2){
+		q->state &= ~Qflow;
+		dowakeup = 1;
+	} else
+		dowakeup = 0;
 
 	unlock(q);
 
+	if(dowakeup)
+		wakeup(&q->wr);
+
+	/* discard the block if we're done with it */
 	if((q->state & Qmsg) || len == n)
 		ifree(b);
 
 	return len;
 }
 
-static int
-qproduce0(Queue *q, uchar *p, int len)
+int
+qproduce(Queue *q, uchar *p, int len)
 {
 	Block *b;
-	int n;
+	int dowakeup;
 
 	/* sync with qread */
 	lock(q);
 
-	b = q->rfirst;
-	if(b){
-		/* hand to waiting receiver */
-		q->rfirst = b->next;
-		unlock(q);
-		n = b->lim - b->wp;
-		if(n < len)
-			len = n;
-		memmove(b->wp, p, len);
-		b->wp += len;
-		b->flag |= Bfilled;
-		wakeup(&b->r);
-		return len;
-	}
-
 	/* no waiting receivers, room in buffer? */
 	if(q->len >= q->limit){
 		unlock(q);


@@ 279,7 273,6 @@ qproduce0(Queue *q, uchar *p, int len)
 		}
 		memmove(b->wp, p, len);
 		b->wp += len;
-		b->flag |= Bfilled;
 		if(q->bfirst)
 			q->blast->next = b;
 		else


@@ 287,24 280,17 @@ qproduce0(Queue *q, uchar *p, int len)
 		q->blast = b;
 	}
 	q->len += len;
+	if(q->state & Qstarve){
+		q->state &= ~Qstarve;
+		dowakeup = 1;
+	} else
+		dowakeup = 0;
 	unlock(q);
 
-	return len;
-}
-
-int
-qproduce(Queue *q, uchar *p, int len)
-{
-	int n, sofar;
+	if(dowakeup)
+		wakeup(&q->rr);
 
-	sofar = 0;
-	do {
-		n = qproduce0(q, p + sofar, len - sofar);
-		if(n < 0)
-			break;
-		sofar += n;
-	} while(sofar < len && (q->state & Qmsg) == 0);
-	return sofar;
+	return len;
 }
 
 /*


@@ 317,7 303,9 @@ qopen(int limit, void (*kick)(void*), void *arg)
 
 	q = malloc(sizeof(Queue));
 	if(q == 0)
-		exhausted("Queues");
+		return 0;
+
+	memset(q, 0, sizeof(Queue));
 	q->limit = limit;
 	q->kick = kick;
 	q->arg = arg;


@@ 326,248 314,207 @@ qopen(int limit, void (*kick)(void*), void *arg)
 	return q;
 }
 
-ulong qrtoomany;
-ulong qrtoofew;
-
 static int
-bfilled(void *a)
+notempty(void *a)
 {
-	Block *b = a;
+	Queue *q = a;
 
-	return b->flag & Bfilled;
+	return q->bfirst != 0;
 }
 
+/*
+ *  read a queue.  if no data is queued, post a Block
+ *  and wait on its Rendez.
+ */
 long
 qread(Queue *q, char *p, int len)
 {
-	Block *b, *bb, **l;
-	int x, n;
+	Block *b;
+	int x, n, dowakeup;
 
 	qlock(&q->rlock);
-	b = 0;
 	if(waserror()){
 		qunlock(&q->rlock);
-		if(b)
-			free(b);
 		nexterror();
 	}
 
-	/*
-	 *  If there are no buffered blocks, allocate a block
-	 *  for the qproducer/qwrite to fill.  This is
-	 *  optimistic and and we will
-	 *  sometimes be wrong: after locking we may either
-	 *  have to throw away or allocate one.
-	 *
-	 *  We hope to replace the allocb with a kmap later on.
-	 */
-retry:
-	if(q->bfirst == 0)
-		b = allocb(len);
-
-	/* sync with qwrite/qproduce */
-	x = splhi();
-	lock(q);
+	/* wait for data */
+	for(;;){
+		/* sync with qwrite/qproduce */
+		x = splhi();
+		lock(q);
 
-	bb = q->bfirst;
-	if(bb == 0){
-		if(b == 0){
-			/* we guessed wrong, drop the locks and try again */
+		if(q->state & Qclosed){
 			unlock(q);
 			splx(x);
-			qrtoofew++;
-			goto retry;
+			return 0;
 		}
 
-		/* add ourselves to the list of readers */
-		if(q->rfirst)
-			q->rlast->next = b;
-		else
-			q->rfirst = b;
-		q->rlast = b;
+		b = q->bfirst;
+		if(b)
+			break;
+		q->state |= Qstarve;
 		unlock(q);
 		splx(x);
-		qunlock(&q->rlock);
-		poperror();
-
-		if(waserror()){
-			/* on error, unlink us from the chain */
-			x = splhi();
-			lock(q);
-			l = &q->rfirst;
-			for(bb = q->rfirst; bb; bb = bb->next){
-				if(b == bb){
-					*l = bb->next;
-					break;
-				} else
-					l = &bb->next;
-			}
-			unlock(q);
-			splx(x);
-			free(b);
-			nexterror();
-		}
-
-		/* wait for the producer */
-		sleep(&b->r, bfilled, b);
-		n = BLEN(b);
-		memmove(p, b->rp, n);
-		poperror();
-		free(b);
-
-		return n;
+		sleep(&q->rr, notempty, q);
 	}
 
-	/* copy from a buffered block */
-	q->bfirst = bb->next;
-	n = BLEN(bb);
-	if(n > len)
-		n = len;
+	/* remove a buffered block */
+	q->bfirst = b->next;
+	n = BLEN(b);
 	q->len -= n;
+
+	/* if writer flow controlled, restart */
+	if((q->state & Qflow) && q->len < q->limit/2){
+		q->state &= ~Qflow;
+		dowakeup = 1;
+	} else
+		dowakeup = 0;
 	unlock(q);
 	splx(x);
 
 	/* do this outside of the lock(q)! */
-	memmove(p, bb->rp, n);
-	bb->rp += n;
+	if(n > len)
+		n = len;
+	memmove(p, b->rp, n);
+	b->rp += n;
 
-	/* free it or put it back on the queue */
-	if(bb->rp >= bb->wp || (q->state&Qmsg))
-		free(bb);
+	/* free it or put it what's left on the queue */
+	if(b->rp >= b->wp || (q->state&Qmsg))
+		free(b);
 	else {
 		x = splhi();
 		lock(q);
-		bb->next = q->bfirst;
-		q->bfirst = bb;
+		b->next = q->bfirst;
+		q->bfirst = b;
+		q->len += BLEN(b);
 		unlock(q);
 		splx(x);
 	}
 
+	/* wakeup flow controlled writers (with a bit of histeresis) */
+	if(dowakeup)
+		wakeup(&q->wr);
+
 	poperror();
 	qunlock(&q->rlock);
-	if(b){
-		qrtoomany++;
-		free(b);
-	}
 	return n;
 }
 
-ulong qwtoomany;
-ulong qwtoofew;
-
-static long
-qwrite0(Queue *q, char *p, int len, Block *b)
+static int
+qnotfull(void *a)
 {
-	Block *bb;
-	int x, n, sofar;
+	Queue *q = a;
 
-	/* sync with qconsume/qread */
-	x = splhi();
-	lock(q);
+	return q->len < q->limit;
+}
 
-	sofar = 0;
-	while(bb = q->rfirst){
-		/* hand to waiting receiver */
-		q->rfirst = bb->next;
-		unlock(q);
-		splx(x);
+/*
+ *  write to a queue.  if no reader blocks are posted
+ *  queue the data.
+ */
+long
+qwrite(Queue *q, char *p, int len)
+{
+	int x, dowakeup;
+	Block *b;
 
-		n = bb->lim - bb->wp;
-		if(n > len-sofar)
-			n = len - sofar;
-		memmove(bb->wp, p+sofar, n);
-		bb->wp += n;
-		bb->flag |= Bfilled;
-		wakeup(&bb->r);
-
-		sofar += n;
-		if(sofar == len){
-			if(b){
-				free(b);	/* we were wrong to allocate */
-				qwtoomany++;
-			}
-			return len;
-		}
+	b = allocb(len);
+	memmove(b->wp, p, len);
+	b->wp += len;
+
+	/* flow control */
+	while(!qnotfull(q)){
+		qlock(&q->wlock);
+		q->state |= Qflow;
+		sleep(&q->wr, qnotfull, q);
+		qunlock(&q->wlock);
 	}
 
-	/* buffer what ever is left */
-	if(b == 0){
-		/* we should have alloc'd, return to qwrite and have it do it */
+	x = splhi();
+	lock(q);
+
+	if(q->state & Qclosed){
 		unlock(q);
 		splx(x);
-		qwtoofew++;
-		return sofar;
+		error(Ehungup);
 	}
-	b->rp += sofar;
 
-	x = splhi();
-	lock(q);
-	if(q->bfirst)
-		q->blast->next = b;
-	else
-		q->bfirst = b;
-	q->blast = b;
+	b->next = q->bfirst;
+	q->bfirst = b;
 	q->len += len;
-	if((q->state & Qstarve) && q->kick){
+
+	if(q->state & Qstarve){
 		q->state &= ~Qstarve;
-		(*q->kick)(q->arg);
-	}
+		dowakeup = 1;
+	} else
+		dowakeup = 0;
+
 	unlock(q);
 	splx(x);
 
+	if(dowakeup)
+		wakeup(&q->rr);
+
 	return len;
 }
 
-static int
-qnotfull(void *a)
+/*
+ *  Mark a queue as closed.  No further IO is permitted.
+ *  All blocks are released.
+ */
+void
+qclose(Queue *q)
 {
-	Queue *q = a;
+	int x;
+	Block *b, *bfirst;
 
-	return q->len < q->limit;
-}
+	/* mark it */
+	x = splhi();
+	lock(q);
+	q->state |= Qclosed;
+	bfirst = q->bfirst;
+	q->bfirst = 0;
+	unlock(q);
+	splx(x);
 
-long
-qwrite(Queue *q, char *p, int len)
-{
-	int n, i;
-	Block *b;
+	/* free queued blocks */
+	while(b = bfirst){
+		bfirst = b->next;
+		free(b);
+	}
 
-	/*
-	 *  If there are no readers, grab a buffer and copy
-	 *  into it before locking anything down.  This
-	 *  provides the highest concurrency but we will
-	 *  sometimes be wrong: after locking we may either
-	 *  have to throw away or allocate one.
-	 */
-	if(q->rfirst == 0){
-		b = allocb(len);
-		memmove(b->wp, p, len);
-		b->wp += len;
-	} else
-		b = 0;
+	/* wake up readers/writers */
+	wakeup(&q->rr);
+	wakeup(&q->wr);
+}
 
-	/* ensure atomic writes */
-	qlock(&q->wlock);
-	if(waserror()){
-		qunlock(&q->wlock);
-		nexterror();
-	}
+/*
+ *  Mark a queue as closed.  Wakeup any readers.  Don't remove queued
+ *  blocks.
+ */
+void
+qhangup(Queue *q)
+{
+	int x;
 
-	/* flow control */
-	sleep(&q->r, qnotfull, q);
-
-	n = qwrite0(q, p, len, b);
-	if(n != len){
-		/* no readers and we need a buffer */
-		i = len - n;
-		b = allocb(i);
-		memmove(b->wp, p + n, i);
-		b->wp += n;
-		n += qwrite0(q, p + n, i, b);
-	}
+	/* mark it */
+	x = splhi();
+	lock(q);
+	q->state |= Qclosed;
+	unlock(q);
+	splx(x);
 
-	qunlock(&q->wlock);
-	poperror();
+	/* wake up readers/writers */
+	wakeup(&q->rr);
+	wakeup(&q->wr);
+}
 
-	return n;
+/*
+ *  mark a queue as no longer hung up
+ */
+void
+qreopen(Queue *q)
+{
+	q->state &= ~Qclosed;
 }

@@ 13,7 13,7 @@ lock(Lock *l)
 
 	pc = getcallerpc(((uchar*)&l) - sizeof(l));
 
-	for(i = 0; i < 10000000; i++){
+	for(i = 0; i < 20000000; i++){
     		if (tas(&l->key) == 0){
 			l->pc = pc;
 			return;