## diffname port/stream.c 1990/0227
## diff -e /dev/null /n/bootesdump/1990/0227/sys/src/9/mips/stream.c
0a
#include	"u.h"
#include	"lib.h"
#include	"mem.h"
#include	"dat.h"
#include	"fns.h"
#include	"io.h"
#include	"errno.h"
#include	"devtab.h"

static void stputq(Queue*, Block*);
Qinfo procinfo = { stputq, nullput, 0, 0, "process" } ;
extern Qinfo noetherinfo;

static Qinfo *lds[] = {
	&noetherinfo,
	0
};

enum {
	Nclass=4,
};

/*
 *  All stream structures are ialloc'd at boot time
 */
Stream *slist;
Queue *qlist;
Block *blist;
static Lock garbagelock;

/*
 *  The block classes.  There are Nclass block sizes, each with its own free list.
 *  All are ialloced at qinit() time.
 */
typedef struct {
	int	size;
	Queue;
} Bclass;
Bclass bclass[Nclass]={
	{ 0 },
	{ 64 },
	{ 512 },
	{ 4096 },
};

/*
 *  Allocate streams, queues, and blocks.  Allocate n block classes with
 *	1/2(m+1) to class m < n-1
 *	1/2(n-1) to class n-1
 */
void
streaminit(void)
{
	int class, i, n;
	Block *bp;
	Bclass *bcp;

	slist = (Stream *)ialloc(conf.nstream * sizeof(Stream), 0);
	qlist = (Queue *)ialloc(conf.nqueue * sizeof(Queue), 0);
	blist = (Block *)ialloc(conf.nblock * sizeof(Block), 0);
	bp = blist;
	n = conf.nblock;
	for(class = 0; class < Nclass; class++){
		if(class < Nclass-1)
			n = n/2;
		bcp = &bclass[class];
		for(i = 0; i < n; i++) {
			if(bcp->size)
				bp->base = (uchar *)ialloc(bcp->size, 0);
			bp->lim = bp->base + bcp->size;
			bp->flags = class;
			freeb(bp);
			bp++;
		}
	}
}


/*
 *  allocate a block
 */
static int
isblock(void *arg)
{
	Bclass *bcp;

	bcp = (Bclass *)arg;
	return bcp->first!=0;
}
Block *
allocb(ulong size)
{
	Block *bp;
	Bclass *bcp;
	int i;


	/*
	 *  map size to class
	 */
	for(bcp=bclass; bcp->size<size && bcp<&bclass[Nclass-1]; bcp++)
		;

	/*
	 *  look for a free block, garbage collect if there are none
	 */
	lock(bcp);
	while(bcp->first == 0){
		unlock(bcp);
		print("waiting for blocks\n");
		sleep(&bcp->r, isblock, (void *)bcp);
		lock(bcp);
	}
	bp = bcp->first;
	bcp->first = bp->next;
	if(bcp->first == 0)
		bcp->last = 0;
	unlock(bcp);

	/*
	 *  return an empty block
	 */
	bp->rptr = bp->wptr = bp->base;
	bp->next = 0;
	bp->type = M_DATA;
	bp->flags &= S_CLASS;
	return bp;
}

/*
 *  Free a block.  Poison its pointers so that someone trying to access
 *  it after freeing will cause a dump.
 */
void
freeb(Block *bp)
{
	Bclass *bcp;

	bcp = &bclass[bp->flags & S_CLASS];
	bp->rptr = bp->wptr = 0;
	lock(bcp);
	if(bcp->first)
		bcp->last->next = bp;
	else
		bcp->first = bp;
	bcp->last = bp;
	bp->next = 0;
	wakeup(&bcp->r);
	unlock(bcp);
}

/*
 *  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;

	for(q=qlist; q<&qlist[conf.nqueue]; q++, q++) {
		if(q->flag == 0){
			if(canlock(q)){
				if(q->flag == 0)
					break;
				unlock(q);
			}
		}
	}

	if(q == &qlist[conf.nqueue]){
		print("no more queues\n");
		error(0, Enoqueue);
	}

	q->flag = QINUSE;
	q->r.p = 0;
	q->info = qi;
	q->put = qi->iput;
	wq = q->other = q + 1;

	wq->r.p = 0;
	wq->info = qi;
	wq->put = qi->oput;
	wq->other = q;

	unlock(q);

	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);
	RD(q)->flag = 0;
}

/*
 *  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
	 */
	RD(nq)->next = q;
	RD(WR(q)->next)->next = RD(nq);
	WR(nq)->next = WR(q)->next;
	WR(q)->next = WR(nq);

	if(qi->open)
		(*qi->open)(RD(nq), s);

	return WR(nq)->next;
}

/*
 *  pop off the top line discipline
 */
static void
popq(Stream *s)
{
	Queue *q;

	if(s->procq->next == WR(s->devq))
		error(0, 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);
	freeq(q);
}

/*
 *  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;

	delim = 0;
	lock(q);
	if(q->first)
		q->last->next = bp;
	else
		q->first = bp;
	q->len += bp->wptr - bp->rptr;
	delim = bp->flags & S_DELIM;
	while(bp->next) {
		bp = bp->next;
		q->len += bp->wptr - bp->rptr;
		delim |= bp->flags & S_DELIM;
	}
	q->last = bp;
	if(q->len >= Streamhi)
		q->flag |= QHIWAT;
	unlock(q);
	return delim;
}
int
putb(Blist *q, Block *bp)
{
	int delim;

	delim = 0;
	if(q->first)
		q->last->next = bp;
	else
		q->first = bp;
	q->len += bp->wptr - bp->rptr;
	delim = bp->flags & S_DELIM;
	while(bp->next) {
		bp = bp->next;
		q->len += bp->wptr - bp->rptr;
		delim |= bp->flags & S_DELIM;
	}
	q->last = bp;
	bp->next = 0;
	return delim;
}

/*
 *  add a block to the start of a queue 
 */
static void
putbq(Blist *q, Block *bp)
{
	lock(q);
	if(q->first)
		bp->next = q->first;
	else
		q->last = bp;
	q->first = bp;
	q->len += bp->wptr - bp->rptr;
	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 -= bp->wptr - bp->rptr;
		if((q->flag&QHIWAT) && q->len < Streamhi/2){
			wakeup(&q->other->next->other->r);
			q->flag &= ~QHIWAT;
		}
		bp->next = 0;
	}
	unlock(q);
	return bp;
}
Block *
getb(Blist *q)
{
	Block *bp;

	bp = q->first;
	if(bp) {
		q->first = bp->next;
		if(q->first == 0)
			q->last = 0;
		q->len -= bp->wptr - bp->rptr;
		bp->next = 0;
	}
	return bp;
}

/*
 *  put a block into the bit bucket
 */
void
nullput(Queue *q, Block *bp)
{
	freeb(bp);
	error(0, Ehungup);
}

/*
 *  find the info structure for line discipline 'name'
 */
static Qinfo *
qinfofind(char *name)
{
	Qinfo **qip;

	if(name == 0)
		error(0, Ebadld);
	for(qip = lds; *qip; qip++)
		if(strcmp((*qip)->name, name)==0)
			return *qip;
	error(0, Ebadld);
}

/*
 *  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(bp->wptr - bp->rptr < 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;
		return 1;
	}
	return 0;
}

/*
 *  the per stream directory structure
 */
Dirtab streamdir[]={
	"data",		Sdataqid,	0,			0600,
	"ctl",		Sctlqid,	0,			0600,
};

/*
 *  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)
{
	Proc *p;
	char buf[NAMELEN];

	if(s < ntab)
		tab = &tab[s];
	else if(s < ntab + Shighqid - Slowqid + 1)
		tab = &streamdir[s - ntab];
	else
		return -1;

	devdir(c, STREAMQID(STREAMID(c->qid),tab->qid), tab->name, tab->length,
		tab->perm, dp);
	return 1;
}

/*
 *  create a new stream
 */
Stream *
streamnew(Chan *c, Qinfo *qi)
{
	Stream *s;
	Queue *q;

	/*
	 *  find a free stream struct
	 */
	for(s = slist; s < &slist[conf.nstream]; s++) {
		if(s->inuse == 0){
			if(canlock(s)){
				if(s->inuse == 0)
					break;
				unlock(s);
			}
		}
	}
	if(s == &slist[conf.nstream]){
		print("no more streams\n");
		error(0, Enostream);
	}
	if(waserror()){
		unlock(s);
		streamclose(c);
		nexterror();
	}

	/*
	 *  marry a stream and a channel
	 */
	if(c){
		c->stream = s;
		s->type = c->type;
		s->dev = c->dev;
		s->id = STREAMID(c->qid);
	} else
		s->type = -1;

	/*
 	 *  hang a device and process q off the stream
	 */
	s->inuse = 1;
	s->tag[0] = 0;
	q = allocq(&procinfo);
	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);

	c->flag |= COPEN;
	unlock(s);
	poperror();
	return s;
}

/*
 *  (Re)open a stream.  If this is the first open, create a stream.
 */
void
streamopen(Chan *c, Qinfo *qi)
{
	Stream *s;
	Queue *q;

	/*
	 *  if the stream already exists, just up the reference count.
	 */
	for(s = slist; s < &slist[conf.nstream]; s++) {
		if(s->inuse && s->type == c->type && s->dev == c->dev
		   && s->id == STREAMID(c->qid)){
			lock(s);
			if(s->inuse && s->type == c->type
			&& s->dev == c->dev
		 	&& s->id == STREAMID(c->qid)){
				s->inuse++;
				c->stream = s;
				unlock(s);
				return;
			}
			unlock(s);
		}
	}

	/*
	 *  create a new stream
	 */
	streamnew(c, qi);
}

/*
 *  On the last close of a stream, for each queue on the
 *  stream release its blocks and call its close routine.
 */
void
streamclose(Chan *c)
{
	Queue *q, *nq;
	Block *bp;

	/*
	 *  if not open, ignore it
	 */
	if(!(c->flag & COPEN))
		return;

	/*
	 *  decrement the reference cound
	 */
	lock(c->stream);
	if(--(c->stream->inuse) != 0){
		unlock(c->stream);
		return;
	}

	/*
	 *  descend the stream closing the queues
	 */
	for(q = c->stream->procq; q; q = q->next){
		if(q->info->close)
			(*q->info->close)(q->other);
		if(q == c->stream->devq->other)
			break;
	}
	/*
	 *  ascend the stream freeing the queues
	 */
	for(q = c->stream->devq; q; q = nq){
		nq = q->next;
		freeq(q);
	}
	c->stream->id = c->stream->dev = c->stream->type = 0;
	unlock(c->stream);
}

/*
 *  put a block to be read into the queue.  wakeup any waiting reader
 */
void
stputq(Queue *q, Block *bp)
{
	int i;

	if(bp->type == M_HANGUP){
		freeb(bp);
		q->flag |= QHUNGUP;
		q->other->flag |= QHUNGUP;
	} else {
		lock(q);
		if(q->first)
			q->last->next = bp;
		else
			q->first = bp;
		q->last = bp;
		q->len += bp->wptr - bp->rptr;
		if(q->len >= Streamhi)
			q->flag |= QHIWAT;
		unlock(q);
	}
	wakeup(&q->r);
}

/*
 *  read a string.  update the offset accordingly.
 */
long
stringread(Chan *c, uchar *buf, long n, char *str)
{
	long i;

	i = strlen(str);
	i -= c->offset;
	if(i<n)
		n = i;
	if(n<0)
		return 0;
	memcpy(buf + c->offset, str, n);
	c->offset += n;
	return n;
}

/*
 *  return true if there is an output buffer available
 */
static int
isinput(void *x)
{
	return ((Queue *)x)->first != 0;
}

/*
 *  read until we fill the buffer or until a DELIM is encountered
 */
long
streamread(Chan *c, void *vbuf, long n)
{
	Block *bp;
	Stream *s;
	Queue *q;
	long rv = 0;
	int left, i, x;
	uchar *buf = vbuf;
	char num[32];

	s = c->stream;
	switch(STREAMTYPE(c->qid)){
	case Sdataqid:
		break;
	case Sctlqid:
		sprint(num, "%d", s->id);
		return stringread(c, buf, n, num);
	default:
		if(CHDIR & c->qid)
			return devdirread(c, vbuf, n, 0, 0, streamgen);
		else
			panic("streamread");
	}

	/*
	 *  one reader at a time
	 */
	qlock(&s->rdlock);
	if(waserror()){
		qunlock(&s->rdlock);
		nexterror();
	}

	/*
	 *  sleep till data is available
	 */
	q = RD(s->procq);
	left = n;
	while(left){
		bp = getq(q);
		if(bp == 0){
			if(q->flag & QHUNGUP)
				break;
			sleep(&q->r, &isinput, (void *)q);
			continue;
		}

		i = bp->wptr - bp->rptr;
		if(i <= left){
			memcpy(buf, bp->rptr, i);
			left -= i;
			buf += i;
			if(bp->flags & S_DELIM){
				freeb(bp);
				break;
			} else
				freeb(bp);
		} else {
			memcpy(buf, bp->rptr, left);
			bp->rptr += left;
			putbq(q, bp);
			left = 0;
		}
	};

	qunlock(&s->rdlock);
	poperror();
	return n - left;	
}

/*
 *  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
 *
 *  This routing is entrered with s->wrlock'ed and must unlock.
 */
static long
streamctlwrite(Stream *s, void *a, long n)
{
	Qinfo *qi;
	Block *bp;

	/*
	 *  package
	 */
	bp = allocb(n+1);
	memcpy(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);
		pushq(s, qi);
		freeb(bp);
	} else if(streamparse("pop", bp)){
		popq(s);
		freeb(bp);
	} 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)
{
	Queue *q;

	q = (Queue *)arg;
	return q->len < Streamhi;
}
void
flowctl(Queue *q)
{
	if(q->next->len >= Streamhi)
		sleep(&q->r, notfull, q->next);
}

/*
 *  send the request as a single delimited block
 */
long
streamwrite(Chan *c, void *a, long n)
{
	Stream *s;
	Block *bp;
	Queue *q;
	long rem;
	int i;

	/*
	 *  one writer at a time
	 */
	s = c->stream;
	qlock(&s->wrlock);
	if(waserror()){
		qunlock(&s->wrlock);
		nexterror();
	}

	/*
	 *  decode the qid
	 */
	switch(STREAMTYPE(c->qid)){
	case Sdataqid:
		break;
	case Sctlqid:
		n = streamctlwrite(s, a, n);
		qunlock(&s->wrlock);
		poperror();
		return n;
	default:
		panic("bad stream qid\n");
	}

	/*
	 *  No writes allowed on hungup channels
	 */
	q = s->procq;
	if(q->other->flag & QHUNGUP)
		error(0, Ehungup);

	if(GLOBAL(a) || n==0){
		/*
		 *  `a' is global to the whole system, just create a
		 *  pointer to it and pass it on.
		 */
		flowctl(q);
		bp = allocb(0);
		bp->rptr = bp->base = (uchar *)a;
		bp->wptr = bp->lim = (uchar *)a+n;
		bp->flags |= S_DELIM;
		bp->type = M_DATA;
		PUTNEXT(q, bp);
	} else {
		/*
		 *  `a' is in the user's address space, copy it into
		 *  system buffers and pass the buffers on.
		 */
		for(rem = n; ; rem -= i) {
			flowctl(q);
			bp = allocb(rem);
			i = bp->lim - bp->wptr;
			if(i >= rem){
				memcpy(bp->wptr, a, rem);
				bp->flags |= S_DELIM;
				bp->wptr += rem;
				bp->type = M_DATA;
				PUTNEXT(q, bp);
				break;
			} else {
				memcpy(bp->wptr, a, i);
				bp->wptr += i;
				bp->type = M_DATA;
				PUTNEXT(q, bp);
				a = ((char*)a) + i;
			}
		}
	}
	qunlock(&s->wrlock);
	poperror();
	return n;
}
.
## diffname port/stream.c 1990/03013
## diff -e /n/bootesdump/1990/0227/sys/src/9/mips/stream.c /n/bootesdump/1990/03013/sys/src/9/mips/stream.c
606,607c
	s->id = s->dev = s->type = 0;
	s->inuse--;
	unlock(s);
.
602c
	for(q = s->devq; q; q = nq){
.
596c
		if(q == s->devq->other)
.
593c
	for(q = s->procq; q; q = q->next){
.
584,585c
	lock(s);
	if(s->inuse != 1){
		s->inuse--;
.
573a
	Stream *s = c->stream;
.
## diffname port/stream.c 1990/0312
## diff -e /n/bootesdump/1990/03013/sys/src/9/mips/stream.c /n/bootesdump/1990/0312/sys/src/9/mips/stream.c
888a
}

/*
 *  like andrew's getmfields but no hidden state
 */
int
getfields(char *lp,	/* to be parsed */
	char **fields,	/* where to put pointers */
	int n,		/* number of pointers */
	char sep	/* separator */
)
{
	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;
.
849c
	if((GLOBAL(a) && !docopy) || n==0){
.
809c
streamwrite(Chan *c, void *a, long n, int docopy)
.
719c
		i = BLEN(bp);
.
654,655c
	memcpy(buf, str + c->offset, n);
.
632d
630a
		q->len += BLEN(bp);
		while(bp->next) {
			bp = bp->next;
			q->len += BLEN(bp);
		}
.
624a
		wakeup(&q->other->r);
.
418c
	if(BLEN(bp) < len)
.
363a
 *  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) {
			memcpy(bp->wptr, nbp->rptr, n);
			bp->wptr += n;
			nbp->rptr += n;
			return bp;
		} else {
			memcpy(bp->wptr, nbp->rptr, i);
			bp->wptr += i;
			bp->next = nbp->next;
			nbp->next = 0;
			freeb(nbp);
		}
	}
	freeb(bp);
	return 0;
}

/*
 *  grow the front of a list of blocks by n bytes
 */
Block *
prepend(Block *bp, int n)
{
	Block *nbp;

	if(bp->base && (bp->rptr - bp->base)>=n){
		/*
		 *  room for channel number in first block of message
		 */
		bp->rptr -= n;
		return bp;
	} else {
		/*
		 *  make new block, put message number at end
		 */
		nbp = allocb(2);
		nbp->next = bp;
		nbp->wptr = nbp->lim;
		nbp->rptr = nbp->wptr - n;
		return nbp;
	}
}

/*
.
357c
		q->len -= BLEN(bp);
.
346a

/*
 *  remove the first block from a list of blocks
 */
.
337c
		q->len -= BLEN(bp);
.
324c
 *  remove the first block from a queue
.
319c
	q->len += BLEN(bp);
.
310c
void
.
303d
299c
		q->len += BLEN(bp);
.
295c
	q->len += BLEN(bp);
.
276c
		q->len += BLEN(bp);
.
272c
	q->len += BLEN(bp);
.
149a
	wakeup(&bcp->r);
.
147,148d
145a
	tries = 0;
	while(bp->next){
		if(++tries > 10){
			dumpstack();
			panic("freeb");
		}
		bp = bp->next;
	}
.
141a
	bp->rptr = bp->wptr = 0;
.
140d
137a
	int tries;
.
131,132c
 *  Free a block (or list of blocks).  Poison its pointers so that
 *  someone trying to access it after freeing will cause a dump.
.
111a
		qunlock(bcp);
.
110a
		qlock(bcp);
.
37c
	Blist;
	QLock;		/* qlock for sleepers on r */
	Rendez	r;	/* sleep here waiting for blocks */
.
15a
	&dkmuxinfo,
	&urpinfo,
.
13a
/*
 *  line disciplines that can be pushed
 *
 *  WARNING: this table should be the result of configuration
 */
extern Qinfo noetherinfo;
extern Qinfo dkmuxinfo;
extern Qinfo urpinfo;
.
11,12c
Qinfo procinfo = { stputq, nullput, 0, 0, "process" };
.
9a
/*
 *  process end line discipline
 */
.
## diffname port/stream.c 1990/0321
## diff -e /n/bootesdump/1990/0312/sys/src/9/mips/stream.c /n/bootesdump/1990/0321/sys/src/9/mips/stream.c
743c
	if(delim)
		wakeup(&q->r);
.
740a
			delim = 1;
		}
.
739c
		if(q->len >= Streamhi){
.
736a
			delim |= bp->flags & S_DELIM;
.
733a
		delim = bp->flags & S_DELIM;
.
727a
		delim = 0;
.
726a
		delim = 1;
.
720c
	int delim;
.
173c
	if(bcp->r.p)
		wakeup(&bcp->r);
.
92d
55,56c
	{ 68 },
	{ 260 },
.
34a
	for(i=0; i<Nlds && lds[i]; i++)
		if(lds[i] == qi)
			return;
	if(i == Nlds)
		panic("pushable");
	lds[i] = qi;
}

.
31,33c
void
newqinfo(Qinfo *qi)
{
	int i;
.
21,29c
static Qinfo *lds[Nlds+1];
.
18,19d
9a
enum {
	Nclass=4,	/* number of block classes */
	Nlds=32,	/* max number of pushable line disciplines */
};

.
## diffname port/stream.c 1990/0322
## diff -e /n/bootesdump/1990/0321/sys/src/9/mips/stream.c /n/bootesdump/1990/0322/sys/src/9/mips/stream.c
127,130c
		print("waiting for block %d\n", size);
		if(loop++ > 10)
			panic("waiting for blocks");
		qlock(&bcp->q);
		tsleep(&bcp->r, isblock, (void *)bcp, 250);
		qunlock(&bcp->q);
.
114d
112c
	int loop=0;
.
54c
	QLock	q;	/* qlock for sleepers on r */
.
## diffname port/stream.c 1990/03292
## diff -e /n/bootesdump/1990/0322/sys/src/9/mips/stream.c /n/bootesdump/1990/03292/sys/src/9/mips/stream.c
685c
	if(!c->stream)
.
631d
131c
		qunlock(bcp);
.
128,129c
			panic("waiting for blocks\n");
		qlock(bcp);
.
126d
54c
	QLock;		/* qlock for sleepers on r */
.
## diffname port/stream.c 1990/0331
## diff -e /n/bootesdump/1990/03292/sys/src/9/mips/stream.c /n/bootesdump/1990/0331/sys/src/9/mips/stream.c
828,829c
			if(q->flag & QHUNGUP){
				if(s->hread++ < 3)
					break;
				else
					error(0, Ehungup);
			}
.
616a
	s->hread = 0;
.
127a
		}
.
126c
		if(loop++ > 10){
			dumpqueues();
			dumpstack();
.
64a
 *  Dump all block information of how many blocks are in which queues
 */
void
dumpqueues(void)
{
	Queue *q;
	int count;
	Block *bp;

	for(q = qlist; q < qlist + conf.nqueue; q++, q++){
		if(!(q->flag & QINUSE))
			continue;
		for(count = 0, bp = q->first; bp; bp = bp->next)
			count++;
		print("%s %ux  RD count %d len %d", q->info->name, q, count, q->len);
		for(count = 0, bp = WR(q)->first; bp; bp = bp->next)
			count++;
		print("  WR count %d len %d\n", count, WR(q)->len);
	}
}

/*
.
## diffname port/stream.c 1990/0403
## diff -e /n/bootesdump/1990/0331/sys/src/9/mips/stream.c /n/bootesdump/1990/0403/sys/src/9/mips/stream.c
1012c
			FLOWCTL(q);
.
999c
		FLOWCTL(q);
.
946,947c
	sleep(&q->r, notfull, q->next);
.
941c
	return !QFULL(q->next);
.
772c
		if(q->len >= Streamhi || q->nb >= Streambhi){
.
768a
			q->nb++;
.
764a
		q->nb++;
.
390c
		q->nb--;
		if((q->flag&QHIWAT) && q->len < Streamhi/2 && q->nb < Streambhi){
.
371a
	q->nb++;
.
333c
	if(q->len >= Streamhi || q->nb >= Streambhi)
.
329a
		q->nb++;
.
325a
	q->nb++;
.
238a
	wq->len = wq->nb = 0;
.
232a
	q->len = q->nb = 0;
.
150,151c
			print("waiting for blocks\n");
.
148c
		if(loop++ == 10){
.
61c
	{ 1024 },
.
## diffname port/stream.c 1990/0406
## diff -e /n/bootesdump/1990/0403/sys/src/9/mips/stream.c /n/bootesdump/1990/0406/sys/src/9/mips/stream.c
945,948c
	return !QFULL((Queue *)arg);
.
395c
		if((q->flag&QHIWAT) && q->len<Streamhi/2 && q->nb<Streambhi/2){
.
## diffname port/stream.c 1990/0409
## diff -e /n/bootesdump/1990/0406/sys/src/9/mips/stream.c /n/bootesdump/1990/0409/sys/src/9/mips/stream.c
61c
	{ 4096 },
.
## diffname port/stream.c 1990/0509
## diff -e /n/bootesdump/1990/0409/sys/src/9/mips/stream.c /n/bootesdump/1990/0509/sys/src/9/mips/stream.c
83a
	print("\n");
	for(bcp=bclass; bcp<&bclass[Nclass-1]; bcp++){
		lock(bcp);
		for(count = 0, bp = bcp->first; bp; count++, bp = bp->next)
			;
		unlock(bcp);
		print("%d blocks of size %d\n", count, bcp->size);
	}
	print("\n");
.
73a
	print("\n");
.
72a
	Bclass *bcp;
.
## diffname port/stream.c 1990/0511
## diff -e /n/bootesdump/1990/0509/sys/src/9/mips/stream.c /n/bootesdump/1990/0511/sys/src/9/mips/stream.c
87c
	for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
.
84c
		print("  W c %d l %d f %ux\n", count, WR(q)->len, WR(q)->flag);
.
81c
		print("%s %ux  R c %d l %d f %ux", q->info->name, q, count,
			q->len, q->flag);
.
## diffname port/stream.c 1990/0513
## diff -e /n/bootesdump/1990/0511/sys/src/9/mips/stream.c /n/bootesdump/1990/0513/sys/src/9/mips/stream.c
194a
	if((bp->flags&S_CLASS) >= Nclass)
		panic("freeb class");
.
## diffname port/stream.c 1990/0629
## diff -e /n/bootesdump/1990/0513/sys/src/9/mips/stream.c /n/bootesdump/1990/0629/sys/src/9/mips/stream.c
744,758c
	streamexit(s, 1);
.
742c
	 *  leave it and free it
.
739a
	s->opens--;
.
735,738c
	if(s->opens == 1){
		/*
		 *  descend the stream closing the queues
		 */
		for(q = s->procq; q; q = q->next){
			if(q->info->close)
				(*q->info->close)(q->other);
			/* 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);
		}
.
732c
	 *  decrement the reference count
.
714a
 *  Enter a stream.  Increment the reference count so it can't disappear
 *  under foot.
 */
int
streamenter(Stream *s)
{
	lock(s);
	if(s->opens == 0){
		unlock(s);
		return -1;
	}
	s->inuse++;
	unlock(s);
	return 0;
}

/*
 *  Decrement the reference count on a stream.  If the count is
 *  zero, free the stream.
 */
void
streamexit(Stream *s, int locked)
{
	Queue *q;
	Queue *nq;

	if(!locked)
		lock(s);
	if(s->inuse == 1){
		/*
		 *  ascend the stream freeing the queues
		 */
		for(q = s->devq; q; q = nq){
			nq = q->next;
			freeq(q);
		}
		s->id = s->dev = s->type = 0;
	}
	s->inuse--;
	if(!locked)
		unlock(s);
}

/*
.
699a
				s->opens++;
.
662d
660a
	s->opens = 1;
.
518,519c
	if(bp->type == M_HANGUP)
		freeb(bp);
	else {
		freeb(bp);
		error(0, Ehungup);
	}
.
260a
 *  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);
}

/*
.
## diffname port/stream.c 1990/0702
## diff -e /n/bootesdump/1990/0629/sys/src/9/mips/stream.c /n/bootesdump/1990/0702/sys/src/9/mips/stream.c
248a
	wq->flag = QINUSE;
.
## diffname port/stream.c 1990/0707
## diff -e /n/bootesdump/1990/0702/sys/src/9/mips/stream.c /n/bootesdump/1990/0707/sys/src/9/mips/stream.c
79,85c
		print("%s %ux  R n %d l %d f %ux r %ux", q->info->name, q, q->nb,
			q->len, q->flag, &(q->r));
		print("  W n %d l %d f %ux r %ux\n", WR(q)->nb, WR(q)->len, WR(q)->flag,
			&(WR(q)->r));
		dumpblocks(q, 'R');
		dumpblocks(WR(q), 'W');
.
67a
dumpblocks(Queue *q, char c)
{
	Block *bp;
	uchar *cp;

	lock(q);
	for(bp = q->first; bp; bp = bp->next){
		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' ');
		for(cp = bp->rptr; cp<bp->wptr && cp<bp->rptr+10; cp++)
			print(" %uo", *cp);
		print("\n");
	}
	unlock(q);
}

void
.
## diffname port/stream.c 1990/0801
## diff -e /n/bootesdump/1990/0707/sys/src/9/mips/stream.c /n/bootesdump/1990/0801/sys/src/9/mips/stream.c
1157a
}

/*
 *  stat a stream.  the length is the number of bytes up to the
 *  first delimiter.
 */
void
streamstat(Chan *c, char *db, char *name)
{
	Dir dir;
	Stream *s;
	Queue *q;
	Block *bp;
	long n;

	s = c->stream;
	if(s == 0)
		panic("streamstat");

	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, 0, &dir);
	convD2M(&dir, db);
.
75c
		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM));
.
8a
#include	"fcall.h"
.
## diffname port/stream.c 1990/08272
## diff -e /n/bootesdump/1990/0801/sys/src/9/mips/stream.c /n/bootesdump/1990/08272/sys/src/9/mips/stream.c
137c
				bp->base = (uchar *)ialloc(bcp->size, i == 0);
.
135a
			/*
			 *  The i == 0 means that each allocation range
			 *  starts on a page boundary.  This makes sure
			 *  no block crosses a page boundary.
			 */
.
60,61c
	{ 64 },
	{ 256 },
.
50a
 *
 *  NOTE: to help the mappings on the IO2 and IO3 boards, the data pointed
 *	  to by a block must not cross a 4k boundary.  Therefore:
 *	  1) all the following block sizes divide evenly into 4k
 *	  2) all the blocks are ialloc'd to not cross 4k boundaries
.
## diffname port/stream.c 1990/0905
## diff -e /n/bootesdump/1990/08272/sys/src/9/mips/stream.c /n/bootesdump/1990/0905/sys/src/9/mips/stream.c
1199a

/*
 *  announce a line discipline that can be pushed
 */
void
newqinfo(Qinfo *qi)
{
	int i;

	for(i=0; i<Nlds && lds[i]; i++)
		if(lds[i] == qi)
			return;
	if(i == Nlds)
		panic("pushable");
	lds[i] = qi;
}

.
629,636d
482,529d
146,147c
			if(bcp->size){
				if(bcp->size > left){
					left = bcp->size>4096 ? bcp->size : 4096;
					ptr = (uchar *)ialloc(left, 1);
				}
				bp->base = ptr;
				ptr += bcp->size;
				left -= bcp->size;
			}
.
135a
	left = 0;
.
129a
	uchar *ptr;
.
127c
	int class, i, n, left;
.
122a
 *
 *  All data areas are alligned to their size.
 *
 *  No data area crosses a 4k boundary.  This allows us to use the
 *  VME/SCSI/LANCE to MP bus maps on the SGI power series machines.
.
70a
 *  the per stream directory structure
 */
Dirtab streamdir[]={
	"data",		Sdataqid,	0,			0600,
	"ctl",		Sctlqid,	0,			0600,
};

/*
.
27,39d
## diffname port/stream.c 1990/09051
## diff -e /n/bootesdump/1990/0905/sys/src/9/mips/stream.c /n/bootesdump/1990/09051/sys/src/9/mips/stream.c
53a
	{ 2048 },
.
12c
	Nclass=5,	/* number of block classes */
.
## diffname port/stream.c 1990/0907
## diff -e /n/bootesdump/1990/09051/sys/src/9/mips/stream.c /n/bootesdump/1990/0907/sys/src/9/mips/stream.c
1154,1170d
590a
 *  the per stream directory structure
 */
Dirtab streamdir[]={
	"data",		Sdataqid,	0,			0600,
	"ctl",		Sctlqid,	0,			0600,
};

/*
.
491a
 *  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) {
			memcpy(bp->wptr, nbp->rptr, n);
			bp->wptr += n;
			nbp->rptr += n;
			return bp;
		} else {
			memcpy(bp->wptr, nbp->rptr, i);
			bp->wptr += i;
			bp->next = nbp->next;
			nbp->next = 0;
			freeb(nbp);
		}
	}
	freeb(bp);
	return 0;
}

/*
.
149,157c
			if(bcp->size)
				bp->base = (uchar *)ialloc(bcp->size, i == 0);
.
138d
131d
128c
	int class, i, n;
.
119,123d
59,66d
54d
26a
void
newqinfo(Qinfo *qi)
{
	int i;

	for(i=0; i<Nlds && lds[i]; i++)
		if(lds[i] == qi)
			return;
	if(i == Nlds)
		panic("pushable");
	lds[i] = qi;
}

.
12c
	Nclass=4,	/* number of block classes */
.
## diffname port/stream.c 1990/0911
## diff -e /n/bootesdump/1990/0907/sys/src/9/mips/stream.c /n/bootesdump/1990/0911/sys/src/9/mips/stream.c
1198a
}

/*
 *  Dump all block information of how many blocks are in which queues
 */
void
dumpblocks(Queue *q, char c)
{
	Block *bp;
	uchar *cp;

	lock(q);
	for(bp = q->first; bp; bp = bp->next){
		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM));
		for(cp = bp->rptr; cp<bp->wptr && cp<bp->rptr+10; cp++)
			print(" %uo", *cp);
		print("\n");
	}
	unlock(q);
}

void
dumpqueues(void)
{
	Queue *q;
	int count;
	Block *bp;
	Bclass *bcp;

	print("\n");
	for(q = qlist; q < qlist + conf.nqueue; q++, q++){
		if(!(q->flag & QINUSE))
			continue;
		print("%s %ux  R n %d l %d f %ux r %ux", q->info->name, q, q->nb,
			q->len, q->flag, &(q->r));
		print("  W n %d l %d f %ux r %ux\n", WR(q)->nb, WR(q)->len, WR(q)->flag,
			&(WR(q)->r));
		dumpblocks(q, 'R');
		dumpblocks(WR(q), 'W');
	}
	print("\n");
	for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
		lock(bcp);
		for(count = 0, bp = bcp->first; bp; count++, bp = bp->next)
			;
		unlock(bcp);
		print("%d blocks of size %d\n", count, bcp->size);
	}
	print("\n");
.
1142,1143c
out:
/*	qunlock(&s->wrlock);
	poperror(); /**/
.
1091,1093c
		goto out;
.
1081a
	 */
.
1075,1076d
1072a
	s = c->stream;

.
579,581c
	for(qi = lds; qi; qi = qi->next)
		if(strcmp(qi->name, name)==0)
			return qi;
.
575c
	Qinfo *qi;
.
244a
 *  pad a block to the front with n bytes
 */
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;
	}
} 

/*
.
186,189d
172d
156a
 *  make known a stream module and call its initialization routine, if
 *  it has one.
 */
void
newqinfo(Qinfo *qi)
{
	qi->next = lds;
	lds = qi;
	if(qi->reset)
		(*qi->reset)();
}

/*
.
153a

	/*
	 *  make stream modules available
	 */
	streaminit0();
.
147c
				bp->base = (uchar *)ialloc(bcp->size, 0);
.
141,145d
130a
	/*
	 *  allocate blocks, queues, and streams
	 */
.
79,118d
70,77c
#include "stream.h"
.
65,66c
	{ 68 },
	{ 260 },
.
51,55d
27,39d
25c
static Qinfo *lds;
.
13d
## diffname port/stream.c 1990/0914
## diff -e /n/bootesdump/1990/0911/sys/src/9/mips/stream.c /n/bootesdump/1990/0914/sys/src/9/mips/stream.c
1160d
1151,1158c
		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);
.
## diffname port/stream.c 1990/0930
## diff -e /n/bootesdump/1990/0914/sys/src/9/mips/stream.c /n/bootesdump/1990/0930/sys/src/9/mips/stream.c
1106,1108d
1069c
	if(!docopy && GLOBAL(a)){
.
1052,1060c
	if(STREAMTYPE(c->qid) != Sdataqid)
		return streamctlwrite(c, a, n);
.
1041,1049d
981a
	if(STREAMTYPE(c->qid) != Sctlqid)
		panic("streamctlwrite %lux", c->qid);
	s = c->stream;

.
980a
	Stream *s;
.
977c
streamctlwrite(Chan *c, void *a, long n)
.
919a
	s = c->stream;
.
903,915c
	if(STREAMTYPE(c->qid) != Sdataqid)
		return streamctlread(c, vbuf, n);
.
901d
898,899c
	int left, i;
.
886c
	Queue *q;

	q = (Queue *)x;
	return (q->flag&QHUNGUP) || q->first!=0;
.
880a
 *  return the stream id
 */
long
streamctlread(Chan *c, void *vbuf, long n)
{
	uchar *buf = vbuf;
	char num[32];
	Stream *s;

	s = c->stream;
	if(STREAMTYPE(c->qid) == Sctlqid){
		sprint(num, "%d", s->id);
		return stringread(c, buf, n, num);
	} else {
		if(CHDIR & c->qid)
			return devdirread(c, vbuf, n, 0, 0, streamgen);
		else
			panic("streamctlread");
	}
}

/*
.
130c
	 *  look for a free block
.
## diffname port/stream.c 1990/1009
## diff -e /n/bootesdump/1990/0930/sys/src/9/mips/stream.c /n/bootesdump/1990/1009/sys/src/9/mips/stream.c
1177c
		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' ');
.
819a
void
streamclose(Chan *c)
{
	/*
	 *  if no stream, ignore it
	 */
	if(!c->stream)
		return;
	streamclose1(c->stream);
}
.
783,788d
780d
776c
streamclose1(Stream *s)
.
724c
	c->stream = streamnew(c->type, c->dev, STREAMID(c->qid), qi, 0);
.
702c
	 *  if the stream already exists, just increment the reference counts.
.
673c
	if(noopen)
		s->opens = 0;
	else
		s->opens = 1;
.
661,667c
	s->type = type;
	s->dev = dev;
	s->id = id;
.
659c
	 *  identify the stream
.
654c
		streamclose1(s);
.
631c
streamnew(ushort type, ushort dev, ushort id, Qinfo *qi, int noopen)
.
628c
 *  create a new stream, if noopen is non-zero, don't increment the open count
.
244a
	wq->ptr = 0;
.
237a
	q->ptr = 0;
.
## diffname port/stream.c 1990/1011
## diff -e /n/bootesdump/1990/1009/sys/src/9/mips/stream.c /n/bootesdump/1990/1011/sys/src/9/mips/stream.c
787,795c
		if(!waserror()){
			/*
			 *  descend the stream closing the queues
			 */
			for(q = s->procq; q; q = q->next){
				if(q->info->close)
					(*q->info->close)(q->other);
				/*
				 *  this may be 2 streams joined device end to device end
				 */
				if(q == s->devq->other)
					break;
			}
			poperror();
.
## diffname port/stream.c 1990/1018
## diff -e /n/bootesdump/1990/1011/sys/src/9/mips/stream.c /n/bootesdump/1990/1018/sys/src/9/mips/stream.c
1050a
	qunlock(&q->rlock);
.
1049a
	qlock(&q->rlock);
.
## diffname port/stream.c 1990/1101
## diff -e /n/bootesdump/1990/1018/sys/src/9/mips/stream.c /n/bootesdump/1990/1101/sys/src/9/mips/stream.c
793a
				WR(q)->put = nullput;

.
769a
	return rv;
.
767a
	rv = s->inuse;
.
753a
	int rv;
.
749c
int
.
## diffname port/stream.c 1990/1104
## diff -e /n/bootesdump/1990/1101/sys/src/9/mips/stream.c /n/bootesdump/1990/1104/sys/src/9/mips/stream.c
758a
		if(s->opens != 0)
			print("streamexit %d %s\n", s->opens, s->devq->info->name);

.
754a
	char *name;
.
## diffname port/stream.c 1990/1113
## diff -e /n/bootesdump/1990/1104/sys/src/9/mips/stream.c /n/bootesdump/1990/1113/sys/src/9/mips/stream.c
1061a
	poperror();
.
1059a
	if(waserror()){
		qunlock(&q->rlock);
		nexterror();
	}
.
137a
		poperror();
.
135a
		if(waserror()){
			qunlock(bcp);
			nexterror();
		}
.
## diffname port/stream.c 1990/11151
## diff -e /n/bootesdump/1990/1113/sys/src/9/mips/stream.c /n/bootesdump/1990/11151/sys/src/9/mips/stream.c
766c
			panic("streamexit %d %s\n", s->opens, s->devq->info->name);
.
19c
Qinfo procinfo =
{
	stputq,
	nullput,
	0,
	0,
	"process"
};
.
## diffname port/stream.c 1990/11161
## diff -e /n/bootesdump/1990/11151/sys/src/9/mips/stream.c /n/bootesdump/1990/11161/sys/src/9/mips/stream.c
838c
	qunlock(s);
.
821c
			WR(q)->put = nullput;

			/*
			 *  this may be 2 streams joined device end to device end
			 */
			if(q == s->devq->other)
				break;
.
813,819c
				poperror();
.
806,810c
		/*
		 *  descend the stream closing the queues
		 */
		for(q = s->procq; q; q = q->next){
			if(!waserror()){
.
804c
	qlock(s);
.
787c
		qunlock(s);
.
770c
		qlock(s);
.
753c
	qunlock(s);
.
749c
		qunlock(s);
.
747c
	qlock(s);
.
730c
			qunlock(s);
.
727c
				qunlock(s);
.
720c
			qlock(s);
.
700c
	qunlock(s);
.
667c
		qunlock(s);
.
658c
				qunlock(s);
.
655c
			if(canqlock(s)){
.
## diffname port/stream.c 1990/11211
## diff -e /n/bootesdump/1990/11161/sys/src/9/mips/stream.c /n/bootesdump/1990/11211/sys/src/9/mips/stream.c
1106c
		error(Ehungup);
.
1098c
	if(STREAMTYPE(c->qid.path) != Sdataqid)
.
1026c
	if(STREAMTYPE(c->qid.path) != Sctlqid)
.
981c
					error(Ehungup);
.
956c
	if(STREAMTYPE(c->qid.path) != Sdataqid)
.
925c
		if(CHDIR & c->qid.path)
.
921c
	if(STREAMTYPE(c->qid.path) == Sctlqid){
.
847,848c
		return 1;
	return streamclose1(c->stream);
.
840c
int
.
838a
	return rv;
.
832c
	rv = --(s->opens);
.
799a
	int rv;
.
795c
int
.
737c
	c->stream = streamnew(c->type, c->dev, STREAMID(c->qid.path), qi, 0);
.
723c
		 	&& s->id == STREAMID(c->qid.path)){
.
719c
		   && s->id == STREAMID(c->qid.path)){
.
664c
		error(Enostream);
.
636c
	devdir(c, (Qid){STREAMQID(STREAMID(c->qid.path),tab->qid.path), 0}, tab->name, tab->length,
.
611,612c
	"data",		{Sdataqid},	0,			0600,
	"ctl",		{Sctlqid},	0,			0600,
.
561c
	error(Ebadld);
.
557c
		error(Ebadld);
.
544c
		error(Ehungup);
.
338c
		error(Ebadld);
.
242c
		error(Enoqueue);
.
## diffname port/stream.c 1990/1127
## diff -e /n/bootesdump/1990/11211/sys/src/9/mips/stream.c /n/bootesdump/1990/1127/sys/src/9/mips/stream.c
849c
		return;
.
140a
		if(newblock(bcp) == 0)
			continue;
.
113a
 *  upgrade a block 0 block to another class (called with bcp qlocked)
 */
newblock(Bclass *bcp)
{
	Page *page;
	int n;
	Block *bp;
	uchar *cp;

	if(bcp->made > bcp->lim)
		return;

	if(bcp == bclass){
		/*
		 *  create some level zero blocks and return
		 */
		page = newpage(1, 0, 0);
		page->va = VA(kmap(page));
		n = BY2PG/sizeof(Block);
		bp = (Block *)(page->va);
		while(n-- > 0){
			bp->flags = 0;
			bp->base = bp->lim = bp->rptr = bp->wptr = 0;
			if(bcp->first)
				bcp->last->next = bp;
			else
				bcp->first = bp;
			bcp->last = bp;
			bcp->made++;
			bp++;
		}
	} else {
		/*
		 *  create a page worth of new blocks
		 */
		page = newpage(1, 0, 0);
		page->va = VA(kmap(page));
		n = BY2PG/bcp->size;
		cp = (uchar *)(page->va);
		
		while(n-- > 0){
			/*
			 *  upgrade a level 0 block
			 */
			bp = allocb(0);
			qlock(bclass);
			bclass->made--;
			bcp->made++;
			bp->flags = bcp - bclass;
			qunlock(bclass);

			/*
			 *  tack on the data area
			 */
			bp->base = bp->rptr = bp->wptr = cp;
			cp += bcp->size;
			bp->lim = cp;
			if(bcp->first)
				bcp->last->next = bp;
			else
				bcp->first = bp;
			bcp->last = bp;
		}
	}
	return;
}

/*
.
84,91c
		bcp->lim = n;
		bcp->made = 0;
.
77,78c

	/*
	 *  set limits on blocks
	 */
.
73c
	 *  allocate queues, streams
.
69d
46a
	int	lim;
	int	made;
.
38d
## diffname port/stream.c 1990/1202
## diff -e /n/bootesdump/1990/1127/sys/src/9/mips/stream.c /n/bootesdump/1990/1202/sys/src/9/mips/stream.c
1117a
		freeb(bp);
	} else if(streamparse("look", bp)){
		qlook(s, (char *)bp->rptr);
.
1083a
 *	look ldname		-- look for a line discipline
.
1078a
 *  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;
	}
	errors("not found");
}

/*
.
## diffname port/stream.c 1990/1212
## diff -e /n/bootesdump/1990/1210/sys/src/9/mips/stream.c /n/bootesdump/1990/1212/sys/src/9/port/stream.c
1170c
	q->rp = &q->r;
	sleep(q->rp, notfull, q->next);
.
1051c
			q->rp = &q->r;
			sleep(q->rp, &isinput, (void *)q);
.
957c
		wakeup(q->rp);
.
931c
		wakeup(q->other->rp);
.
498c
			wakeup(q->other->next->other->rp);
.
325a
	wq->rp = &wq->r;
.
316a
	q->rp = &q->r;
.
## diffname port/stream.c 1990/1214
## diff -e /n/bootesdump/1990/1212/sys/src/9/port/stream.c /n/bootesdump/1990/1214/sys/src/9/port/stream.c
936d
917c
		return 0;
.
450d
424d
174c
	return 0;
.
120c
		return -1;
.
111a
int
.
## diffname port/stream.c 1990/1219
## diff -e /n/bootesdump/1990/1214/sys/src/9/port/stream.c /n/bootesdump/1990/1219/sys/src/9/port/stream.c
216a
		splhi();
.
## diffname port/stream.c 1990/1229
## diff -e /n/bootesdump/1990/1219/sys/src/9/port/stream.c /n/bootesdump/1990/1229/sys/src/9/port/stream.c
244a
	int x;
.
217d
## diffname port/stream.c 1991/0115
## diff -e /n/bootesdump/1990/1229/sys/src/9/port/stream.c /n/bootesdump/1991/0115/sys/src/9/port/stream.c
59,60d
## diffname port/stream.c 1991/0316
## diff -e /n/bootesdump/1991/0115/sys/src/9/port/stream.c /n/bootesdump/1991/0316/sys/src/9/port/stream.c
598a

.
444a
blen(Block *bp)
{
	int len;

	len = 0;
	while(bp) {
		len += BLEN(bp);
		bp = bp->next;
	}

	return len;
}

/*
 * bround - round a block to chain to some 2^n number of bytes
 */
int
bround(Block *bp, int amount)
{
	Block *last;
	int len, pad;

	len = 0;
	SET(last);
	while(bp) {
		len += BLEN(bp);
		last = bp;
		bp = bp->next;
	}

	pad = ((len + amount) & ~amount) - len;
	if(pad) {
		last->next = allocb(pad);
		memset(last->next->rptr, 0, pad);
		last->next->flags |= S_DELIM;
		last->flags &= ~S_DELIM;
	}

	return len + pad;
}

int
.
443a

.
## diffname port/stream.c 1991/0317
## diff -e /n/bootesdump/1991/0316/sys/src/9/port/stream.c /n/bootesdump/1991/0317/sys/src/9/port/stream.c
1380c
	print("%d queues\n", qcount);
.
1372a
		qcount++;
.
1369a
	qcount = 0;
.
1365c
	int count, qcount;
.
481a
		last = last->next;
		memset(last->wptr, 0, pad);
		last->wptr += pad;
		last->flags |= S_DELIM;
		
.
479,480d
469c
	SET(last);	/* Ken's magic */

.
460c
 * bround - round a block chain to some 2^n number of bytes
.
## diffname port/stream.c 1991/0318
## diff -e /n/bootesdump/1991/0317/sys/src/9/port/stream.c /n/bootesdump/1991/0318/sys/src/9/port/stream.c
1280c
				memmove(bp->wptr, a, i);
.
1273c
				memmove(bp->wptr, a, rem);
.
1173c
	memmove(bp->wptr, a, n);
.
1114c
			memmove(buf, bp->rptr, left);
.
1105c
			memmove(buf, bp->rptr, i);
.
1021c
	memmove(buf, str + c->offset, n);
.
610c
			memmove(bp->wptr, nbp->rptr, i);
.
605c
			memmove(bp->wptr, nbp->rptr, n);
.
## diffname port/stream.c 1991/0320
## diff -e /n/bootesdump/1991/0318/sys/src/9/port/stream.c /n/bootesdump/1991/0320/sys/src/9/port/stream.c
1381,1382c
		print("  W n %d l %d f %ux r %ux next %lux put %lux Rz %lux\n", 
			WR(q)->nb, WR(q)->len,
			WR(q)->flag, &(WR(q)->r), q->next, q->put, q->rp);
.
## diffname port/stream.c 1991/0323
## diff -e /n/bootesdump/1991/0320/sys/src/9/port/stream.c /n/bootesdump/1991/0323/sys/src/9/port/stream.c
1379,1380c
		print("%10s %ux  R n %d l %d f %ux r %ux", q->info->name, q,
			q->nb, q->len, q->flag, &(q->r));
.
## diffname port/stream.c 1991/0328
## diff -e /n/bootesdump/1991/0323/sys/src/9/port/stream.c /n/bootesdump/1991/0328/sys/src/9/port/stream.c
261,264d
246,259c
	for(; bp; bp = nbp){
		bcp = &bclass[bp->flags & S_CLASS];
		lock(bcp);
		bp->rptr = bp->wptr = 0;
		if(bcp->first)
			bcp->last->next = bp;
		else
			bcp->first = bp;
		bcp->last = bp;
		nbp = bp->next;
		bp->next = 0;
		unlock(bcp);
		if(bcp->r.p)
			wakeup(&bcp->r);
.
241d
239a
	Block *nbp;
.
229a
	if(bp->lim-bp->rptr<size && size<4096)
		panic("allocb %lux %lux %d %ux %d", bp->lim, bp->rptr,
			size, bp->flags, bcp-bclass);
.
## diffname port/stream.c 1991/0401
## diff -e /n/bootesdump/1991/0328/sys/src/9/port/stream.c /n/bootesdump/1991/0401/sys/src/9/port/stream.c
714a
		while(*bp->rptr==' ' && bp->wptr>bp->rptr)
			bp->rptr++;
.
## diffname port/stream.c 1991/0404
## diff -e /n/bootesdump/1991/0401/sys/src/9/port/stream.c /n/bootesdump/1991/0404/sys/src/9/port/stream.c
1252c
	if(!docopy && isphys(a)){
.
## diffname port/stream.c 1991/0411
## diff -e /n/bootesdump/1991/0404/sys/src/9/port/stream.c /n/bootesdump/1991/0411/sys/src/9/port/stream.c
1249,1250c
	if(q->other->flag & QHUNGUP){
		if(s->err)
			errors((char*)(s->err->rptr));
		else
			error(Ehungup);
	}
.
1094c
				if(s->err)
					errors((char*)s->err->rptr);
				else if(s->hread++<3)
.
1039c
		return stringread(c, buf, n, num, c->offset);
.
1022c
	memmove(buf, str + offset, n);
.
1017c
	i -= offset;
.
1012c
stringread(Chan *c, uchar *buf, long n, char *str, ulong offset)
.
977c
		s = q->ptr;
		if(bp->rptr<bp->wptr && s->err==0)
			s->err = bp;
		else
			freeb(bp);
.
974a
	Stream *s;
.
897a
		if(s->err)
			freeb(s->err);
.
803a
	WR(q)->ptr = s;
	RD(q)->ptr = s;
.
792a
	s->err = 0;
.
## diffname port/stream.c 1991/0413
## diff -e /n/bootesdump/1991/0411/sys/src/9/port/stream.c /n/bootesdump/1991/0413/sys/src/9/port/stream.c
959a
	if(*err)
		errors(err);
.
931c
			if(waserror()){
				if(*err == 0)
					strncpy(err, u->error, ERRLEN-1);
			} else {
.
925a
	*err = 0;
.
920a
	char err[ERRLEN];
.
410a
	qunlock(s);
.
403a
	if(waserror()){
		qunlock(s);
		nexterror();
	}
	qlock(s);
.
388a
	qunlock(s);
.
384a
	qlock(s);
.
## diffname port/stream.c 1991/0419
## diff -e /n/bootesdump/1991/0413/sys/src/9/port/stream.c /n/bootesdump/1991/0419/sys/src/9/port/stream.c
598a
	 */
.
250a
		bp->flags = S_CLASS;			/* Check for doulbe free */
.
225a
	bp->flags = bcp - bclass;
.
## diffname port/stream.c 1991/0420
## diff -e /n/bootesdump/1991/0419/sys/src/9/port/stream.c /n/bootesdump/1991/0420/sys/src/9/port/stream.c
252c
		bp->flags = bp->flags|S_CLASS;		/* Check for doulbe free */
.
249a

.
243a
	ulong mark[1];
.
## diffname port/stream.c 1991/0421
## diff -e /n/bootesdump/1991/0420/sys/src/9/port/stream.c /n/bootesdump/1991/0421/sys/src/9/port/stream.c
55c
	{ 268 },
.
## diffname port/stream.c 1991/0501
## diff -e /n/bootesdump/1991/0421/sys/src/9/port/stream.c /n/bootesdump/1991/0501/sys/src/9/port/stream.c
1152c
	}
.
## diffname port/stream.c 1991/0502
## diff -e /n/bootesdump/1991/0501/sys/src/9/port/stream.c /n/bootesdump/1991/0502/sys/src/9/port/stream.c
1324a
	poperror();
.
1288c
	/*
	 *  if an error occurs during write n,
	 *  force a delim before write n+1
	 */
	if(waserror()){
		s->forcedelim = 1;
		nexterror();
	}
	if(s->forcedelim){
		FLOWCTL(q);
		bp = allocb(0);
		bp->flags |= S_DELIM;
		bp->type = M_DATA;
		PUTNEXT(q, bp);
		s->forcedelim = 0;
	}

	if(0 && !docopy && isphys(a)){
.
1270a
	if(n == 1 && *((char*)a) == 1)
		print("u->p->pid %d %s\n", u->p->pid, u->p->text);

.
1012a
		if(BLEN(bp) == 1 && *(bp->rptr) == 1)
			print("stputq u->p->pid %d %s\n", u->p->pid, u->p->text);

.
## diffname port/stream.c 1991/0504
## diff -e /n/bootesdump/1991/0502/sys/src/9/port/stream.c /n/bootesdump/1991/0504/sys/src/9/port/stream.c
810a
	s->forcedelim = 0;
.
## diffname port/stream.c 1991/0507
## diff -e /n/bootesdump/1991/0504/sys/src/9/port/stream.c /n/bootesdump/1991/0507/sys/src/9/port/stream.c
1014,1016d
## diffname port/stream.c 1991/0511
## diff -e /n/bootesdump/1991/0507/sys/src/9/port/stream.c /n/bootesdump/1991/0511/sys/src/9/port/stream.c
1272,1274d
## diffname port/stream.c 1991/0516
## diff -e /n/bootesdump/1991/0511/sys/src/9/port/stream.c /n/bootesdump/1991/0516/sys/src/9/port/stream.c
254c
		bp->flags = bp->flags|S_CLASS;		/* Check for double free */
.
## diffname port/stream.c 1991/0614
## diff -e /n/bootesdump/1991/0516/sys/src/9/port/stream.c /n/bootesdump/1991/0614/sys/src/9/port/stream.c
421a
	poperror();
.
## diffname port/stream.c 1991/0705
## diff -e /n/bootesdump/1991/0614/sys/src/9/port/stream.c /n/bootesdump/1991/0705/sys/src/9/port/stream.c
422d
144,145d
125,126d
120a
	page = newpage(1, 0, 0);
	page->va = VA(kmap(page));
.
## diffname port/stream.c 1991/0809
## diff -e /n/bootesdump/1991/0705/sys/src/9/port/stream.c /n/bootesdump/1991/0809/sys/src/9/port/stream.c
1131a
			continue;
		}

		if(s->flushmsg){
			if(bp->flags & S_DELIM)
				s->flushmsg = 0;
			freeb(bp);
.
1118d
1109a
		/*
		 *  notes will flush the rest of any partially
		 *  read message.
		 */
		if(n != left)
			s->flushmsg = 1;
.
1107a
	left = n;
.
825a
	s->flushmsg = 0;
.
## diffname port/stream.c 1991/0811
## diff -e /n/bootesdump/1991/0809/sys/src/9/port/stream.c /n/bootesdump/1991/0811/sys/src/9/port/stream.c
1165a

	/*
	 *  free completely read blocks
	 */
	if(tofree)
		freeb(tofree);
.
1157,1158d
1154,1155c
			bp->next = tofree;
			tofree = bp;
			if(bp->flags & S_DELIM)
.
1142,1148d
1116,1117c
		while(tofree){
			bp = tofree;
			tofree = bp->next;
			bp->next = 0;
			putbq(q, bp);
		}
.
1113,1114c
		 *  put any partially read message back into the
		 *  queue
.
1110a
	tofree = 0;
.
1096a
	Block *tofree;
.
## diffname port/stream.c 1991/0831
## diff -e /n/bootesdump/1991/0811/sys/src/9/port/stream.c /n/bootesdump/1991/0831/sys/src/9/port/stream.c
1465c
		print("%d byte blocks: %d made %d free\n", bcp->size,
			bcp->made, count);
.
213d
210,211c
		newblock(bcp);
.
207c
			unlock(bcp);
.
202,205d
156c
			unlock(bclass);
.
152c
			lock(bclass);
.
118,120d
84d
77c
	 *  set block classes
.
49,50d
46d
## diffname port/stream.c 1991/0901
## diff -e /n/bootesdump/1991/0831/sys/src/9/port/stream.c /n/bootesdump/1991/0901/sys/src/9/port/stream.c
250,251d
## diffname port/stream.c 1991/0904
## diff -e /n/bootesdump/1991/0901/sys/src/9/port/stream.c /n/bootesdump/1991/0904/sys/src/9/port/stream.c
1097a
	q = 0;
.
## diffname port/stream.c 1991/0926
## diff -e /n/bootesdump/1991/0904/sys/src/9/port/stream.c /n/bootesdump/1991/0926/sys/src/9/port/stream.c
1308,1345c
	/*
	 *  send it down stream
	 */
	last->flags |= S_DELIM;
	FLOWCTL(q);
	PUTNEXT(q, first);
.
1304,1305c
		a = ((char*)a) + i;
		if(first == 0)
			first = bp;
		else
			last->next = bp;
		last = bp;
		if(i == rem)
			break;
.
1295,1302c
	first = last = 0;
	for(rem = n; ; rem -= i) {
		bp = allocb(rem);
		i = bp->lim - bp->wptr;
		if(i >= rem)
			i = rem;
		memmove(bp->wptr, a, i);
		bp->wptr += i;
.
1292,1293c
	 *  copy the whole write into kernel space
.
1270a
	Block *bp, *first, *last;
.
1267d
811d
794d
## diffname port/stream.c 1991/1012
## diff -e /n/bootesdump/1991/0926/sys/src/9/port/stream.c /n/bootesdump/1991/1012/sys/src/9/port/stream.c
96a
	if(qi->next)
		panic("newqinfo: already configured");

.
## diffname port/stream.c 1991/1027
## diff -e /n/bootesdump/1991/1012/sys/src/9/port/stream.c /n/bootesdump/1991/1027/sys/src/9/port/stream.c
1327,1331c
getfields(char *lp, char **fields, int n, char sep)
.
## diffname port/stream.c 1991/1105
## diff -e /n/bootesdump/1991/1027/sys/src/9/port/stream.c /n/bootesdump/1991/1105/sys/src/9/port/stream.c
1193c
 *  This routing is entered with s->wrlock'ed and must unlock.
.
## diffname port/stream.c 1991/1107
## diff -e /n/bootesdump/1991/1105/sys/src/9/port/stream.c /n/bootesdump/1991/1107/sys/src/9/port/stream.c
1420c
		qunlock(bcp);
.
1417c
		qlock(bcp);
.
619a
 *  expand a block list to be one byte, len bytes long
 */
Block*
expandb(Block *bp, int len)
{
	Block *nbp, *new;
	int i;

	new = allocb(len);
	if(new == 0){
		freeb(bp);
		return 0;
	}

	/*
	 *  copy bytes into new block
	 */
	for(nbp = bp; len>0 && nbp; nbp = nbp->next){
		i = BLEN(bp);
		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;
	}
	freeb(bp);
	return new;

}

/*
.
252c
		qunlock(bcp);
.
243c
		qlock(bcp);
.
209c
	qunlock(bcp);
.
199c
			qunlock(bcp);
.
196c
	qlock(bcp);
.
152c
			qunlock(bclass);
.
148c
			qlock(bclass);
.
46a
	QLock;
.
## diffname port/stream.c 1991/1109
## diff -e /n/bootesdump/1991/1107/sys/src/9/port/stream.c /n/bootesdump/1991/1109/sys/src/9/port/stream.c
1410c
	devdir(c, c->qid, name, n, eve, 0, &dir);
.
790,791c
	devdir(c, (Qid){STREAMQID(STREAMID(c->qid.path),tab->qid.path), 0}, 
		tab->name, tab->length, eve, tab->perm, dp);
.
## diffname port/stream.c 1991/1115
## diff -e /n/bootesdump/1991/1109/sys/src/9/port/stream.c /n/bootesdump/1991/1115/sys/src/9/port/stream.c
1095c
		return stringread(buf, n, num, c->offset);
.
1068c
stringread(uchar *buf, long n, char *str, ulong offset)
.
691a
	USED(q);
.
## diffname port/stream.c 1991/1121
## diff -e /n/bootesdump/1991/1115/sys/src/9/port/stream.c /n/bootesdump/1991/1121/sys/src/9/port/stream.c
653a
	new->flags |= delim;
.
639c
		delim = nbp->flags & S_DELIM;
		i = BLEN(nbp);
.
627a
	ulong delim = 0;
.
## diffname port/stream.c 1991/1122
## diff -e /n/bootesdump/1991/1121/sys/src/9/port/stream.c /n/bootesdump/1991/1122/sys/src/9/port/stream.c
217a
	bp->list = 0;
.
## diffname port/stream.c 1991/1126
## diff -e /n/bootesdump/1991/1122/sys/src/9/port/stream.c /n/bootesdump/1991/1126/sys/src/9/port/stream.c
543c
		if((q->flag&QHIWAT) && q->len<Streamhi/2 && q->nb<Streambhi/2 &&q->other){
.
## diffname port/stream.c 1991/1227
## diff -e /n/bootesdump/1991/1126/sys/src/9/port/stream.c /n/bootesdump/1991/1227/sys/src/9/port/stream.c
1009,1010d
977,980c
			if(!waserror()){
.
971d
965d
## diffname port/stream.c 1992/0101
## diff -e /n/bootesdump/1991/1227/sys/src/9/port/stream.c /n/bootesdump/1992/0101/sys/src/9/port/stream.c
119c
	page->va = VA(kmapperm(page));
.
## diffname port/stream.c 1992/0111
## diff -e /n/bootesdump/1992/0101/sys/src/9/port/stream.c /n/bootesdump/1992/0111/sys/src/9/port/stream.c
7c
#include	"../port/error.h"
.
## diffname port/stream.c 1992/0114
## diff -e /n/bootesdump/1992/0111/sys/src/9/port/stream.c /n/bootesdump/1992/0114/sys/src/9/port/stream.c
1325c
			error((char*)(s->err->rptr));
.
1220c
	error(Ebadarg);
.
1162c
					error((char*)s->err->rptr);
.
823c
		exhausted("streams");
.
299c
		exhausted("queues");
.
## diffname port/stream.c 1992/0207
## diff -e /n/bootesdump/1992/0114/sys/src/9/port/stream.c /n/bootesdump/1992/0207/sys/src/9/port/stream.c
1169c
			sleep(q->rp, isinput, (void *)q);
.
## diffname port/stream.c 1992/0222
## diff -e /n/bootesdump/1992/0207/sys/src/9/port/stream.c /n/bootesdump/1992/0222/sys/src/9/port/stream.c
239,240c
	pc = getcallerpc(((uchar*)&bp) - sizeof(bp));
	if((bp->flags&S_CLASS) >= Nclass)		/* Check for double free */
		panic("freeb class last(%lux) this(%lux)", bp->pc, pc);
	bp->pc = pc;
.
237a
	ulong pc;
.
## diffname port/stream.c 1992/0305
## diff -e /n/bootesdump/1992/0222/sys/src/9/port/stream.c /n/bootesdump/1992/0305/sys/src/9/port/stream.c
1359,1360c
	FLOWCTL(q, first);
.
1296,1299c
	PUTNEXT(q, bp);
.
1294c
		poperror();
.
1291,1292c
	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);
.
1289c
flowctl(Queue *q, Block *bp)
.
## diffname port/stream.c 1992/0318
## diff -e /n/bootesdump/1992/0305/sys/src/9/port/stream.c /n/bootesdump/1992/0318/sys/src/9/port/stream.c
1415a
}

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;
.
1262a
		if(qi == 0)
			error(Ebadld);
.
721c
	return 0;
.
717c
		return 0;
.
711c
Qinfo *
.
## diffname port/stream.c 1992/0319
## diff -e /n/bootesdump/1992/0318/sys/src/9/port/stream.c /n/bootesdump/1992/0319/sys/src/9/port/stream.c
1510a
	for(i = 0; i < 100; i++) {
		if(refsa[i] != refsf[i])
			print("%d alloc %lux %d free %lux\n", refsa[i], apcs[i], refsf[i], fpcs[i]);
	}
.
1497a
			if(q->pg)
				print("get %d %s ", q->pg->pid, q->pg->text);
			if(q->pp)
				print("put %d %s ", q->pp->pid, q->pp->text);
		print("\n");
.
1495c
			if(q->pg)
				print("get %d %s ", q->pg->pid, q->pg->text);
			if(q->pp)
				print("put %d %s ", q->pp->pid, q->pp->text);
		print("  W n %d l %d f %ux r %ux next %lux put %lux Rz %lux", 
.
1493c
		print("%10s %ux  R n %d l %d f %ux r %ux ",
			q->info->name, q,
.
1486c
int i;
.
1471,1473c
		print("%c %c%d%c", c, bp->type == M_DATA ? 'd' : 'c',
			bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' ');
		for(cp = bp->rptr; cp<bp->wptr && cp<bp->rptr+30; cp++)
			print(" %.2x", *cp);
.
1040a
		q->pp = u->p;
.
538a
	q->pg = u->p;
.
426a
	q->pp = u->p;
.
242a

	bcp = &bclass[bp->flags & S_CLASS];
	if(bcp->size == 68) {
		for(i = 0; i < 100; i++)
			if(apcs[i] == bp->pc) {
				refsf[i]++;
				fpcs[i] = pc;
			}
	}

.
238a
int i;
.
223a
	bp->pc = pc;
.
209a

	if(bcp->size == 68) {
		int i, hole, fnd;
		fnd = 0;
		hole = -1;
		for(i = 0; i < 100; i++) {
			if(apcs[i] == pc) {
				refsa[i]++;
				fnd = 1;
				break;
			}
			if(refsa[i] == 0 && hole<0)
				hole = i;
		}
		if(fnd == 0 && hole>=0) {
			refsa[hole] = 1;
			apcs[hole] = pc;
		}
	}
.
187a
	pc = getcallerpc(((uchar*)&size) - sizeof(size));

.
186a
	ulong pc;
.
10a
ulong fpcs[100], refsa[100], refsf[100], apcs[100];

.
## diffname port/stream.c 1992/0321
## diff -e /n/bootesdump/1992/0319/sys/src/9/port/stream.c /n/bootesdump/1992/0321/sys/src/9/port/stream.c
2c
#include	"../port/lib.h"
.
## diffname port/stream.c 1992/0326
## diff -e /n/bootesdump/1992/0321/sys/src/9/port/stream.c /n/bootesdump/1992/0326/sys/src/9/port/stream.c
1561,1564d
1543,1546d
1536,1539d
1526c

.
1079d
576d
463d
269,278d
264d
248d
215,233d
191,192d
189d
11,12d
## diffname port/stream.c 1992/0509
## diff -e /n/bootesdump/1992/0326/sys/src/9/port/stream.c /n/bootesdump/1992/0509/sys/src/9/port/stream.c
1509d
1506d
## diffname port/stream.c 1992/0520
## diff -e /n/bootesdump/1992/0509/sys/src/9/port/stream.c /n/bootesdump/1992/0520/sys/src/9/port/stream.c
1102a
	return 0;	/* not reached */
.
## diffname port/stream.c 1992/0529
## diff -e /n/bootesdump/1992/0520/sys/src/9/port/stream.c /n/bootesdump/1992/0529/sys/src/9/port/stream.c
625c
 *  expand a block list to be one block, len bytes long
.
607c
		if(i >= n) {
.
## diffname port/stream.c 1992/0603
## diff -e /n/bootesdump/1992/0529/sys/src/9/port/stream.c /n/bootesdump/1992/0603/sys/src/9/port/stream.c
244c
#endif asdf
.
239a
#ifdef asdf
.
220,223d
174,181d
## diffname port/stream.c 1992/0609
## diff -e /n/bootesdump/1992/0603/sys/src/9/port/stream.c /n/bootesdump/1992/0609/sys/src/9/port/stream.c
545a
 *  grab all the blocks in a queue
 */
Block *
grabq(Queue *q)
{
	Block *bp;

	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;
		}
	}
	unlock(q);
	return bp;
}

/*
.
## diffname port/stream.c 1992/0619
## diff -e /n/bootesdump/1992/0609/sys/src/9/port/stream.c /n/bootesdump/1992/0619/sys/src/9/port/stream.c
1519,1525d
1500d
228,246c
	while(bp){
		bp->rptr = 0;
		bp->wptr = 0;
		next = bp->next;
		free(bp);
		bp = next;	
.
222,226c
	Block *next;
.
186,208c
	data = (uchar*)bp + sizeof(Block);
	bp->rptr = data;
	bp->wptr = data;
	bp->base = data;
	bp->lim = data+size;
	bp->flags = 0;
.
180,184c
	bp = smalloc(sizeof(Block)+size);
.
178c
	uchar *data;
.
108,171d
75,85d
71,72c
	slist = (Stream *)xalloc(conf.nstream * sizeof(Stream));
	qlist = (Queue *)xalloc(conf.nqueue * sizeof(Queue));
.
65,66d
41,57d
33,35d
9d
## diffname port/stream.c 1992/0623
## diff -e /n/bootesdump/1992/0619/sys/src/9/port/stream.c /n/bootesdump/1992/0623/sys/src/9/port/stream.c
1361,1381d
1355c
	return i;
.
1347,1353c
	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++;
.
1344,1345c
	int i;
.
1341,1342c
int
getfields(char *lp, char **fields, int n, char sep)
.
1339c
 *  like andrew's getmfields but no hidden state
.
1335c
/*
 *  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;
.
1304,1333c
	bp = allocb(0);
	bp->type = M_HANGUP;
	(*s->devq->put)(s->devq, bp);
}
.
1301,1302c
	Block *bp;
.
1298,1299c
/*
 *  send a hangup up a stream
 */
static void
hangup(Stream *s)
.
1248,1267d
973c
		return readstr(c->offset, buf, n, num);
.
966c
	char *buf = vbuf;
.
943,960d
883d
877c
	rv = s->opens;
	qunlock(s);
.
850c
	if(s->opens-- == 1){
.
847c
	 *  decrement the open count
.
836,837c
 *  Decrement the open count.  When it goes to zero, call the close
 *  routines for each queue in the stream.
.
828,832c
	qunlock(hb);
.
826a

		/*
		 *  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);
.
824d
811,813c
	hb = hash(s->type, s->dev, s->id);
	qlock(hb);
	if(s->inuse-- == 1){
.
809a
	Sthash *hb;
	Stream **l, *ns;
.
808d
803c
void
.
789,796c
	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, 1);
				return -1;
			}
			return 0;
		}
	qunlock(hb);
	return -1;
.
783,784c
 *  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.
.
753,778d
748c
 *  Associate a stream with a channel
.
727d
722d
719a
	 *  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();
	}

	/*
.
717a
	s->hread = 0;
	s->next = hb->s;
	hb->s = s;
.
713a
	s = smalloc(sizeof(Stream));
	s->inuse = 1;
.
712c
	 *  create and init a new stream
.
701,709d
698a
			return s;
.
692,696c
	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++;
.
690c
	 *  if the stream already exists, just increment the reference counts.
.
688a
	hb = hash(type, dev, id);

.
687a
	Sthash *hb;
.
680a
 *  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];
}

/*
.
647,654d
629,643c
static void	hangup(Stream*);

void
streaminit(void)
{
	/*
	 *  make stream modules available
	 */
	streaminit0();
.
627c
	Nbits=	5,
	Nhash=	1<<Nbits,
	Nmask=	Nhash-1,
};
typedef struct Sthash Sthash;
struct Sthash
{
	QLock;
	Stream	*s;
};
static Sthash ht[Nhash];
.
624,625c
enum
.
615,622c
 *  hash buckets containing all streams
.
609,613d
604,607c
Dirtab streamdir[]={
	"data",		{Sdataqid},	0,			0600,
	"ctl",		{Sctlqid},	0,			0600,
};
.
602c
 *  the per stream directory structure
.
593,600d
588,591d
586c
 *  Part 3) Streams
.
502,570d
461,500d
452,459d
301,346d
211d
201c
flushq(Queue *q)
.
198c
 *  flush a queue
.
194a
	free(RD(q));
.
185c
freeq(Queue *q)
.
182c
 *  free a queue
.
176,177d
153,157d
143,151c
	q = smalloc(2*sizeof(Queue));
.
134a
 *  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 void stputq(Queue*, Block*);
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)();
}

/*
 *  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;
}

/*
.
115c
 *  Pad a block to the front with n bytes.  This is used to add protocol
 *  headers to the front of blocks.
.
98c
 *  someone trying to access it after freeing will cause a panic.
.
84,88c
	base = (uchar*)bp + sizeof(Block);
	lim = (uchar*)bp + msize(bp);
	bp->wptr = bp->rptr = lim - size;
	bp->base = base;
	bp->lim = lim;
.
80c
	uchar *base, *lim;
.
30,75d
28c
 *  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.
.
17,25d
15c
 *  Part 1) Blocks
.
10,13d
## diffname port/stream.c 1992/0625
## diff -e /n/bootesdump/1992/0623/sys/src/9/port/stream.c /n/bootesdump/1992/0625/sys/src/9/port/stream.c
1325,1329d
1222,1227d
1220a
		}
.
1212,1219c
		va += i;
		rem -= i;
		if(rem > 0){
			FLOWCTL(q, bp);
		} else {
			bp->flags |= S_DELIM;
			FLOWCTL(q, bp);
.
1210c
		bp = allocb(i);
		memmove(bp->wptr, va, i);
.
1204,1208c
	va = a;
	rem = n;
	for(;;){
		if(rem > Streamhi)
			i = Streamhi;
		else
.
1202c
	 *  Write the message using blocks <= Streamhi bytes longs
.
1180c
	Block *bp;
	char *va;
.
940c
	if(awaken)
.
936c
			awaken = 1;
.
931c
			awaken |= bp->flags & S_DELIM;
.
926c
		awaken = bp->flags & S_DELIM;
.
917c
		awaken = 1;
.
905c
	int awaken;
.
837a
 *  nail down a stream so that it can't be closed
 */
void
naildownstream(Stream *s)
{
	s->opens++;
	s->inuse++;
}

/*
.
629,637d
278c
static Streamput stputq;
.
## diffname port/stream.c 1992/0711
## diff -e /n/bootesdump/1992/0625/sys/src/9/port/stream.c /n/bootesdump/1992/0711/sys/src/9/port/stream.c
1182a

	/*
	 *  docopy will get used if I ever figure out when to avoid copying
	 *  data. -- presotto
	 */
	USED(docopy);
.
886c
	streamexit(s);
.
846d
792d
788c
streamexit(Stream *s)
.
774c
				streamexit(s);
.
640,642d
## diffname port/stream.c 1992/0826
## diff -e /n/bootesdump/1992/0711/sys/src/9/port/stream.c /n/bootesdump/1992/0826/sys/src/9/port/stream.c
1257c
	devdir(c, c->qid, name, n, eve, perm, &dir);
.
1235c
streamstat(Chan *c, char *db, char *name, long perm)
.
## diffname port/stream.c 1993/0511
## diff -e /n/bootesdump/1992/0826/sys/src/9/port/stream.c /n/fornaxdump/1993/0511/sys/src/brazil/port/stream.c
955,1323c
		memmove(p, b->rp, len);
.
936,953c
	n = BLEN(b);
	if(n < len){
		memmove(p, b->rp, n);
.
934a
		return -1;
.
844,933c
	lock(q);
	b = q->first;
	if(b == 0){
		q->state |= Qcsleep;
.
841,842c
	Block *b;
	int n;
.
839c
consume(Queue *q, uchar *p, int len, int drop)
.
827,837d
825c
 *  copy out of a queue, returns # bytes copied
.
780,821c
	return b;
.
764,778c
	b->base = (uchar*)(b+1);
	b->rp = b->wp = b->base;
	b->lim = b->base + size;
.
752,762c
	b = alloc(sizeof(Block) + size);
	if(b == 0)
		exhausted("blocks");
.
749,750c
	Block *b;
.
743,747c
Block*
allocb(int size)
.
671,740c
	q = smalloc(sizeof(Queue));
	q->limit = limit;
.
669d
615,667d
601,613c
Queue*
allocq(int limit)
.
599c
 *  allocate queues and blocks
.
589,595c
	/* start garbage collector */
	kproc("buffer", bgc, 0);
.
587c
blockinit(void)
.
582,585d
569,579c
	lock(&freed);
	b->next = freed->first;
	freed->first = b;
	unlock(&freed);
.
563,567c
void
freeb(Block *b)
.
559,560d
382,556c
		for(; b; b = nb){
			nb = b->next;
			free(b);
.
374,380c
		x = slphi();
		lock(&freed);
		b = freed->first;
		freed->first = freed->last = 0;;
		unlock(&freed);
		spllo();
.
365,372c
	USED(arg);
	for(;;){
		tsleep(&freed->r, return0, 0, 500);
		if(freed->first == 0)
			continue;
.
363c
	Block *b, *nb;
.
361c
bgc(void *arg)
.
19,359d
15,17c
 *  Interrupt handlers use freeb() to release blocks.  They are
 *  garbage collected by the kproc running bgc().
.
10,12c
static Queue *freed;
.
8d
6d
1d
## diffname port/stream.c 1993/0512
## diff -e /n/fornaxdump/1993/0511/sys/src/brazil/port/stream.c /n/fornaxdump/1993/0512/sys/src/brazil/port/stream.c
99,102c
	if(n < len)
		len = n;
	memmove(p, b->rp, len);
	if(len == n || drop){
		q->first = b->next;
		ifree(b);
	} else
		b->rp += len;
	unlock(q);
	return len;
}

int
produce(Queue *q, uchar *p, int len)
{
	Block *b;

	b = ialloc(sizeof(Block)
.
94c
		q->state |= Qstarve;
.
83c
 *  Interrupt level copy out of a queue, return # bytes copied.  If drop is
 *  set, any bytes left in a block afer a consume are discarded.
.
65a
/*
 *  allocate queues and blocks
 */
.
62,63c
	cl = &arena.freed;
	p = a;
	lock(cl);
	p->next = cl->first;
	cl->first = p;
	unlock(cl);
.
60c
	Chunk *p;
	Chunkl *cl;
.
54,58c
void
ifree(void *a)
.
50,51c
	int pow;
	Chunkl *cl;
	Chunk *p;

	for(pow = Min; pow <= Maxpow; pow++)
		if(size <= (1<<pow)){
			cl = &arena.alloc[pow];
			lock(cl);
			p = cl->first;
			if(p){
				cl->have--;
				cl->first = p->next;
			}
			unlock(cl);
			return (void*)p;
		}
	panic("ialloc %d\n", size);
.
47,48c
void*
ialloc(int size)
.
41,44c
	int pow;
	Chunkl *cl;

	for(pow = Minpow; pow <= Maxpow; pow++){
		cl = &arena.alloc[pow];
		cl->goal = Maxpow-pow + 4;
	}

	/* start garbage collector */
	kproc("iallockproc", iallockproc, 0);
.
39c
iallocinit(void)
.
31,33c
		/* make sure we have blocks available for interrupt level */
		for(pow = Minpow; pow <= Maxpow; pow++){
			cl = &arena.alloc[pow];
			if(cl->have >= cl->goal){
				cl->had = cl->have;
				continue;
			}

			/* increase goal if we've been drained twice in a row */
			if(cl->have == 0 && cl->had == 0)
				cl->goal += cl->goal>>2;
			cl->had = cl->have;
			l = &first;
			for(i = x = cl->goal - cl->have; x > 0; x--){
				p = alloc(1<<pow);
				if(p == 0)
					break;
				*l = p;
				l = &p->next;
			}
			if(first){
				x = splhi();
				lock(cl);
				*l = cl->first;
				cl->first = first;
				cl->have += i;
				unlock(cl);
				spllo(x);
			}
.
24,29c
		/* really free what was freed at interrupt level */
		cl = &arena.freed;
		if(cl->first){
			x = slphi();
			lock(cl);
			first = cl->first;
			cl->first = 0;
			unlock(cl);
			spllo();
	
			for(; first; first = p){
				p = first->next;
				free(first);
			}
		}
.
21,22d
16c
	Chunk *p, *first, **l;
	Chunkl *cl;
	int pow, x, i;
.
14c
iallockproc(void *arg)
.
10,11c
 *  Manage interrupt level memory allocation.
.
8a
enum
{
	Minpow= 7,
	Maxpow=	12,
};

struct Chunk
{
	Chunk	*next;
};

struct Alloc
{
	Lock;
	Chunk	*first;
	int	had;
	int	goal;
	int	last;
};

struct Arena
{
	Chunkl	alloc[Maxpow-Minpow+1];
	Chunkl	freed;
};

static Arena arena;

.
7c
typedef struct Chunk	Chunk;
typedef	struct Chunkl	Chunkl;
typedef	struct Arena	Arena;
.
## diffname port/stream.c 1993/0513
## diff -e /n/fornaxdump/1993/0512/sys/src/brazil/port/stream.c /n/fornaxdump/1993/0513/sys/src/brazil/port/stream.c
207c
	lock(q);
	b = q->rfirst;
	if(b){
		/* hand to waiting receiver */
		n = b->lim - b->wp;
		if(n < len)
			len = n;
		memmove(b->wp, p, len);
		b->wp += len;
		q->rfirst = b->next;
		wakeup(&b->r);
		unlock(q);
		return len;
	}

	/* no waiting receivers, buffer */
	if(q->len >= q->limit)
		return -1;
	b = ialloc(sizeof(Block)+len);
	if(b == 0)
		return -1;
	b->base = (uchar*)(b+1);
	b->rp = b->base;
	b->wp = b->lim = b->base + len;
	memmove(b->rp, p, len);
	if(q->bfirst)
		q->blast->next = b;
	else
		q->bfirst = b;
	q->last = b;
	q->len += len;
	unlock(q);
	return len;
}

/*
 *  called by non-interrupt code
 */
Queue*
qopen(int limit)
{
	Queue *q;

	q = malloc(sizeof(Queue));
	if(q == 0)
		exhausted("Queues");
	q->limit = limit;
}

static int
bfilled(void *a)
{
	Block *b = a;

	return b->wp - b->rp;
}

long
qread(Queue *q, char *p, int len, int drop)
{
	Block *b, *bb;
	int x, n;

	/* ... to be replaced by a mapping */
	b = allocb(len);

	x = splhi();
	lock(q);
	bb = q->bfirst;
	if(bb == 0){
		/* wait for our block to be filled */
		if(q->rfirst)
			q->rlast->next = b;
		else
			q->rfirst = b;
		q->rlast = b;
		unlock(q);
		splx(x);
		sleep(&b->r, bfilled, b);
		n = BLEN(b);
		memmove(p, b->rp, n);
		return n;
	}

	/* grab a block from the buffer */
	n = BLEN(b);
	if(drop || n <= len){
		q->bfirst = b->next;
		q->len -= n;
		unlock(q);
		slpx(x);
		memmove(p, b->rp, n);
	} else {
		n = len;
		q->len -= n;
		memmove(p, b->rp, n);
		b->rp += n;
		unlock(q);
		slpx(x);
	}
	free(b);
	return n;
}

static int
qnotfull(void *a)
{
	Queue *q = a;

	return q->len < q->limit;
}

long
qwrite(Queue *q, char *p, int len)
{
	Block *b;
	int x, n;

	b = allocb(len);
	memmove(b->rp, p, len);
	b->wp += len;

	/* flow control */
	if(!qnotfull(q)){
		qlock(&q->wlock);
		sleep(&q->r, qnotfull, q);
		qunlock(&q->wlock);
	}
		
	x = splhi();
	lock(q);
	if(q->bfirst)
		q->blast->next = b;
	else
		q->bfirst = b;
	q->blast = b;
	q->len += len;
	unlock(q);
	splx(x);

	return len;
.
198a

	if(drop || len == n)
		ifree(b);

.
197a
	q->len -= len;

	/* wakeup flow controlled writers */
	if(q->len+len >= q->limit && q->len < q->limit)
		wakeup(&q->r);

.
193,196c
	if(drop || len == n)
		q->bfirst = b->next;
	else
.
183c
	b = q->bfirst;
.
163c
		exhausted("Blocks");
.
161c
	b = malloc(sizeof(Block) + size);
.
83c
				p = malloc(1<<pow);
.
79a
			else {
				x = cl->goal/2;
				if(cl->goal > 4 && cl->had > x && cl->have > x)
					cl->goal--;
			}

.
77,78c
			/*
			 *  increase goal if we've been drained, decrease
			 *  goal if we've had lots of blocks twice in a row.
			 */
			if(cl->have == 0)
.
## diffname port/stream.c 1993/0515
## diff -e /n/fornaxdump/1993/0513/sys/src/brazil/port/stream.c /n/fornaxdump/1993/0515/sys/src/brazil/port/stream.c
243,254c
	b = q->first;
	if(append && b && b->lim-b->wp <= len){
		memmove(b->wp, p, len);
		b->wp += len;
	} else {
		b = ialloc(sizeof(Block)+len);
		if(b == 0)
			return -1;
		b->base = (uchar*)(b+1);
		b->rp = b->base;
		b->wp = b->lim = b->base + len;
		memmove(b->rp, p, len);
		if(q->bfirst)
			q->blast->next = b;
		else
			q->bfirst = b;
		q->last = b;
	}
.
221c
produce(Queue *q, uchar *p, int len, int append)
.
## diffname port/stream.c 1993/0522
## diff -e /n/fornaxdump/1993/0515/sys/src/brazil/port/stream.c /n/fornaxdump/1993/0522/sys/src/brazil/port/stream.c
367a
	if((q->state & Qstarve) && q->kick){
		q->stat &= ~Qstarve;
		(*q->kick)(q->arg);
	}
.
277a
	q->kick = kick;
	q->arg = arg;
.
270c
qopen(int limit, void (*kick)(void*), void *arg)
.
## diffname port/stream.c 1993/0525
## diff -e /n/fornaxdump/1993/0522/sys/src/brazil/port/stream.c /n/fornaxdump/1993/0525/sys/src/brazil/port/stream.c
371c
		q->state &= ~Qstarve;
.
349c
	int x;
.
331c
		splx(x);
.
327,329c
	q->len -= n;
	unlock(q);
	splx(x);
	memmove(p, bb->rp, n);
	bb->rp += n;

	/* free it or put it back */
	if(drop || bb->rp == bb->wp)
		free(bb);
	else {
		x = splhi();
		lock(q);
		bb->next = q->bfirst;
		q->bfirst = bb;
.
317,325c
	/* copy from a buffered block */
	q->bfirst = bb->next;
	n = BLEN(bb);
	if(n > len)
.
313a
		free(b);
.
296c
	/* ... to be replaced by a kmapping if need be */
.
291c
qread(Queue *q, char *p, int len)
.
279a
	q->state = Qmsg;

	return q;
.
259c
		q->blast = b;
.
250a
		}
.
249c
		if(b == 0){
			unlock(q);
.
243,244c
	}

	/* save in buffer */
	b = q->bfirst;
	if((q->state&Qmsg)==0 && b && b->lim-b->wp <= len){
.
240,241c
	/* no waiting receivers, room in buffer? */
	if(q->len >= q->limit){
		unlock(q);
.
223a
	int n;
.
221c
qproduce(Queue *q, uchar *p, int len)
.
214c
	if((q->state&Qmsg) || len == n)
.
208,209c
	/* wakeup flow controlled writers (with a bit of histeresis) */
	if(q->len+len >= q->limit && q->len < q->limit/2)
.
202c
	if((q->state&Qmsg) || len == n)
.
186c
qconsume(Queue *q, uchar *p, int len)
.
145a
	return 0;			/* not reached */
.
133c
	for(pow = Minpow; pow <= Maxpow; pow++)
.
105c
				splx(x);
.
89c
			first = 0;
.
83,87d
75c
			} else
				cl->hist <<= 1;
.
73c
				cl->hist = ((cl->hist<<1) | 1) & 0xff;
				if(cl->hist == 0xff && cl->goal > 8)
					cl->goal--;
.
71a

			/*
			 *  if we've been ahead of the game for a while
			 *  start giving blocks back to the general pool
			 */
.
61c
			splx(x);
.
56c
			x = splhi();
.
51c
		tsleep(&arena.r, return0, 0, 500);
.
34a
	Rendez r;
.
33c
	Chunkl	alloc[Maxpow+1];
.
28c
	int	hist;
.
26c
	int	have;
.
22c
struct Chunkl
.
0a
#include	"u.h"
.
## diffname port/stream.c 1993/0526 # deleted
## diff -e /n/fornaxdump/1993/0525/sys/src/brazil/port/stream.c /n/fornaxdump/1993/0526/sys/src/brazil/port/stream.c
1,401d