5 files changed, 72 insertions(+), 61 deletions(-)

M port/devpipe.c
M port/netif.c
M port/portdat.h
M port/portfns.h
M port/qio.c

@@ 65,19 65,17 @@ pipeattach(char *spec)
 		exhausted("memory");
 	p->ref = 1;
 
-	p->q[0] = qopen(64*1024, 0, 0);
+	p->q[0] = qopen(64*1024, 0, 0, 0);
 	if(p->q[0] == 0){
 		free(p);
 		exhausted("memory");
 	}
-	p->q[0]->state &= ~Qmsg;
-	p->q[1] = qopen(32*1024, 0, 0);
+	p->q[1] = qopen(32*1024, 0, 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;


@@ 139,10 137,10 @@ pipestat(Chan *c, char *db)
 		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);
+		devdir(c, c->qid, "data", qlen(p->q[0]), eve, 0660, &dir);
 		break;
 	case Qdata1:
-		devdir(c, c->qid, "data1", p->q[1]->len, eve, 0660, &dir);
+		devdir(c, c->qid, "data1", qlen(p->q[1]), eve, 0660, &dir);
 		break;
 	default:
 		panic("pipestat");

@@ 24,7 24,7 @@ netifinit(Netif *nif, char *name, int nfile, ulong limit)
 	nif->f = xalloc(nfile*sizeof(Netfile*));
 	memset(nif->f, 0, nfile*sizeof(Netfile*));
 	nif->limit = limit;
-	nif->out = qopen(limit, 0, 0);
+	nif->out = qopen(limit, 1, 0, 0);
 }
 
 /*


@@ 339,7 339,7 @@ openfile(Netif *nif, int id)
 				error(Enodev);
 			}
 			*fp = f;
-			f->in = qopen(nif->limit, 0, 0);
+			f->in = qopen(nif->limit, 1, 0, 0);
 			qlock(f);
 		} else {
 			qlock(f);

@@ 636,56 636,6 @@ struct Proc
 	PMMU;
 };
 
-/*
- *  IO queues
- */
-struct Block
-{
-	Block	*next;
-
-	uchar	*rp;			/* first unconsumed byte */
-	uchar	*wp;			/* first empty byte */
-	uchar	*lim;			/* 1 past the end of the buffer */
-	uchar	*base;			/* start of the buffer */
-	uchar	flag;
-
-	Rendez	r;			/* waiting reader */
-};
-#define BLEN(b)		((b)->wp - (b)->rp)
-
-struct Queue
-{
-	Lock;
-
-	Block	*bfirst;	/* buffer */
-	Block	*blast;
-
-	int	len;		/* bytes in queue */
-	int	limit;		/* max bytes in queue */
-	int	state;
-
-	void	(*kick)(void*);	/* restart output */
-	void	*arg;		/* argument to kick */
-
-	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
-{
-	/* Block.flag */
-	Bfilled=1,		/* block filled */
-
-	/* Queue.state */	
-	Qstarve=	(1<<0),		/* consumer starved */
-	Qmsg=		(1<<1),		/* message stream */
-	Qclosed=	(1<<2),
-	Qflow=		(1<<3),
-};
-
-
 enum
 {
 	PRINTSIZE =	256,

@@ 173,8 173,9 @@ ulong		pwait(Waitmsg*);
 void		qclose(Queue*);
 int		qconsume(Queue*, uchar*, int);
 void		qhangup(Queue*);
+int		qlen(Queue*);
 void		qlock(QLock*);
-Queue*		qopen(int, void (*)(void*), void*);
+Queue*		qopen(int, int, void (*)(void*), void*);
 int		qproduce(Queue*, uchar*, int);
 long		qread(Queue*, char*, int);
 void		qreopen(Queue*);

@@ 5,6 5,9 @@
 #include	"fns.h"
 #include	"../port/error.h"
 
+/*
+ *  interrupt level memory allocation
+ */
 typedef struct Chunk	Chunk;
 typedef	struct Chunkl	Chunkl;
 typedef	struct Arena	Arena;


@@ 39,6 42,56 @@ struct Arena
 static Arena arena;
 
 /*
+ *  IO queues
+ */
+typedef struct Block	Block;
+typedef struct Queue	Queue;
+
+struct Block
+{
+	Block	*next;
+
+	uchar	*rp;			/* first unconsumed byte */
+	uchar	*wp;			/* first empty byte */
+	uchar	*lim;			/* 1 past the end of the buffer */
+	uchar	*base;			/* start of the buffer */
+	uchar	flag;
+};
+#define BLEN(b)		((b)->wp - (b)->rp)
+
+struct Queue
+{
+	Lock;
+
+	Block	*bfirst;	/* buffer */
+	Block	*blast;
+
+	int	len;		/* bytes in queue */
+	int	limit;		/* max bytes in queue */
+	int	state;
+
+	void	(*kick)(void*);	/* restart output */
+	void	*arg;		/* argument to kick */
+
+	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
+{
+	/* Block.flag */
+	Bfilled=1,		/* block filled */
+
+	/* Queue.state */	
+	Qstarve=	(1<<0),		/* consumer starved */
+	Qmsg=		(1<<1),		/* message stream */
+	Qclosed=	(1<<2),
+	Qflow=		(1<<3),
+};
+
+/*
  *  Manage interrupt level memory allocation.
  */
 static void


@@ 297,7 350,7 @@ qproduce(Queue *q, uchar *p, int len)
  *  called by non-interrupt code
  */
 Queue*
-qopen(int limit, void (*kick)(void*), void *arg)
+qopen(int limit, int msg, void (*kick)(void*), void *arg)
 {
 	Queue *q;
 


@@ 309,7 362,7 @@ qopen(int limit, void (*kick)(void*), void *arg)
 	q->limit = limit;
 	q->kick = kick;
 	q->arg = arg;
-	q->state = Qmsg;
+	q->state = msg ? Qmsg : 0;
 
 	return q;
 }


@@ 518,3 571,12 @@ qreopen(Queue *q)
 {
 	q->state &= ~Qclosed;
 }
+
+/*
+ *  return bytes queued
+ */
+int
+qlen(Queue *q)
+{
+	return q->len;
+}