plan 9 kernel history: overview | file list | diff list

1990/0312/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 | prev | next)
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    
}


source code copyright © 1990-2005 Lucent Technologies; see license
Plan 9 distribution
comments to russ cox (rsc@swtch.com)