4 files changed, 42 insertions(+), 30 deletions(-)

M port/net.c
M port/portdat.h
M port/qlock.c
M port/stream.c

@@ 60,7 60,7 @@ netgen(Chan *c, void *vp, int ntab, int i, Dir *dp)
 	}
 
 	/* second level contains clone plus all the conversations */
-	t = STREAMTYPE(c->qid.path);
+	t = NETTYPE(c->qid.path);
 	if(t == Q2nd || t == Qclone){
 		if(i == 0){
 			q.path = Qclone;


@@ 68,7 68,7 @@ netgen(Chan *c, void *vp, int ntab, int i, Dir *dp)
 		}else if(i <= np->nconv){
 			if(findprot(np, i-1) == 0)
 				return 0;
-			q.path = CHDIR|STREAMQID(i-1, Q3rd);
+			q.path = CHDIR|NETQID(i-1, Q3rd);
 			sprint(buf, "%d", i-1);
 			devdir(c, q, buf, 0, eve, 0555, dp);
 		}else


@@ 77,7 77,7 @@ netgen(Chan *c, void *vp, int ntab, int i, Dir *dp)
 	}
 
 	/* third level depends on the number of info files */
-	p = findprot(np, STREAMID(c->qid.path));
+	p = findprot(np, NETID(c->qid.path));
 	if(p == 0)
 		return 0;
 	if(*p->owner){


@@ 89,24 89,24 @@ netgen(Chan *c, void *vp, int ntab, int i, Dir *dp)
 	}
 	switch(i){
 	case 0:
-		q.path = STREAMQID(STREAMID(c->qid.path), Sdataqid);
+		q.path = NETQID(NETID(c->qid.path), Ndataqid);
 		devdir(c, q, "data", 0, o, perm, dp);
 		break;
 	case 1:
-		q.path = STREAMQID(STREAMID(c->qid.path), Sctlqid);
+		q.path = NETQID(NETID(c->qid.path), Nctlqid);
 		devdir(c, q, "ctl", 0, o, perm, dp);
 		break;
 	case 2:
 		if(np->listen == 0)
 			return 0;
-		q.path = STREAMQID(STREAMID(c->qid.path), Qlisten);
+		q.path = NETQID(NETID(c->qid.path), Qlisten);
 		devdir(c, q, "listen", 0, o, perm, dp);
 		break;
 	default:
 		i -= 3;
 		if(i >= np->ninfo)
 			return -1;
-		q.path = STREAMQID(STREAMID(c->qid.path), Qinf+i);
+		q.path = NETQID(NETID(c->qid.path), Qinf+i);
 		devdir(c, q, np->info[i].name, 0, eve, 0444, dp);
 		break;
 	}


@@ 117,7 117,7 @@ int
 netwalk(Chan *c, char *name, Network *np)
 {
 	if(strcmp(name, "..") == 0) {
-		switch(STREAMTYPE(c->qid.path)){
+		switch(NETTYPE(c->qid.path)){
 		case Q2nd:
 			c->qid.path = CHDIR;
 			break;


@@ 176,29 176,29 @@ netopen(Chan *c, int omode, Network *np)
 		if(omode != OREAD)
 			error(Eperm);
 	} else {
-		switch(STREAMTYPE(c->qid.path)){
-		case Sdataqid:
-		case Sctlqid:
-			id = STREAMID(c->qid.path);
+		switch(NETTYPE(c->qid.path)){
+		case Ndataqid:
+		case Nctlqid:
+			id = NETID(c->qid.path);
 			id = (*np->open)(c, id);
 			break;
 		case Qlisten:
 			id = (*np->listen)(c);
-			c->qid.path = STREAMQID(id, Sctlqid);
+			c->qid.path = NETQID(id, Nctlqid);
 			ptclone(c, 1, id);
 			break;
 		case Qclone:
 			id = (*np->open)(c, -1);
-			c->qid.path = STREAMQID(id, Sctlqid);
+			c->qid.path = NETQID(id, Nctlqid);
 			ptclone(c, 0, id);
 			break;
 		default:
 			if(omode != OREAD)
 				error(Ebadarg);
 		}
-		switch(STREAMTYPE(c->qid.path)){
-		case Sdataqid:
-		case Sctlqid:
+		switch(NETTYPE(c->qid.path)){
+		case Ndataqid:
+		case Nctlqid:
 			p = findprot(np, id);
 			if(netown(p, up->user, omode&7) < 0)
 				error(Eperm);


@@ 220,7 220,10 @@ netread(Chan *c, void *a, long n, ulong offset, Network *np)
 	if(c->qid.path&CHDIR)
 		return devdirread(c, a, n, (Dirtab*)np, 0, netgen);
 
-	t = STREAMTYPE(c->qid.path);
+	t = NETTYPE(c->qid.path);
+	if(t == Nctlqid)
+		readnum(offset, a, n, NETID(c->qid.path), NUMSIZE);
+
 	if(t < Qinf || t >= Qinf + np->ninfo)
 		error(Ebadusefd);
 


@@ 294,7 297,7 @@ netwstat(Chan *c, char *db, Network *np)
 	Dir dir;
 	Netprot *p;
 
-	p = findprot(np, STREAMID(c->qid.path));
+	p = findprot(np, NETID(c->qid.path));
 	if(p == 0)
 		error(Enonexist);
 	lock(np);

@@ 668,6 668,9 @@ struct Queue
 	int	limit;		/* max bytes in queue */
 	int	state;
 
+	void	(*kick)(void*);	/* restart output */
+	void	*arg;		/* argument to kick */
+
 	QLock	wlock;		/* mutex for r */
 	Rendez	r;
 };


@@ 676,23 679,22 @@ enum
 {
 	Qstarve=1,	/* consumer starved */
 };
+#define BLEN(b)		((b)->wp - (b)->rp)
 
 /*
  *  Macros to manage Qid's used for multiplexed devices
  */
-#define STREAMTYPE(x)	((x)&0x1f)
-#define STREAMID(x)	(((x)&~CHDIR)>>5)
-#define STREAMQID(i,t)	(((i)<<5)|(t))
-#define BLEN(b)		((b)->wp - (b)->rp)
+#define NETTYPE(x)	((x)&0x1f)
+#define NETID(x)	(((x)&~CHDIR)>>5)
+#define NETQID(i,t)	(((i)<<5)|(t))
 enum
 {
-	Shighqid	= STREAMQID(1,0) - 1,
-	Sdataqid	= Shighqid,
-	Sctlqid		= Sdataqid-1,
-	Slowqid		= Sctlqid,
+	Nhighqid	= NETQID(1,0) - 1,
+	Ndataqid	= Nhighqid,
+	Nctlqid		= Ndataqid-1,
+	Nlowqid		= Nctlqid,
 };
 
-
 /*
  *  a multiplexed network
  */

@@ 32,7 32,8 @@ qlock(QLock *q)
 int
 canqlock(QLock *q)
 {
-	lock(&q->use);
+	if(!canlock(&q->use))
+		return 0;
 	if(q->locked){
 		unlock(&q->use);
 		return 0;

@@ 267,7 267,7 @@ produce(Queue *q, uchar *p, int len, int append)
  *  called by non-interrupt code
  */
 Queue*
-qopen(int limit)
+qopen(int limit, void (*kick)(void*), void *arg)
 {
 	Queue *q;
 


@@ 275,6 275,8 @@ qopen(int limit)
 	if(q == 0)
 		exhausted("Queues");
 	q->limit = limit;
+	q->kick = kick;
+	q->arg = arg;
 }
 
 static int


@@ 365,6 367,10 @@ qwrite(Queue *q, char *p, int len)
 		q->bfirst = b;
 	q->blast = b;
 	q->len += len;
+	if((q->state & Qstarve) && q->kick){
+		q->stat &= ~Qstarve;
+		(*q->kick)(q->arg);
+	}
 	unlock(q);
 	splx(x);