4 files changed, 81 insertions(+), 1408 deletions(-)

M port/chan.c
M port/portdat.h
M port/portfns.h
M port/stream.c

@@ 96,7 96,6 @@ newchan(void)
 	c->dev = 0;
 	c->offset = 0;
 	c->mnt = 0;
-	c->stream = 0;
 	c->aux = 0;
 	c->mchan = 0;
 	c->path = 0;

@@ 1,6 1,5 @@
 typedef struct Alarms	Alarms;
 typedef struct Block	Block;
-typedef struct Blist	Blist;
 typedef struct Chan	Chan;
 typedef struct Crypt	Crypt;
 typedef struct Dev	Dev;


@@ 30,7 29,6 @@ typedef struct Physseg	Physseg;
 typedef struct Proc	Proc;
 typedef struct Pte	Pte;
 typedef struct Pthash	Pthash;
-typedef struct Qinfo	Qinfo;
 typedef struct QLock	QLock;
 typedef struct Queue	Queue;
 typedef struct Ref	Ref;


@@ 42,14 40,9 @@ typedef struct Scsibuf	Scsibuf;
 typedef struct Scsidata	Scsidata;
 typedef struct Segment	Segment;
 typedef struct Session	Session;
-typedef struct Stream	Stream;
 typedef struct Talarm	Talarm;
 typedef struct Waitq	Waitq;
 typedef int    Devgen(Chan*, Dirtab*, int, int, Dir*);
-typedef	void   Streamput(Queue*, Block*);
-typedef	void   Streamopen(Queue*, Stream*);
-typedef	void   Streamclose(Queue*);
-typedef	void   Streamreset(void);
 
 #include <auth.h>
 #include <fcall.h>


@@ 100,37 93,6 @@ struct Sargs
 	ulong	args[MAXSYSARG];
 };
 
-/* Block.flags */
-#define S_DELIM 0x80
-#define S_CLASS 0x07
-
-/* Block.type */
-#define M_DATA 0
-#define M_CTL 1
-#define M_HANGUP 2
-
-struct Block
-{
-	Block	*next;
-	Block	*list;			/* chain of block lists */
-	uchar	*rptr;			/* first unconsumed byte */
-	uchar	*wptr;			/* first empty byte */
-	uchar	*lim;			/* 1 past the end of the buffer */
-	uchar	*base;			/* start of the buffer */
-	uchar	flags;
-	uchar	type;
-	ulong	pc;			/* pc of caller */
-};
-
-struct Blist
-{
-	Lock;
-	Block	*first;			/* first data block */
-	Block	*last;			/* last data block */
-	long	len;			/* length of list in bytes */
-	int	nb;			/* number of blocks in list */
-};
-
 /*
  * Access types in namec & channel flags
  */


@@ 172,7 134,6 @@ struct Chan
 	ushort	flag;
 	Qid	qid;
 	int	fid;			/* for devmnt */
-	Stream	*stream;		/* for stream channels */
 	Path	*path;
 	Mount	*mnt;			/* mount point that derived Chan */
 	Mount	*xmnt;			/* Last mount point crossed */


@@ 679,86 640,54 @@ struct Proc
 };
 
 /*
- *  operations available to a queue
+ *  IO queues
  */
-struct Qinfo
+struct Block
 {
-	Streamput	*iput;		/* input routine */
-	Streamput	*oput;		/* output routine */
-	Streamopen	*open;
-	Streamclose	*close;
-	char		*name;
-	Streamreset	*reset;		/* initialization */
-	char		nodelim;	/* True if stream does not preserve delimiters */
-	Qinfo		*next;
+	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 */
 };
 
-/* Queue.flag */
-enum
+struct Queue
 {
-	QHUNGUP	=	0x1,	/* stream has been hung up */
-	QINUSE =	0x2,	/* allocation check */
-	QHIWAT =	0x4,	/* queue has gone past the high water mark */	
-	QDEBUG =	0x8,
+	Lock;
+
+	Block	*first;
+	Block	*last;
+	int	nbytes;		/* bytes in queue */
+	int	limit;		/* max bytes in queue */
+	int	state;
+
+	QLock	rlock;		/* mutex for readers */
+	QLock	wlock;		/* mutex for writers */
+	Rendez	r;
 };
 
-struct Queue
+enum
 {
-	Blist;
-	int	flag;
-	Qinfo	*info;			/* line discipline definition */
-	Queue	*other;			/* opposite direction, same line discipline */
-	Queue	*next;			/* next queue in the stream */
-	void	(*put)(Queue*, Block*);
-	QLock	rlock;			/* mutex for processes sleeping at r */
-	Rendez	r;			/* standard place to wait for flow control */
-	Rendez	*rp;			/* where flow control wakeups go to */
-	void	*ptr;			/* private info for the queue */
-};
-
-struct Stream
-{
-	QLock;				/* structure lock */
-	Stream	*next;
-	short	inuse;			/* number of processes in stream */
-	short	opens;			/* number of processes with stream open */
-	ushort	hread;			/* number of reads after hangup */
-	ushort	type;			/* correlation with Chan */
-	ushort	dev;			/* ... */
-	ushort	id;			/* ... */
-	QLock	rdlock;			/* read lock */
-	Queue	*procq;			/* write queue at process end */
-	Queue	*devq;			/* read queue at device end */
-	Block	*err;			/* error message from down stream */
-	int	flushmsg;		/* flush up till the next delimiter */
+	Qcsleep=1,	/* consumer sleeping */
 };
 
 /*
- *  useful stream macros
+ *  Macros to manage Qid's used for multiplexed devices
  */
-#define	RD(q)		((q)->other < (q) ? (q->other) : q)
-#define	WR(q)		((q)->other > (q) ? (q->other) : q)
 #define STREAMTYPE(x)	((x)&0x1f)
 #define STREAMID(x)	(((x)&~CHDIR)>>5)
 #define STREAMQID(i,t)	(((i)<<5)|(t))
-#define PUTNEXT(q,b)	(*(q)->next->put)((q)->next, b)
-#define BLEN(b)		((b)->wptr - (b)->rptr)
-#define QFULL(q)	((q)->flag & QHIWAT)
-#define FLOWCTL(q,b)	{ if(QFULL(q->next)) flowctl(q,b); else PUTNEXT(q,b);}
-
-/*
- *  stream file qid's & high water mark
- */
+#define BLEN(b)		((b)->wp - (b)->rp)
 enum
 {
 	Shighqid	= STREAMQID(1,0) - 1,
 	Sdataqid	= Shighqid,
 	Sctlqid		= Sdataqid-1,
 	Slowqid		= Sctlqid,
-	Streamhi	= (32*1024),	/* byte count high water mark */
-	Streambhi	= 128,		/* block count high water mark */
 };
 
+
 /*
  *  a multiplexed network
  */

@@ 1,12 1,9 @@
 void		alarmkproc(void*);
-Block*		allocb(ulong);
 int		anyready(void);
 Image*		attachimage(int, Chan*, ulong, ulong);
 void		bitdebug(void);
 void		bitdepth(void);
 void		bitreverse(uchar*, int);
-int		blen(Block *);
-int		bround(Block *, int);
 void		buzz(int, int);
 void		cachedel(Image*, ulong);
 void		cachepage(Page*, Image*);


@@ 32,7 29,6 @@ void		confinit(void);
 void		confinit1(int);
 int		consactive(void);
 void		consdebug(void);
-Block*		copyb(Block*, int);
 void		copypage(Page*, Page*);
 void		cursoron(int);
 void		cursoroff(int);


@@ 63,29 59,23 @@ void		error(char*);
 long		execregs(ulong, ulong, ulong);
 void		exhausted(char*);
 void		exit(int);
-Block*		expandb(Block*, int);
 int		fault(ulong, int);
 void		fdclose(int, int);
 Chan*		fdtochan(int, int, int, int);
 int		fixfault(Segment*, ulong, int, int);
-void		flowctl(Queue*, Block*);
 void		flushmmu(void);
 void		forkchild(Proc*, Ureg*);
 void		forkret(void);
 void		free(void*);
-void		freeb(Block*);
 int		freebroken(void);
 void		freechan(Chan*);
 void		freepte(Segment*, Pte*);
 void		freesegs(int);
-Block*		getb(Blist*);
 int		getc(IOQ*);
 void		getcolor(ulong, ulong*, ulong*, ulong*);
 int		getfields(char*, char**, int, char);
-Block*		getq(Queue*);
 int		gets(IOQ*, void*, int);
 void		gotolabel(Label*);
-Block*		grabq(Queue*);
 int		haswaitq(void*);
 int		hwcursmove(int, int);
 int		hwcursset(uchar*, uchar*, int, int);


@@ 145,17 135,13 @@ Mount*		newmount(Mhead*, Chan*, int, char*);
 Page*		newpage(int, Segment **, ulong);
 Pgrp*		newpgrp(void);
 Proc*		newproc(void);
-void		newqinfo(Qinfo*);
 Segment*	newseg(int, ulong, ulong);
 char*		nextelem(char*, char*);
 void		nexterror(void);
-int		nodelims(Stream*);
 int		notify(Ureg*);
 int		nrand(int);
-void		nullput(Queue*, Block*);
 int		okaddr(ulong, ulong, int);
 int		openmode(ulong);
-Block*		padb(Block*, int);
 void		pageinit(void);
 void		panic(char*, ...);
 void		pexit(char*, int);


@@ 175,15 161,10 @@ void		procinit0(void);
 Proc*		proctab(int);
 Pte*		ptealloc(void);
 Pte*		ptecpy(Pte*);
-Block*		pullup(Block *, int);
-Queue*		pushq(Stream*, Qinfo*);
-int		putb(Blist*, Block*);
-void		putbq(Blist*, Block*);
 int		putc(IOQ*, int);
 void		putimage(Image*);
 void		putmmu(ulong, ulong, Page*);
 void		putpage(Page*);
-int		putq(Queue*, Block*);
 void		puts(IOQ*, void*, int);
 void		putseg(Segment*);
 void		putstr(char*);


@@ 191,7 172,6 @@ void		putstr(char*);
 void		putstrn(char*, long);
 void		putswap(Page*);
 ulong		pwait(Waitmsg*);
-Qinfo*		qinfofind(char*);
 void		qlock(QLock*);
 void		qunlock(QLock*);
 int		readnum(ulong, char*, ulong, ulong, int);


@@ 244,20 224,6 @@ void*		smalloc(ulong);
 int		splhi(void);
 int		spllo(void);
 void		splx(int);
-int		streamclose(Chan*);
-int		streamclose1(Stream*);
-int		streamenter(Stream*);
-void		streamexit(Stream*);
-Devgen		streamgen;
-void		streaminit(void);
-void		streaminit0(void);
-Stream*		streamnew(ushort, ushort, ushort, Qinfo*, int);
-void		streamopen(Chan*, Qinfo*);
-int		streamparse(char*, Block*);
-long		streamread(Chan*, void*, long);
-void		streamstat(Chan*, char*, char*, long);
-long		streamwrite(Chan*, void*, long, int);
-void		naildownstream(Stream*);
 void		swapinit(void);
 long		syscall(Ureg*);
 void		tsleep(Rendez*, int (*)(void*), void*, int);

@@ 1,1324 1,103 @@
-#include	"u.h"
 #include	"../port/lib.h"
 #include	"mem.h"
 #include	"dat.h"
 #include	"fns.h"
-#include	"io.h"
 #include	"../port/error.h"
-#include	"devtab.h"
 
-/*
- *  Part 1) Blocks
- */
-
-/*
- *  Allocate a block.  Put the data portion at the end of the smalloc'd
- *  chunk so that it can easily grow from the front to add protocol
- *  headers.  Thank Larry Peterson for the suggestion.
- */
-Block *
-allocb(ulong size)
-{
-	Block *bp;
-	uchar *base, *lim;
-
-	bp = smalloc(sizeof(Block)+size);
-
-	base = (uchar*)bp + sizeof(Block);
-	lim = (uchar*)bp + msize(bp);
-	bp->wptr = bp->rptr = lim - size;
-	bp->base = base;
-	bp->lim = lim;
-	bp->flags = 0;
-	bp->next = 0;
-	bp->list = 0;
-	bp->type = M_DATA;
-	return bp;
-}
-
-/*
- *  Free a block (or list of blocks).  Poison its pointers so that
- *  someone trying to access it after freeing will cause a panic.
- */
-void
-freeb(Block *bp)
-{
-	Block *next;
-
-	while(bp){
-		bp->rptr = 0;
-		bp->wptr = 0;
-		next = bp->next;
-		free(bp);
-		bp = next;	
-	}
-}
-
-/*
- *  Pad a block to the front with n bytes.  This is used to add protocol
- *  headers to the front of blocks.
- */
-Block *
-padb(Block *bp, int n)
-{
-	Block *nbp;
-
-	if(bp->base && bp->rptr-bp->base>=n){
-		bp->rptr -= n;
-		return bp;
-	} else {
-		nbp = allocb(n);
-		nbp->wptr = nbp->lim;
-		nbp->rptr = nbp->wptr - n;
-		nbp->next = bp;
-		return nbp;
-	}
-} 
-
-/*
- *  make sure the first block has n bytes
- */
-Block *
-pullup(Block *bp, int n)
-{
-	Block *nbp;
-	int i;
-
-	/*
-	 *  this should almost always be true, the rest it
-	 *  just for to avoid every caller checking.
-	 */
-	if(BLEN(bp) >= n)
-		return bp;
-
-	/*
-	 *  if not enough room in the first block,
-	 *  add another to the front of the list.
-	 */
-	if(bp->lim - bp->rptr < n){
-		nbp = allocb(n);
-		nbp->next = bp;
-		bp = nbp;
-	}
-
-	/*
-	 *  copy bytes from the trailing blocks into the first
-	 */
-	n -= BLEN(bp);
-	while(nbp = bp->next){
-		i = BLEN(nbp);
-		if(i >= n) {
-			memmove(bp->wptr, nbp->rptr, n);
-			bp->wptr += n;
-			nbp->rptr += n;
-			return bp;
-		} else {
-			memmove(bp->wptr, nbp->rptr, i);
-			bp->wptr += i;
-			bp->next = nbp->next;
-			nbp->next = 0;
-			freeb(nbp);
-			n -= i;
-		}
-	}
-	freeb(bp);
-	return 0;
-}
-
-/*
- *  return the number of data bytes of a list of blocks
- */
-int
-blen(Block *bp)
-{
-	int len;
-
-	len = 0;
-	while(bp) {
-		len += BLEN(bp);
-		bp = bp->next;
-	}
-
-	return len;
-}
-
-/*
- *  round a block chain to some even number of bytes.  Used
- *  by devip.c becuase all IP packets must have an even number
- *  of bytes.
- *
- *  The last block in the returned chain will have S_DELIM set.
- */
-int
-bround(Block *bp, int amount)
-{
-	Block *last;
-	int len, pad;
-
-	len = 0;
-	SET(last);	/* Ken's magic */
-
-	while(bp) {
-		len += BLEN(bp);
-		last = bp;
-		bp = bp->next;
-	}
-
-	pad = ((len + amount) & ~amount) - len;
-	if(pad) {
-		if(last->lim - last->wptr >= pad){
-			memset(last->wptr, 0, pad);
-			last->wptr += pad;
-		} else {
-			last->next = allocb(pad);
-			last->flags &= ~S_DELIM;
-			last = last->next;
-			last->wptr += pad;
-			last->flags |= S_DELIM;
-		}
-	}
-
-	return len + pad;
-}
-
-/*
- *  expand a block list to be one block, len bytes long.  used by
- *  ethernet routines.
- */
-Block*
-expandb(Block *bp, int len)
-{
-	Block *nbp, *new;
-	int i;
-	ulong delim = 0;
-
-	new = allocb(len);
-	if(new == 0){
-		freeb(bp);
-		return 0;
-	}
-
-	/*
-	 *  copy bytes into new block
-	 */
-	for(nbp = bp; len>0 && nbp; nbp = nbp->next){
-		delim = nbp->flags & S_DELIM;
-		i = BLEN(nbp);
-		if(i > len) {
-			memmove(new->wptr, nbp->rptr, len);
-			new->wptr += len;
-			break;
-		} else {
-			memmove(new->wptr, nbp->rptr, i);
-			new->wptr += i;
-			len -= i;
-		}
-	}
-	if(len){
-		memset(new->wptr, 0, len);
-		new->wptr += len;
-	}
-	new->flags |= delim;
-	freeb(bp);
-	return new;
-
-}
-
-/*
- *  make a copy of the first 'count' bytes of a block chain.  Use
- *  by transport protocols.
- */
-Block *
-copyb(Block *bp, int count)
-{
-	Block *nb, *head, **p;
-	int l;
-
-	p = &head;
-	while(count) {
-		l = BLEN(bp);
-		if(count < l)
-			l = count;
-		nb = allocb(l);
-		if(nb == 0)
-			panic("copyb.1");
-		memmove(nb->wptr, bp->rptr, l);
-		nb->wptr += l;
-		count -= l;
-		if(bp->flags & S_DELIM)
-			nb->flags |= S_DELIM;
-		*p = nb;
-		p = &nb->next;
-		bp = bp->next;
-		if(bp == 0)
-			break;
-	}
-	if(count) {
-		nb = allocb(count);
-		if(nb == 0)
-			panic("copyb.2");
-		memset(nb->wptr, 0, count);
-		nb->wptr += count;
-		nb->flags |= S_DELIM;
-		*p = nb;
-	}
-	if(blen(head) == 0)
-		print("copyb: zero length\n");
-
-	return head;
-}
-
-/*
- *  Part 2) Queues
- */
-
-/*
- *  process end line discipline
- */
-static Streamput stputq;
-Qinfo procinfo =
-{
-	stputq,
-	nullput,
-	0,
-	0,
-	"process"
-};
-
-/*
- *  line disciplines that can be pushed
- */
-static Qinfo *lds;
-
-/*
- *  make known a stream module and call its initialization routine, if
- *  it has one.
- */
-void
-newqinfo(Qinfo *qi)
-{
-	if(qi->next)
-		panic("newqinfo: already configured");
-
-	qi->next = lds;
-	lds = qi;
-	if(qi->reset)
-		(*qi->reset)();
-}
+static Queue *freed;
 
 /*
- *  find the info structure for line discipline 'name'
- */
-Qinfo *
-qinfofind(char *name)
-{
-	Qinfo *qi;
-
-	if(name == 0)
-		return 0;
-	for(qi = lds; qi; qi = qi->next)
-		if(strcmp(qi->name, name)==0)
-			return qi;
-	return 0;
-}
-
-/*
- *  allocate a pair of queues.  flavor them with the requested put routines.
- *  the `QINUSE' flag on the read side is the only one used.
- */
-static Queue *
-allocq(Qinfo *qi)
-{
-	Queue *q, *wq;
-
-	q = smalloc(2*sizeof(Queue));
-
-	q->flag = QINUSE;
-	q->r.p = 0;
-	q->info = qi;
-	q->put = qi->iput;
-	q->len = q->nb = 0;
-	q->ptr = 0;
-	q->rp = &q->r;
-	wq = q->other = q + 1;
-
-	wq->flag = QINUSE;
-	wq->r.p = 0;
-	wq->info = qi;
-	wq->put = qi->oput;
-	wq->other = q;
-	wq->ptr = 0;
-	wq->len = wq->nb = 0;
-	wq->rp = &wq->r;
-
-	return q;
-}
-
-/*
- *  free a queue
- */
-static void
-freeq(Queue *q)
-{
-	Block *bp;
-
-	q = RD(q);
-	while(bp = getq(q))
-		freeb(bp);
-	q = WR(q);
-	while(bp = getq(q))
-		freeb(bp);
-	free(RD(q));
-}
-
-/*
- *  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);
-}
-
-/*
- *  push a queue onto a stream referenced by the proc side write q
- */
-Queue *
-pushq(Stream* s, Qinfo *qi)
-{
-	Queue *q;
-	Queue *nq;
-
-	q = RD(s->procq);
-
-	/*
-	 *  make the new queue
-	 */
-	nq = allocq(qi);
-
-	/*
-	 *  push
-	 */
-	qlock(s);
-	RD(nq)->next = q;
-	RD(WR(q)->next)->next = RD(nq);
-	WR(nq)->next = WR(q)->next;
-	WR(q)->next = WR(nq);
-	qunlock(s);
-
-	if(qi->open)
-		(*qi->open)(RD(nq), s);
-
-	return WR(nq)->next;
-}
-
-/*
- *  pop off the top line discipline
+ *  Interrupt handlers use freeb() to release blocks.  They are
+ *  garbage collected by the kproc running bgc().
  */
 static void
-popq(Stream *s)
+bgc(void *arg)
 {
-	Queue *q;
-
-	if(waserror()){
-		qunlock(s);
-		nexterror();
-	}
-	qlock(s);
-	if(s->procq->next == WR(s->devq))
-		error(Ebadld);
-	q = s->procq->next;
-	if(q->info->close)
-		(*q->info->close)(RD(q));
-	s->procq->next = q->next;
-	RD(q->next)->next = RD(s->procq);
-	qunlock(s);
-	freeq(q);
-}
+	Block *b, *nb;
 
-/*
- *  add a block (or list of blocks) to the end of a queue.  return true
- *  if one of the blocks contained a delimiter. 
- */
-int
-putq(Queue *q, Block *bp)
-{
-	int delim;
-
-	lock(q);
-	if(q->first)
-		q->last->next = bp;
-	else
-		q->first = bp;
-	q->len += BLEN(bp);
-	q->nb++;
-	delim = bp->flags & S_DELIM;
-	while(bp->next) {
-		bp = bp->next;
-		q->len += BLEN(bp);
-		q->nb++;
-		delim |= bp->flags & S_DELIM;
-	}
-	q->last = bp;
-	if(q->len >= Streamhi || q->nb >= Streambhi)
-		q->flag |= QHIWAT;
-	unlock(q);
-	return delim;
-}
-
-int
-putb(Blist *q, Block *bp)
-{
-	int delim;
-
-	if(q->first)
-		q->last->next = bp;
-	else
-		q->first = bp;
-	q->len += BLEN(bp);
-	delim = bp->flags & S_DELIM;
-	while(bp->next) {
-		bp = bp->next;
-		q->len += BLEN(bp);
-		delim |= bp->flags & S_DELIM;
-	}
-	q->last = bp;
-	return delim;
-}
-
-/*
- *  add a block to the start of a queue 
- */
-void
-putbq(Blist *q, Block *bp)
-{
-	lock(q);
-	if(q->first)
-		bp->next = q->first;
-	else
-		q->last = bp;
-	q->first = bp;
-	q->len += BLEN(bp);
-	q->nb++;
-	unlock(q);
-}
-
-/*
- *  remove the first block from a queue
- */
-Block *
-getq(Queue *q)
-{
-	Block *bp;
-
-	lock(q);
-	bp = q->first;
-	if(bp) {
-		q->first = bp->next;
-		if(q->first == 0)
-			q->last = 0;
-		q->len -= BLEN(bp);
-		q->nb--;
-		if((q->flag&QHIWAT) && q->len<Streamhi/2 && q->nb<Streambhi/2 &&q->other){
-			wakeup(q->other->next->other->rp);
-			q->flag &= ~QHIWAT;
-		}
-		bp->next = 0;
-	}
-	unlock(q);
-	return bp;
-}
+	USED(arg);
+	for(;;){
+		tsleep(&freed->r, return0, 0, 500);
+		if(freed->first == 0)
+			continue;
 
-/*
- *  grab all the blocks in a queue
- */
-Block *
-grabq(Queue *q)
-{
-	Block *bp;
+		x = slphi();
+		lock(&freed);
+		b = freed->first;
+		freed->first = freed->last = 0;;
+		unlock(&freed);
+		spllo();
 
-	lock(q);
-	bp = q->first;
-	if(bp){
-		q->first = 0;
-		q->last = 0;
-		q->len = 0;
-		q->nb = 0;
-		if(q->flag&QHIWAT){
-			wakeup(q->other->next->other->rp);
-			q->flag &= ~QHIWAT;
+		for(; b; b = nb){
+			nb = b->next;
+			free(b);
 		}
 	}
-	unlock(q);
-	return bp;
 }
 
-/*
- *  remove the first block from a list of blocks
- */
-Block *
-getb(Blist *q)
-{
-	Block *bp;
-
-	bp = q->first;
-	if(bp) {
-		q->first = bp->next;
-		if(q->first == 0)
-			q->last = 0;
-		q->len -= BLEN(bp);
-		bp->next = 0;
-	}
-	return bp;
-}
-
-
-/*
- *  put a block into the bit bucket
- */
 void
-nullput(Queue *q, Block *bp)
+freeb(Block *b)
 {
-	USED(q);
-	if(bp->type == M_HANGUP)
-		freeb(bp);
-	else {
-		freeb(bp);
-		error(Ehungup);
-	}
+	lock(&freed);
+	b->next = freed->first;
+	freed->first = b;
+	unlock(&freed);
 }
 
-/*
- *  Part 3) Streams
- */
-
-/*
- *  the per stream directory structure
- */
-Dirtab streamdir[]={
-	"data",		{Sdataqid},	0,			0600,
-	"ctl",		{Sctlqid},	0,			0600,
-};
-
-/*
- *  hash buckets containing all streams
- */
-enum
-{
-	Nbits=	5,
-	Nhash=	1<<Nbits,
-	Nmask=	Nhash-1,
-};
-typedef struct Sthash Sthash;
-struct Sthash
-{
-	QLock;
-	Stream	*s;
-};
-static Sthash ht[Nhash];
-
-static void	hangup(Stream*);
-
-/*
- *  A stream device consists of the contents of streamdir plus
- *  any directory supplied by the actual device.
- *
- *  values of s:
- * 	0 to ntab-1 apply to the auxiliary directory.
- *	ntab to ntab+Shighqid-Slowqid+1 apply to streamdir.
- */
-int
-streamgen(Chan *c, Dirtab *tab, int ntab, int s, Dir *dp)
-{
-	if(s < ntab)
-		tab = &tab[s];
-	else if(s < ntab + Shighqid - Slowqid + 1)
-		tab = &streamdir[s - ntab];
-	else
-		return -1;
-
-	devdir(c, (Qid){STREAMQID(STREAMID(c->qid.path),tab->qid.path), 0}, 
-		tab->name, tab->length, eve, tab->perm, dp);
-	return 1;
-}
-
-/*
- *  return a hash bucket for a stream
- */
-static Sthash*
-hash(int type, int dev, int id)
-{
-	return &ht[(type*7*7 + dev*7 + id) & Nmask];
-}
-
-/*
- *  create a new stream, if noopen is non-zero, don't increment the open count
- */
-Stream *
-streamnew(ushort type, ushort dev, ushort id, Qinfo *qi, int noopen)
-{
-	Stream *s;
-	Queue *q;
-	Sthash *hb;
-
-	hb = hash(type, dev, id);
-
-	/*
-	 *  if the stream already exists, just increment the reference counts.
-	 */
-	qlock(hb);
-	for(s = hb->s; s; s = s->next) {
-		if(s->type == type && s->dev == dev && s->id == id){
-			s->inuse++;
-			qunlock(hb);
-			if(noopen == 0){
-				qlock(s);
-				s->opens++;
-				qunlock(s);
-			}
-			return s;
-		}
-	}
-
-	/*
-	 *  create and init a new stream
-	 */
-	s = smalloc(sizeof(Stream));
-	s->inuse = 1;
-	s->type = type;
-	s->dev = dev;
-	s->id = id;
-	s->err = 0;
-	s->hread = 0;
-	s->next = hb->s;
-	hb->s = s;
-
-	/*
-	 *  The ordering of these 2 instructions is very important.
-	 *  It makes sure we finish the stream initialization before
-	 *  anyone else can access it.
-	 */
-	qlock(s);
-	qunlock(hb);
-
-	if(waserror()){
-		qunlock(s);
-		streamclose1(s);
-		nexterror();
-	}
-
-	/*
- 	 *  hang a device and process q off the stream
-	 */
-	if(noopen)
-		s->opens = 0;
-	else
-		s->opens = 1;
-	q = allocq(&procinfo);
-	WR(q)->ptr = s;
-	RD(q)->ptr = s;
-	s->procq = WR(q);
-	q = allocq(qi);
-	s->devq = RD(q);
-	WR(s->procq)->next = WR(s->devq);
-	RD(s->procq)->next = 0;
-	RD(s->devq)->next = RD(s->procq);
-	WR(s->devq)->next = 0;
-
-	if(qi->open)
-		(*qi->open)(RD(s->devq), s);
-
-	qunlock(s);
-	poperror();
-	return s;
-}
-
-/*
- *  Associate a stream with a channel
- */
 void
-streamopen(Chan *c, Qinfo *qi)
+blockinit(void)
 {
-	c->stream = streamnew(c->type, c->dev, STREAMID(c->qid.path), qi, 0);
+	/* start garbage collector */
+	kproc("buffer", bgc, 0);
 }
 
 /*
- *  Enter a stream only if the stream exists and is open.  Increment the
- *  reference count so it can't disappear under foot.
- *
- *  Return -1 if the stream no longer exists or is not opened.
+ *  allocate queues and blocks
  */
-int
-streamenter(Stream *s)
-{
-	Sthash *hb;
-	Stream *ns;
-
-	hb = hash(s->type, s->dev, s->id);
-	qlock(hb);
-	for(ns = hb->s; ns; ns = ns->next)
-		if(s->type == ns->type && s->dev == ns->dev && s->id == ns->id){
-			s->inuse++;
-			qunlock(hb);
-			if(s->opens == 0){
-				streamexit(s);
-				return -1;
-			}
-			return 0;
-		}
-	qunlock(hb);
-	return -1;
-}
-
-/*
- *  Decrement the reference count on a stream.  If the count is
- *  zero, free the stream.
- */
-void
-streamexit(Stream *s)
+Queue*
+allocq(int limit)
 {
 	Queue *q;
-	Queue *nq;
-	Sthash *hb;
-	Stream **l, *ns;
-
-	hb = hash(s->type, s->dev, s->id);
-	qlock(hb);
-	if(s->inuse-- == 1){
-		if(s->opens != 0)
-			panic("streamexit %d %s\n", s->opens, s->devq->info->name);
-
-		/*
-		 *  ascend the stream freeing the queues
-		 */
-		for(q = s->devq; q; q = nq){
-			nq = q->next;
-			freeq(q);
-		}
-		if(s->err)
-			freeb(s->err);
-
-		/*
-		 *  unchain it from the hash bucket and free
-		 */
-		l = &hb->s;
-		for(ns = hb->s; ns; ns = ns->next){
-			if(s == ns){
-				*l = s->next;
-				break;
-			}
-			l = &ns->next;
-		}
-		free(s);
-	}
-	qunlock(hb);
-}
 
-/*
- *  nail down a stream so that it can't be closed
- */
-void
-naildownstream(Stream *s)
-{
-	s->opens++;
-	s->inuse++;
+	q = smalloc(sizeof(Queue));
+	q->limit = limit;
 }
 
-/*
- *  Decrement the open count.  When it goes to zero, call the close
- *  routines for each queue in the stream.
- */
-int
-streamclose1(Stream *s)
+Block*
+allocb(int size)
 {
-	Queue *q, *nq;
-	int rv;
+	Block *b;
 
-	/*
-	 *  decrement the open count
-	 */
-	qlock(s);
-	if(s->opens-- == 1){
-		/*
-		 *  descend the stream closing the queues
-		 */
-		for(q = s->procq; q; q = q->next){
-			if(!waserror()){
-				if(q->info->close)
-					(*q->info->close)(q->other);
-				poperror();
-			}
-			WR(q)->put = nullput;
+	b = alloc(sizeof(Block) + size);
+	if(b == 0)
+		exhausted("blocks");
 
-			/*
-			 *  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);
-		}
-	}
-	rv = s->opens;
-	qunlock(s);
+	b->base = (uchar*)(b+1);
+	b->rp = b->wp = b->base;
+	b->lim = b->base + size;
 
-	/*
-	 *  leave it and free it
-	 */
-	streamexit(s);
-	return rv;
-}
-int
-streamclose(Chan *c)
-{
-	/*
-	 *  if no stream, ignore it
-	 */
-	if(!c->stream)
-		return 0;
-	return streamclose1(c->stream);
+	return b;
 }
 
 /*
- *  put a block to be read into the queue.  wakeup any waiting reader
+ *  copy out of a queue, returns # bytes copied
  */
-void
-stputq(Queue *q, Block *bp)
+int
+consume(Queue *q, uchar *p, int len, int drop)
 {
-	int awaken;
-	Stream *s;
+	Block *b;
+	int n;
 
-	if(bp->type == M_HANGUP){
-		s = q->ptr;
-		if(bp->rptr<bp->wptr && s->err==0)
-			s->err = bp;
-		else
-			freeb(bp);
-		q->flag |= QHUNGUP;
-		q->other->flag |= QHUNGUP;
-		wakeup(q->other->rp);
-		awaken = 1;
-	} else {
-		lock(q);
-		if(q->first)
-			q->last->next = bp;
-		else
-			q->first = bp;
-		q->len += BLEN(bp);
-		q->nb++;
-		awaken = bp->flags & S_DELIM;
-		while(bp->next) {
-			bp = bp->next;
-			q->len += BLEN(bp);
-			q->nb++;
-			awaken |= bp->flags & S_DELIM;
-		}
-		q->last = bp;
-		if(q->len >= Streamhi || q->nb >= Streambhi){
-			q->flag |= QHIWAT;
-			awaken = 1;
-		}
+	lock(q);
+	b = q->first;
+	if(b == 0){
+		q->state |= Qcsleep;
 		unlock(q);
+		return -1;
 	}
-	if(awaken)
-		wakeup(q->rp);
-}
-
-/*
- *  return the stream id
- */
-long
-streamctlread(Chan *c, void *vbuf, long n)
-{
-	char *buf = vbuf;
-	char num[32];
-	Stream *s;
-
-	s = c->stream;
-	if(STREAMTYPE(c->qid.path) == Sctlqid){
-		sprint(num, "%d", s->id);
-		return readstr(c->offset, buf, n, num);
-	} else {
-		if(CHDIR & c->qid.path)
-			return devdirread(c, vbuf, n, 0, 0, streamgen);
-		else
-			panic("streamctlread");
-	}
-	return 0;	/* not reached */
-}
-
-/*
- *  return true if there is an output buffer available
- */
-static int
-isinput(void *x)
-{
-	Queue *q;
-
-	q = (Queue *)x;
-	return (q->flag&QHUNGUP) || q->first!=0;
-}
-
-/*
- *  read until we fill the buffer or until a DELIM is encountered
- */
-long
-streamread(Chan *c, void *vbuf, long n)
-{
-	Block *bp;
-	Block *tofree;
-	Stream *s;
-	Queue *q;
-	int left, i;
-	uchar *buf = vbuf;
-
-	if(STREAMTYPE(c->qid.path) != Sdataqid)
-		return streamctlread(c, vbuf, n);
-
-	/*
-	 *  one reader at a time
-	 */
-	s = c->stream;
-	left = n;
-	qlock(&s->rdlock);
-	tofree = 0;
-	q = 0;
-	if(waserror()){
-		/*
-		 *  put any partially read message back into the
-		 *  queue
-		 */
-		while(tofree){
-			bp = tofree;
-			tofree = bp->next;
-			bp->next = 0;
-			putbq(q, bp);
-		}
-		qunlock(&s->rdlock);
-		nexterror();
-	}
-
-	/*
-	 *  sleep till data is available
-	 */
-	q = RD(s->procq);
-	while(left){
-		bp = getq(q);
-		if(bp == 0){
-			if(q->flag & QHUNGUP){
-				if(s->err)
-					error((char*)s->err->rptr);
-				else if(s->hread++<3)
-					break;
-				else
-					error(Ehungup);
-			}
-			q->rp = &q->r;
-			sleep(q->rp, isinput, (void *)q);
-			continue;
-		}
-
-		i = BLEN(bp);
-		if(i <= left){
-			memmove(buf, bp->rptr, i);
-			left -= i;
-			buf += i;
-			bp->next = tofree;
-			tofree = bp;
-			if(bp->flags & S_DELIM)
-				break;
-		} else {
-			memmove(buf, bp->rptr, left);
-			bp->rptr += left;
-			putbq(q, bp);
-			left = 0;
-		}
-	}
-
-	/*
-	 *  free completely read blocks
-	 */
-	if(tofree)
-		freeb(tofree);
-
-	qunlock(&s->rdlock);
-	poperror();
-	return n - left;	
-}
-
-/*
- *  look for an instance of the line discipline `name' on
- *  the stream `s'
- */
-void
-qlook(Stream *s, char *name)
-{
-	Queue *q;
-
-	for(q = s->procq; q; q = q->next){
-		if(strcmp(q->info->name, name) == 0)
-			return;
-
-		/*
-		 *  this may be 2 streams joined device end to device end
-		 */
-		if(q == s->devq->other)
-			break;
-	}
-	error(Ebadarg);
-}
-
-/*
- *  Handle a ctl request.  Streamwide requests are:
- *
- *	hangup			-- send an M_HANGUP up the stream
- *	push ldname		-- push the line discipline named ldname
- *	pop			-- pop a line discipline
- *	look ldname		-- look for a line discipline
- *
- *  This routing is entered with s->wrlock'ed and must unlock.
- */
-static long
-streamctlwrite(Chan *c, void *a, long n)
-{
-	Qinfo *qi;
-	Block *bp;
-	Stream *s;
-
-	if(STREAMTYPE(c->qid.path) != Sctlqid)
-		panic("streamctlwrite %lux", c->qid);
-	s = c->stream;
-
-	/*
-	 *  package
-	 */
-	bp = allocb(n+1);
-	memmove(bp->wptr, a, n);
-	bp->wptr[n] = 0;
-	bp->wptr += n + 1;
-
-	/*
-	 *  check for standard requests
-	 */
-	if(streamparse("hangup", bp)){
-		hangup(s);
-		freeb(bp);
-	} else if(streamparse("push", bp)){
-		qi = qinfofind((char *)bp->rptr);
-		if(qi == 0)
-			error(Ebadld);
-		pushq(s, qi);
-		freeb(bp);
-	} else if(streamparse("pop", bp)){
-		popq(s);
-		freeb(bp);
-	} else if(streamparse("look", bp)){
-		qlook(s, (char *)bp->rptr);
-		freeb(bp);
+	n = BLEN(b);
+	if(n < len){
+		memmove(p, b->rp, n);
 	} else {
-		bp->type = M_CTL;
-		bp->flags |= S_DELIM;
-		PUTNEXT(s->procq, bp);
-	}
-
-	return n;
-}
-
-/*
- *  wait till there's room in the next stream
- */
-static int
-notfull(void *arg)
-{
-	return !QFULL((Queue *)arg);
-}
-void
-flowctl(Queue *q, Block *bp)
-{
-	if(bp->type != M_HANGUP){
-		qlock(&q->rlock);
-		if(waserror()){
-			qunlock(&q->rlock);
-			freeb(bp);
-			nexterror();
-		}
-		q->rp = &q->r;
-		sleep(q->rp, notfull, q->next);
-		qunlock(&q->rlock);
-		poperror();
-	}
-	PUTNEXT(q, bp);
-}
-
-/*
- *  send the request as a single delimited block
- */
-long
-streamwrite(Chan *c, void *a, long n, int docopy)
-{
-	Stream *s;
-	Queue *q;
-	long rem;
-	int i;
-	Block *bp;
-	char *va;
-
-	/*
-	 *  docopy will get used if I ever figure out when to avoid copying
-	 *  data. -- presotto
-	 */
-	USED(docopy);
-
-	s = c->stream;
-
-	/*
-	 *  decode the qid
-	 */
-	if(STREAMTYPE(c->qid.path) != Sdataqid)
-		return streamctlwrite(c, a, n);
-
-	/*
-	 *  No writes allowed on hungup channels
-	 */
-	q = s->procq;
-	if(q->other->flag & QHUNGUP){
-		if(s->err)
-			error((char*)(s->err->rptr));
-		else
-			error(Ehungup);
-	}
-
-	/*
-	 *  Write the message using blocks <= Streamhi bytes longs
-	 */
-	va = a;
-	rem = n;
-	for(;;){
-		if(rem > Streamhi)
-			i = Streamhi;
-		else
-			i = rem;
-		bp = allocb(i);
-		memmove(bp->wptr, va, i);
-		bp->wptr += i;
-		va += i;
-		rem -= i;
-		if(rem > 0){
-			FLOWCTL(q, bp);
-		} else {
-			bp->flags |= S_DELIM;
-			FLOWCTL(q, bp);
-			break;
-		}
-	}
-	return n;
-}
-
-/*
- *  stat a stream.  the length is the number of bytes up to the
- *  first delimiter.
- */
-void
-streamstat(Chan *c, char *db, char *name, long perm)
-{
-	Dir dir;
-	Stream *s;
-	Queue *q;
-	Block *bp;
-	long n;
-
-	s = c->stream;
-	if(s == 0)
-		n = 0;
-	else {
-		q = RD(s->procq);
-		lock(q);
-		for(n=0, bp=q->first; bp; bp = bp->next){
-			n += BLEN(bp);
-			if(bp->flags&S_DELIM)
-				break;
-		}
-		unlock(q);
-	}
-
-	devdir(c, c->qid, name, n, eve, perm, &dir);
-	convD2M(&dir, db);
-}
-
-/*
- *  send a hangup up a stream
- */
-static void
-hangup(Stream *s)
-{
-	Block *bp;
-
-	bp = allocb(0);
-	bp->type = M_HANGUP;
-	(*s->devq->put)(s->devq, bp);
-}
-
-/*
- *  parse a string and return a pointer to the second element if the 
- *  first matches name.  bp->rptr will be updated to point to the
- *  second element.
- *
- *  return 0 if no match.
- *
- *  it is assumed that the block data is null terminated.  streamwrite
- *  guarantees this.
- */
-int
-streamparse(char *name, Block *bp)
-{
-	int len;
-
-	len = strlen(name);
-	if(BLEN(bp) < len)
-		return 0;
-	if(strncmp(name, (char *)bp->rptr, len)==0){
-		if(bp->rptr[len] == ' ')
-			bp->rptr += len+1;
-		else if(bp->rptr[len])
-			return 0;
-		else
-			bp->rptr += len;
-		while(*bp->rptr==' ' && bp->wptr>bp->rptr)
-			bp->rptr++;
-		return 1;
-	}
-	return 0;
-}
-
-/*
- *  like andrew's getmfields but no hidden state
- */
-int
-getfields(char *lp, char **fields, int n, char sep)
-{
-	int i;
-
-	for(i=0; lp && *lp && i<n; i++){
-		while(*lp == sep)
-			*lp++=0;
-		if(*lp == 0)
-			break;
-		fields[i]=lp;
-		while(*lp && *lp != sep)
-			lp++;
-	}
-	return i;
+		memmove(p, b->rp, len);
 }