1990/1113/gnot/stream.c (diff list | history)

1990/03091/sys/src/9/gnot/stream.c:7,18 – 1990/0312/sys/src/9/gnot/stream.c:7,29 (short | long)
1990/03091		#include "errno.h" #include "devtab.h"
1990/0312		/* * process end line discipline */
1990/03091		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/03091
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/03091		static Qinfo *lds[] = { /* &noetherinfo, */
1990/0312		&dkmuxinfo, &urpinfo,
1990/03091		0 };
1990/03091/sys/src/9/gnot/stream.c:34,40 – 1990/0312/sys/src/9/gnot/stream.c:45,53
1990/03091		*/ typedef struct { int size; Queue;
1990/0312		Blist; QLock; /* qlock for sleepers on r */ Rendez r; /* sleep here waiting for blocks */
1990/03091		} Bclass; Bclass bclass[Nclass]={ { 0 },
1990/03091/sys/src/9/gnot/stream.c:108,114 – 1990/0312/sys/src/9/gnot/stream.c:121,129
1990/03091		while(bcp->first == 0){ unlock(bcp); print("waiting for blocks\n");
1990/0312		qlock(bcp);
1990/03091		sleep(&bcp->r, isblock, (void *)bcp);
1990/0312		qunlock(bcp);
1990/03091		lock(bcp); } bp = bcp->first;
1990/03091/sys/src/9/gnot/stream.c:128,152 – 1990/0312/sys/src/9/gnot/stream.c:143,175
1990/03091		} /* * 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/03091		*/ void freeb(Block *bp) { Bclass *bcp;
1990/0312		int tries;
1990/03091		bcp = &bclass[bp->flags & S_CLASS]; bp->rptr = bp->wptr = 0; lock(bcp);
1990/0312		bp->rptr = bp->wptr = 0;
1990/03091		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/03091		bcp->last = bp; bp->next = 0; wakeup(&bcp->r); unlock(bcp);
1990/0312		wakeup(&bcp->r);
1990/03091		} /*
1990/03091/sys/src/9/gnot/stream.c:269,279 – 1990/0312/sys/src/9/gnot/stream.c:292,302
1990/03091		q->last->next = bp; else q->first = bp; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/03091		delim = bp->flags & S_DELIM; while(bp->next) { bp = bp->next; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/03091		delim |= bp->flags & S_DELIM; } q->last = bp;
1990/03091/sys/src/9/gnot/stream.c:292,306 – 1990/0312/sys/src/9/gnot/stream.c:315,328
1990/03091		q->last->next = bp; else q->first = bp; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/03091		delim = bp->flags & S_DELIM; while(bp->next) { bp = bp->next; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/03091		delim |= bp->flags & S_DELIM; } q->last = bp; bp->next = 0; return delim; }
1990/03091/sys/src/9/gnot/stream.c:307,313 – 1990/0312/sys/src/9/gnot/stream.c:329,335
1990/03091		/* * add a block to the start of a queue */ static void
1990/0312		void
1990/03091		putbq(Blist *q, Block *bp) { lock(q);
1990/03091/sys/src/9/gnot/stream.c:316,327 – 1990/0312/sys/src/9/gnot/stream.c:338,349
1990/03091		else q->last = bp; q->first = bp; q->len += bp->wptr - bp->rptr;
1990/0312		q->len += BLEN(bp);
1990/03091		unlock(q); } /* * remove the first block from a queue
1990/0312		* remove the first block from a queue
1990/03091		*/ Block * getq(Queue *q)
1990/03091/sys/src/9/gnot/stream.c:334,340 – 1990/0312/sys/src/9/gnot/stream.c:356,362
1990/03091		q->first = bp->next; if(q->first == 0) q->last = 0; q->len -= bp->wptr - bp->rptr;
1990/0312		q->len -= BLEN(bp);
1990/03091		if((q->flag&QHIWAT) && q->len < Streamhi/2){ wakeup(&q->other->next->other->r); q->flag &= ~QHIWAT;
1990/03091/sys/src/9/gnot/stream.c:344,349 – 1990/0312/sys/src/9/gnot/stream.c:366,375
1990/03091		unlock(q); return bp; }
1990/0312		/* * remove the first block from a list of blocks */
1990/03091		Block * getb(Blist *q) {
1990/03091/sys/src/9/gnot/stream.c:354,360 – 1990/0312/sys/src/9/gnot/stream.c:380,386
1990/03091		q->first = bp->next; if(q->first == 0) q->last = 0; q->len -= bp->wptr - bp->rptr;
1990/0312		q->len -= BLEN(bp);
1990/03091		bp->next = 0; } return bp;
1990/03091/sys/src/9/gnot/stream.c:361,366 – 1990/0312/sys/src/9/gnot/stream.c:387,466
1990/03091		} /*
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/03091		* put a block into the bit bucket */ void
1990/03091/sys/src/9/gnot/stream.c:415,421 – 1990/0312/sys/src/9/gnot/stream.c:515,521
1990/03091		int len; len = strlen(name); if(bp->wptr - bp->rptr < len)
1990/0312		if(BLEN(bp) < len)
1990/03091		return 0; if(strncmp(name, (char *)bp->rptr, len)==0){ if(bp->rptr[len] == ' ')
1990/03091/sys/src/9/gnot/stream.c:622,627 – 1990/0312/sys/src/9/gnot/stream.c:722,728
1990/03091		freeb(bp); q->flag |= QHUNGUP; q->other->flag |= QHUNGUP;
1990/0312		wakeup(&q->other->r);
1990/03091		} else { lock(q); if(q->first)
1990/03091/sys/src/9/gnot/stream.c:628,635 – 1990/0312/sys/src/9/gnot/stream.c:729,740
1990/03091		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/03091		q->last = bp; q->len += bp->wptr - bp->rptr; if(q->len >= Streamhi) q->flag |= QHIWAT; unlock(q);
1990/03091/sys/src/9/gnot/stream.c:651,658 – 1990/0312/sys/src/9/gnot/stream.c:756,762
1990/03091		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/03091		return n; }
1990/03091/sys/src/9/gnot/stream.c:716,722 – 1990/0312/sys/src/9/gnot/stream.c:820,826
1990/03091		continue; } i = bp->wptr - bp->rptr;
1990/0312		i = BLEN(bp);
1990/03091		if(i <= left){ memcpy(buf, bp->rptr, i); left -= i;
1990/03091/sys/src/9/gnot/stream.c:806,812 – 1990/0312/sys/src/9/gnot/stream.c:910,916
1990/03091		* 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/03091		{ Stream *s; Block *bp;
1990/03091/sys/src/9/gnot/stream.c:846,852 – 1990/0312/sys/src/9/gnot/stream.c:950,956
1990/03091		if(q->other->flag & QHUNGUP) error(0, Ehungup); if(GLOBAL(a) || n==0){
1990/0312		if((GLOBAL(a) && !docopy) || n==0){
1990/03091		/* * `a' is global to the whole system, just create a * pointer to it and pass it on.
1990/03091/sys/src/9/gnot/stream.c:886,889 – 1990/0312/sys/src/9/gnot/stream.c:990,1017
1990/03091		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/03091		}
1990/0312/sys/src/9/gnot/stream.c:7,12 – 1990/0321/sys/src/9/gnot/stream.c:7,17 (short | long)
1990/03091		#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/gnot/stream.c:15,36 – 1990/0321/sys/src/9/gnot/stream.c:20,41
1990/03091
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/03091		static Qinfo *lds[] = {
1990/0312		&dkmuxinfo, &urpinfo,
1990/03091		0 };
1990/0321		static Qinfo *lds[Nlds+1];
1990/03091		enum { Nclass=4, };
1990/0321		void newqinfo(Qinfo *qi) { int i;
1990/03091
1990/0321		for(i=0; i<Nlds && lds[i]; i++) if(lds[i] == qi) return; if(i == Nlds) panic("pushable"); lds[i] = qi; }
1990/03091		/* * All stream structures are ialloc'd at boot time */
1990/0312/sys/src/9/gnot/stream.c:51,58 – 1990/0321/sys/src/9/gnot/stream.c:56,63
1990/03091		} Bclass; Bclass bclass[Nclass]={ { 0 }, { 64 }, { 512 },
1990/0321		{ 68 }, { 260 },
1990/03091		{ 4096 }, };
1990/0312/sys/src/9/gnot/stream.c:88,94 – 1990/0321/sys/src/9/gnot/stream.c:93,98
1990/03091		} }                  /* * allocate a block */
1990/0312/sys/src/9/gnot/stream.c:169,175 – 1990/0321/sys/src/9/gnot/stream.c:173,180
1990/0312		}
1990/03091		bcp->last = bp; unlock(bcp);
1990/0312		wakeup(&bcp->r);
1990/0321		if(bcp->r.p) wakeup(&bcp->r);
1990/03091		} /*
1990/0312/sys/src/9/gnot/stream.c:716,722 – 1990/0321/sys/src/9/gnot/stream.c:721,727
1990/03091		void stputq(Queue *q, Block *bp) { int i;
1990/0321		int delim;
1990/03091		if(bp->type == M_HANGUP){ freeb(bp);
1990/0312/sys/src/9/gnot/stream.c:723,729 – 1990/0321/sys/src/9/gnot/stream.c:728,736
1990/03091		q->flag |= QHUNGUP; q->other->flag |= QHUNGUP;
1990/0312		wakeup(&q->other->r);
1990/0321		delim = 1;
1990/03091		} else {
1990/0321		delim = 0;
1990/03091		lock(q); if(q->first) q->last->next = bp;
1990/0312/sys/src/9/gnot/stream.c:730,745 – 1990/0321/sys/src/9/gnot/stream.c:737,757
1990/03091		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/03091		q->last = bp; if(q->len >= Streamhi)
1990/0321		if(q->len >= Streamhi){
1990/03091		q->flag |= QHIWAT;
1990/0321		delim = 1; }
1990/03091		unlock(q); } wakeup(&q->r);
1990/0321		if(delim) wakeup(&q->r);
1990/03091		} /*
1990/0321/sys/src/9/gnot/stream.c:109,117 – 1990/0322/sys/src/9/gnot/stream.c:109,116 (short | long)
1990/03091		{ Block *bp; Bclass *bcp; int i;
1990/0322		int loop=0;
1990/03091		/* * map size to class */
1990/0321/sys/src/9/gnot/stream.c:124,132 – 1990/0322/sys/src/9/gnot/stream.c:123,132
1990/03091		lock(bcp); while(bcp->first == 0){ unlock(bcp); print("waiting for blocks\n");
1990/0322		if(loop++ > 10) panic("waiting for blocks\n");
1990/0312		qlock(bcp);
1990/03091		sleep(&bcp->r, isblock, (void *)bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/0312		qunlock(bcp);
1990/03091		lock(bcp); }
1990/0322/sys/src/9/gnot/stream.c:627,633 – 1990/03292/sys/src/9/gnot/stream.c:627,632 (short | long)
1990/03091		if(qi->open) (*qi->open)(RD(s->devq), s); c->flag |= COPEN; unlock(s); poperror(); return s;
1990/0322/sys/src/9/gnot/stream.c:681,687 – 1990/03292/sys/src/9/gnot/stream.c:680,686
1990/03091		/* * if not open, ignore it */ if(!(c->flag & COPEN))
1990/03292		if(!c->stream)
1990/03091		return; /*
1990/03292/sys/src/9/gnot/stream.c:58,67 – 1990/0331/sys/src/9/gnot/stream.c:58,89 (short | long)
1990/03091		{ 0 },
1990/0321		{ 68 }, { 260 },
1990/03091		{ 4096 },
1990/0331		{ 1024 },
1990/03091		}; /*
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/03091		* 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/gnot/stream.c:123,130 – 1990/0331/sys/src/9/gnot/stream.c:145,155
1990/03091		lock(bcp); while(bcp->first == 0){ unlock(bcp);
1990/0322		if(loop++ > 10)
1990/0331		if(loop++ > 10){ dumpqueues(); dumpstack();
1990/0322		panic("waiting for blocks\n");
1990/0331		}
1990/0312		qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/0312		qunlock(bcp);
1990/03292/sys/src/9/gnot/stream.c:614,619 – 1990/0331/sys/src/9/gnot/stream.c:639,645
1990/03091		* hang a device and process q off the stream */ s->inuse = 1;
1990/0331		s->hread = 0;
1990/03091		s->tag[0] = 0; q = allocq(&procinfo); s->procq = WR(q);
1990/03292/sys/src/9/gnot/stream.c:825,832 – 1990/0331/sys/src/9/gnot/stream.c:851,862
1990/03091		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/03091		sleep(&q->r, &isinput, (void *)q); continue; }
1990/0331/sys/src/9/gnot/stream.c:145,154 – 1990/0403/sys/src/9/gnot/stream.c:145,153 (short | long)
1990/03091		lock(bcp); while(bcp->first == 0){ unlock(bcp);
1990/0331		if(loop++ > 10){
1990/0403		if(loop++ == 10){
1990/0331		dumpqueues(); dumpstack();
1990/0322		panic("waiting for blocks\n");
1990/0403		print("waiting for blocks\n");
1990/0331		}
1990/0312		qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/0331/sys/src/9/gnot/stream.c:230,235 – 1990/0403/sys/src/9/gnot/stream.c:229,235
1990/03091		q->r.p = 0; q->info = qi; q->put = qi->iput;
1990/0403		q->len = q->nb = 0;
1990/03091		wq = q->other = q + 1; wq->r.p = 0;
1990/0331/sys/src/9/gnot/stream.c:236,241 – 1990/0403/sys/src/9/gnot/stream.c:236,242
1990/03091		wq->info = qi; wq->put = qi->oput; wq->other = q;
1990/0403		wq->len = wq->nb = 0;
1990/03091		unlock(q);
1990/0331/sys/src/9/gnot/stream.c:323,336 – 1990/0403/sys/src/9/gnot/stream.c:324,339
1990/03091		else q->first = bp;
1990/0312		q->len += BLEN(bp);
1990/0403		q->nb++;
1990/03091		delim = bp->flags & S_DELIM; while(bp->next) { bp = bp->next;
1990/0312		q->len += BLEN(bp);
1990/0403		q->nb++;
1990/03091		delim |= bp->flags & S_DELIM; } q->last = bp; if(q->len >= Streamhi)
1990/0403		if(q->len >= Streamhi || q->nb >= Streambhi)
1990/03091		q->flag |= QHIWAT; unlock(q); return delim;
1990/0331/sys/src/9/gnot/stream.c:387,393 – 1990/0403/sys/src/9/gnot/stream.c:390,397
1990/03091		if(q->first == 0) q->last = 0;
1990/0312		q->len -= BLEN(bp);
1990/03091		if((q->flag&QHIWAT) && q->len < Streamhi/2){
1990/0403		q->nb--; if((q->flag&QHIWAT) && q->len < Streamhi/2 && q->nb < Streambhi){
1990/03091		wakeup(&q->other->next->other->r); q->flag &= ~QHIWAT; }
1990/0331/sys/src/9/gnot/stream.c:762,775 – 1990/0403/sys/src/9/gnot/stream.c:766,781
1990/03091		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/03091		q->last = bp;
1990/0321		if(q->len >= Streamhi){
1990/0403		if(q->len >= Streamhi || q->nb >= Streambhi){
1990/03091		q->flag |= QHIWAT;
1990/0321		delim = 1; }
1990/0331/sys/src/9/gnot/stream.c:938,950 – 1990/0403/sys/src/9/gnot/stream.c:944,955
1990/03091		Queue *q; q = (Queue *)arg; return q->len < Streamhi;
1990/0403		return !QFULL(q->next);
1990/03091		} void flowctl(Queue *q) { if(q->next->len >= Streamhi) sleep(&q->r, notfull, q->next);
1990/0403		sleep(&q->r, notfull, q->next);
1990/03091		} /*
1990/0331/sys/src/9/gnot/stream.c:996,1002 – 1990/0403/sys/src/9/gnot/stream.c:1001,1007
1990/03091		* `a' is global to the whole system, just create a * pointer to it and pass it on. */ flowctl(q);
1990/0403		FLOWCTL(q);
1990/03091		bp = allocb(0); bp->rptr = bp->base = (uchar *)a; bp->wptr = bp->lim = (uchar *)a+n;
1990/0331/sys/src/9/gnot/stream.c:1009,1015 – 1990/0403/sys/src/9/gnot/stream.c:1014,1020
1990/03091		* system buffers and pass the buffers on. */ for(rem = n; ; rem -= i) { flowctl(q);
1990/0403		FLOWCTL(q);
1990/03091		bp = allocb(rem); i = bp->lim - bp->wptr; if(i >= rem){
1990/0403/sys/src/9/gnot/stream.c:372,377 – 1990/0406/sys/src/9/gnot/stream.c:372,378 (short | long)
1990/03091		q->last = bp; q->first = bp;
1990/0312		q->len += BLEN(bp);
1990/0406		q->nb++;
1990/03091		unlock(q); }
1990/0403/sys/src/9/gnot/stream.c:391,397 – 1990/0406/sys/src/9/gnot/stream.c:392,398
1990/03091		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/03091		wakeup(&q->other->next->other->r); q->flag &= ~QHIWAT; }
1990/0403/sys/src/9/gnot/stream.c:941,950 – 1990/0406/sys/src/9/gnot/stream.c:942,948
1990/03091		static int notfull(void *arg) { Queue *q;                  q = (Queue *)arg;
1990/0403		return !QFULL(q->next);
1990/0406		return !QFULL((Queue *)arg);
1990/03091		} void flowctl(Queue *q)
1990/0406/sys/src/9/gnot/stream.c:58,64 – 1990/05313/sys/src/9/gnot/stream.c:58,64 (short | long)
1990/03091		{ 0 },
1990/0321		{ 68 }, { 260 },
1990/0331		{ 1024 },
1990/05313		{ 4096 },
1990/03091		}; /*
1990/0406/sys/src/9/gnot/stream.c:70,86 – 1990/05313/sys/src/9/gnot/stream.c:70,98
1990/0331		Queue *q; int count; Block *bp;
1990/05313		Bclass *bcp;
1990/0331
1990/05313		print("\n");
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++; print("%s %ux RD count %d len %d", q->info->name, q, count, q->len);
1990/05313		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/05313		print(" W c %d l %d f %ux\n", count, WR(q)->len, WR(q)->flag);
1990/0331		}
1990/05313		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/0331		} /*
1990/0406/sys/src/9/gnot/stream.c:180,185 – 1990/05313/sys/src/9/gnot/stream.c:192,199
1990/03091		Bclass *bcp;
1990/0312		int tries;
1990/03091
1990/05313		if((bp->flags&S_CLASS) >= Nclass) panic("freeb class");
1990/03091		bcp = &bclass[bp->flags & S_CLASS]; lock(bcp);
1990/0312		bp->rptr = bp->wptr = 0;
1990/05313/sys/src/9/gnot/stream.c:258,263 – 1990/0620/sys/src/9/gnot/stream.c:258,279 (short | long)
1990/03091		} /*
1990/0620		* 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/03091		* free a queue */ static void
1990/05313/sys/src/9/gnot/stream.c:658,665 – 1990/0620/sys/src/9/gnot/stream.c:674,681
1990/03091		* hang a device and process q off the stream */ s->inuse = 1;
1990/0620		s->opens = 1;
1990/0331		s->hread = 0;
1990/03091		s->tag[0] = 0; q = allocq(&procinfo); s->procq = WR(q); q = allocq(qi);
1990/05313/sys/src/9/gnot/stream.c:697,702 – 1990/0620/sys/src/9/gnot/stream.c:713,719
1990/03091		&& s->dev == c->dev && s->id == STREAMID(c->qid)){ s->inuse++;
1990/0620		s->opens++;
1990/03091		c->stream = s; unlock(s); return;
1990/05313/sys/src/9/gnot/stream.c:712,717 – 1990/0620/sys/src/9/gnot/stream.c:729,782
1990/03091		} /*
1990/0620		* 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); s->inuse--; if(s->inuse != 0){ if(!locked) unlock(s); return; } /* * 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; if(!locked) unlock(s); } /*
1990/03091		* On the last close of a stream, for each queue on the * stream release its blocks and call its close routine. */
1990/05313/sys/src/9/gnot/stream.c:729,739 – 1990/0620/sys/src/9/gnot/stream.c:794,804
1990/03091		return; /* * decrement the reference cound
1990/0620		* decrement the reference count
1990/03091		*/ lock(s); if(s->inuse != 1){ s->inuse--;
1990/0620		if(s->opens != 1){ s->opens--;
1990/03091		unlock(c->stream); return; }
1990/05313/sys/src/9/gnot/stream.c:747,761 – 1990/0620/sys/src/9/gnot/stream.c:812,830
1990/03091		if(q == s->devq->other) break; }
1990/0620
1990/03091		/* * ascend the stream freeing the queues
1990/0620		* ascend the stream flushing the queues
1990/03091		*/ for(q = s->devq; q; q = nq){ nq = q->next; freeq(q);
1990/0620		flushq(q);
1990/03091		} s->id = s->dev = s->type = 0; s->inuse--;
1990/0620		/* * leave it and free it */ streamexit(s, 1);
1990/03091		unlock(s); }
1990/0620/sys/src/9/gnot/stream.c:531,538 – 1990/0621/sys/src/9/gnot/stream.c:531,542 (short | long)
1990/03091		void nullput(Queue *q, Block *bp) { freeb(bp); error(0, Ehungup);
1990/0621		if(bp->type == M_HANGUP) freeb(bp); else { freeb(bp); error(0, Ehungup); }
1990/03091		} /*
1990/0620/sys/src/9/gnot/stream.c:757,777 – 1990/0621/sys/src/9/gnot/stream.c:761,777
1990/0620		if(!locked) lock(s); s->inuse--; if(s->inuse != 0){ if(!locked) unlock(s); return;
1990/0621		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;
1990/0620		}                  /* * 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;
1990/0621		s->inuse--;
1990/0620		if(!locked) unlock(s); }
1990/0620/sys/src/9/gnot/stream.c:797,825 – 1990/0621/sys/src/9/gnot/stream.c:797,823
1990/0620		* decrement the reference count
1990/03091		*/ lock(s);
1990/0620		if(s->opens != 1){ s->opens--;
1990/03091		unlock(c->stream); return;
1990/0621		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/03091		}                  /* * descend the stream closing the queues */ for(q = s->procq; q; q = q->next){ if(q->info->close) (*q->info->close)(q->other); if(q == s->devq->other) break; }
1990/0620
1990/03091		/*
1990/0620		* ascend the stream flushing the queues
1990/03091		*/ for(q = s->devq; q; q = nq){ nq = q->next;
1990/0620		flushq(q);
1990/03091		}
1990/0621		s->opens--;
1990/0620		/* * leave it and free it
1990/0621/sys/src/9/gnot/stream.c:78,88 – 1990/0702/sys/src/9/gnot/stream.c:78,89 (short | long)
1990/0331		continue; for(count = 0, bp = q->first; bp; bp = bp->next) count++;
1990/05313		print("%s %ux R c %d l %d f %ux", q->info->name, q, count, q->len, q->flag);
1990/0702		print("%s %ux R c %d l %d f %ux r %ux", q->info->name, q, count, q->len, q->flag, &(q->r));
1990/0331		for(count = 0, bp = WR(q)->first; bp; bp = bp->next) count++;
1990/05313		print(" W c %d l %d f %ux\n", count, WR(q)->len, WR(q)->flag);
1990/0702		print(" W c %d l %d f %ux r %ux\n", count, WR(q)->len, WR(q)->flag, &(WR(q)->r));
1990/0331		}
1990/05313		print("\n"); for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
1990/0621/sys/src/9/gnot/stream.c:246,251 – 1990/0702/sys/src/9/gnot/stream.c:247,253
1990/0403		q->len = q->nb = 0;
1990/03091		wq = q->other = q + 1;
1990/0702		wq->flag = QINUSE;
1990/03091		wq->r.p = 0; wq->info = qi; wq->put = qi->oput;
1990/0702/sys/src/9/gnot/stream.c:65,70 – 1990/0703/sys/src/9/gnot/stream.c:65,86 (short | long)
1990/0331		* Dump all block information of how many blocks are in which queues */ void
1990/0703		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
1990/0331		dumpqueues(void) { Queue *q;
1990/0702/sys/src/9/gnot/stream.c:76,89 – 1990/0703/sys/src/9/gnot/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/0702		print("%s %ux R c %d l %d f %ux r %ux", q->info->name, q, count,
1990/0703		print("%s %ux R n %d l %d f %ux r %ux", q->info->name, q, q->nb,
1990/0702		q->len, q->flag, &(q->r));
1990/0331		for(count = 0, bp = WR(q)->first; bp; bp = bp->next) count++;
1990/0702		print(" W c %d l %d f %ux r %ux\n", count, WR(q)->len, WR(q)->flag,
1990/0703		print(" W n %d l %d f %ux r %ux\n", WR(q)->nb, WR(q)->len, WR(q)->flag,
1990/0702		&(WR(q)->r));
1990/0703		dumpblocks(q, 'R'); dumpblocks(q, 'W');
1990/0331		}
1990/05313		print("\n"); for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
1990/0703/sys/src/9/gnot/stream.c:97,103 – 1990/0704/sys/src/9/gnot/stream.c:97,103 (short | long)
1990/0703		print(" W n %d l %d f %ux r %ux\n", WR(q)->nb, WR(q)->len, WR(q)->flag,
1990/0702		&(WR(q)->r));
1990/0703		dumpblocks(q, 'R'); dumpblocks(q, 'W');
1990/0704		dumpblocks(WR(q), 'W');
1990/0331		}
1990/05313		print("\n"); for(bcp=bclass; bcp<&bclass[Nclass]; bcp++){
1990/0704/sys/src/9/gnot/stream.c:6,11 – 1990/0801/sys/src/9/gnot/stream.c:6,12 (short | long)
1990/03091		#include "io.h" #include "errno.h" #include "devtab.h"
1990/0801		#include "fcall.h"
1990/03091
1990/0321		enum { Nclass=4, /* number of block classes */
1990/0704/sys/src/9/gnot/stream.c:1155,1158 – 1990/0801/sys/src/9/gnot/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/03091		}
1990/0801/sys/src/9/gnot/stream.c:986,991 – 1990/08272/sys/src/9/gnot/stream.c:986,995 (short | long)
1990/03091		qunlock(&s->rdlock); poperror();
1990/08272		if(left<0 || left>n){ print("streamread returns %d for a %d read\n", n-left, n); panic("streamread"); }
1990/03091		return n - left; }
1990/08272/sys/src/9/gnot/stream.c:232,237 – 1990/0905/sys/src/9/gnot/stream.c:232,257 (short | long)
1990/03091		} /*
1990/0905		* 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 = nbp; return nbp; } } /*
1990/03091		* 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/0905/sys/src/9/gnot/stream.c:246,252 – 1990/09051/sys/src/9/gnot/stream.c:246,252 (short | long)
1990/0905		nbp = allocb(n); nbp->wptr = nbp->lim; nbp->rptr = nbp->wptr - n; nbp->next = nbp;
1990/09051		nbp->next = bp;
1990/0905		return nbp; } }
1990/09051/sys/src/9/gnot/stream.c:10,16 – 1990/0911/sys/src/9/gnot/stream.c:10,15 (short | long)
1990/03091
1990/0321		enum { Nclass=4, /* number of block classes */ Nlds=32, /* max number of pushable line disciplines */ };
1990/0312		/*
1990/09051/sys/src/9/gnot/stream.c:22,42 – 1990/0911/sys/src/9/gnot/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/03091
1990/0321		void newqinfo(Qinfo *qi) { int i;
1990/03091
1990/0321		for(i=0; i<Nlds && lds[i]; i++) if(lds[i] == qi) return; if(i == Nlds) panic("pushable"); lds[i] = qi; }
1990/03091		/* * All stream structures are ialloc'd at boot time */
1990/09051/sys/src/9/gnot/stream.c:62,116 – 1990/0911/sys/src/9/gnot/stream.c:48,55
1990/05313		{ 4096 },
1990/03091		}; /*
1990/0331		* Dump all block information of how many blocks are in which queues */ void
1990/0703		dumpblocks(Queue *q, char c) { Block *bp; uchar *cp;
1990/0911		#include "stream.h"
1990/0703		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
1990/0331		dumpqueues(void) { Queue *q; int count; Block *bp;
1990/05313		Bclass *bcp;
1990/0331
1990/05313		print("\n");
1990/0331		for(q = qlist; q < qlist + conf.nqueue; q++, q++){ if(!(q->flag & QINUSE)) continue;
1990/0703		print("%s %ux R n %d l %d f %ux r %ux", q->info->name, q, q->nb,
1990/0702		q->len, q->flag, &(q->r));
1990/0703		print(" W n %d l %d f %ux r %ux\n", WR(q)->nb, WR(q)->len, WR(q)->flag,
1990/0702		&(WR(q)->r));
1990/0703		dumpblocks(q, 'R');
1990/0704		dumpblocks(WR(q), 'W');
1990/0331		}
1990/05313		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/0331		}                  /*
1990/03091		* Allocate streams, queues, and blocks. Allocate n block classes with * 1/2(m+1) to class m < n-1
1990/09051/sys/src/9/gnot/stream.c:123,128 – 1990/0911/sys/src/9/gnot/stream.c:62,70
1990/03091		Block *bp; Bclass *bcp;
1990/0911		/* * allocate blocks, queues, and streams */
1990/03091		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/09051/sys/src/9/gnot/stream.c:141,149 – 1990/0911/sys/src/9/gnot/stream.c:83,109
1990/03091		bp++; } }
1990/0911		/* * make stream modules available */ streaminit0();
1990/03091		} /*
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/03091		* allocate a block */ static int
1990/09051/sys/src/9/gnot/stream.c:159,165 – 1990/0911/sys/src/9/gnot/stream.c:119,124
1990/03091		{ Block *bp; Bclass *bcp;
1990/0322		int loop=0;
1990/03091		/* * map size to class
1990/09051/sys/src/9/gnot/stream.c:173,182 – 1990/0911/sys/src/9/gnot/stream.c:132,137
1990/03091		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/0312		qlock(bcp);
1990/0322		tsleep(&bcp->r, isblock, (void *)bcp, 250);
1990/0312		qunlock(bcp);
1990/09051/sys/src/9/gnot/stream.c:582,594 – 1990/0911/sys/src/9/gnot/stream.c:537,549
1990/03091		static Qinfo * qinfofind(char *name) { Qinfo **qip;
1990/0911		Qinfo *qi;
1990/03091		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/03091		error(0, Ebadld); }
1990/09051/sys/src/9/gnot/stream.c:1006,1015 – 1990/0911/sys/src/9/gnot/stream.c:961,966
1990/03091		qunlock(&s->rdlock); poperror();
1990/08272		if(left<0 || left>n){ print("streamread returns %d for a %d read\n", n-left, n); panic("streamread"); }
1990/03091		return n - left; }
1990/09051/sys/src/9/gnot/stream.c:1210,1213 – 1990/0911/sys/src/9/gnot/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/03091		}
1990/0911/sys/src/9/gnot/stream.c:1035,1049 – 1990/0914/sys/src/9/gnot/stream.c:1035,1050 (short | long)
1990/03091		long rem; int i;
1990/0914		s = c->stream;
1990/03091		/* * one writer at a time */ s = c->stream; qlock(&s->wrlock); if(waserror()){ qunlock(&s->wrlock); nexterror(); }
1990/0914		*/
1990/03091		/* * decode the qid
1990/0911/sys/src/9/gnot/stream.c:1053,1061 – 1990/0914/sys/src/9/gnot/stream.c:1054,1060
1990/03091		break; case Sctlqid: n = streamctlwrite(s, a, n); qunlock(&s->wrlock); poperror(); return n;
1990/0914		goto out;
1990/03091		default: panic("bad stream qid\n"); }
1990/0911/sys/src/9/gnot/stream.c:1104,1111 – 1990/0914/sys/src/9/gnot/stream.c:1103,1111
1990/03091		} } } qunlock(&s->wrlock); poperror();
1990/0914		out: /* qunlock(&s->wrlock); poperror(); /**/
1990/03091		return n;
1990/0312		}
1990/0911/sys/src/9/gnot/stream.c:1148,1163 – 1990/0914/sys/src/9/gnot/stream.c:1148,1164
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/gnot/stream.c:127,133 – 1990/0925/sys/src/9/gnot/stream.c:127,133 (short | long)
1990/03091		; /* * look for a free block, garbage collect if there are none
1990/0925		* look for a free block
1990/03091		*/ lock(bcp); while(bcp->first == 0){
1990/0914/sys/src/9/gnot/stream.c:1038,1052 – 1990/0925/sys/src/9/gnot/stream.c:1038,1043
1990/0914		s = c->stream;
1990/03091		/* * one writer at a time qlock(&s->wrlock); if(waserror()){ qunlock(&s->wrlock); nexterror(); }
1990/0914		*/
1990/03091		/* * decode the qid */ switch(STREAMTYPE(c->qid)){
1990/0914/sys/src/9/gnot/stream.c:1104,1111 – 1990/0925/sys/src/9/gnot/stream.c:1095,1100
1990/03091		} }
1990/0914		out: /* qunlock(&s->wrlock); poperror(); /**/
1990/03091		return n;
1990/0312		}
1990/0925/sys/src/9/gnot/stream.c:878,883 – 1990/0928/sys/src/9/gnot/stream.c:878,905 (short | long)
1990/03091		} /*
1990/0928		* 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/03091		* return true if there is an output buffer available */ static int
1990/0925/sys/src/9/gnot/stream.c:895,922 – 1990/0928/sys/src/9/gnot/stream.c:917,932
1990/03091		Block *bp; Stream *s; Queue *q; long rv = 0; int left, i, x;
1990/0928		int left, i;
1990/03091		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/0928		if(STREAMTYPE(c->qid) != Sdataqid) return streamctlread(c, vbuf, n);
1990/03091		/* * one reader at a time */
1990/0928		s = c->stream;
1990/03091		qlock(&s->rdlock); if(waserror()){ qunlock(&s->rdlock);
1990/0925/sys/src/9/gnot/stream.c:974,984 – 1990/0928/sys/src/9/gnot/stream.c:984,999
1990/03091		* This routing is entrered with s->wrlock'ed and must unlock. */ static long streamctlwrite(Stream *s, void *a, long n)
1990/0928		streamctlwrite(Chan *c, void *a, long n)
1990/03091		{ Qinfo *qi; Block *bp;
1990/0928		Stream *s;
1990/03091
1990/0928		if(STREAMTYPE(c->qid) != Sctlqid) panic("streamctlwrite %lux", c->qid); s = c->stream;
1990/03091		/* * package */
1990/0925/sys/src/9/gnot/stream.c:1040,1054 – 1990/0928/sys/src/9/gnot/stream.c:1055,1062
1990/03091		/* * decode the qid */ switch(STREAMTYPE(c->qid)){ case Sdataqid: break; case Sctlqid: n = streamctlwrite(s, a, n);
1990/0914		goto out;
1990/03091		default: panic("bad stream qid\n"); }
1990/0928		if(STREAMTYPE(c->qid) != Sdataqid) return streamctlwrite(c, a, n);
1990/03091		/* * No writes allowed on hungup channels
1990/0925/sys/src/9/gnot/stream.c:1057,1063 – 1990/0928/sys/src/9/gnot/stream.c:1065,1071
1990/03091		if(q->other->flag & QHUNGUP) error(0, Ehungup);
1990/0312		if((GLOBAL(a) && !docopy) || n==0){
1990/0928		if(!docopy && GLOBAL(a)){
1990/03091		/* * `a' is global to the whole system, just create a * pointer to it and pass it on.
1990/0925/sys/src/9/gnot/stream.c:1094,1100 – 1990/0928/sys/src/9/gnot/stream.c:1102,1107
1990/03091		} } }
1990/0914		out:
1990/03091		return n;
1990/0312		}
1990/0928/sys/src/9/gnot/stream.c:905,911 – 1990/0930/sys/src/9/gnot/stream.c:905,914 (short | long)
1990/03091		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/03091		} /*
1990/0930/sys/src/9/gnot/stream.c:235,240 – 1990/1009/sys/src/9/gnot/stream.c:235,241 (short | long)
1990/03091		q->info = qi; q->put = qi->iput;
1990/0403		q->len = q->nb = 0;
1990/1009		q->ptr = 0;
1990/03091		wq = q->other = q + 1;
1990/0702		wq->flag = QINUSE;
1990/0930/sys/src/9/gnot/stream.c:242,247 – 1990/1009/sys/src/9/gnot/stream.c:243,249
1990/03091		wq->info = qi; wq->put = qi->oput; wq->other = q;
1990/1009		wq->ptr = 0;
1990/0403		wq->len = wq->nb = 0;
1990/03091		unlock(q);
1990/0930/sys/src/9/gnot/stream.c:625,634 – 1990/1009/sys/src/9/gnot/stream.c:627,636
1990/03091		} /* * create a new stream
1990/1009		* create a new stream, if noopen is non-zero, don't increment the open count
1990/03091		*/ Stream * streamnew(Chan *c, Qinfo *qi)
1990/1009		streamnew(ushort type, ushort dev, ushort id, Qinfo *qi, int noopen)
1990/03091		{ Stream *s; Queue *q;
1990/0930/sys/src/9/gnot/stream.c:651,676 – 1990/1009/sys/src/9/gnot/stream.c:653,677
1990/03091		} if(waserror()){ unlock(s); streamclose(c);
1990/1009		streamclose1(s);
1990/03091		nexterror(); } /* * marry a stream and a channel
1990/1009		* identify the stream
1990/03091		*/ 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/03091		/* * hang a device and process q off the stream */ s->inuse = 1;
1990/0620		s->opens = 1;
1990/1009		if(noopen) s->opens = 0; else s->opens = 1;
1990/0331		s->hread = 0;
1990/03091		q = allocq(&procinfo); s->procq = WR(q);
1990/0930/sys/src/9/gnot/stream.c:699,705 – 1990/1009/sys/src/9/gnot/stream.c:700,706
1990/03091		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/03091		*/ for(s = slist; s < &slist[conf.nstream]; s++) { if(s->inuse && s->type == c->type && s->dev == c->dev
1990/0930/sys/src/9/gnot/stream.c:721,727 – 1990/1009/sys/src/9/gnot/stream.c:722,728
1990/03091		/* * create a new stream */ streamnew(c, qi);
1990/1009		c->stream = streamnew(c->type, c->dev, STREAMID(c->qid), qi, 0);
1990/03091		} /*
1990/0930/sys/src/9/gnot/stream.c:773,791 – 1990/1009/sys/src/9/gnot/stream.c:774,785
1990/03091		* stream release its blocks and call its close routine. */ void streamclose(Chan *c)
1990/1009		streamclose1(Stream *s)
1990/03091		{ Queue *q, *nq; Block *bp; Stream *s = c->stream; /* * if not open, ignore it */
1990/03292		if(!c->stream)
1990/03091		return;                  /*
1990/0620		* decrement the reference count
1990/03091		*/ lock(s);
1990/0930/sys/src/9/gnot/stream.c:817,822 – 1990/1009/sys/src/9/gnot/stream.c:811,826
1990/0620		streamexit(s, 1);
1990/03091		unlock(s); }
1990/1009		void streamclose(Chan *c) { /* * if no stream, ignore it */ if(!c->stream) return; streamclose1(c->stream); }
1990/03091		/* * put a block to be read into the queue. wakeup any waiting reader
1990/0930/sys/src/9/gnot/stream.c:1174,1180 – 1990/1009/sys/src/9/gnot/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/gnot/stream.c:784,798 – 1990/1011/sys/src/9/gnot/stream.c:784,803 (short | long)
1990/03091		*/ lock(s);
1990/0621		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/0621		} /*
1990/1011/sys/src/9/gnot/stream.c:1047,1053 – 1990/1018/sys/src/9/gnot/stream.c:1047,1055 (short | long)
1990/03091		void flowctl(Queue *q) {
1990/1018		qlock(&q->rlock);
1990/0403		sleep(&q->r, notfull, q->next);
1990/1018		qunlock(&q->rlock);
1990/03091		} /*
Too many diffs (26 > 25). Stopping.