1990/0331/port/sturp.c (diff list | history)

1990/0227/sys/src/9/port/sturp.c:1,23 – 1990/0312/sys/src/9/port/sturp.c:1,23 (short | long)
1990/0227		#include "syslibc.h" #include "lock.h" #include "chan.h" #include "proc.h" #include "user.h" #include "errno.h" #include "lint.h" #include "mem.h" #include "mempool.h" #include "stream.h" #include "dkparam.h" #include "misc.h"
1990/0312		#include "u.h" #include "lib.h" #include "mem.h" #include "dat.h" #include "fns.h" #include "io.h" #include "errno.h"
1990/0227		enum { Nurp= 32, MSrexmit= 1000,
1990/0312		Nmask= 0x7,
1990/0227		};
1990/0312		#define DPRINT if(0)
1990/0227		typedef struct Urp Urp;
1990/0312		#define NOW (MACHP(0)->ticks*MS2HZ)
1990/0227		/* * URP status */
1990/0227/sys/src/9/port/sturp.c:34,45 – 1990/0312/sys/src/9/port/sturp.c:34,46
1990/0227		} urpstat; struct Urp { Qlock;
1990/0312		QLock;
1990/0227		short state; /* flags */
1990/0312		Rendez r; /* process waiting for close */
1990/0227		/* input */ ulong iwindow; /* input window */
1990/0312		Queue *rq; /* input queue */
1990/0227		uchar iseq; /* last good input sequence number */ uchar lastecho; /* last echo/rej sent */ uchar trbuf[3]; /* trailer being collected */
1990/0227/sys/src/9/port/sturp.c:47,68 – 1990/0312/sys/src/9/port/sturp.c:48,70
1990/0227		/* output */ Queue *rq; Queue *wq; int XW; /* blocks per xmit window */
1990/0312		QLock xmit; /* output lock, only one process at a time */ Queue *wq; /* output queue */ int maxout; /* maximum outstanding unacked blocks */
1990/0227		int maxblock; /* max block size */ int timeout; /* a timeout has occurred */ int WS; /* start of current window */ int WACK; /* first non-acknowledged message */ int WNX; /* next message to be sent */ Rendez r; /* process waiting in urpoput */ Rendez kr; /* process waiting in urpoput */
1990/0312		int next; /* next block to send */ int unechoed; /* first unechoed block */ int unacked; /* first unacked block */ int nxb; /* next xb to use */
1990/0227		Block *xb[8]; /* the xmit window buffer */ uchar timer; /* timeout for xmit */
1990/0312		QLock xl[8]; ulong timer; /* timeout for xmit */ int kstarted;
1990/0227		}; #define WINDOW(u) ((u->WS + u->XW - u->WNX)%8) #define BETWEEN(x, s, n) (s<=n ? x>=s && x<n : x<n || x>=s) #define UNACKED(x, u) (BETWEEN(x, u->WACK, u->WNX)
1990/0312		#define WINDOW(u) ((u->unechoed + u->maxout - u->next)%8) #define IN(x, f, n) (f<=n ? x>=f && x<n : x<n || x>=f) #define NEXT(x) (((x)+1)&Nmask)
1990/0227		/* * Protocol control bytes
1990/0227/sys/src/9/port/sturp.c:87,95 – 1990/0312/sys/src/9/port/sturp.c:89,99
1990/0227		#define DELAY 0100 /* real-time printing delay */ #define BREAK 0110 /* Send/receive break (new style) */ #define REXMITING 0001 #define REXMIT 0002 #define INITING 0004
1990/0312		#define REJECTING 0x1 #define INITING 0x2 #define HUNGUP 0x4 #define OPEN 0x8 #define CLOSING 0x10
1990/0227		Urp urp[Nurp];
1990/0227/sys/src/9/port/sturp.c:96,119 – 1990/0312/sys/src/9/port/sturp.c:100,127
1990/0227		/* * predeclared */ static void urpiput(Queue*, Block*, Rendez*); static void urpoput(Queue*, Block*, Rendez*);
1990/0312		static void urpciput(Queue*, Block*); static void urpiput(Queue*, Block*); static void urpoput(Queue*, Block*);
1990/0227		static void urpopen(Queue*, Stream*); static void urpclose(Queue *);
1990/0312		static void output(Urp*); static void sendblock(Urp*, int);
1990/0227		static void rcvack(Urp*, int); static void flushinput(Urp*); static void sendctl(Queue*, int); static void queuectl(Urp*, int);
1990/0312		static void sendctl(Urp*, int);
1990/0227		static void initoutput(Urp*, int); static void initinput(Urp*, int);
1990/0312		static void urpkproc(void *arg);
1990/0227		Qinfo urpinfo = { urpiput, urpoput, urpopen, urpclose, "urp" };
1990/0312		Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
1990/0227		int
1990/0312		static void
1990/0227		urpopen(Queue *q, Stream *s) { Urp *up; int i;
1990/0312		char name[128];
1990/0227		/* * find a free urp structure
1990/0227/sys/src/9/port/sturp.c:127,152 – 1990/0312/sys/src/9/port/sturp.c:135,178
1990/0227		if(up == &urp[Nurp]) error(0, Egreg); q->ptr = = q->other->ptr = up;
1990/0312		q->ptr = q->other->ptr = up;
1990/0227		up->rq = q; up->wq = WR(q); q->put = urpciput;
1990/0312		up->wq = q->other; up->state = OPEN;
1990/0227		qunlock(up); initinput(up, 0); initoutput(up, 0);
1990/0312		/* * start the ack/(re)xmit process */ if(up->kstarted == 0){ up->kstarted = 1; sprint(name, "**urp%d**", up - urp); kproc(name, urpkproc, up); }
1990/0227		} /* * Shut it down.
1990/0312		* Shut down the connection and kill off the kernel process
1990/0227		*/ static int allacked(void *a)
1990/0312		isflushed(void *a)
1990/0227		{ Urp *up; up = (Urp *)a; return up->WACK == up->WNX;
1990/0312		return (up->state&HUNGUP) || (up->unechoed==up->next && up->wq->len==0);
1990/0227		}
1990/0312		static int isdead(void *a) { Urp *up; up = (Urp *)a; return up->kstarted == 0; } static void
1990/0227		urpclose(Queue *q) { Block *bp;
1990/0227/sys/src/9/port/sturp.c:154,167 – 1990/0312/sys/src/9/port/sturp.c:180,210
1990/0227		int i; up = (Urp *)q->ptr; up->state |= LCLOSE; /* * wait for output to get acked
1990/0312		* wait for all outstanding messages to drain, tell kernel * process we're closing.
1990/0227		*/ while(!urpdone(up)) sleep(&up->r, urpdone, up); up->state = 0;
1990/0312		up->state |= CLOSING; sleep(&up->r, isflushed, up); /* * ack all outstanding messages */ qlock(&up->xmit); up->state |= HUNGUP; i = up->next - 1; if(i < 0) i = 7; rcvack(up, ECHO+i); qunlock(&up->xmit); DPRINT("urpclose(%ux)\n", up); /* * kill off the kernel process */ wakeup(&up->rq->r); DPRINT("urpclosed(%ux)\n", up);
1990/0227		} /*
1990/0227/sys/src/9/port/sturp.c:172,200 – 1990/0312/sys/src/9/port/sturp.c:215,239
1990/0227		{ switch(bp->type){ case M_HANGUP: /* * ack all outstanding messages */ lock(up); up->state &= ~(INITING); up->state |= RCLOSE; unlock(up); if(up->WS<up->WNX) urprack(up, ECHO+((up->WNX-1)&07));
1990/0312		up->state |= HUNGUP; wakeup(&up->r); wakeup(&up->rq->r); break;
1990/0227		} PUTNEXT(q, bp); } /* * character mode input. the last character in EVERY block is * a control character.
1990/0312		* character mode input. * * the first byte in every message is a ctl byte (which belongs at the end).
1990/0227		*/ void urpciput(Queue *q, Block *bp, Rendez *rp)
1990/0312		urpciput(Queue *q, Block *bp)
1990/0227		{ Urp *up; int i, full; Block *nbp;
1990/0312		int i; int ctl;
1990/0227		up = (Urp *)q->ptr; if(bp->type != M_DATA){
1990/0227/sys/src/9/port/sturp.c:205,219 – 1990/0312/sys/src/9/port/sturp.c:244,257
1990/0227		/* * get the control character */ if(bp->wptr > bp->rptr) ctl = *(--bp->wptr); else ctl = 0;
1990/0312		ctl = *bp->rptr++; if(ctl < 0) return;
1990/0227		/* * send the block upstream
1990/0312		* take care of any data
1990/0227		*/ if(bp->wptr > bp->rptr)
1990/0312		if(BLEN(bp)>0 && q->next->len<Streamhi)
1990/0227		PUTNEXT(q, bp); else freeb(bp);
1990/0227/sys/src/9/port/sturp.c:224,252 – 1990/0312/sys/src/9/port/sturp.c:262,288
1990/0227		switch(ctl){ case 0: break;                  case ENQ: urpstat.enqsr++; queuectl(up, up->lastecho); queuectl(up, ACK+up->iseq); flushinput(up);
1990/0312		sendctl(up, up->lastecho); sendctl(up, ACK+up->iseq);
1990/0227		break; case CHECK: queuectl(up, ACK+up->iseq);
1990/0312		sendctl(up, ACK+up->iseq);
1990/0227		break; case AINIT: up->state &= ~INITING; flushinput(up); wakeup(&cp->kr);
1990/0312		wakeup(&up->rq->r);
1990/0227		break; case INIT0: case INIT1: queuectl(up, AINIT); if(*bp->rptr == INIT1) q->put = urpbiput;
1990/0312		sendctl(up, AINIT); if(ctl == INIT1) q->put = urpiput;
1990/0227		initinput(up, 0); break;
1990/0227/sys/src/9/port/sturp.c:257,262 – 1990/0312/sys/src/9/port/sturp.c:293,303
1990/0227		case BREAK: break;
1990/0312		case REJ+0: case REJ+1: case REJ+2: case REJ+3: case REJ+4: case REJ+5: case REJ+6: case REJ+7: rcvack(up, ctl); break;
1990/0227		case ACK+0: case ACK+1: case ACK+2: case ACK+3: case ACK+4: case ACK+5: case ACK+6: case ACK+7: case ECHO+0: case ECHO+1: case ECHO+2: case ECHO+3:
1990/0227/sys/src/9/port/sturp.c:266,294 – 1990/0312/sys/src/9/port/sturp.c:307,335
1990/0227		case SEQ+0: case SEQ+1: case SEQ+2: case SEQ+3: case SEQ+4: case SEQ+5: case SEQ+6: case SEQ+7: /* * acknowledge receipt */ ctl = ctl & 07; if(q->next->len < Streamhi){ queuectl(up, ECHO+ctl); up->lastecho = ECHO+ctl; wakeup(&cp->kr); } up->iseq = ctl;
1990/0312		i = ctl & Nmask; if(q->next->len < Streamhi) sendctl(up, up->lastecho = ECHO+i); up->iseq = i;
1990/0227		break; } } /* * block mode input. the last character in EVERY block is a control character.
1990/0312		* block mode input. * * the first byte in every message is a ctl byte (which belongs at the end). * * Simplifying assumption: one put == one message && the control byte * is in the first block. If this isn't true, strange bytes will be * used as control bytes.
1990/0227		*/ void urpbiput(Queue *q, Block *bp, Rendez *rp)
1990/0312		urpiput(Queue *q, Block *bp)
1990/0227		{ Urp *up; int i, full; Block *nbp;
1990/0312		int i; int ctl;
1990/0227		up = (Urp *)q->ptr; if(bp->type != M_DATA){
1990/0227/sys/src/9/port/sturp.c:299,313 – 1990/0312/sys/src/9/port/sturp.c:340,353
1990/0227		/* * get the control character */ if(bp->wptr > bp->rptr) ctl = *(--bp->wptr); else ctl = 0;
1990/0312		ctl = *bp->rptr++;
1990/0227		/* * take care of any block count(trx) */ while(bp->wptr > bp->rptr && up->trx){
1990/0312		while(up->trx){ if(BLEN(bp)<=0) break;
1990/0227		switch (up->trx) { case 1: case 2:
1990/0227/sys/src/9/port/sturp.c:315,335 – 1990/0312/sys/src/9/port/sturp.c:355,374
1990/0227		continue; default: up->trx = 0; case 0: break; } } /* * queue the block
1990/0312		* queue the block(s)
1990/0227		*/ if(bp->wptr > bp->rptr){ if(q->len > up->iwindow){
1990/0312		if(BLEN(bp) > 0){ putq(q, bp); q->last->flags &= ~S_DELIM; if(q->len > 4*1024){
1990/0227		flushinput(up); freeb(bp); return; } putq(q, bp); } else freeb(bp);
1990/0227/sys/src/9/port/sturp.c:340,365 – 1990/0312/sys/src/9/port/sturp.c:379,405
1990/0227		case 0: break; case ENQ:
1990/0312		print("rENQ\n");
1990/0227		urpstat.enqsr++; queuectl(up, up->lastecho); queuectl(up, ACK+up->iseq);
1990/0312		sendctl(up, up->lastecho); sendctl(up, ACK+up->iseq);
1990/0227		flushinput(up); break; case CHECK: queuectl(WR(q)->next, ACK+up->iseq);
1990/0312		sendctl(up, ACK+up->iseq);
1990/0227		break; case AINIT: up->state &= ~INITING; flushinput(up); wakeup(&cp->kr);
1990/0312		wakeup(&up->rq->r);
1990/0227		break; case INIT0: case INIT1: queuectl(up, AINIT); if(*bp->rptr == INIT0)
1990/0312		sendctl(up, AINIT); if(ctl == INIT0)
1990/0227		q->put = urpciput; initinput(up, 0); break;
1990/0227/sys/src/9/port/sturp.c:380,385 – 1990/0312/sys/src/9/port/sturp.c:420,426
1990/0227		case REJ+0: case REJ+1: case REJ+2: case REJ+3: case REJ+4: case REJ+5: case REJ+6: case REJ+7:
1990/0312		print("rREJ\n");
1990/0227		rcvack(up, ctl); break;
1990/0227/sys/src/9/port/sturp.c:391,412 – 1990/0312/sys/src/9/port/sturp.c:432,462
1990/0227		break; /* * this case is extremely ugliferous
1990/0312		* if the seuence number is the next expected * and te trailer length == 3 * and the block count matches the bytes received * then send the bytes upstream.
1990/0227		*/ case SEQ+0: case SEQ+1: case SEQ+2: case SEQ+3: case SEQ+4: case SEQ+5: case SEQ+6: case SEQ+7: i = ctl & 07;
1990/0312		i = ctl & Nmask;
1990/0227		if(up->trx != 3){ urpstat.rjtrs++; flushinput(up);
1990/0312		print("sREJ\n"); sendctl(up, up->lastecho = REJ+up->iseq);
1990/0227		break; } else if(q->len != up->trbuf[1] + (up->trbuf[2]<<8)){ urpstat.rjpks++; flushinput(up);
1990/0312		print("sREJ\n"); sendctl(up, up->lastecho = REJ+up->iseq);
1990/0227		break; } else if(i != ((up->iseq+1)&07))) {
1990/0312		} else if(i != ((up->iseq+1)&Nmask)) {
1990/0227		urpstat.rjseq++; flushinput(up);
1990/0312		print("sREJ\n"); sendctl(up, up->lastecho = REJ+up->iseq);
1990/0227		break; }
1990/0227/sys/src/9/port/sturp.c:414,439 – 1990/0312/sys/src/9/port/sturp.c:464,489
1990/0227		* send data upstream */ if(q->first) { q->first->flags |= S_DELIM;
1990/0312		if(up->trbuf[0] != BOTM) q->last->flags |= S_DELIM;
1990/0227		while(bp = getq(q)) PUTNEXT(q, nbp);
1990/0312		PUTNEXT(q, bp);
1990/0227		} else { bp = allocb(0); bp->flag |= S_DELIM;
1990/0312		bp->flags |= S_DELIM;
1990/0227		PUTNEXT(q, bp); {
1990/0312		}
1990/0227		up->trx = 0; /* * acknowledge receipt */ ctl = ctl & 07; if(q->next->len < Streamhi){ queuectl(up, ECHO+ctl); up->lastecho = ECHO+ctl; wakeup(&cp->kr);
1990/0312		sendctl(up, ECHO+i); up->lastecho = ECHO+i; wakeup(&up->rq->r);
1990/0227		} up->iseq = ctl;
1990/0312		up->iseq = i;
1990/0227		break; } }
1990/0227/sys/src/9/port/sturp.c:444,450 – 1990/0312/sys/src/9/port/sturp.c:494,500
1990/0227		static void urpctloput(Urp *up, Queue *q, Block *bp) { int fields[2];
1990/0312		char *fields[2];
1990/0227		int n; int inwin=0, outwin=0;
1990/0227/sys/src/9/port/sturp.c:451,463 – 1990/0312/sys/src/9/port/sturp.c:501,514
1990/0227		switch(bp->type){ case M_CTL: if(streamparse("init", bp)){ switch(getfields(bp->rptr, fields, 2, ' ')){
1990/0312		switch(getfields((char *)bp->rptr, fields, 2, ' ')){
1990/0227		case 2: inwin = strtoul(fields[1], 0, 0); case 1: outwin = strtoul(fields[0], 0, 0); } initinput(up, inwin);
1990/0312		/* initinput(up, inwin); */ DPRINT("initoutput %d\n", outwin);
1990/0227		initoutput(up, outwin); freeb(bp); return;
1990/0227/sys/src/9/port/sturp.c:467,474 – 1990/0312/sys/src/9/port/sturp.c:518,526
1990/0227		} /* * accept data from writer
1990/0312		* accept data from a writer
1990/0227		*/
1990/0312		static void
1990/0227		urpoput(Queue *q, Block *bp) { Urp *up;
1990/0227/sys/src/9/port/sturp.c:480,671 – 1990/0312/sys/src/9/port/sturp.c:532,679
1990/0227		return; } urpstat.output + = bp->wptr - bp->rptr;                  for(;;){ }
1990/0312		urpstat.output += BLEN(bp); putq(q, bp); output(up);
1990/0227		} /* * wait for an AINIT
1990/0312		* start output
1990/0227		*/ void urpopenwait(Urp *up, Queue *q)
1990/0312		static void output(Urp *up)
1990/0227		{ while((up->state&ISOPENING) && !(up->state&RCLOSE)) { proc->state = Waiting; up->proc = proc; putctl1(q->next, M_RDATA, INIT1); if((up->state&ISOPENING) == 0){ up->proc = 0; if(proc->state == Waiting){ proc->state = Running; return; } } sched(); up->proc = 0; } }
1990/0312		Block *bp, *nbp; ulong now; Queue *q; int n;
1990/0227		/* * wait till transmission is complete */ void urpxmitwait(Urp *up, Queue *q) { Block *bp; int debug;
1990/0312		if(!canqlock(&up->xmit)) return;
1990/0227		debug = (q->flag&QDEBUG);
1990/0312		if(waserror()){ print("urp output error\n"); qunlock(&up->xmit); nexterror(); }
1990/0227		up->timeout = 0; while(!EMPTY(q) || up->WS<up->WNX){ /* * clean up if the channel closed */ if (up->state & RCLOSE) { while(bp = getq(q)) freeb(bp); up->WACK = up->WS = up->WNX;
1990/0312		/* * if still initing and it's time to rexmit, send an INIT1 */ now = NOW; if(up->state & INITING){ if(now > up->timer){ sendctl(up, INIT1); up->timer = now + MSrexmit;
1990/0227		} /* * free acked blocks */ urpfreeacked(up); /* * retransmit if requested */ if(up->state&REXMIT) urprexmit(up, q); /* * fill up the window */ if(!EMPTY(q) && WINDOW(up)) urpfillwindow(up, q); /* * ask other end for its status */ if(up->timeout){ urpstat.enqsx++; putctl1(q->next, M_RDATA, ENQ); up->timeout = 0; } lock(up->olock); if((up->state&RCLOSE) == 0 && up->WS! = up->WNX) { proc->state = Waiting; up->proc = proc; unlock(up->olock); sched(); } else unlock(up->olock);
1990/0312		qunlock(&up->xmit); poperror(); return;
1990/0227		} urpfreeacked(up); up->WS = up->WACK = up->WF = up->WNX = up->WNX&07; } /* * fill up the urp output window */ void urpfillwindow(Urp *up, Queue *q) { Block *bp, *xbp; int debug;                  debug = (q->flag&QDEBUG);                  /* * now process any thing that fits in the flow control window
1990/0312		* fill the transmit buffers
1990/0227		*/ while (WINDOW(up)) { bp = getq(q); if (bp == NULL) break; /* force MAXBLOCK length segments */ if (bp->rptr+up->maxblock < bp->wptr) { xbp = allocb(0); if (xbp == NULL) { putbq(q, bp); break; } xbp->rptr = bp->rptr; bp->rptr + = up->maxblock; xbp->wptr = bp->rptr; putbq(q, bp); bp = xbp;
1990/0312		q = up->wq; for(bp = getq(q); bp && up->xb[up->nxb]==0; up->nxb = NEXT(up->nxb)){ if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr; nbp->wptr = bp->rptr = bp->rptr + up->maxblock; } else { up->xb[up->nxb] = bp; bp = getq(q);
1990/0227		} /* * put new block in the block array. if something is already * there, it should be because we haven't gotten around to freeing * it yet. If not, complain. */ if (up->xb[up->WNX&07]) { urpfreeacked(up); if (up->xb[up->WNX&07]) { uprint(up, "urpfillwindow: overlap"); freeb(up->xb[up->WNX&07]); } } up->xb[up->WNX&07] = bp; up->WNX++; urpxmit(q, bp, up->WNX-1); }
1990/0312		if(bp) putbq(q, bp); /* print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n", WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state); /**/ /* * if a retransmit time has elapsed since a transmit, send an ENQ */ if(up->unechoed != up->next && NOW > up->timer){ print("sENQ\n"); up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ); qunlock(&up->xmit); poperror(); return; } /* * if there's a window open, push some blocks out */ while(WINDOW(up)>0 && up->xb[up->next]!=0 && canqlock(&up->xl[up->next])){ if(up->xb[up->next]) sendblock(up, up->next); qunlock(&up->xl[up->next]); up->next = NEXT(up->next); } qunlock(&up->xmit); poperror();
1990/0227		} /* * Send out a message, with trailer.
1990/0312		* send a control byte, put the byte at the end of the allocated * space in case a lower layer needs header room.
1990/0227		*/ void urpxmit(Queue *q, Block *bp, int seqno)
1990/0312		static void sendctl(Urp *up, int ctl)
1990/0227		{ Urp *up = (Urp *)q->ptr; int size; Block *xbp; int debug;
1990/0312		Block *bp;
1990/0227		if (bp == NULL) { print("null bp in urpxmit\n");
1990/0312		if(up->wq->next->len > Streamhi)
1990/0227		return; } debug = (q->flag&QDEBUG); size = bp->wptr - bp->rptr; seqno & = 07; /* send ptr to block, if non-empty */ if (size) { if ((xbp = allocb(0)) == NULL){ print("can't xmit\n"); return; } xbp->rptr = bp->rptr; xbp->wptr = bp->wptr; xbp->type = bp->type; PUTNEXT(q, xbp); } /* send trailer */ if ((xbp = allocb(3)) == NULL){ print("can't xmit2\n"); return; } xbp->type = M_RDATA; *xbp->wptr++ = (bp->class&S_DELIM) ? BOT: BOTM; *xbp->wptr++ = size; *xbp->wptr++ = size >> 8; PUTNEXT(q, xbp); putctl1(q->next, M_RDATA, SEQ + seqno); up->timer = DKPTIME;
1990/0312		bp = allocb(1); bp->wptr = bp->lim; bp->rptr = bp->lim-1; *bp->rptr = ctl; bp->flags |= S_DELIM; PUTNEXT(up->wq, bp);
1990/0227		} void urprexmit(Urp *up, Queue *q)
1990/0312		/* * send a block. */ static void sendblock(Urp *up, int bn)
1990/0227		{ int i;
1990/0312		Block *bp, *m, *nbp; int n;
1990/0227		for (i = up->WACK; i<up->WNX; i++) { urpxmit(q, up->xb[i&07], i); urpstat.rxmit++; } up->state & = ~REXMIT;
1990/0312		up->timer = NOW + MSrexmit; if(up->wq->next->len > Streamhi) return; /* * message 1, the BOT and the data */ bp = up->xb[bn]; m = allocb(1); m->rptr = m->lim - 1; m->wptr = m->lim; *m->rptr = (bp->flags & S_DELIM) ? BOT : BOTM; nbp = m->next = allocb(0); nbp->rptr = bp->rptr; nbp->wptr = bp->wptr; nbp->flags |= S_DELIM; PUTNEXT(up->wq, m); /* * message 2, the block length and the SEQ */ m = allocb(3); m->rptr = m->lim - 3; m->wptr = m->lim; n = BLEN(bp); m->rptr[0] = SEQ | bn; m->rptr[1] = n; m->rptr[2] = n<<8; m->flags |= S_DELIM; PUTNEXT(up->wq, m);
1990/0227		} /*
1990/0227/sys/src/9/port/sturp.c:677,706 – 1990/0312/sys/src/9/port/sturp.c:685,736
1990/0227		int seqno; int next; seqno = msg&07; next = (seqno+1) & 0x7
1990/0312		seqno = msg&Nmask; next = NEXT(seqno);
1990/0227		lock(up); if(BETWEEN(seqno, up->WACK, up->WNX)) up->WACK = next;
1990/0312		/* * release any acknowledged blocks */ if(IN(seqno, up->unacked, up->next)){ for(; up->unacked != next; up->unacked = NEXT(up->unacked)){ qlock(&up->xl[up->unacked]); if(up->xb[up->unacked]) freeb(up->xb[up->unacked]); up->xb[up->unacked] = 0; qunlock(&up->xl[up->unacked]); } }
1990/0227		switch(msg & 0370){ case ECHO: up->timer = MSrexmit; /* push off ENQ timeout */ if(BETWEEN(seqno, up->WS, up->WNX)){ up->WS = next; wakeup(&up->r);
1990/0312		if(IN(seqno, up->unechoed, up->next)) { up->unechoed = next;
1990/0227		}
1990/0312		/* * the next reject at the start of a window starts a * retransmission. */ up->state &= ~REJECTING;
1990/0227		break; case REJ:
1990/0312		if(IN(seqno, up->unechoed, up->next)) up->unechoed = next; /* * ... FALL THROUGH ... */
1990/0227		case ACK: if(up->WACK == next){ up->state |= REXMIT; wakeup(&up->r);
1990/0312		/* * start a retransmission if we aren't retransmitting * and this is the start of a window. */ if(up->unechoed==next && !(up->state & REJECTING)){ up->state |= REJECTING; up->next = next;
1990/0227		} break; } unlock(up);
1990/0312		wakeup(&up->rq->r);
1990/0227		} /*
1990/0227/sys/src/9/port/sturp.c:717,770 – 1990/0312/sys/src/9/port/sturp.c:747,758
1990/0227		} /* * send a control character down stream */ static void sendctl(Queue *q, int x) { Block *bp;                  /* * send anything queued */ while(bp = getq(q)) PUTNEXT(q, bp);                  /* * send the new byte */ bp = allocb(1); *bp->wptr++ = x; PUTNEXT(q, bp); }                  /* * queue a control character to be sent down stream */ static void queuectl(Urp *up, int x) { Block *bp; Queue *q;                  q = up->wq; bp = allocb(1); *bp->wptr++ = x; putq(q, bp); wakeup(&up->r); }                  /* * initialize output */ static void initoutput(Urp *up, int window) { /* * ack any outstanding blocks */ if(up->WS<up->WNX) urprack(up, ECHO+((up->WNX-1)&07));
1990/0312		int i;
1990/0227		/* * set output window
1990/0227/sys/src/9/port/sturp.c:774,794 – 1990/0312/sys/src/9/port/sturp.c:762,794
1990/0227		up->maxblock = 64; if(up->maxblock > Streamhi/4) up->maxblock = Streamhi/4; up->XW = 4;
1990/0312		up->maxblock -= 4; up->maxout = 3;
1990/0227		/* * set sequence varialbles */ up->WS = 1; up->WACK = 1; up->WNX = 1;
1990/0312		up->unechoed = 1; up->unacked = 1; up->next = 1; up->nxb = 1;
1990/0227		/*
1990/0312		* free any outstanding blocks */ for(i = 0; i < 8; i++){ qlock(&up->xl[i]); if(up->xb[i]) freeb(up->xb[i]); qunlock(&up->xl[i]); } /*
1990/0227		* tell the other side we've inited */ up->state |= ISOPENING; up->timer = MSrexmit; sendctl(q->next, INIT1);
1990/0312		up->state |= INITING; up->timer = NOW + MSrexmit; sendctl(up, INIT1);
1990/0227		} /*
1990/0227/sys/src/9/port/sturp.c:803,808 – 1990/0312/sys/src/9/port/sturp.c:803,842
1990/0227		up->trx = 0; up->iseq = 0; up->lastecho = ECHO+0; up->WF = 1; flushinput(up);
1990/0312		} /* * do retransmissions etc */ static int todo(void *arg) { Urp *up; up = (Urp *)arg; return (WINDOW(up)>0 && up->wq->len>0 && !(up->state&INITING)); } static void urpkproc(void *arg) { Urp *up; up = (Urp *)arg; for(;;){ if(up->state & (HUNGUP|CLOSING)){ if(isflushed(up)) wakeup(&up->r); if(up->state & HUNGUP) break; } if((up->lastecho&Nmask)!=up->iseq && up->rq->next->len<Streamhi) sendctl(up, up->lastecho = ECHO+up->iseq); output(up); tsleep(&up->rq->r, todo, up, MSrexmit/2); } DPRINT("urpkproc exiting %ux\n", up); up->kstarted = 0; up->state = 0;
1990/0227		}
1990/0312/sys/src/9/port/sturp.c:123,128 – 1990/0315/sys/src/9/port/sturp.c:123,130 (short | long)
1990/0227		int i;
1990/0312		char name[128];
1990/0227
1990/0315		DPRINT("urpopen\n");
1990/0227		/* * find a free urp structure */
1990/0312/sys/src/9/port/sturp.c:379,385 – 1990/0315/sys/src/9/port/sturp.c:381,387
1990/0227		case 0: break; case ENQ:
1990/0312		print("rENQ\n");
1990/0315		print("rENQ %uo %uo\n", up->lastecho, ACK+up->iseq);
1990/0227		urpstat.enqsr++;
1990/0312		sendctl(up, up->lastecho); sendctl(up, ACK+up->iseq);
1990/0312/sys/src/9/port/sturp.c:443,461 – 1990/0315/sys/src/9/port/sturp.c:445,463
1990/0227		if(up->trx != 3){ urpstat.rjtrs++; flushinput(up);
1990/0312		print("sREJ\n");
1990/0315		print("sREJ1 %d\n", up->iseq);
1990/0312		sendctl(up, up->lastecho = REJ+up->iseq);
1990/0227		break; } else if(q->len != up->trbuf[1] + (up->trbuf[2]<<8)){ urpstat.rjpks++; flushinput(up);
1990/0312		print("sREJ\n");
1990/0315		print("sREJ2 %d\n", up->iseq);
1990/0312		sendctl(up, up->lastecho = REJ+up->iseq);
1990/0227		break;
1990/0312		} else if(i != ((up->iseq+1)&Nmask)) {
1990/0227		urpstat.rjseq++; flushinput(up);
1990/0312		print("sREJ\n");
1990/0315		print("sREJ3 %d %d\n", i, up->iseq);
1990/0312		sendctl(up, up->lastecho = REJ+up->iseq);
1990/0227		break; }
1990/0312/sys/src/9/port/sturp.c:780,785 – 1990/0315/sys/src/9/port/sturp.c:782,788
1990/0312		qlock(&up->xl[i]); if(up->xb[i]) freeb(up->xb[i]);
1990/0315		up->xb[i] = 0;
1990/0312		qunlock(&up->xl[i]); }
1990/0312/sys/src/9/port/sturp.c:823,828 – 1990/0315/sys/src/9/port/sturp.c:826,832
1990/0312		Urp *up; up = (Urp *)arg;
1990/0315		DPRINT("urpkproc started\n");
1990/0312		for(;;){ if(up->state & (HUNGUP|CLOSING)){
1990/0315/sys/src/9/port/sturp.c:39,50 – 1990/0321/sys/src/9/port/sturp.c:39,51 (short | long)
1990/0312		Rendez r; /* process waiting for close */
1990/0227		/* input */
1990/0321		QLock ack; /* ack lock */
1990/0312		Queue *rq; /* input queue */
1990/0227		uchar iseq; /* last good input sequence number */ uchar lastecho; /* last echo/rej sent */ uchar trbuf[3]; /* trailer being collected */ short trx; /* # bytes in trailer being collected */
1990/0321		int blocks;
1990/0227		/* output */
1990/0315/sys/src/9/port/sturp.c:110,115 – 1990/0321/sys/src/9/port/sturp.c:111,118
1990/0227		static void rcvack(Urp*, int); static void flushinput(Urp*);
1990/0312		static void sendctl(Urp*, int);
1990/0321		static void sendack(Urp*); static void sendrej(Urp*);
1990/0227		static void initoutput(Urp*, int); static void initinput(Urp*, int);
1990/0312		static void urpkproc(void *arg);
1990/0315/sys/src/9/port/sturp.c:123,130 – 1990/0321/sys/src/9/port/sturp.c:126,131
1990/0227		int i;
1990/0312		char name[128];
1990/0227
1990/0315		DPRINT("urpopen\n");
1990/0227		/* * find a free urp structure */
1990/0315/sys/src/9/port/sturp.c:200,212 – 1990/0321/sys/src/9/port/sturp.c:201,211
1990/0312		i = 7; rcvack(up, ECHO+i); qunlock(&up->xmit); DPRINT("urpclose(%ux)\n", up); /* * kill off the kernel process */ wakeup(&up->rq->r); DPRINT("urpclosed(%ux)\n", up);
1990/0227		} /*
1990/0315/sys/src/9/port/sturp.c:309,318 – 1990/0321/sys/src/9/port/sturp.c:308,319
1990/0227		case SEQ+0: case SEQ+1: case SEQ+2: case SEQ+3: case SEQ+4: case SEQ+5: case SEQ+6: case SEQ+7:
1990/0321		qlock(&up->ack);
1990/0312		i = ctl & Nmask; if(q->next->len < Streamhi) sendctl(up, up->lastecho = ECHO+i); up->iseq = i;
1990/0321		qunlock(&up->ack);
1990/0227		break; } }
1990/0315/sys/src/9/port/sturp.c:381,387 – 1990/0321/sys/src/9/port/sturp.c:382,389
1990/0227		case 0: break; case ENQ:
1990/0315		print("rENQ %uo %uo\n", up->lastecho, ACK+up->iseq);
1990/0321		DPRINT("rENQ %d %uo %uo\n", up->blocks, up->lastecho, ACK+up->iseq); up->blocks = 0;
1990/0227		urpstat.enqsr++;
1990/0312		sendctl(up, up->lastecho); sendctl(up, ACK+up->iseq);
1990/0315/sys/src/9/port/sturp.c:422,428 – 1990/0321/sys/src/9/port/sturp.c:424,430
1990/0227		case REJ+0: case REJ+1: case REJ+2: case REJ+3: case REJ+4: case REJ+5: case REJ+6: case REJ+7:
1990/0312		print("rREJ\n");
1990/0321		DPRINT("rREJ\n");
1990/0227		rcvack(up, ctl); break;
1990/0315/sys/src/9/port/sturp.c:444,464 – 1990/0321/sys/src/9/port/sturp.c:446,460
1990/0312		i = ctl & Nmask;
1990/0227		if(up->trx != 3){ urpstat.rjtrs++; flushinput(up);
1990/0315		print("sREJ1 %d\n", up->iseq);
1990/0312		sendctl(up, up->lastecho = REJ+up->iseq);
1990/0321		sendrej(up);
1990/0227		break; } else if(q->len != up->trbuf[1] + (up->trbuf[2]<<8)){ urpstat.rjpks++; flushinput(up);
1990/0315		print("sREJ2 %d\n", up->iseq);
1990/0312		sendctl(up, up->lastecho = REJ+up->iseq);
1990/0321		sendrej(up);
1990/0227		break;
1990/0312		} else if(i != ((up->iseq+1)&Nmask)) {
1990/0227		urpstat.rjseq++; flushinput(up);
1990/0315		print("sREJ3 %d %d\n", i, up->iseq);
1990/0312		sendctl(up, up->lastecho = REJ+up->iseq);
1990/0321		sendrej(up);
1990/0227		break; }
1990/0315/sys/src/9/port/sturp.c:480,491 – 1990/0321/sys/src/9/port/sturp.c:476,486
1990/0227		/* * acknowledge receipt */ if(q->next->len < Streamhi){
1990/0312		sendctl(up, ECHO+i); up->lastecho = ECHO+i; wakeup(&up->rq->r);
1990/0227		}
1990/0321		qlock(&up->ack);
1990/0312		up->iseq = i;
1990/0321		if(q->next->len < Streamhi) sendctl(up, up->lastecho = ECHO|i); qunlock(&up->ack);
1990/0227		break; } }
1990/0315/sys/src/9/port/sturp.c:510,516 – 1990/0321/sys/src/9/port/sturp.c:505,510
1990/0227		outwin = strtoul(fields[0], 0, 0); }
1990/0312		/* initinput(up, inwin); */ DPRINT("initoutput %d\n", outwin);
1990/0227		initoutput(up, outwin); freeb(bp); return;
1990/0315/sys/src/9/port/sturp.c:596,602 – 1990/0321/sys/src/9/port/sturp.c:590,596
1990/0312		* if a retransmit time has elapsed since a transmit, send an ENQ */ if(up->unechoed != up->next && NOW > up->timer){ print("sENQ\n");
1990/0321		DPRINT("sENQ\n");
1990/0312		up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ);
1990/0315/sys/src/9/port/sturp.c:638,643 – 1990/0321/sys/src/9/port/sturp.c:632,684
1990/0227		}
1990/0312		/*
1990/0321		* send a reject */ static void sendrej(Urp *up) { flushinput(up); qlock(&up->ack); if((up->lastecho&~Nmask) == ECHO){ DPRINT("REJ %d\n", up->iseq); sendctl(up, up->lastecho = REJ|up->iseq); } qunlock(&up->ack); } /* * send an acknowledge */ static void sendack(Urp *up) { Block *bp; /* * check the precondition for acking */ if(up->rq->next->len>=Streamhi || (up->lastecho&Nmask)==up->iseq) return; if(!canqlock(&up->ack)) return; /* * check again now that we've locked */ if(up->rq->next->len>=Streamhi || (up->lastecho&Nmask)==up->iseq){ qunlock(&up->ack); return; } /* * send the ack */ sendctl(up, up->lastecho = ECHO|up->iseq); qunlock(&up->ack); } /*
1990/0312		* send a block. */ static void
1990/0315/sys/src/9/port/sturp.c:803,808 – 1990/0321/sys/src/9/port/sturp.c:844,850
1990/0227		/* * restart all sequence parameters */
1990/0321		up->blocks = 0;
1990/0227		up->trx = 0; up->iseq = 0; up->lastecho = ECHO+0;
1990/0315/sys/src/9/port/sturp.c:826,832 – 1990/0321/sys/src/9/port/sturp.c:868,873
1990/0312		Urp *up; up = (Urp *)arg;
1990/0315		DPRINT("urpkproc started\n");
1990/0312		for(;;){ if(up->state & (HUNGUP|CLOSING)){
1990/0315/sys/src/9/port/sturp.c:835,846 – 1990/0321/sys/src/9/port/sturp.c:876,885
1990/0312		if(up->state & HUNGUP) break; } if((up->lastecho&Nmask)!=up->iseq && up->rq->next->len<Streamhi) sendctl(up, up->lastecho = ECHO+up->iseq);
1990/0321		sendack(up);
1990/0312		output(up); tsleep(&up->rq->r, todo, up, MSrexmit/2); } DPRINT("urpkproc exiting %ux\n", up); up->kstarted = 0; up->state = 0;
1990/0227		}
1990/0321/sys/src/9/port/sturp.c:63,69 – 1990/0331/sys/src/9/port/sturp.c:63,70 (short | long)
1990/0312		int kstarted;
1990/0227		};
1990/0312		#define WINDOW(u) ((u->unechoed + u->maxout - u->next)%8)
1990/0331		#define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\ (u)->unechoed+(u)->maxout-(u)->next)
1990/0312		#define IN(x, f, n) (f<=n ? x>=f && x<n : x<n || x>=f) #define NEXT(x) (((x)+1)&Nmask)
1990/0227
1990/0331/sys/src/9/port/sturp.c:207,212 – 1990/0403/sys/src/9/port/sturp.c:207,215 (short | long)
1990/0312		* kill off the kernel process */ wakeup(&up->rq->r);
1990/0403		if(up->kstarted == 0) up->state = 0;
1990/0227		} /*
1990/0331/sys/src/9/port/sturp.c:253,259 – 1990/0403/sys/src/9/port/sturp.c:256,262
1990/0227		/*
1990/0312		* take care of any data
1990/0227		*/
1990/0312		if(BLEN(bp)>0 && q->next->len<Streamhi)
1990/0403		if(BLEN(bp)>0 && !QFULL(q->next))
1990/0227		PUTNEXT(q, bp); else freeb(bp);
1990/0331/sys/src/9/port/sturp.c:311,317 – 1990/0403/sys/src/9/port/sturp.c:314,320
1990/0227		case SEQ+4: case SEQ+5: case SEQ+6: case SEQ+7:
1990/0321		qlock(&up->ack);
1990/0312		i = ctl & Nmask; if(q->next->len < Streamhi)
1990/0403		if(!QFULL(q->next))
1990/0312		sendctl(up, up->lastecho = ECHO+i); up->iseq = i;
1990/0321		qunlock(&up->ack);
1990/0331/sys/src/9/port/sturp.c:479,485 – 1990/0403/sys/src/9/port/sturp.c:482,488
1990/0227		*/
1990/0321		qlock(&up->ack);
1990/0312		up->iseq = i;
1990/0321		if(q->next->len < Streamhi)
1990/0403		if(!QFULL(q->next))
1990/0321		sendctl(up, up->lastecho = ECHO|i); qunlock(&up->ack);
1990/0227		break;
1990/0331/sys/src/9/port/sturp.c:572,578 – 1990/0403/sys/src/9/port/sturp.c:575,583
1990/0312		* fill the transmit buffers
1990/0227		*/
1990/0312		q = up->wq; for(bp = getq(q); bp && up->xb[up->nxb]==0; up->nxb = NEXT(up->nxb)){
1990/0403		for(bp = getq(q); q->first && up->xb[up->nxb]==0; up->nxb = NEXT(up->nxb)){ if(bp == 0) bp = getq(q);
1990/0312		if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr;
1990/0331/sys/src/9/port/sturp.c:579,585 – 1990/0403/sys/src/9/port/sturp.c:584,590
1990/0312		nbp->wptr = bp->rptr = bp->rptr + up->maxblock; } else { up->xb[up->nxb] = bp; bp = getq(q);
1990/0403		bp = 0;
1990/0227		} }
1990/0312		if(bp)
1990/0331/sys/src/9/port/sturp.c:622,628 – 1990/0403/sys/src/9/port/sturp.c:627,633
1990/0227		{
1990/0312		Block *bp;
1990/0227
1990/0312		if(up->wq->next->len > Streamhi)
1990/0403		if(QFULL(up->wq->next))
1990/0227		return;
1990/0312		bp = allocb(1); bp->wptr = bp->lim;
1990/0331/sys/src/9/port/sturp.c:658,664 – 1990/0403/sys/src/9/port/sturp.c:663,669
1990/0321		/* * check the precondition for acking */ if(up->rq->next->len>=Streamhi || (up->lastecho&Nmask)==up->iseq)
1990/0403		if(QFULL(up->rq->next) || (up->lastecho&Nmask)==up->iseq)
1990/0321		return; if(!canqlock(&up->ack))
1990/0331/sys/src/9/port/sturp.c:667,673 – 1990/0403/sys/src/9/port/sturp.c:672,678
1990/0321		/* * check again now that we've locked */ if(up->rq->next->len>=Streamhi || (up->lastecho&Nmask)==up->iseq){
1990/0403		if(QFULL(up->rq->next) || (up->lastecho&Nmask)==up->iseq){
1990/0321		qunlock(&up->ack); return; }
1990/0331/sys/src/9/port/sturp.c:689,695 – 1990/0403/sys/src/9/port/sturp.c:694,700
1990/0312		int n;
1990/0227
1990/0312		up->timer = NOW + MSrexmit; if(up->wq->next->len > Streamhi)
1990/0403		if(QFULL(up->wq->next))
1990/0312		return; /*
1990/0331/sys/src/9/port/sturp.c:804,811 – 1990/0403/sys/src/9/port/sturp.c:809,814
1990/0227		up->maxblock = window/4; if(up->maxblock < 64) up->maxblock = 64; if(up->maxblock > Streamhi/4) up->maxblock = Streamhi/4;
1990/0312		up->maxblock -= 4; up->maxout = 3;
1990/0227
1990/0331/sys/src/9/port/sturp.c:870,875 – 1990/0403/sys/src/9/port/sturp.c:873,884
1990/0312		up = (Urp *)arg;
1990/0403		if(waserror()){ up->state = 0; up->kstarted = 0; wakeup(&up->r); return; }
1990/0312		for(;;){ if(up->state & (HUNGUP|CLOSING)){ if(isflushed(up))
1990/0331/sys/src/9/port/sturp.c:881,886 – 1990/0403/sys/src/9/port/sturp.c:890,895
1990/0312		output(up); tsleep(&up->rq->r, todo, up, MSrexmit/2); } up->kstarted = 0; up->state = 0;
1990/0403		up->kstarted = 0;
1990/0227		}
1990/0403/sys/src/9/port/sturp.c:575,594 – 1990/0406/sys/src/9/port/sturp.c:575,594 (short | long)
1990/0312		* fill the transmit buffers
1990/0227		*/
1990/0312		q = up->wq;
1990/0403		for(bp = getq(q); q->first && up->xb[up->nxb]==0; up->nxb = NEXT(up->nxb)){ if(bp == 0) bp = getq(q);
1990/0312		if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr; nbp->wptr = bp->rptr = bp->rptr + up->maxblock; } else { up->xb[up->nxb] = bp;
1990/0403		bp = 0;
1990/0406		if(up->xb[up->nxb]==0) { for(bp=getq(q); bp && up->xb[up->nxb]==0; up->nxb=NEXT(up->nxb)){ if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr; nbp->wptr = bp->rptr = bp->rptr + up->maxblock; } else { up->xb[up->nxb] = bp; bp = getq(q); }
1990/0227		}
1990/0406		if(bp) putbq(q, bp);
1990/0227		}
1990/0312		if(bp) putbq(q, bp); /* print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n", WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state); /**/
1990/0406/sys/src/9/port/sturp.c:370,377 – 1990/0424/sys/src/9/port/sturp.c:370,377 (short | long)
1990/0312		* queue the block(s)
1990/0227		*/
1990/0312		if(BLEN(bp) > 0){
1990/0424		bp->flags &= ~S_DELIM;
1990/0312		putq(q, bp); q->last->flags &= ~S_DELIM; if(q->len > 4*1024){
1990/0227		flushinput(up); return;
1990/0406/sys/src/9/port/sturp.c:472,478 – 1990/0424/sys/src/9/port/sturp.c:472,479
1990/0312		PUTNEXT(q, bp);
1990/0227		} else { bp = allocb(0);
1990/0312		bp->flags |= S_DELIM;
1990/0424		if(up->trbuf[0] != BOTM) bp->flags |= S_DELIM;
1990/0227		PUTNEXT(q, bp);
1990/0312		}
1990/0227		up->trx = 0;
1990/0424/sys/src/9/port/sturp.c:190,196 – 1990/0505/sys/src/9/port/sturp.c:190,196 (short | long)
1990/0312		* process we're closing.
1990/0227		*/
1990/0312		up->state |= CLOSING; sleep(&up->r, isflushed, up);
1990/0505		tsleep(&up->r, isflushed, up, 60*1000);
1990/0312		/* * ack all outstanding messages
1990/0505/sys/src/9/port/sturp.c:166,172 – 1990/0509/sys/src/9/port/sturp.c:166,172 (short | long)
1990/0227		Urp *up; up = (Urp *)a;
1990/0312		return (up->state&HUNGUP) || (up->unechoed==up->next && up->wq->len==0);
1990/0509		return (up->state&HUNGUP) || (up->unechoed==up->nxb && up->wq->len==0);
1990/0227		}
1990/0312		static int isdead(void *a)
1990/0505/sys/src/9/port/sturp.c:188,212 – 1990/0509/sys/src/9/port/sturp.c:188,223
1990/0227		/*
1990/0312		* wait for all outstanding messages to drain, tell kernel * process we're closing.
1990/0509		* * if 2 minutes elapse, give it up
1990/0227		*/
1990/0312		up->state |= CLOSING;
1990/0505		tsleep(&up->r, isflushed, up, 60*1000);
1990/0509		tsleep(&up->r, isflushed, up, 2*60*1000);
1990/0312		/* * ack all outstanding messages
1990/0509		* kill off the kernel process
1990/0312		*/ qlock(&up->xmit); up->state |= HUNGUP;
1990/0509		wakeup(&up->rq->r); qlock(&up->xmit); /* * ack all outstanding messages */
1990/0312		i = up->next - 1; if(i < 0) i = 7; rcvack(up, ECHO+i); qunlock(&up->xmit); /* * kill off the kernel process
1990/0509		* free all staged but unsent messages
1990/0312		*/ wakeup(&up->rq->r);
1990/0509		for(i = 0; i < 7; i++) if(up->xb[i]){ freeb(up->xb[i]); up->xb[i] = 0; } qunlock(&up->xmit);
1990/0403		if(up->kstarted == 0) up->state = 0;
1990/0509/sys/src/9/port/sturp.c:48,54 – 1990/0511/sys/src/9/port/sturp.c:48,53 (short | long)
1990/0321		int blocks;
1990/0227		/* output */
1990/0312		QLock xmit; /* output lock, only one process at a time */ Queue *wq; /* output queue */ int maxout; /* maximum outstanding unacked blocks */
1990/0509/sys/src/9/port/sturp.c:60,65 – 1990/0511/sys/src/9/port/sturp.c:59,65
1990/0227		Block *xb[8]; /* the xmit window buffer */
1990/0312		QLock xl[8]; ulong timer; /* timeout for xmit */
1990/0511		int rexmit;
1990/0312		int kstarted;
1990/0227		};
1990/0509/sys/src/9/port/sturp.c:168,181 – 1990/0511/sys/src/9/port/sturp.c:168,173
1990/0227		up = (Urp *)a;
1990/0509		return (up->state&HUNGUP) || (up->unechoed==up->nxb && up->wq->len==0);
1990/0227		}
1990/0312		static int isdead(void *a) { Urp *up;                  up = (Urp *)a; return up->kstarted == 0; } static void
1990/0227		urpclose(Queue *q) {
1990/0509/sys/src/9/port/sturp.c:219,226 – 1990/0511/sys/src/9/port/sturp.c:211,220
1990/0509		} qunlock(&up->xmit);
1990/0403		if(up->kstarted == 0)
1990/0511		if(up->kstarted == 0){ DPRINT("urpclose %ux\n", up);
1990/0403		up->state = 0;
1990/0511		}
1990/0227		} /*
1990/0509/sys/src/9/port/sturp.c:471,476 – 1990/0511/sys/src/9/port/sturp.c:465,474
1990/0227		urpstat.rjseq++;
1990/0321		sendrej(up);
1990/0227		break;
1990/0511		} else if(q->next->len > (3*Streamhi)/2 || q->next->nb > (3*Streambhi)/2) { flushinput(up); break;
1990/0227		} /*
1990/0509/sys/src/9/port/sturp.c:559,564 – 1990/0511/sys/src/9/port/sturp.c:557,563
1990/0312		ulong now; Queue *q; int n;
1990/0511		int i;
1990/0227
1990/0312		if(!canqlock(&up->xmit)) return;
1990/0509/sys/src/9/port/sturp.c:607,614 – 1990/0511/sys/src/9/port/sturp.c:606,612
1990/0312		/* * if a retransmit time has elapsed since a transmit, send an ENQ */ if(up->unechoed != up->next && NOW > up->timer){
1990/0321		DPRINT("sENQ\n");
1990/0511		if(up->unechoed!=up->next && NOW>up->timer){
1990/0312		up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ);
1990/0509/sys/src/9/port/sturp.c:620,630 – 1990/0511/sys/src/9/port/sturp.c:618,638
1990/0312		/* * if there's a window open, push some blocks out */ while(WINDOW(up)>0 && up->xb[up->next]!=0 && canqlock(&up->xl[up->next])){ if(up->xb[up->next]) sendblock(up, up->next); qunlock(&up->xl[up->next]);
1990/0511		if(up->rexmit){ up->rexmit = 0; up->next = up->unechoed; } while(WINDOW(up)>0 && up->xb[up->next]!=0){ i = up->next; qlock(&up->xl[i]); if(waserror()){ qunlock(&up->xl[i]); nexterror(); } sendblock(up, i); qunlock(&up->xl[i]);
1990/0312		up->next = NEXT(up->next);
1990/0511		poperror();
1990/0312		} qunlock(&up->xmit); poperror();
1990/0509/sys/src/9/port/sturp.c:745,750 – 1990/0511/sys/src/9/port/sturp.c:753,759
1990/0227		{ int seqno; int next;
1990/0511		int i;
1990/0227
1990/0312		seqno = msg&Nmask; next = NEXT(seqno);
1990/0509/sys/src/9/port/sturp.c:754,764 – 1990/0511/sys/src/9/port/sturp.c:763,774
1990/0312		*/ if(IN(seqno, up->unacked, up->next)){ for(; up->unacked != next; up->unacked = NEXT(up->unacked)){ qlock(&up->xl[up->unacked]); if(up->xb[up->unacked]) freeb(up->xb[up->unacked]); up->xb[up->unacked] = 0; qunlock(&up->xl[up->unacked]);
1990/0511		i = up->unacked; qlock(&up->xl[i]); if(up->xb[i]) freeb(up->xb[i]); up->xb[i] = 0; qunlock(&up->xl[i]);
1990/0312		} }
1990/0227
1990/0509/sys/src/9/port/sturp.c:786,792 – 1990/0511/sys/src/9/port/sturp.c:796,802
1990/0312		*/ if(up->unechoed==next && !(up->state & REJECTING)){ up->state |= REJECTING; up->next = next;
1990/0511		up->rexmit = 1;
1990/0227		} break; }
1990/0509/sys/src/9/port/sturp.c:831,836 – 1990/0511/sys/src/9/port/sturp.c:841,847
1990/0312		up->unacked = 1; up->next = 1; up->nxb = 1;
1990/0511		up->rexmit = 0;
1990/0227		/*
1990/0312		* free any outstanding blocks
1990/0509/sys/src/9/port/sturp.c:886,891 – 1990/0511/sys/src/9/port/sturp.c:897,903
1990/0312		up = (Urp *)arg;
1990/0403		if(waserror()){
1990/0511		print("urpkproc error %ux\n", up);
1990/0403		up->state = 0; up->kstarted = 0; wakeup(&up->r);
1990/0509/sys/src/9/port/sturp.c:898,907 – 1990/0511/sys/src/9/port/sturp.c:910,925
1990/0312		if(up->state & HUNGUP) break; }
1990/0321		sendack(up);
1990/0511		if(up->state == 0){ DPRINT("urpkproc: %ux->state == 0\n", up); break; } if(!QFULL(up->rq->next)) sendack(up);
1990/0312		output(up); tsleep(&up->rq->r, todo, up, MSrexmit/2); } up->state = 0;
1990/0403		up->kstarted = 0;
1990/0511		DPRINT("urpkproc %ux\n", up);
1990/0227		}
1990/0511/sys/src/9/port/sturp.c:261,269 – 1990/0601/sys/src/9/port/sturp.c:261,270 (short | long)
1990/0227		/*
1990/0312		* take care of any data
1990/0227		*/
1990/0403		if(BLEN(bp)>0 && !QFULL(q->next))
1990/0601		if(BLEN(bp)>0 && q->next->len<2*Streamhi && q->next->nb<2*Streambhi){ bp->flags |= S_DELIM;
1990/0227		PUTNEXT(q, bp); else
1990/0601		} else
1990/0227		freeb(bp); /*
1990/0601/sys/src/9/port/sturp.c:65,71 – 1990/0629/sys/src/9/port/sturp.c:65,71 (short | long)
1990/0227		};
1990/0331		#define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\ (u)->unechoed+(u)->maxout-(u)->next)
1990/0312		#define IN(x, f, n) (f<=n ? x>=f && x<n : x<n || x>=f)
1990/0629		#define IN(x, f, n) (f<=n ? (x>=f && x<n) : (x<n || x>=f))
1990/0312		#define NEXT(x) (((x)+1)&Nmask)
1990/0227		/*
1990/0601/sys/src/9/port/sturp.c:117,122 – 1990/0629/sys/src/9/port/sturp.c:117,123
1990/0227		static void initoutput(Urp*, int); static void initinput(Urp*, int);
1990/0312		static void urpkproc(void *arg);
1990/0629		static void urpvomit(char*, Urp*);
1990/0227
1990/0312		Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
1990/0227
1990/0601/sys/src/9/port/sturp.c:584,594 – 1990/0629/sys/src/9/port/sturp.c:585,598
1990/0227		} /*
1990/0312		* fill the transmit buffers
1990/0629		* fill the transmit buffers, `nxb' can never overtake `unechoed'
1990/0227		*/
1990/0312		q = up->wq;
1990/0406		if(up->xb[up->nxb]==0) { for(bp=getq(q); bp && up->xb[up->nxb]==0; up->nxb=NEXT(up->nxb)){
1990/0629		i = NEXT(up->nxb); if(i != up->unechoed) { for(bp = getq(q); bp && i!=up->unechoed; i = NEXT(i)){ if(up->xb[up->nxb] != 0) urpvomit("output", up);
1990/0406		if(BLEN(bp) > up->maxblock){ nbp = up->xb[up->nxb] = allocb(0); nbp->rptr = bp->rptr;
1990/0601/sys/src/9/port/sturp.c:597,613 – 1990/0629/sys/src/9/port/sturp.c:601,627
1990/0406		up->xb[up->nxb] = bp; bp = getq(q); }
1990/0629		up->nxb = i;
1990/0227		}
1990/0406		if(bp) putbq(q, bp);
1990/0227		}
1990/0312		/* print("output w(%d) up->xb[%d](%ux) up->nxb(%d) up->state(%ux)\n", WINDOW(up), up->next, up->xb[up->next], up->nxb, up->state); /**/
1990/0629
1990/0312		/* * if a retransmit time has elapsed since a transmit, send an ENQ
1990/0629		* retransmit cruft
1990/0312		*/
1990/0511		if(up->unechoed!=up->next && NOW>up->timer){
1990/0629		if(up->rexmit){ /* * if a retransmit is requested, move next back to * the unacked blocks */ up->rexmit = 0; up->next = up->unacked; } else if(up->unacked!=up->next && NOW>up->timer){ /* * if a retransmit time has elapsed since a transmit, * send an ENQ */
1990/0312		up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ);
1990/0601/sys/src/9/port/sturp.c:618,629 – 1990/0629/sys/src/9/port/sturp.c:632,642
1990/0312		/* * if there's a window open, push some blocks out
1990/0629		* * the lock is to synchronize with acknowledges that free * blocks.
1990/0312		*/
1990/0511		if(up->rexmit){ up->rexmit = 0; up->next = up->unechoed; } while(WINDOW(up)>0 && up->xb[up->next]!=0){
1990/0629		while(WINDOW(up)>0 && up->next!=up->nxb){
1990/0511		i = up->next; qlock(&up->xl[i]); if(waserror()){
1990/0601/sys/src/9/port/sturp.c:722,727 – 1990/0629/sys/src/9/port/sturp.c:735,744
1990/0312		* message 1, the BOT and the data */ bp = up->xb[bn];
1990/0629		if(bp == 0){ urpvomit("sendblock", up); return; }
1990/0312		m = allocb(1); m->rptr = m->lim - 1; m->wptr = m->lim;
1990/0601/sys/src/9/port/sturp.c:768,773 – 1990/0629/sys/src/9/port/sturp.c:785,792
1990/0511		qlock(&up->xl[i]); if(up->xb[i]) freeb(up->xb[i]);
1990/0629		else urpvomit("rcvack", up);
1990/0511		up->xb[i] = 0; qunlock(&up->xl[i]);
1990/0312		}
1990/0601/sys/src/9/port/sturp.c:923,926 – 1990/0629/sys/src/9/port/sturp.c:942,973
1990/0312		up->state = 0;
1990/0403		up->kstarted = 0;
1990/0511		DPRINT("urpkproc %ux\n", up);
1990/0629		} /* * urp got very confused, complain */ static void urpvomit(char *msg, Urp* up) { print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n", msg, up, up->next, up->unechoed, up->unacked, up->nxb); print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n", up->xb[0], up->xb[1], up->xb[2], up->xb[3], up->xb[4], up->xb[5], up->xb[6], up->xb[7]); print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n", up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1], up->trbuf[2]); print("\tupq: %ux %d %d\n", up->rq->next->first, up->rq->next->nb, up->rq->next->len); } int urpdump(void) { Urp *up; for(up = urp; up < &urp[Nurp]; up++) if(up->rq) urpvomit("", up);
1990/0227		}
1990/0629/sys/src/9/port/sturp.c:950,955 – 1990/0702/sys/src/9/port/sturp.c:950,958 (short | long)
1990/0629		static void urpvomit(char *msg, Urp* up) {
1990/0702		Block *bp; uchar *cp;
1990/0629		print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n", msg, up, up->next, up->unechoed, up->unacked, up->nxb); print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n",
1990/0629/sys/src/9/port/sturp.c:958,965 – 1990/0702/sys/src/9/port/sturp.c:961,977
1990/0629		print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n", up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1], up->trbuf[2]); print("\tupq: %ux %d %d\n", up->rq->next->first, up->rq->next->nb,
1990/0702		print("\tupq: %ux %d %d\n", &up->rq->next->r, up->rq->next->nb,
1990/0629		up->rq->next->len);
1990/0702		lock(up->rq->next); for(bp = up->rq->next->first; bp; bp = bp->next){ print("%d%c:\t", 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(up->rq->next);
1990/0629		} int
1990/0702/sys/src/9/port/sturp.c:950,958 – 1990/0707/sys/src/9/port/sturp.c:950,955 (short | long)
1990/0629		static void urpvomit(char *msg, Urp* up) {
1990/0702		Block *bp; uchar *cp;
1990/0629		print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n", msg, up, up->next, up->unechoed, up->unacked, up->nxb); print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n",
1990/0702/sys/src/9/port/sturp.c:963,977 – 1990/0707/sys/src/9/port/sturp.c:960,965
1990/0629		up->trbuf[2]);
1990/0702		print("\tupq: %ux %d %d\n", &up->rq->next->r, up->rq->next->nb,
1990/0629		up->rq->next->len);
1990/0702		lock(up->rq->next); for(bp = up->rq->next->first; bp; bp = bp->next){ print("%d%c:\t", 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(up->rq->next);
1990/0629		} int
1990/0707/sys/src/9/port/sturp.c:7,13 – 1990/0717/sys/src/9/port/sturp.c:7,12 (short | long)
1990/0312		#include "errno.h"
1990/0227		enum { Nurp= 32, MSrexmit= 1000,
1990/0312		Nmask= 0x7,
1990/0227		};
1990/0707/sys/src/9/port/sturp.c:97,103 – 1990/0717/sys/src/9/port/sturp.c:96,102
1990/0312		#define OPEN 0x8 #define CLOSING 0x10
1990/0227		Urp urp[Nurp];
1990/0717		Urp *urp;
1990/0227		/* * predeclared
1990/0707/sys/src/9/port/sturp.c:121,126 – 1990/0717/sys/src/9/port/sturp.c:120,132
1990/0227
1990/0312		Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
1990/0227
1990/0717		void urpreset(void) { newqinfo(&urpinfo); urp = (Urp *)ialloc(conf.nurp*sizeof(Urp), 0); }
1990/0312		static void
1990/0227		urpopen(Queue *q, Stream *s) {
1990/0707/sys/src/9/port/sturp.c:131,144 – 1990/0717/sys/src/9/port/sturp.c:137,153
1990/0227		/* * find a free urp structure */ for(up = urp; up < &urp[Nurp]; up++){
1990/0717		for(up = urp; up < &urp[conf.nurp]; up++){
1990/0227		qlock(up); if(up->state == 0) break; qunlock(up); } if(up == &urp[Nurp])
1990/0717		if(up == &urp[conf.nurp]){ q->ptr = 0; WR(q)->ptr = 0;
1990/0227		error(0, Egreg);
1990/0717		}
1990/0227
1990/0312		q->ptr = q->other->ptr = up;
1990/0227		up->rq = q;
1990/0707/sys/src/9/port/sturp.c:177,182 – 1990/0717/sys/src/9/port/sturp.c:186,193
1990/0227		int i; up = (Urp *)q->ptr;
1990/0717		if(up == 0) return;
1990/0227		/*
1990/0312		* wait for all outstanding messages to drain, tell kernel
1990/0707/sys/src/9/port/sturp.c:967,973 – 1990/0717/sys/src/9/port/sturp.c:978,984
1990/0629		{ Urp *up; for(up = urp; up < &urp[Nurp]; up++)
1990/0717		for(up = urp; up < &urp[conf.nurp]; up++)
1990/0629		if(up->rq) urpvomit("", up);
1990/0227		}
1990/0717/sys/src/9/port/sturp.c:11,17 – 1990/0721/sys/src/9/port/sturp.c:11,17 (short | long)
1990/0312		Nmask= 0x7,
1990/0227		};
1990/0312		#define DPRINT if(0)
1990/0721		#define DPRINT /*if(q->flag&QDEBUG)kprint*/
1990/0312
1990/0227		typedef struct Urp Urp;
1990/0717/sys/src/9/port/sturp.c:438,445 – 1990/0721/sys/src/9/port/sturp.c:438,446
1990/0227		break; case BOT: case BOTS: case BOTM:
1990/0721		case BOTS: DPRINT("rBOT%d...", ctl-BOT);
1990/0227		up->trx = 1; up->trbuf[0] = ctl; break;
1990/0717/sys/src/9/port/sturp.c:458,465 – 1990/0721/sys/src/9/port/sturp.c:459,466
1990/0227		break; /*
1990/0312		* if the seuence number is the next expected * and te trailer length == 3
1990/0721		* if the sequence number is the next expected * and the trailer length == 3
1990/0312		* and the block count matches the bytes received * then send the bytes upstream.
1990/0227		*/
1990/0717/sys/src/9/port/sturp.c:483,488 – 1990/0721/sys/src/9/port/sturp.c:484,490
1990/0511		flushinput(up); break;
1990/0227		}
1990/0721		DPRINT("rSEQ%d accept %d\n", i, q->len);
1990/0227		/* * send data upstream
1990/0717/sys/src/9/port/sturp.c:536,541 – 1990/0721/sys/src/9/port/sturp.c:538,558
1990/0227		freeb(bp); return; }
1990/0721		if(streamparse("debug", bp)){ switch(getfields((char *)bp->rptr, fields, 2, ' ')){ case 1: if (strcmp(fields[0], "on") == 0) { q->flag |= QDEBUG; q->other->flag |= QDEBUG; } if (strcmp(fields[0], "off") == 0) { q->flag &= ~QDEBUG; q->other->flag &= ~QDEBUG; } } freeb(bp); return; }
1990/0227		} PUTNEXT(q, bp); }
1990/0717/sys/src/9/port/sturp.c:688,693 – 1990/0721/sys/src/9/port/sturp.c:705,711
1990/0321		static void sendrej(Urp *up) {
1990/0721		Queue *q = up->wq;
1990/0321		flushinput(up); qlock(&up->ack); if((up->lastecho&~Nmask) == ECHO){
1990/0717/sys/src/9/port/sturp.c:923,931 – 1990/0721/sys/src/9/port/sturp.c:941,950
1990/0312		static void urpkproc(void *arg) { Urp *up;
1990/0721		Urp *up; Queue *q;
1990/0312		up = (Urp *)arg;
1990/0721		q = up->wq;
1990/0312
1990/0403		if(waserror()){
1990/0511		print("urpkproc error %ux\n", up);
1990/0721/sys/src/9/port/sturp.c:162,168 – 1990/0722/sys/src/9/port/sturp.c:162,168 (short | long)
1990/0312		*/ if(up->kstarted == 0){ up->kstarted = 1; sprint(name, "**urp%d**", up - urp);
1990/0722		sprint(name, "urp%d", up - urp);
1990/0312		kproc(name, urpkproc, up); }
1990/0227		}
1990/0722/sys/src/9/port/sturp.c:68,73 – 1990/0725/sys/src/9/port/sturp.c:68,79 (short | long)
1990/0312		#define NEXT(x) (((x)+1)&Nmask)
1990/0227		/*
1990/0725		* Alarm for urptiming */ Alarm *urptiming; Lock urptlock; /*
1990/0227		* Protocol control bytes */ #define SEQ 0010 /* sequence number, ends trailers */
1990/0722/sys/src/9/port/sturp.c:116,121 – 1990/0725/sys/src/9/port/sturp.c:122,128
1990/0227		static void initoutput(Urp*, int); static void initinput(Urp*, int);
1990/0312		static void urpkproc(void *arg);
1990/0725		static void urptimer(Alarm*);
1990/0629		static void urpvomit(char*, Urp*);
1990/0227
1990/0312		Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
1990/0722/sys/src/9/port/sturp.c:165,170 – 1990/0725/sys/src/9/port/sturp.c:172,188
1990/0722		sprint(name, "urp%d", up - urp);
1990/0312		kproc(name, urpkproc, up); }
1990/0725		/* * start the urptimer if it isn't already */ if(urptiming==0){ if(canlock(&urptlock)){ if(urptiming == 0) urptiming = alarm(500, urptimer, 0); unlock(&urptlock); } }
1990/0227		} /*
1990/0722/sys/src/9/port/sturp.c:589,594 – 1990/0725/sys/src/9/port/sturp.c:607,623
1990/0312		int n;
1990/0511		int i;
1990/0227
1990/0725		/* * start the urptimer if it isn't already */ if(urptiming==0){ if(canlock(&urptlock)){ if(urptiming == 0) urptiming = alarm(500, urptimer, 0); unlock(&urptlock); } }
1990/0312		if(!canqlock(&up->xmit)) return;
1990/0227
1990/0722/sys/src/9/port/sturp.c:967,977 – 1990/0725/sys/src/9/port/sturp.c:996,1028
1990/0511		if(!QFULL(up->rq->next)) sendack(up);
1990/0312		output(up); tsleep(&up->rq->r, todo, up, MSrexmit/2);
1990/0725		sleep(&up->rq->r, todo, up);
1990/0312		} up->state = 0;
1990/0403		up->kstarted = 0;
1990/0511		DPRINT("urpkproc %ux\n", up);
1990/0725		} /* * timer to wakeup urpkproc's for retransmissions */ static void urptimer(Alarm *a) { Urp *up; Urp *last; Queue *q; urptiming = 0; for(up = urp, last = &urp[conf.nurp]; up < last; up++){ if(up->state==0) continue; if(up->unacked!=up->next && NOW>up->timer){ q = up->rq; if(q) wakeup(&q->r); } }
1990/0629		} /*
1990/0725/sys/src/9/port/sturp.c:1013,1018 – 1990/0726/sys/src/9/port/sturp.c:1013,1019 (short | long)
1990/0725		Urp *last; Queue *q;
1990/0726		cancel(a);
1990/0725		urptiming = 0; for(up = urp, last = &urp[conf.nurp]; up < last; up++){ if(up->state==0)
1990/0726/sys/src/9/port/sturp.c:68,79 – 1990/0728/sys/src/9/port/sturp.c:68,73 (short | long)
1990/0312		#define NEXT(x) (((x)+1)&Nmask)
1990/0227		/*
1990/0725		* Alarm for urptiming */ Alarm *urptiming; Lock urptlock;                  /*
1990/0227		* Protocol control bytes */ #define SEQ 0010 /* sequence number, ends trailers */
1990/0726/sys/src/9/port/sturp.c:132,137 – 1990/0728/sys/src/9/port/sturp.c:126,132
1990/0717		{ newqinfo(&urpinfo); urp = (Urp *)ialloc(conf.nurp*sizeof(Urp), 0);
1990/0728		alarm(500, urptimer, 0);
1990/0717		}
1990/0312		static void
1990/0726/sys/src/9/port/sturp.c:167,188 – 1990/0728/sys/src/9/port/sturp.c:162,169
1990/0312		/* * start the ack/(re)xmit process */ if(up->kstarted == 0){ up->kstarted = 1;
1990/0722		sprint(name, "urp%d", up - urp);
1990/0312		kproc(name, urpkproc, up); }
1990/0725		/* * start the urptimer if it isn't already */ if(urptiming==0){ if(canlock(&urptlock)){ if(urptiming == 0) urptiming = alarm(500, urptimer, 0); unlock(&urptlock); } }
1990/0728		sprint(name, "urp%d", up - urp); kproc(name, urpkproc, up);
1990/0227		} /*
1990/0726/sys/src/9/port/sturp.c:189,194 – 1990/0728/sys/src/9/port/sturp.c:170,183
1990/0312		* Shut down the connection and kill off the kernel process
1990/0227		*/ static int
1990/0728		isdead(void *a) { Urp *up; up = (Urp *)a; return up->kstarted==0; } static int
1990/0312		isflushed(void *a)
1990/0227		{ Urp *up;
1990/0726/sys/src/9/port/sturp.c:217,223 – 1990/0728/sys/src/9/port/sturp.c:206,212
1990/0509		tsleep(&up->r, isflushed, up, 2*60*1000);
1990/0312		/*
1990/0509		* kill off the kernel process
1990/0728		* tell kernel process to die
1990/0312		*/ up->state |= HUNGUP;
1990/0509		wakeup(&up->rq->r);
1990/0726/sys/src/9/port/sturp.c:241,250 – 1990/0728/sys/src/9/port/sturp.c:230,242
1990/0509		} qunlock(&up->xmit);
1990/0403
1990/0511		if(up->kstarted == 0){ DPRINT("urpclose %ux\n", up);
1990/0403		up->state = 0;
1990/0511		}
1990/0728		/* * wait for kernel process to die */ while(up->kstarted) sleep(&up->r, isdead, up); up->state = 0;
1990/0227		} /*
1990/0726/sys/src/9/port/sturp.c:607,623 – 1990/0728/sys/src/9/port/sturp.c:599,604
1990/0312		int n;
1990/0511		int i;
1990/0227
1990/0725		/* * start the urptimer if it isn't already */ if(urptiming==0){ if(canlock(&urptlock)){ if(urptiming == 0) urptiming = alarm(500, urptimer, 0); unlock(&urptlock); } }
1990/0312		if(!canqlock(&up->xmit)) return;
1990/0227
1990/0726/sys/src/9/port/sturp.c:636,644 – 1990/0728/sys/src/9/port/sturp.c:617,623
1990/0312		sendctl(up, INIT1); up->timer = now + MSrexmit;
1990/0227		}
1990/0312		qunlock(&up->xmit); poperror(); return;
1990/0728		goto out;
1990/0227		} /*
1990/0726/sys/src/9/port/sturp.c:682,690 – 1990/0728/sys/src/9/port/sturp.c:661,667
1990/0312		up->timer = NOW + MSrexmit; up->state &= ~REJECTING; sendctl(up, ENQ); qunlock(&up->xmit); poperror(); return;
1990/0728		goto out;
1990/0312		} /*
1990/0726/sys/src/9/port/sturp.c:705,710 – 1990/0728/sys/src/9/port/sturp.c:682,688
1990/0312		up->next = NEXT(up->next);
1990/0511		poperror();
1990/0312		}
1990/0728		out:
1990/0312		qunlock(&up->xmit); poperror();
1990/0227		}
1990/0726/sys/src/9/port/sturp.c:970,1005 – 1990/0728/sys/src/9/port/sturp.c:948,975
1990/0312		static void urpkproc(void *arg) {
1990/0721		Urp *up; Queue *q;
1990/0728		Urp *up;
1990/0312		up = (Urp *)arg;
1990/0721		q = up->wq;
1990/0728		up->kstarted = 1;
1990/0312
1990/0403		if(waserror()){
1990/0511		print("urpkproc error %ux\n", up);
1990/0403		up->state = 0; up->kstarted = 0; wakeup(&up->r); return; }
1990/0312		for(;;){ if(up->state & (HUNGUP|CLOSING)){ if(isflushed(up)) wakeup(&up->r); if(up->state & HUNGUP) break; }
1990/0511		if(up->state == 0){ DPRINT("urpkproc: %ux->state == 0\n", up);
1990/0728		if(up->state & HUNGUP)
1990/0511		break; } if(!QFULL(up->rq->next)) sendack(up);
1990/0312		output(up);
1990/0725		sleep(&up->rq->r, todo, up);
1990/0312		} up->state = 0;
1990/0403		up->kstarted = 0;
1990/0728		wakeup(&up->r); poperror();
1990/0511		DPRINT("urpkproc %ux\n", up);
1990/0725		}
1990/0726/sys/src/9/port/sturp.c:1014,1020 – 1990/0728/sys/src/9/port/sturp.c:984,990
1990/0725		Queue *q;
1990/0726		cancel(a);
1990/0725		urptiming = 0;
1990/0728		alarm(500, urptimer, 0);
1990/0725		for(up = urp, last = &urp[conf.nurp]; up < last; up++){ if(up->state==0) continue;
1990/0728/sys/src/9/port/sturp.c:988,994 – 1990/0731/sys/src/9/port/sturp.c:988,994 (short | long)
1990/0725		for(up = urp, last = &urp[conf.nurp]; up < last; up++){ if(up->state==0) continue; if(up->unacked!=up->next && NOW>up->timer){
1990/0731		if((up->unacked!=up->next || (up->state&INITING)) && NOW>up->timer){
1990/0725		q = up->rq; if(q) wakeup(&q->r);
1990/0731/sys/src/9/port/sturp.c:943,949 – 1990/0804/sys/src/9/port/sturp.c:943,953 (short | long)
1990/0312		Urp *up; up = (Urp *)arg; return (WINDOW(up)>0 && up->wq->len>0 && !(up->state&INITING));
1990/0804		return (up->state&INITING) ? NOW>up->timer /* time to INIT1 */ : ((up->unacked!=up->next && NOW>up->timer) /* time to ENQ */ || (!QFULL(up->rq->next) && up->iseq!=(up->lastecho&7))); /* time to ECHO */
1990/0312		} static void urpkproc(void *arg)
1990/0731/sys/src/9/port/sturp.c:970,976 – 1990/0804/sys/src/9/port/sturp.c:974,979
1990/0403		up->kstarted = 0;
1990/0728		wakeup(&up->r); poperror();
1990/0511		DPRINT("urpkproc %ux\n", up);
1990/0725		} /*
1990/0731/sys/src/9/port/sturp.c:988,998 – 1990/0804/sys/src/9/port/sturp.c:991,998
1990/0725		for(up = urp, last = &urp[conf.nurp]; up < last; up++){ if(up->state==0) continue;
1990/0731		if((up->unacked!=up->next || (up->state&INITING)) && NOW>up->timer){
1990/0725		q = up->rq; if(q) wakeup(&q->r); }
1990/0804		if(up->rq && todo(up)) wakeup(&up->rq->r);
1990/0725		}
1990/0629		}
1990/0804/sys/src/9/port/sturp.c:771,780 – 1990/0814/sys/src/9/port/sturp.c:771,778 (short | long)
1990/0312		* message 1, the BOT and the data */ bp = up->xb[bn];
1990/0629		if(bp == 0){ urpvomit("sendblock", up);
1990/0814		if(bp == 0)
1990/0629		return; }
1990/0312		m = allocb(1); m->rptr = m->lim - 1; m->wptr = m->lim;
1990/0804/sys/src/9/port/sturp.c:947,952 – 1990/0814/sys/src/9/port/sturp.c:945,951
1990/0804		return (up->state&INITING) ? NOW>up->timer /* time to INIT1 */ : ((up->unacked!=up->next && NOW>up->timer) /* time to ENQ */
1990/0814		|| WINDOW(up)>0 && up->next!=up->nxb
1990/0804		|| (!QFULL(up->rq->next) && up->iseq!=(up->lastecho&7))); /* time to ECHO */
1990/0312		} static void
1990/0814/sys/src/9/port/sturp.c:11,17 – 1990/0911/sys/src/9/port/sturp.c:11,17 (short | long)
1990/0312		Nmask= 0x7,
1990/0227		};
1990/0721		#define DPRINT /*if(q->flag&QDEBUG)kprint*/
1990/0911		#define DPRINT if(q->flag&QDEBUG)print
1990/0312
1990/0227		typedef struct Urp Urp;
1990/0814/sys/src/9/port/sturp.c:101,106 – 1990/0911/sys/src/9/port/sturp.c:101,107
1990/0227		/* * predeclared */
1990/0911		static void urpreset(void);
1990/0312		static void urpciput(Queue*, Block*); static void urpiput(Queue*, Block*); static void urpoput(Queue*, Block*);
1990/0814/sys/src/9/port/sturp.c:119,130 – 1990/0911/sys/src/9/port/sturp.c:120,130
1990/0725		static void urptimer(Alarm*);
1990/0629		static void urpvomit(char*, Urp*);
1990/0227
1990/0312		Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp" };
1990/0911		Qinfo urpinfo = { urpciput, urpoput, urpopen, urpclose, "urp", urpreset };
1990/0227
1990/0717		void
1990/0911		static void
1990/0717		urpreset(void) { newqinfo(&urpinfo); urp = (Urp *)ialloc(conf.nurp*sizeof(Urp), 0);
1990/0728		alarm(500, urptimer, 0);
1990/0717		}
1990/0911/sys/src/9/port/sturp.c:653,659 – 1990/0930/sys/src/9/port/sturp.c:653,659 (short | long)
1990/0629		*/ up->rexmit = 0; up->next = up->unacked; } else if(up->unacked!=up->next && NOW>up->timer){
1990/0930		} else if(up->unechoed!=up->next && NOW>up->timer){
1990/0629		/* * if a retransmit time has elapsed since a transmit, * send an ENQ
1990/0911/sys/src/9/port/sturp.c:944,950 – 1990/0930/sys/src/9/port/sturp.c:944,950
1990/0804		return (up->state&INITING) ? NOW>up->timer /* time to INIT1 */ : ((up->unacked!=up->next && NOW>up->timer) /* time to ENQ */
1990/0930		: ((up->unechoed!=up->next && NOW>up->timer) /* time to ENQ */
1990/0814		|| WINDOW(up)>0 && up->next!=up->nxb
1990/0804		|| (!QFULL(up->rq->next) && up->iseq!=(up->lastecho&7))); /* time to ECHO */
1990/0312		}
Too many diffs (26 > 25). Stopping.