1990/1202/port/stream.c (diff list | history)

1990/0227/sys/src/9/port/stream.c:571,576 – 1990/03013/sys/src/9/port/stream.c:571,577 (short | long)
1990/0227		{ Queue *q, *nq; Block *bp;
1990/03013		Stream *s = c->stream;
1990/0227		/* * if not open, ignore it
1990/0227/sys/src/9/port/stream.c:581,588 – 1990/03013/sys/src/9/port/stream.c:582,590
1990/0227		/* * decrement the reference cound */ lock(c->stream); if(--(c->stream->inuse) != 0){
1990/03013		lock(s); if(s->inuse != 1){ s->inuse--;
1990/0227		unlock(c->stream); return; }
1990/0227/sys/src/9/port/stream.c:590,610 – 1990/03013/sys/src/9/port/stream.c:592,613
1990/0227		/* * descend the stream closing the queues */ for(q = c->stream->procq; q; q = q->next){
1990/03013		for(q = s->procq; q; q = q->next){
1990/0227		if(q->info->close) (*q->info->close)(q->other); if(q == c->stream->devq->other)
1990/03013		if(q == s->devq->other)
1990/0227		break; } /* * ascend the stream freeing the queues */ for(q = c->stream->devq; q; q = nq){
1990/03013		for(q = s->devq; q; q = nq){
1990/0227		nq = q->next; freeq(q); } c->stream->id = c->stream->dev = c->stream->type = 0; unlock(c->stream);
1990/03013		s->id = s->dev = s->type = 0; s->inuse--; unlock(s);
1990/0227		} /*
1990/03013/sys/src/9/port/stream.c:7,18 – 1990/0312/sys/src/9/port/stream.c:7,30 (short | long)
1990/0227		#include "errno.h" #include "devtab.h"
1990/0312		/* * process end line discipline */
1990/0227		static void stputq(Queue*, Block*); Qinfo procinfo = { stputq, nullput, 0, 0, "process" } ; extern Qinfo noetherinfo;
1990/0312		Qinfo procinfo = { stputq, nullput, 0, 0, "process" };
1990/0227
1990/0312		/* * line disciplines that can be pushed * * WARNING: this table should be the result of configuration */ extern Qinfo noetherinfo; extern Qinfo dkmuxinfo; extern Qinfo urpinfo;
1990/0227		static Qinfo *lds[] = { &noetherinfo,
1990/0312		&dkmuxinfo, &urpinfo,
1990/0227		0 };
1990/03013/sys/src/9/port/stream.c:34,40 – 1990/0312/sys/src/9/port/stream.c:46,54
1990/0227		*/ typedef struct { int size; Queue;
1990/0312		Blist; QLock; /* qlock for sleepers on r */ Rendez r; /* sleep here waiting for blocks */
1990/0227		} Bclass; Bclass bclass[Nclass]={ { 0 },
1990/03013/sys/src/9/port/stream.c:108,114 – 1990/0312/sys/src/9/port/stream.c:122,130
1990/0227		while(bcp->first == 0){ unlock(bcp); print("waiting for blocks\n");
1990/0312		qlock(bcp);
1990/0227		sleep(&bcp->r, isblock, (void *)bcp);
1990/0312		qunlock(bcp);
1990/0227		lock(bcp); } bp = bcp->first;
1990/03013/sys/src/9/port/stream.c:128,152 – 1990/0312/sys/src/9/port/stream.c:144,176
1990/0227		} /* * Free a block. Poison its pointers so that someone trying to access * it after freeing will cause a dump.
1990/0312		* Free a block (or list of blocks). Poison its pointers so that * someone trying to access it after freeing will cause a dump.
1990/0227		*/ void freeb(Block *bp) { Bclass *bcp;
1990/0312		int tries;
1990/0227		bcp = &bclass[bp->flags & S_CLASS]; bp->rptr = bp->wptr = 0; lock(bcp);
1990/0312		bp->rptr = bp->wptr = 0;
1990/0227		if(bcp->first) bcp->last->next = bp; else bcp->first = bp;
1990/0312		tries = 0; while(bp->next){ if(++tries > 10){ dumpstack(); panic("freeb"); } bp = bp->next; }
1990/0227		bcp->last = bp; bp->next = 0; wakeup(&bcp->r); unlock(bcp);
1990/0312		wakeup(&bcp->r);
1990/0227		} /*
1990/03013/sys/src/9/port/stream.c:269,279 – 1990/0312/sys/src/9/port/stream.c:293,303
1990/0227		q->last->next = bp; else q->first = bp; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/0227		delim = bp->flags & S_DELIM; while(bp->next) { bp = bp->next; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/0227		delim |= bp->flags & S_DELIM; } q->last = bp;
1990/03013/sys/src/9/port/stream.c:292,306 – 1990/0312/sys/src/9/port/stream.c:316,329
1990/0227		q->last->next = bp; else q->first = bp; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/0227		delim = bp->flags & S_DELIM; while(bp->next) { bp = bp->next; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/0227		delim |= bp->flags & S_DELIM; } q->last = bp; bp->next = 0; return delim; }
1990/03013/sys/src/9/port/stream.c:307,313 – 1990/0312/sys/src/9/port/stream.c:330,336
1990/0227		/* * add a block to the start of a queue */ static void
1990/0312		void
1990/0227		putbq(Blist *q, Block *bp) { lock(q);
1990/03013/sys/src/9/port/stream.c:316,327 – 1990/0312/sys/src/9/port/stream.c:339,350
1990/0227		else q->last = bp; q->first = bp; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/0227		unlock(q); } /* * remove the first block from a queue
1990/0312		* remove the first block from a queue
1990/0227		*/ Block * getq(Queue *q)
1990/03013/sys/src/9/port/stream.c:334,340 – 1990/0312/sys/src/9/port/stream.c:357,363
1990/0227		q->first = bp->next; if(q->first == 0) q->last = 0; q->len -= bp->wptr - bp->rptr;
1990/0312		q->len -= BLEN(bp);
1990/0227		if((q->flag&QHIWAT) && q->len < Streamhi/2){ wakeup(&q->other->next->other->r); q->flag &= ~QHIWAT;
1990/03013/sys/src/9/port/stream.c:344,349 – 1990/0312/sys/src/9/port/stream.c:367,376
1990/0227		unlock(q); return bp; }
1990/0312		/* * remove the first block from a list of blocks */
1990/0227		Block * getb(Blist *q) {
1990/03013/sys/src/9/port/stream.c:354,360 – 1990/0312/sys/src/9/port/stream.c:381,387
1990/0227		q->first = bp->next; if(q->first == 0) q->last = 0; q->len -= bp->wptr - bp->rptr;
1990/0312		q->len -= BLEN(bp);
1990/0227		bp->next = 0; } return bp;
1990/03013/sys/src/9/port/stream.c:361,366 – 1990/0312/sys/src/9/port/stream.c:388,467
1990/0227		} /*
1990/0312		* 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; } } /*
1990/0227		* put a block into the bit bucket */ void
1990/03013/sys/src/9/port/stream.c:415,421 – 1990/0312/sys/src/9/port/stream.c:516,522
1990/0227		int len; len = strlen(name); if(bp->wptr - bp->rptr < len)
1990/0312		if(BLEN(bp) < len)
1990/0227		return 0; if(strncmp(name, (char *)bp->rptr, len)==0){ if(bp->rptr[len] == ' ')
1990/03013/sys/src/9/port/stream.c:622,627 – 1990/0312/sys/src/9/port/stream.c:723,729
1990/0227		freeb(bp); q->flag |= QHUNGUP; q->other->flag |= QHUNGUP;
1990/0312		wakeup(&q->other->r);
1990/0227		} else { lock(q); if(q->first)
1990/03013/sys/src/9/port/stream.c:628,635 – 1990/0312/sys/src/9/port/stream.c:730,741
1990/0227		q->last->next = bp; else q->first = bp;
1990/0312		q->len += BLEN(bp); while(bp->next) { bp = bp->next; q->len += BLEN(bp); }
1990/0227		q->last = bp; q->len += bp->wptr - bp->rptr; if(q->len >= Streamhi) q->flag |= QHIWAT; unlock(q);
1990/03013/sys/src/9/port/stream.c:651,658 – 1990/0312/sys/src/9/port/stream.c:757,763
1990/0227		n = i; if(n<0) return 0; memcpy(buf + c->offset, str, n); c->offset += n;
1990/0312		memcpy(buf, str + c->offset, n);
1990/0227		return n; }
1990/03013/sys/src/9/port/stream.c:716,722 – 1990/0312/sys/src/9/port/stream.c:821,827
1990/0227		continue; } i = bp->wptr - bp->rptr;
1990/0312		i = BLEN(bp);
1990/0227		if(i <= left){ memcpy(buf, bp->rptr, i); left -= i;
1990/03013/sys/src/9/port/stream.c:806,812 – 1990/0312/sys/src/9/port/stream.c:911,917
1990/0227		* send the request as a single delimited block */ long streamwrite(Chan *c, void *a, long n)
1990/0312		streamwrite(Chan *c, void *a, long n, int docopy)
1990/0227		{ Stream *s; Block *bp;
1990/03013/sys/src/9/port/stream.c:846,852 – 1990/0312/sys/src/9/port/stream.c:951,957
1990/0227		if(q->other->flag & QHUNGUP) error(0, Ehungup); if(GLOBAL(a) || n==0){
1990/0312		if((GLOBAL(a) && !docopy) || n==0){
1990/0227		/* * `a' is global to the whole system, just create a * pointer to it and pass it on.
1990/03013/sys/src/9/port/stream.c:886,889 – 1990/0312/sys/src/9/port/stream.c:991,1018
1990/0227		qunlock(&s->wrlock); poperror(); return n;
1990/0312		} /* * 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;
1990/0227		}
1990/0312/sys/src/9/port/stream.c:7,12 – 1990/0321/sys/src/9/port/stream.c:7,17 (short | long)
1990/0227		#include "errno.h" #include "devtab.h"
1990/0321		enum { Nclass=4, /* number of block classes */ Nlds=32, /* max number of pushable line disciplines */ };
1990/0312		/* * process end line discipline */
1990/0312/sys/src/9/port/stream.c:15,37 – 1990/0321/sys/src/9/port/stream.c:20,41
1990/0227
1990/0312		/* * line disciplines that can be pushed * * WARNING: this table should be the result of configuration */ extern Qinfo noetherinfo; extern Qinfo dkmuxinfo; extern Qinfo urpinfo;
1990/0227		static Qinfo *lds[] = { &noetherinfo,
1990/0312		&dkmuxinfo, &urpinfo,
1990/0227		0 };
1990/0321		static Qinfo *lds[Nlds+1];
1990/0227		enum { Nclass=4, };
1990/0321		void newqinfo(Qinfo *qi) { int i;
1990/0227
1990/0321		for(i=0; i<Nlds && lds[i]; i++) if(lds[i] == qi) return; if(i == Nlds) panic("pushable"); lds[i] = qi; }
1990/0227		/* * All stream structures are ialloc'd at boot time */
1990/0312/sys/src/9/port/stream.c:52,59 – 1990/0321/sys/src/9/port/stream.c:56,63
1990/0227		} Bclass; Bclass bclass[Nclass]={ { 0 }, { 64 }, { 512 },
1990/0321		{ 68 }, { 260 },
1990/0227		{ 4096 }, };
1990/0312/sys/src/9/port/stream.c:89,95 – 1990/0321/sys/src/9/port/stream.c:93,98
1990/0227		} }                  /* * allocate a block */
1990/0312/sys/src/9/port/stream.c:170,176 – 1990/0321/sys/src/9/port/stream.c:173,180
1990/0312		}
1990/0227		bcp->last = bp; unlock(bcp);
1990/0312		wakeup(&bcp->r);
1990/0321		if(bcp->r.p) wakeup(&bcp->r);
1990/0227		} /*
1990/0312/sys/src/9/port/stream.c:717,723 – 1990/0321/sys/src/9/port/stream.c:721,727
1990/0227		void stputq(Queue *q, Block *bp) { int i;
1990/0321		int delim;
1990/0227		if(bp->type == M_HANGUP){ freeb(bp);
1990/0312/sys/src/9/port/stream.c:724,730 – 1990/0321/sys/src/9/port/stream.c:728,736
1990/0227		q->flag |= QHUNGUP; q->other->flag |= QHUNGUP;
1990/0312		wakeup(&q->other->r);
1990/0321		delim = 1;
1990/0227		} else {
1990/0321		delim = 0;
1990/0227		lock(q); if(q->first) q->last->next = bp;
1990/0312/sys/src/9/port/stream.c:731,746 – 1990/0321/sys/src/9/port/stream.c:737,757
1990/0227		else q->first = bp;
1990/0312		q->len += BLEN(bp);
1990/0321		delim = bp->flags & S_DELIM;
1990/0312		while(bp->next) { bp = bp->next; q->len += BLEN(bp);
1990/0321		delim |= bp->flags & S_DELIM;
1990/0312		}
1990/0227		q->last = bp; if(q->len >= Streamhi)
1990/0321		if(q->len >= Streamhi){
1990/0227		q->flag |= QHIWAT;
1990/0321		delim = 1; }
1990/0227		unlock(q); } wakeup(&q->r);
1990/0321		if(delim) wakeup(&q->r);
1990/0227		} /*
1990/0321/sys/src/9/port/stream.c:51,57 – 1990/0322/sys/src/9/port/stream.c:51,57 (short | long)
1990/0227		typedef struct { int size;
1990/0312		Blist; QLock; /* qlock for sleepers on r */
1990/0322		QLock q; /* qlock for sleepers on r */
1990/0312		Rendez r; /* sleep here waiting for blocks */
1990/0227		} Bclass; Bclass bclass[Nclass]={
1990/0321/sys/src/9/port/stream.c:109,117 – 1990/0322/sys/src/9/port/stream.c:109,116
1990/0227		{ Block *bp; Bclass *bcp; int i;
1990/0322		int loop=0;
1990/0227		/* * map size to class */
1990/0321/sys/src/9/port/stream.c:124,133 – 1990/0322/sys/src/9/port/stream.c:123,134
1990/0227		lock(bcp); while(bcp->first == 0){ unlock(bcp); print("waiting for blocks\n");
1990/0312		qlock(bcp);
1990/0227		sleep(&bcp->r, isblock, (void *)bcp);
1990/0312		qunlock(bcp);
1990/0322		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);
1990/0227		lock(bcp); } bp = bcp->first;
1990/0322/sys/src/9/port/stream.c:51,57 – 1990/03292/sys/src/9/port/stream.c:51,57 (short | long)
1990/0227		typedef struct { int size;
1990/0312		Blist;
1990/0322		QLock q; /* qlock for sleepers on r */
1990/03292		QLock; /* qlock for sleepers on r */
1990/0312		Rendez r; /* sleep here waiting for blocks */
1990/0227		} Bclass; Bclass bclass[Nclass]={
1990/0322/sys/src/9/port/stream.c:123,134 – 1990/03292/sys/src/9/port/stream.c:123,133
1990/0227		lock(bcp); while(bcp->first == 0){ unlock(bcp);
1990/0322		print("waiting for block %d\n", size); if(loop++ > 10) panic("waiting for blocks"); qlock(&bcp->q);
1990/03292		panic("waiting for blocks\n"); qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250); qunlock(&bcp->q);
1990/03292		qunlock(bcp);
1990/0227		lock(bcp); } bp = bcp->first;
1990/0322/sys/src/9/port/stream.c:628,634 – 1990/03292/sys/src/9/port/stream.c:627,632
1990/0227		if(qi->open) (*qi->open)(RD(s->devq), s); c->flag |= COPEN; unlock(s); poperror(); return s;
1990/0322/sys/src/9/port/stream.c:682,688 – 1990/03292/sys/src/9/port/stream.c:680,686
1990/0227		/* * if not open, ignore it */ if(!(c->flag & COPEN))
1990/03292		if(!c->stream)
1990/0227		return; /*
1990/03292/sys/src/9/port/stream.c:62,67 – 1990/0331/sys/src/9/port/stream.c:62,89 (short | long)
1990/0227		}; /*
1990/0331		* 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); } } /*
1990/0227		* 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
1990/03292/sys/src/9/port/stream.c:123,130 – 1990/0331/sys/src/9/port/stream.c:145,155
1990/0227		lock(bcp); while(bcp->first == 0){ unlock(bcp);
1990/0322		if(loop++ > 10)
1990/0331		if(loop++ > 10){ dumpqueues(); dumpstack();
1990/03292		panic("waiting for blocks\n");
1990/0331		}
1990/03292		qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/03292		qunlock(bcp);
1990/03292/sys/src/9/port/stream.c:614,619 – 1990/0331/sys/src/9/port/stream.c:639,645
1990/0227		* hang a device and process q off the stream */ s->inuse = 1;
1990/0331		s->hread = 0;
1990/0227		s->tag[0] = 0; q = allocq(&procinfo); s->procq = WR(q);
1990/03292/sys/src/9/port/stream.c:825,832 – 1990/0331/sys/src/9/port/stream.c:851,862
1990/0227		while(left){ bp = getq(q); if(bp == 0){ if(q->flag & QHUNGUP) break;
1990/0331		if(q->flag & QHUNGUP){ if(s->hread++ < 3) break; else error(0, Ehungup); }
1990/0227		sleep(&q->r, &isinput, (void *)q); continue; }
1990/0331/sys/src/9/port/stream.c:58,64 – 1990/0403/sys/src/9/port/stream.c:58,64 (short | long)
1990/0227		{ 0 },
1990/0321		{ 68 }, { 260 },
1990/0227		{ 4096 },
1990/0403		{ 1024 },
1990/0227		}; /*
1990/0331/sys/src/9/port/stream.c:145,154 – 1990/0403/sys/src/9/port/stream.c:145,153
1990/0227		lock(bcp); while(bcp->first == 0){ unlock(bcp);
1990/0331		if(loop++ > 10){
1990/0403		if(loop++ == 10){
1990/0331		dumpqueues(); dumpstack();
1990/03292		panic("waiting for blocks\n");
1990/0403		print("waiting for blocks\n");
1990/0331		}
1990/03292		qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/0331/sys/src/9/port/stream.c:230,235 – 1990/0403/sys/src/9/port/stream.c:229,235
1990/0227		q->r.p = 0; q->info = qi; q->put = qi->iput;
1990/0403		q->len = q->nb = 0;
1990/0227		wq = q->other = q + 1; wq->r.p = 0;
1990/0331/sys/src/9/port/stream.c:236,241 – 1990/0403/sys/src/9/port/stream.c:236,242
1990/0227		wq->info = qi; wq->put = qi->oput; wq->other = q;
1990/0403		wq->len = wq->nb = 0;
1990/0227		unlock(q);
1990/0331/sys/src/9/port/stream.c:323,336 – 1990/0403/sys/src/9/port/stream.c:324,339
1990/0227		else q->first = bp;
1990/0312		q->len += BLEN(bp);
1990/0403		q->nb++;
1990/0227		delim = bp->flags & S_DELIM; while(bp->next) { bp = bp->next;
1990/0312		q->len += BLEN(bp);
1990/0403		q->nb++;
1990/0227		delim |= bp->flags & S_DELIM; } q->last = bp; if(q->len >= Streamhi)
1990/0403		if(q->len >= Streamhi || q->nb >= Streambhi)
1990/0227		q->flag |= QHIWAT; unlock(q); return delim;
1990/0331/sys/src/9/port/stream.c:369,374 – 1990/0403/sys/src/9/port/stream.c:372,378
1990/0227		q->last = bp; q->first = bp;
1990/0312		q->len += BLEN(bp);
1990/0403		q->nb++;
1990/0227		unlock(q); }
1990/0331/sys/src/9/port/stream.c:387,393 – 1990/0403/sys/src/9/port/stream.c:391,398
1990/0227		if(q->first == 0) q->last = 0;
1990/0312		q->len -= BLEN(bp);
1990/0227		if((q->flag&QHIWAT) && q->len < Streamhi/2){
1990/0403		q->nb--; if((q->flag&QHIWAT) && q->len < Streamhi/2 && q->nb < Streambhi){
1990/0227		wakeup(&q->other->next->other->r); q->flag &= ~QHIWAT; }
1990/0331/sys/src/9/port/stream.c:762,775 – 1990/0403/sys/src/9/port/stream.c:767,782
1990/0227		else q->first = bp;
1990/0312		q->len += BLEN(bp);
1990/0403		q->nb++;
1990/0321		delim = bp->flags & S_DELIM;
1990/0312		while(bp->next) { bp = bp->next; q->len += BLEN(bp);
1990/0403		q->nb++;
1990/0321		delim |= bp->flags & S_DELIM;
1990/0312		}
1990/0227		q->last = bp;
1990/0321		if(q->len >= Streamhi){
1990/0403		if(q->len >= Streamhi || q->nb >= Streambhi){
1990/0227		q->flag |= QHIWAT;
1990/0321		delim = 1; }
1990/0331/sys/src/9/port/stream.c:938,950 – 1990/0403/sys/src/9/port/stream.c:945,956
1990/0227		Queue *q; q = (Queue *)arg; return q->len < Streamhi;
1990/0403		return !QFULL(q->next);
1990/0227		} void flowctl(Queue *q) { if(q->next->len >= Streamhi) sleep(&q->r, notfull, q->next);
1990/0403		sleep(&q->r, notfull, q->next);
1990/0227		} /*
1990/0331/sys/src/9/port/stream.c:996,1002 – 1990/0403/sys/src/9/port/stream.c:1002,1008
1990/0227		* `a' is global to the whole system, just create a * pointer to it and pass it on. */ flowctl(q);
1990/0403		FLOWCTL(q);
1990/0227		bp = allocb(0); bp->rptr = bp->base = (uchar *)a; bp->wptr = bp->lim = (uchar *)a+n;
1990/0331/sys/src/9/port/stream.c:1009,1015 – 1990/0403/sys/src/9/port/stream.c:1015,1021
1990/0227		* system buffers and pass the buffers on. */ for(rem = n; ; rem -= i) { flowctl(q);
1990/0403		FLOWCTL(q);
1990/0227		bp = allocb(rem); i = bp->lim - bp->wptr; if(i >= rem){
1990/0403/sys/src/9/port/stream.c:392,398 – 1990/0406/sys/src/9/port/stream.c:392,398 (short | long)
1990/0227		q->last = 0;
1990/0312		q->len -= BLEN(bp);
1990/0403		q->nb--; if((q->flag&QHIWAT) && q->len < Streamhi/2 && q->nb < Streambhi){
1990/0406		if((q->flag&QHIWAT) && q->len<Streamhi/2 && q->nb<Streambhi/2){
1990/0227		wakeup(&q->other->next->other->r); q->flag &= ~QHIWAT; }
1990/0403/sys/src/9/port/stream.c:942,951 – 1990/0406/sys/src/9/port/stream.c:942,948
1990/0227		static int notfull(void *arg) { Queue *q;                  q = (Queue *)arg;
1990/0403		return !QFULL(q->next);
1990/0406		return !QFULL((Queue *)arg);
1990/0227		} void flowctl(Queue *q)
1990/0406/sys/src/9/port/stream.c:58,64 – 1990/0409/sys/src/9/port/stream.c:58,64 (short | long)
1990/0227		{ 0 },
1990/0321		{ 68 }, { 260 },
1990/0403		{ 1024 },
1990/0409		{ 4096 },
1990/0227		}; /*
1990/0409/sys/src/9/port/stream.c:70,76 – 1990/0509/sys/src/9/port/stream.c:70,78 (short | long)
1990/0331		Queue *q; int count; Block *bp;
1990/0509		Bclass *bcp;
1990/0331
1990/0509		print("\n");
1990/0331		for(q = qlist; q < qlist + conf.nqueue; q++, q++){ if(!(q->flag & QINUSE)) continue;
1990/0409/sys/src/9/port/stream.c:81,86 – 1990/0509/sys/src/9/port/stream.c:83,97
1990/0331		count++; print(" WR count %d len %d\n", count, WR(q)->len); }
1990/0509		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");
1990/0331		} /*
1990/0509/sys/src/9/port/stream.c:78,90 – 1990/0511/sys/src/9/port/stream.c:78,91 (short | long)
1990/0331		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);
1990/0511		print("%s %ux R c %d l %d f %ux", q->info->name, q, count, q->len, q->flag);
1990/0331		for(count = 0, bp = WR(q)->first; bp; bp = bp->next) count++; print(" WR count %d len %d\n", count, WR(q)->len);
1990/0511		print(" W c %d l %d f %ux\n", count, WR(q)->len, WR(q)->flag);
1990/0331		}
1990/0509		print("\n"); for(bcp=bclass; bcp<&bclass[Nclass-1]; bcp++){
1990/0511		for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
1990/0509		lock(bcp); for(count = 0, bp = bcp->first; bp; count++, bp = bp->next) ;
1990/0511/sys/src/9/port/stream.c:192,197 – 1990/0513/sys/src/9/port/stream.c:192,199 (short | long)
1990/0227		Bclass *bcp;
1990/0312		int tries;
1990/0227
1990/0513		if((bp->flags&S_CLASS) >= Nclass) panic("freeb class");
1990/0227		bcp = &bclass[bp->flags & S_CLASS]; lock(bcp);
1990/0312		bp->rptr = bp->wptr = 0;
1990/0513/sys/src/9/port/stream.c:258,263 – 1990/0629/sys/src/9/port/stream.c:258,279 (short | long)
1990/0227		} /*
1990/0629		* 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); } /*
1990/0227		* free a queue */ static void
1990/0513/sys/src/9/port/stream.c:515,522 – 1990/0629/sys/src/9/port/stream.c:531,542
1990/0227		void nullput(Queue *q, Block *bp) { freeb(bp); error(0, Ehungup);
1990/0629		if(bp->type == M_HANGUP) freeb(bp); else { freeb(bp); error(0, Ehungup); }
1990/0227		} /*
1990/0513/sys/src/9/port/stream.c:658,665 – 1990/0629/sys/src/9/port/stream.c:678,685
1990/0227		* hang a device and process q off the stream */ s->inuse = 1;
1990/0629		s->opens = 1;
1990/0331		s->hread = 0;
1990/0227		s->tag[0] = 0; q = allocq(&procinfo); s->procq = WR(q); q = allocq(qi);
1990/0513/sys/src/9/port/stream.c:697,702 – 1990/0629/sys/src/9/port/stream.c:717,723
1990/0227		&& s->dev == c->dev && s->id == STREAMID(c->qid)){ s->inuse++;
1990/0629		s->opens++;
1990/0227		c->stream = s; unlock(s); return;
1990/0513/sys/src/9/port/stream.c:712,717 – 1990/0629/sys/src/9/port/stream.c:733,782
1990/0227		} /*
1990/0629		* 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); } /*
1990/0227		* On the last close of a stream, for each queue on the * stream release its blocks and call its close routine. */
1990/0513/sys/src/9/port/stream.c:729,761 – 1990/0629/sys/src/9/port/stream.c:794,828
1990/0227		return; /* * decrement the reference cound
1990/0629		* decrement the reference count
1990/0227		*/
1990/03013		lock(s); if(s->inuse != 1){ s->inuse--;
1990/0227		unlock(c->stream); return;
1990/0629		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); }
1990/0227		}
1990/0629		s->opens--;
1990/0227		/* * descend the stream closing the queues
1990/0629		* leave it and free it
1990/0227		*/
1990/03013		for(q = s->procq; q; q = q->next){
1990/0227		if(q->info->close) (*q->info->close)(q->other);
1990/03013		if(q == s->devq->other)
1990/0227		break; } /* * ascend the stream freeing the queues */
1990/03013		for(q = s->devq; q; q = nq){
1990/0227		nq = q->next; freeq(q); }
1990/03013		s->id = s->dev = s->type = 0; s->inuse--;
1990/0629		streamexit(s, 1);
1990/03013		unlock(s);
1990/0227		}
1990/0629/sys/src/9/port/stream.c:246,251 – 1990/0702/sys/src/9/port/stream.c:246,252 (short | long)
1990/0403		q->len = q->nb = 0;
1990/0227		wq = q->other = q + 1;
1990/0702		wq->flag = QINUSE;
1990/0227		wq->r.p = 0; wq->info = qi; wq->put = qi->oput;
1990/0702/sys/src/9/port/stream.c:65,70 – 1990/0707/sys/src/9/port/stream.c:65,86 (short | long)
1990/0331		* Dump all block information of how many blocks are in which queues */ void
1990/0707		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
1990/0331		dumpqueues(void) { Queue *q;
1990/0702/sys/src/9/port/stream.c:76,88 – 1990/0707/sys/src/9/port/stream.c:92,103
1990/0331		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++;
1990/0511		print("%s %ux R c %d l %d f %ux", q->info->name, q, count, q->len, q->flag);
1990/0331		for(count = 0, bp = WR(q)->first; bp; bp = bp->next) count++;
1990/0511		print(" W c %d l %d f %ux\n", count, WR(q)->len, WR(q)->flag);
1990/0707		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');
1990/0331		}
1990/0509		print("\n");
1990/0511		for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
1990/0707/sys/src/9/port/stream.c:6,11 – 1990/0801/sys/src/9/port/stream.c:6,12 (short | long)
1990/0227		#include "io.h" #include "errno.h" #include "devtab.h"
1990/0801		#include "fcall.h"
1990/0227
1990/0321		enum { Nclass=4, /* number of block classes */
1990/0707/sys/src/9/port/stream.c:72,78 – 1990/0801/sys/src/9/port/stream.c:73,79
1990/0707		lock(q); for(bp = q->first; bp; bp = bp->next){ print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' ');
1990/0801		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM));
1990/0707		for(cp = bp->rptr; cp<bp->wptr && cp<bp->rptr+10; cp++) print(" %uo", *cp); print("\n");
1990/0707/sys/src/9/port/stream.c:1155,1158 – 1990/0801/sys/src/9/port/stream.c:1156,1189
1990/0312		lp++; } return i;
1990/0801		} /* * 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);
1990/0227		}
1990/0801/sys/src/9/port/stream.c:48,53 – 1990/08272/sys/src/9/port/stream.c:48,58 (short | long)
1990/0227		/* * The block classes. There are Nclass block sizes, each with its own free list. * All are ialloced at qinit() time.
1990/08272		* * 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
1990/0227		*/ typedef struct { int size;
1990/0801/sys/src/9/port/stream.c:57,64 – 1990/08272/sys/src/9/port/stream.c:62,69
1990/0227		} Bclass; Bclass bclass[Nclass]={ { 0 },
1990/0321		{ 68 }, { 260 },
1990/08272		{ 64 }, { 256 },
1990/0409		{ 4096 },
1990/0227		};
1990/0801/sys/src/9/port/stream.c:133,140 – 1990/08272/sys/src/9/port/stream.c:138,150
1990/0227		n = n/2; bcp = &bclass[class]; for(i = 0; i < n; i++) {
1990/08272		/* * The i == 0 means that each allocation range * starts on a page boundary. This makes sure * no block crosses a page boundary. */
1990/0227		if(bcp->size) bp->base = (uchar *)ialloc(bcp->size, 0);
1990/08272		bp->base = (uchar *)ialloc(bcp->size, i == 0);
1990/0227		bp->lim = bp->base + bcp->size; bp->flags = class; freeb(bp);
1990/08272/sys/src/9/port/stream.c:24,42 – 1990/0905/sys/src/9/port/stream.c:24,29 (short | long)
1990/0312		*/
1990/0321		static Qinfo *lds[Nlds+1];
1990/0227
1990/0321		void newqinfo(Qinfo *qi) { int i;
1990/0227
1990/0321		for(i=0; i<Nlds && lds[i]; i++) if(lds[i] == qi) return; if(i == Nlds) panic("pushable"); lds[i] = qi; }
1990/0227		/* * All stream structures are ialloc'd at boot time */
1990/08272/sys/src/9/port/stream.c:68,73 – 1990/0905/sys/src/9/port/stream.c:55,68
1990/0227		}; /*
1990/0905		* the per stream directory structure */ Dirtab streamdir[]={ "data", Sdataqid, 0, 0600, "ctl", Sctlqid, 0, 0600, }; /*
1990/0331		* Dump all block information of how many blocks are in which queues */ void
1990/08272/sys/src/9/port/stream.c:120,132 – 1990/0905/sys/src/9/port/stream.c:115,133
1990/0227		* 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
1990/0905		* * 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.
1990/0227		*/ void streaminit(void) { int class, i, n;
1990/0905		int class, i, n, left;
1990/0227		Block *bp; Bclass *bcp;
1990/0905		uchar *ptr;
1990/0227		slist = (Stream *)ialloc(conf.nstream * sizeof(Stream), 0); qlist = (Queue *)ialloc(conf.nqueue * sizeof(Queue), 0);
1990/08272/sys/src/9/port/stream.c:133,138 – 1990/0905/sys/src/9/port/stream.c:134,140
1990/0227		blist = (Block *)ialloc(conf.nblock * sizeof(Block), 0); bp = blist; n = conf.nblock;
1990/0905		left = 0;
1990/0227		for(class = 0; class < Nclass; class++){ if(class < Nclass-1) n = n/2;
1990/08272/sys/src/9/port/stream.c:143,150 – 1990/0905/sys/src/9/port/stream.c:145,159
1990/08272		* starts on a page boundary. This makes sure * no block crosses a page boundary. */
1990/0227		if(bcp->size)
1990/08272		bp->base = (uchar *)ialloc(bcp->size, i == 0);
1990/0905		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; }
1990/0227		bp->lim = bp->base + bcp->size; bp->flags = class; freeb(bp);
1990/08272/sys/src/9/port/stream.c:479,532 – 1990/0905/sys/src/9/port/stream.c:488,493
1990/0227		} /*
1990/0312		* 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 *
1990/08272/sys/src/9/port/stream.c:626,639 – 1990/0905/sys/src/9/port/stream.c:587,592
1990/0227		} /* * 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. *
1990/08272/sys/src/9/port/stream.c:1197,1199 – 1990/0905/sys/src/9/port/stream.c:1150,1169
1990/0801		devdir(c, c->qid, name, n, 0, &dir); convD2M(&dir, db);
1990/0227		}
1990/0905		/* * 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; }
1990/0905/sys/src/9/port/stream.c:9,15 – 1990/09051/sys/src/9/port/stream.c:9,15 (short | long)
1990/0801		#include "fcall.h"
1990/0227
1990/0321		enum { Nclass=4, /* number of block classes */
1990/09051		Nclass=5, /* number of block classes */
1990/0321		Nlds=32, /* max number of pushable line disciplines */ };
1990/0905/sys/src/9/port/stream.c:51,56 – 1990/09051/sys/src/9/port/stream.c:51,57
1990/0227		{ 0 },
1990/08272		{ 64 }, { 256 },
1990/09051		{ 2048 },
1990/0409		{ 4096 },
1990/0227		};
1990/09051/sys/src/9/port/stream.c:9,15 – 1990/0907/sys/src/9/port/stream.c:9,15 (short | long)
1990/0801		#include "fcall.h"
1990/0227
1990/0321		enum {
1990/09051		Nclass=5, /* number of block classes */
1990/0907		Nclass=4, /* number of block classes */
1990/0321		Nlds=32, /* max number of pushable line disciplines */ };
1990/09051/sys/src/9/port/stream.c:24,29 – 1990/0907/sys/src/9/port/stream.c:24,42
1990/0312		*/
1990/0321		static Qinfo *lds[Nlds+1];
1990/0227
1990/0907		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; }
1990/0227		/* * All stream structures are ialloc'd at boot time */
1990/09051/sys/src/9/port/stream.c:51,69 – 1990/0907/sys/src/9/port/stream.c:64,73
1990/0227		{ 0 },
1990/08272		{ 64 }, { 256 },
1990/09051		{ 2048 },
1990/0409		{ 4096 },
1990/0227		}; /*
1990/0905		* the per stream directory structure */ Dirtab streamdir[]={ "data", Sdataqid, 0, 0600, "ctl", Sctlqid, 0, 0600, };                  /*
1990/0331		* Dump all block information of how many blocks are in which queues */ void
1990/09051/sys/src/9/port/stream.c:116,134 – 1990/0907/sys/src/9/port/stream.c:120,132
1990/0227		* 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
1990/0905		* * 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.
1990/0227		*/ void streaminit(void) {
1990/0905		int class, i, n, left;
1990/0907		int class, i, n;
1990/0227		Block *bp; Bclass *bcp;
1990/0905		uchar *ptr;
1990/0227		slist = (Stream *)ialloc(conf.nstream * sizeof(Stream), 0); qlist = (Queue *)ialloc(conf.nqueue * sizeof(Queue), 0);
1990/09051/sys/src/9/port/stream.c:135,141 – 1990/0907/sys/src/9/port/stream.c:133,138
1990/0227		blist = (Block *)ialloc(conf.nblock * sizeof(Block), 0); bp = blist; n = conf.nblock;
1990/0905		left = 0;
1990/0227		for(class = 0; class < Nclass; class++){ if(class < Nclass-1) n = n/2;
1990/09051/sys/src/9/port/stream.c:146,160 – 1990/0907/sys/src/9/port/stream.c:143,150
1990/08272		* starts on a page boundary. This makes sure * no block crosses a page boundary. */
1990/0905		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; }
1990/0907		if(bcp->size) bp->base = (uchar *)ialloc(bcp->size, i == 0);
1990/0227		bp->lim = bp->base + bcp->size; bp->flags = class; freeb(bp);
1990/09051/sys/src/9/port/stream.c:489,494 – 1990/0907/sys/src/9/port/stream.c:479,532
1990/0227		} /*
1990/0907		* 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; } /*
1990/0312		* grow the front of a list of blocks by n bytes */ Block *
1990/09051/sys/src/9/port/stream.c:588,593 – 1990/0907/sys/src/9/port/stream.c:626,639
1990/0227		} /*
1990/0907		* the per stream directory structure */ Dirtab streamdir[]={ "data", Sdataqid, 0, 0600, "ctl", Sctlqid, 0, 0600, }; /*
1990/0227		* A stream device consists of the contents of streamdir plus * any directory supplied by the actual device. *
1990/09051/sys/src/9/port/stream.c:1151,1170 – 1990/0907/sys/src/9/port/stream.c:1197,1199
1990/0801		devdir(c, c->qid, name, n, 0, &dir); convD2M(&dir, db);
1990/0227		}
1990/0905		/* * 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; }
1990/0907/sys/src/9/port/stream.c:10,16 – 1990/0911/sys/src/9/port/stream.c:10,15 (short | long)
1990/0227
1990/0321		enum {
1990/0907		Nclass=4, /* number of block classes */
1990/0321		Nlds=32, /* max number of pushable line disciplines */ };
1990/0312		/*
1990/0907/sys/src/9/port/stream.c:22,42 – 1990/0911/sys/src/9/port/stream.c:21,28
1990/0312		/* * line disciplines that can be pushed */
1990/0321		static Qinfo *lds[Nlds+1];
1990/0911		static Qinfo *lds;
1990/0227
1990/0907		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; }
1990/0227		/* * All stream structures are ialloc'd at boot time */
1990/0907/sys/src/9/port/stream.c:48,58 – 1990/0911/sys/src/9/port/stream.c:34,39
1990/0227		/* * The block classes. There are Nclass block sizes, each with its own free list. * All are ialloced at qinit() time.
1990/08272		* * 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
1990/0227		*/ typedef struct { int size;
1990/0907/sys/src/9/port/stream.c:62,121 – 1990/0911/sys/src/9/port/stream.c:43,55
1990/0227		} Bclass; Bclass bclass[Nclass]={ { 0 },
1990/08272		{ 64 }, { 256 },
1990/0911		{ 68 }, { 260 },
1990/0409		{ 4096 },
1990/0227		}; /*
1990/0331		* Dump all block information of how many blocks are in which queues */ void
1990/0707		dumpblocks(Queue *q, char c) { Block *bp; uchar *cp;
1990/0911		#include "stream.h"
1990/0707		lock(q); for(bp = q->first; bp; bp = bp->next){
1990/0801		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM));
1990/0707		for(cp = bp->rptr; cp<bp->wptr && cp<bp->rptr+10; cp++) print(" %uo", *cp); print("\n"); } unlock(q); }                  void
1990/0331		dumpqueues(void) { Queue *q; int count; Block *bp;
1990/0509		Bclass *bcp;
1990/0331
1990/0509		print("\n");
1990/0331		for(q = qlist; q < qlist + conf.nqueue; q++, q++){ if(!(q->flag & QINUSE)) continue;
1990/0707		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');
1990/0331		}
1990/0509		print("\n");
1990/0511		for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
1990/0509		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");
1990/0331		}                  /*
1990/0227		* Allocate streams, queues, and blocks. Allocate n block classes with * 1/2(m+1) to class m < n-1
1990/0907/sys/src/9/port/stream.c:128,133 – 1990/0911/sys/src/9/port/stream.c:62,70
1990/0227		Block *bp; Bclass *bcp;
1990/0911		/* * allocate blocks, queues, and streams */
1990/0227		slist = (Stream *)ialloc(conf.nstream * sizeof(Stream), 0); qlist = (Queue *)ialloc(conf.nqueue * sizeof(Queue), 0); blist = (Block *)ialloc(conf.nblock * sizeof(Block), 0);
1990/0907/sys/src/9/port/stream.c:138,150 – 1990/0911/sys/src/9/port/stream.c:75,82
1990/0227		n = n/2; bcp = &bclass[class]; for(i = 0; i < n; i++) {
1990/08272		/* * The i == 0 means that each allocation range * starts on a page boundary. This makes sure * no block crosses a page boundary. */
1990/0907		if(bcp->size) bp->base = (uchar *)ialloc(bcp->size, i == 0);
1990/0911		bp->base = (uchar *)ialloc(bcp->size, 0);
1990/0227		bp->lim = bp->base + bcp->size; bp->flags = class; freeb(bp);
1990/0907/sys/src/9/port/stream.c:151,159 – 1990/0911/sys/src/9/port/stream.c:83,109
1990/0227		bp++; } }
1990/0911		/* * make stream modules available */ streaminit0();
1990/0227		} /*
1990/0911		* 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)(); } /*
1990/0227		* allocate a block */ static int
1990/0907/sys/src/9/port/stream.c:169,175 – 1990/0911/sys/src/9/port/stream.c:119,124
1990/0227		{ Block *bp; Bclass *bcp;
1990/0322		int loop=0;
1990/0227		/* * map size to class
1990/0907/sys/src/9/port/stream.c:183,192 – 1990/0911/sys/src/9/port/stream.c:132,137
1990/0227		lock(bcp); while(bcp->first == 0){ unlock(bcp);
1990/0403		if(loop++ == 10){
1990/0331		dumpqueues();
1990/0403		print("waiting for blocks\n");
1990/0331		}
1990/03292		qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/03292		qunlock(bcp);
1990/0907/sys/src/9/port/stream.c:242,247 – 1990/0911/sys/src/9/port/stream.c:187,212
1990/0227		} /*
1990/0911		* 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; } } /*
1990/0227		* allocate a pair of queues. flavor them with the requested put routines. * the `QINUSE' flag on the read side is the only one used. */
1990/0907/sys/src/9/port/stream.c:572,584 – 1990/0911/sys/src/9/port/stream.c:537,549
1990/0227		static Qinfo * qinfofind(char *name) { Qinfo **qip;
1990/0911		Qinfo *qi;
1990/0227		if(name == 0) error(0, Ebadld); for(qip = lds; *qip; qip++) if(strcmp((*qip)->name, name)==0) return *qip;
1990/0911		for(qi = lds; qi; qi = qi->next) if(strcmp(qi->name, name)==0) return qi;
1990/0227		error(0, Ebadld); }
1990/0907/sys/src/9/port/stream.c:1070,1084 – 1990/0911/sys/src/9/port/stream.c:1035,1050
1990/0227		long rem; int i;
1990/0911		s = c->stream;
1990/0227		/* * one writer at a time */ s = c->stream; qlock(&s->wrlock); if(waserror()){ qunlock(&s->wrlock); nexterror(); }
1990/0911		*/
1990/0227		/* * decode the qid
1990/0907/sys/src/9/port/stream.c:1088,1096 – 1990/0911/sys/src/9/port/stream.c:1054,1060
1990/0227		break; case Sctlqid: n = streamctlwrite(s, a, n); qunlock(&s->wrlock); poperror(); return n;
1990/0911		goto out;
1990/0227		default: panic("bad stream qid\n"); }
1990/0907/sys/src/9/port/stream.c:1139,1146 – 1990/0911/sys/src/9/port/stream.c:1103,1111
1990/0227		} } } qunlock(&s->wrlock); poperror();
1990/0911		out: /* qunlock(&s->wrlock); poperror(); /**/
1990/0227		return n;
1990/0312		}
1990/0907/sys/src/9/port/stream.c:1196,1199 – 1990/0911/sys/src/9/port/stream.c:1161,1213
1990/0801		devdir(c, c->qid, name, n, 0, &dir); convD2M(&dir, db);
1990/0911		} /* * 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");
1990/0227		}
1990/0911/sys/src/9/port/stream.c:1148,1163 – 1990/0914/sys/src/9/port/stream.c:1148,1164 (short | long)
1990/0801		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;
1990/0914		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);
1990/0801		} unlock(q); devdir(c, c->qid, name, n, 0, &dir); convD2M(&dir, db);
1990/0914/sys/src/9/port/stream.c:127,133 – 1990/0930/sys/src/9/port/stream.c:127,133 (short | long)
1990/0227		; /* * look for a free block, garbage collect if there are none
1990/0930		* look for a free block
1990/0227		*/ lock(bcp); while(bcp->first == 0){
1990/0914/sys/src/9/port/stream.c:878,889 – 1990/0930/sys/src/9/port/stream.c:878,914
1990/0227		} /*
1990/0930		* 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"); } } /*
1990/0227		* return true if there is an output buffer available */ static int isinput(void *x) { return ((Queue *)x)->first != 0;
1990/0930		Queue *q; q = (Queue *)x; return (q->flag&QHUNGUP) || q->first!=0;
1990/0227		} /*
1990/0914/sys/src/9/port/stream.c:895,922 – 1990/0930/sys/src/9/port/stream.c:920,935
1990/0227		Block *bp; Stream *s; Queue *q; long rv = 0; int left, i, x;
1990/0930		int left, i;
1990/0227		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"); }
1990/0930		if(STREAMTYPE(c->qid) != Sdataqid) return streamctlread(c, vbuf, n);
1990/0227		/* * one reader at a time */
1990/0930		s = c->stream;
1990/0227		qlock(&s->rdlock); if(waserror()){ qunlock(&s->rdlock);
1990/0914/sys/src/9/port/stream.c:974,984 – 1990/0930/sys/src/9/port/stream.c:987,1002
1990/0227		* This routing is entrered with s->wrlock'ed and must unlock. */ static long streamctlwrite(Stream *s, void *a, long n)
1990/0930		streamctlwrite(Chan *c, void *a, long n)
1990/0227		{ Qinfo *qi; Block *bp;
1990/0930		Stream *s;
1990/0227
1990/0930		if(STREAMTYPE(c->qid) != Sctlqid) panic("streamctlwrite %lux", c->qid); s = c->stream;
1990/0227		/* * package */
1990/0914/sys/src/9/port/stream.c:1038,1063 – 1990/0930/sys/src/9/port/stream.c:1056,1065
1990/0911		s = c->stream;
1990/0227		/* * one writer at a time qlock(&s->wrlock); if(waserror()){ qunlock(&s->wrlock); nexterror(); }
1990/0911		*/
1990/0227		/* * decode the qid */ switch(STREAMTYPE(c->qid)){ case Sdataqid: break; case Sctlqid: n = streamctlwrite(s, a, n);
1990/0911		goto out;
1990/0227		default: panic("bad stream qid\n"); }
1990/0930		if(STREAMTYPE(c->qid) != Sdataqid) return streamctlwrite(c, a, n);
1990/0227		/* * No writes allowed on hungup channels
1990/0914/sys/src/9/port/stream.c:1066,1072 – 1990/0930/sys/src/9/port/stream.c:1068,1074
1990/0227		if(q->other->flag & QHUNGUP) error(0, Ehungup);
1990/0312		if((GLOBAL(a) && !docopy) || n==0){
1990/0930		if(!docopy && GLOBAL(a)){
1990/0227		/* * `a' is global to the whole system, just create a * pointer to it and pass it on.
1990/0914/sys/src/9/port/stream.c:1103,1111 – 1990/0930/sys/src/9/port/stream.c:1105,1110
1990/0227		} } }
1990/0911		out: /* qunlock(&s->wrlock); poperror(); /**/
1990/0227		return n;
1990/0312		}
1990/0930/sys/src/9/port/stream.c:235,240 – 1990/1009/sys/src/9/port/stream.c:235,241 (short | long)
1990/0227		q->info = qi; q->put = qi->iput;
1990/0403		q->len = q->nb = 0;
1990/1009		q->ptr = 0;
1990/0227		wq = q->other = q + 1;
1990/0702		wq->flag = QINUSE;
1990/0930/sys/src/9/port/stream.c:242,247 – 1990/1009/sys/src/9/port/stream.c:243,249
1990/0227		wq->info = qi; wq->put = qi->oput; wq->other = q;
1990/1009		wq->ptr = 0;
1990/0403		wq->len = wq->nb = 0;
1990/0227		unlock(q);
1990/0930/sys/src/9/port/stream.c:625,634 – 1990/1009/sys/src/9/port/stream.c:627,636
1990/0227		} /* * create a new stream
1990/1009		* create a new stream, if noopen is non-zero, don't increment the open count
1990/0227		*/ Stream * streamnew(Chan *c, Qinfo *qi)
1990/1009		streamnew(ushort type, ushort dev, ushort id, Qinfo *qi, int noopen)
1990/0227		{ Stream *s; Queue *q;
1990/0930/sys/src/9/port/stream.c:651,676 – 1990/1009/sys/src/9/port/stream.c:653,677
1990/0227		} if(waserror()){ unlock(s); streamclose(c);
1990/1009		streamclose1(s);
1990/0227		nexterror(); } /* * marry a stream and a channel
1990/1009		* identify the stream
1990/0227		*/ if(c){ c->stream = s; s->type = c->type; s->dev = c->dev; s->id = STREAMID(c->qid); } else s->type = -1;
1990/1009		s->type = type; s->dev = dev; s->id = id;
1990/0227		/* * hang a device and process q off the stream */ s->inuse = 1;
1990/0629		s->opens = 1;
1990/1009		if(noopen) s->opens = 0; else s->opens = 1;
1990/0331		s->hread = 0;
1990/0227		q = allocq(&procinfo); s->procq = WR(q);
1990/0930/sys/src/9/port/stream.c:699,705 – 1990/1009/sys/src/9/port/stream.c:700,706
1990/0227		Queue *q; /* * if the stream already exists, just up the reference count.
1990/1009		* if the stream already exists, just increment the reference counts.
1990/0227		*/ for(s = slist; s < &slist[conf.nstream]; s++) { if(s->inuse && s->type == c->type && s->dev == c->dev
1990/0930/sys/src/9/port/stream.c:721,727 – 1990/1009/sys/src/9/port/stream.c:722,728
1990/0227		/* * create a new stream */ streamnew(c, qi);
1990/1009		c->stream = streamnew(c->type, c->dev, STREAMID(c->qid), qi, 0);
1990/0227		} /*
1990/0930/sys/src/9/port/stream.c:773,791 – 1990/1009/sys/src/9/port/stream.c:774,785
1990/0227		* stream release its blocks and call its close routine. */ void streamclose(Chan *c)
1990/1009		streamclose1(Stream *s)
1990/0227		{ Queue *q, *nq; Block *bp;
1990/03013		Stream *s = c->stream;
1990/0227		/* * if not open, ignore it */
1990/03292		if(!c->stream)
1990/0227		return;                  /*
1990/0629		* decrement the reference count
1990/0227		*/
1990/03013		lock(s);
1990/0930/sys/src/9/port/stream.c:817,822 – 1990/1009/sys/src/9/port/stream.c:811,826
1990/0629		streamexit(s, 1);
1990/03013		unlock(s);
1990/0227		}
1990/1009		void streamclose(Chan *c) { /* * if no stream, ignore it */ if(!c->stream) return; streamclose1(c->stream); }
1990/0227		/* * put a block to be read into the queue. wakeup any waiting reader
1990/0930/sys/src/9/port/stream.c:1174,1180 – 1990/1009/sys/src/9/port/stream.c:1178,1184
1990/0911		lock(q); for(bp = q->first; bp; bp = bp->next){ print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM));
1990/1009		print("%c%d%c", c, bp->wptr-bp->rptr, (bp->flags&S_DELIM)?'D':' ');
1990/0911		for(cp = bp->rptr; cp<bp->wptr && cp<bp->rptr+10; cp++) print(" %uo", *cp); print("\n");
1990/1009/sys/src/9/port/stream.c:784,798 – 1990/1011/sys/src/9/port/stream.c:784,803 (short | long)
1990/0227		*/
1990/03013		lock(s);
1990/0629		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;
1990/1011		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();
1990/0629		} /*
Too many diffs (26 > 25). Stopping.