From f4e177f8683e75876b0160ad5257357e61323f9c Mon Sep 17 00:00:00 2001
From: David du Colombier <0intro@gmail.com>
Date: Wed, 26 May 1993 00:00:00 +0000
Subject: [PATCH] Plan 9 from Bell Labs 1993-05-26

---
 port/chan.c              |   4 +-
 port/portdat.h           |   3 +-
 port/{stream.c => qio.c} | 106 ++++++++++++++++++++++++++++++++-------
 3 files changed, 91 insertions(+), 22 deletions(-)
 rename port/{stream.c => qio.c} (82%)

diff --git a/port/chan.c b/port/chan.c
index ae66c23e1832f22bcc1742de4c88ffeecd89fe34..598fb46795d787e079c525b473ddadcc9ceac4c2 100644
--- a/port/chan.c
+++ b/port/chan.c
@@ -55,7 +55,7 @@ chandevreset(void)
 {
 	int i;
 
-	for(i=0; i<devchar[i]; i++)
+	for(i=0; devchar[i]; i++)
 		(*devtab[i].reset)();
 }
 
@@ -64,7 +64,7 @@ chandevinit(void)
 {
 	int i;
 
-	for(i=0; i<devchar[i]; i++)
+	for(i=0; devchar[i]; i++)
 		(*devtab[i].init)();
 }
 
diff --git a/port/portdat.h b/port/portdat.h
index f9b2516b2e1ee73b986f6d4d654c170385f259bf..34d73315107931d27574541c634145b1d8a731fc 100644
--- a/port/portdat.h
+++ b/port/portdat.h
@@ -654,6 +654,8 @@ struct Block
 struct Queue
 {
 	Lock;
+	QLock	rlock;		/* mutex for reading processes */
+	QLock	wlock;		/* mutex for writing processes */
 
 	Block	*rfirst;	/* waiting readers */
 	Block	*rlast;
@@ -668,7 +670,6 @@ struct Queue
 	void	(*kick)(void*);	/* restart output */
 	void	*arg;		/* argument to kick */
 
-	QLock	wlock;		/* mutex for r */
 	Rendez	r;
 };
 
diff --git a/port/stream.c b/port/qio.c
similarity index 82%
rename from port/stream.c
rename to port/qio.c
index 3ab30afff5a7255cd0e230ddafaae49b4a9a385a..a0666b9e1017484713e6b2ede8f015287ad0a566 100644
--- a/port/stream.c
+++ b/port/qio.c
@@ -128,13 +128,15 @@ iallocinit(void)
 	kproc("iallockproc", iallockproc, 0);
 }
 
-void*
-ialloc(int size)
+Block*
+iallocb(int size)
 {
 	int pow;
 	Chunkl *cl;
 	Chunk *p;
+	Block *b;
 
+	size += sizeof(Block);
 	for(pow = Minpow; pow <= Maxpow; pow++)
 		if(size <= (1<<pow)){
 			cl = &arena.alloc[pow];
@@ -145,9 +147,13 @@ ialloc(int size)
 				cl->first = p->next;
 			}
 			unlock(cl);
-			return (void*)p;
+			b = (Block *)p;
+			b->base = (uchar*)(b+1);
+			b->wp = b->rp = b->base;
+			b->lim = b->base + (1<<pow) - sizeof(Block);
+			return b;
 		}
-	panic("ialloc %d\n", size);
+	panic("iallocb %d\n", size);
 	return 0;			/* not reached */
 }
 
@@ -223,8 +229,8 @@ qconsume(Queue *q, uchar *p, int len)
 	return len;
 }
 
-int
-qproduce(Queue *q, uchar *p, int len)
+static int
+qproduce0(Queue *q, uchar *p, int len)
 {
 	Block *b;
 	int n;
@@ -256,14 +262,12 @@ qproduce(Queue *q, uchar *p, int len)
 		memmove(b->wp, p, len);
 		b->wp += len;
 	} else {
-		b = ialloc(sizeof(Block)+len);
+		b = iallocb(len);
 		if(b == 0){
 			unlock(q);
 			return -1;
 		}
-		b->base = (uchar*)(b+1);
-		b->rp = b->base;
-		b->wp = b->lim = b->base + len;
+		b->wp += len;
 		memmove(b->rp, p, len);
 		if(q->bfirst)
 			q->blast->next = b;
@@ -276,6 +280,22 @@ qproduce(Queue *q, uchar *p, int len)
 	return len;
 }
 
+int
+qproduce(Queue *q, uchar *p, int len)
+{
+	int n, sofar;
+
+	if(q->state&Qmsg)
+		return qproduce0(q, p, len);
+
+	for(sofar = 0; sofar < len; sofar += n){
+		n = qproduce0(q, p+sofar, len-sofar);
+		if(n < 0)
+			break;
+	}
+	return sofar;
+}
+
 /*
  *  called by non-interrupt code
  */
@@ -309,6 +329,8 @@ qread(Queue *q, char *p, int len)
 	Block *b, *bb;
 	int x, n;
 
+	qlock(&q->rlock);
+
 	/* ... to be replaced by a kmapping if need be */
 	b = allocb(len);
 
@@ -324,6 +346,7 @@ qread(Queue *q, char *p, int len)
 		q->rlast = b;
 		unlock(q);
 		splx(x);
+		qunlock(&q->rlock);
 		sleep(&b->r, bfilled, b);
 		n = BLEN(b);
 		memmove(p, b->rp, n);
@@ -353,6 +376,7 @@ qread(Queue *q, char *p, int len)
 		unlock(q);
 		splx(x);
 	}
+	qunlock(&q->rlock);
 	free(b);
 	return n;
 }
@@ -365,23 +389,41 @@ qnotfull(void *a)
 	return q->len < q->limit;
 }
 
-long
-qwrite(Queue *q, char *p, int len)
+static long
+qwrite0(Queue *q, char *p, int len)
 {
-	Block *b;
-	int x;
+	Block *b, *bb;
+	int x, n;
 
 	b = allocb(len);
+
+	x = splhi();
+	lock(q);
+	bb = q->rfirst;
+	if(bb){
+		/* hand to waiting receiver */
+		n = bb->lim - bb->wp;
+		q->rfirst = bb->next;
+		unlock(q);
+		splx(x);
+
+		if(n < len)
+			len = n;
+		memmove(bb->wp, p, len);
+		bb->wp += len;
+		wakeup(&bb->r);
+
+		free(b);
+		return len;
+	}
+		
 	memmove(b->rp, p, len);
 	b->wp += len;
 
 	/* flow control */
-	if(!qnotfull(q)){
-		qlock(&q->wlock);
+	if(!qnotfull(q))
 		sleep(&q->r, qnotfull, q);
-		qunlock(&q->wlock);
-	}
-		
+
 	x = splhi();
 	lock(q);
 	if(q->bfirst)
@@ -399,3 +441,29 @@ qwrite(Queue *q, char *p, int len)
 
 	return len;
 }
+
+long
+qwrite(Queue *q, char *p, int len)
+{
+	int n, sofar;
+
+	qlock(&q->wlock);
+	if(waserror()){
+		qunlock(&q->wlock);
+		nexterror();
+	}
+
+	if(q->state&Qmsg){
+		sofar = qwrite0(q, p, len);
+	} else {
+		for(sofar = 0; sofar < len; sofar += n){
+			n = qwrite0(q, p+sofar, len-sofar);
+			if(n < 0)
+				break;
+		}
+	}
+
+	poperror();
+	qunlock(&q->wlock);
+	return sofar;
+}