il.c

这是一个同样来自贝尔实验室的和UNIX有着渊源的操作系统, 其简洁的设计和实现易于我们学习和理解

#include	"u.h"
#include	"../port/lib.h"
#include	"mem.h"
#include	"dat.h"
#include	"fns.h"
#include	"../port/error.h"

#include	"ip.h"

enum				/* Connection state */
{
	Ilclosed,
	Ilsyncer,
	Ilsyncee,
	Ilestablished,
	Illistening,
	Ilclosing,
	Ilopening,		/* only for file server */
};

char	*ilstates[] = 
{ 
	"Closed",
	"Syncer",
	"Syncee",
	"Established",
	"Listen",
	"Closing",
	"Opening",		/* only for file server */
};

enum				/* Packet types */
{
	Ilsync,
	Ildata,
	Ildataquery,
	Ilack,
	Ilquery,
	Ilstate,
	Ilclose,
};

char	*iltype[] = 
{	
	"sync",
	"data",
	"dataquery",
	"ack",
	"query",
	"state",
	"close" 
};

enum
{
	Seconds		= 1000,
	Iltickms 	= 50,		/* time base */
	AckDelay	= 2*Iltickms,	/* max time twixt message rcvd & ack sent */
	MaxTimeout 	= 30*Seconds,	/* max time between rexmit */
	QueryTime	= 10*Seconds,	/* time between subsequent queries */
	DeathTime	= 30*QueryTime,

	MaxRexmit 	= 16,		/* max retransmissions before hangup */
	Defaultwin	= 20,

	LogAGain	= 3,
	AGain		= 1<<LogAGain,
	LogDGain	= 2,
	DGain		= 1<<LogDGain,

	DefByteRate	= 100,		/* assume a megabit link */
	DefRtt		= 50,		/* cross country on a great day */

	Maxrq		= 64*1024,
};

enum
{
	Nqt=	8,
};

typedef struct Ilcb Ilcb;
struct Ilcb			/* Control block */
{
	int	state;		/* Connection state */
	Conv	*conv;
	QLock	ackq;		/* Unacknowledged queue */
	Block	*unacked;
	Block	*unackedtail;
	ulong	unackedbytes;
	QLock	outo;		/* Out of order packet queue */
	Block	*outoforder;
	ulong	next;		/* Id of next to send */
	ulong	recvd;		/* Last packet received */
	ulong	acksent;	/* Last packet acked */
	ulong	start;		/* Local start id */
	ulong	rstart;		/* Remote start id */
	int	window;		/* Maximum receive window */
	int	rxquery;	/* number of queries on this connection */
	int	rxtot;		/* number of retransmits on this connection */
	int	rexmit;		/* number of retransmits of *unacked */
	ulong	qt[Nqt+1];	/* state table for query messages */
	int	qtx;		/* ... index into qt */

	/* if set, fasttimeout causes a connection request to terminate after 4*Iltickms */
	int	fasttimeout;

	/* timers */
	ulong	lastxmit;	/* time of last xmit */
	ulong	lastrecv;	/* time of last recv */
	ulong	timeout;	/* retransmission time for *unacked */
	ulong	acktime;	/* time to send next ack */
	ulong	querytime;	/* time to send next query */

	/* adaptive measurements */
	int	delay;		/* Average of the fixed rtt delay */
	int	rate;		/* Average uchar rate */
	int	mdev;		/* Mean deviation of rtt */
	int	maxrtt;		/* largest rtt seen */
	ulong	rttack;		/* The ack we are waiting for */
	int	rttlen;		/* Length of rttack packet */
	uvlong	rttstart;	/* Time we issued rttack packet */
};

enum
{
	IL_IPSIZE 	= 20,
	IL_HDRSIZE	= 18,	
	IL_LISTEN	= 0,
	IL_CONNECT	= 1,
	IP_ILPROTO	= 40,
};

typedef struct Ilhdr Ilhdr;
struct Ilhdr
{
	uchar	vihl;		/* Version and header length */
	uchar	tos;		/* Type of service */
	uchar	length[2];	/* packet length */
	uchar	id[2];		/* Identification */
	uchar	frag[2];	/* Fragment information */
	uchar	ttl;		/* Time to live */
	uchar	proto;		/* Protocol */
	uchar	cksum[2];	/* Header checksum */
	uchar	src[4];		/* Ip source */
	uchar	dst[4];		/* Ip destination */
	uchar	ilsum[2];	/* Checksum including header */
	uchar	illen[2];	/* Packet length */
	uchar	iltype;		/* Packet type */
	uchar	ilspec;		/* Special */
	uchar	ilsrc[2];	/* Src port */
	uchar	ildst[2];	/* Dst port */
	uchar	ilid[4];	/* Sequence id */
	uchar	ilack[4];	/* Acked sequence */
};

enum
{
	InMsgs,
	OutMsgs,
	CsumErrs,		/* checksum errors */
	HlenErrs,		/* header length error */
	LenErrs,		/* short packet */
	OutOfOrder,		/* out of order */
	Retrans,		/* retransmissions */
	DupMsg,
	DupBytes,
	DroppedMsgs,

	Nstats,
};

static char *statnames[] =
{
[InMsgs]	"InMsgs",
[OutMsgs]	"OutMsgs",
[CsumErrs]	"CsumErrs",
[HlenErrs]	"HlenErr",
[LenErrs]	"LenErrs",
[OutOfOrder]	"OutOfOrder",
[Retrans]	"Retrans",
[DupMsg]	"DupMsg",
[DupBytes]	"DupBytes",
[DroppedMsgs]	"DroppedMsgs",
};

typedef struct Ilpriv Ilpriv;
struct Ilpriv
{
	Ipht	ht;

	ulong	stats[Nstats];

	ulong	csumerr;		/* checksum errors */
	ulong	hlenerr;		/* header length error */
	ulong	lenerr;			/* short packet */
	ulong	order;			/* out of order */
	ulong	rexmit;			/* retransmissions */
	ulong	dup;
	ulong	dupb;

	/* keeping track of the ack kproc */
	int	ackprocstarted;
	QLock	apl;
};

/* state for query/dataquery messages */


void	ilrcvmsg(Conv*, Block*);
void	ilsendctl(Conv*, Ilhdr*, int, ulong, ulong, int);
void	ilackq(Ilcb*, Block*);
void	ilprocess(Conv*, Ilhdr*, Block*);
void	ilpullup(Conv*);
void	ilhangup(Conv*, char*);
void	ilfreeq(Ilcb*);
void	ilrexmit(Ilcb*);
void	ilbackoff(Ilcb*);
void	ilsettimeout(Ilcb*);
char*	ilstart(Conv*, int, int);
void	ilackproc(void*);
void	iloutoforder(Conv*, Ilhdr*, Block*);
void	iliput(Proto*, Ipifc*, Block*);
void	iladvise(Proto*, Block*, char*);
int	ilnextqt(Ilcb*);
void	ilcbinit(Ilcb*);
int	later(ulong, ulong, char*);
void	ilreject(Fs*, Ilhdr*);
void	illocalclose(Conv *c);
	int 	ilcksum = 1;
static 	int 	initseq = 25001;
static	ulong	scalediv, scalemul;
static	char	*etime = "connection timed out";

static char*
ilconnect(Conv *c, char **argv, int argc)
{
	char *e, *p;
	int fast;

	/* huge hack to quickly try an il connection */
	fast = 0;
	if(argc > 1){
		p = strstr(argv[1], "!fasttimeout");
		if(p != nil){
			*p = 0;
			fast = 1;
		}
	}

	e = Fsstdconnect(c, argv, argc);
	if(e != nil)
		return e;
	return ilstart(c, IL_CONNECT, fast);
}

static int
ilstate(Conv *c, char *state, int n)
{
	Ilcb *ic;

	ic = (Ilcb*)(c->ptcl);
	return snprint(state, n, "%s qin %d qout %d del %5.5d Br %5.5d md %5.5d una %5.5lud rex %5.5d rxq %5.5d max %5.5d",
		ilstates[ic->state],
		c->rq ? qlen(c->rq) : 0,
		c->wq ? qlen(c->wq) : 0,
		ic->delay>>LogAGain, ic->rate>>LogAGain, ic->mdev>>LogDGain,
		ic->unackedbytes, ic->rxtot, ic->rxquery, ic->maxrtt);
}

static int
ilinuse(Conv *c)
{
	Ilcb *ic;

	ic = (Ilcb*)(c->ptcl);
	return ic->state != Ilclosed;

}

/* called with c locked */
static char*
ilannounce(Conv *c, char **argv, int argc)
{
	char *e;

	e = Fsstdannounce(c, argv, argc);
	if(e != nil)
		return e;
	e = ilstart(c, IL_LISTEN, 0);
	if(e != nil)
		return e;
	Fsconnected(c, nil);

	return nil;
}

void
illocalclose(Conv *c)
{
	Ilcb *ic;
	Ilpriv *ipriv;

	ipriv = c->p->priv;
	ic = (Ilcb*)c->ptcl;
	ic->state = Ilclosed;
	iphtrem(&ipriv->ht, c);
	ipmove(c->laddr, IPnoaddr);
	c->lport = 0;
}

static void
ilclose(Conv *c)
{
	Ilcb *ic;

	ic = (Ilcb*)c->ptcl;

	qclose(c->rq);
	qclose(c->wq);
	qclose(c->eq);

	switch(ic->state) {
	case Ilclosing:
	case Ilclosed:
		break;
	case Ilsyncer:
	case Ilsyncee:
	case Ilestablished:
		ic->state = Ilclosing;
		ilsettimeout(ic);
		ilsendctl(c, nil, Ilclose, ic->next, ic->recvd, 0);
		break;
	case Illistening:
		illocalclose(c);
		break;
	}
	ilfreeq(ic);
}

void
ilkick(void *x, Block *bp)
{
	Conv *c = x;
	Ilhdr *ih;
	Ilcb *ic;
	int dlen;
	ulong id, ack;
	Fs *f;
	Ilpriv *priv;

	f = c->p->f;
	priv = c->p->priv;
	ic = (Ilcb*)c->ptcl;

	if(bp == nil)
		return;

	switch(ic->state) {
	case Ilclosed:
	case Illistening:
	case Ilclosing:
		freeblist(bp);
		qhangup(c->rq, nil);
		return;
	}

	dlen = blocklen(bp);

	/* Make space to fit il & ip */
	bp = padblock(bp, IL_IPSIZE+IL_HDRSIZE);
	ih = (Ilhdr *)(bp->rp);
	ih->vihl = IP_VER4;

	/* Ip fields */
	ih->frag[0] = 0;
	ih->frag[1] = 0;
	v6tov4(ih->dst, c->raddr);
	v6tov4(ih->src, c->laddr);
	ih->proto = IP_ILPROTO;

	/* Il fields */
	hnputs(ih->illen, dlen+IL_HDRSIZE);
	hnputs(ih->ilsrc, c->lport);
	hnputs(ih->ildst, c->rport);

	qlock(&ic->ackq);
	id = ic->next++;
	hnputl(ih->ilid, id);
	ack = ic->recvd;
	hnputl(ih->ilack, ack);
	ic->acksent = ack;
	ic->acktime = NOW + AckDelay;
	ih->iltype = Ildata;
	ih->ilspec = 0;
	ih->ilsum[0] = 0;
	ih->ilsum[1] = 0;

	/* Checksum of ilheader plus data (not ip & no pseudo header) */
	if(ilcksum)
		hnputs(ih->ilsum, ptclcsum(bp, IL_IPSIZE, dlen+IL_HDRSIZE));

	ilackq(ic, bp);
	qunlock(&ic->ackq);

	/* Start the round trip timer for this packet if the timer is free */
	if(ic->rttack == 0) {
		ic->rttack = id;
		ic->rttstart = fastticks(nil);
		ic->rttlen = dlen + IL_IPSIZE + IL_HDRSIZE;
	}

	if(later(NOW, ic->timeout, nil))
		ilsettimeout(ic);
	ipoput4(f, bp, 0, c->ttl, c->tos, c);
	priv->stats[OutMsgs]++;
}

static void
ilcreate(Conv *c)
{
	c->rq = qopen(Maxrq, 0, 0, c);
	c->wq = qbypass(ilkick, c);
}

int
ilxstats(Proto *il, char *buf, int len)
{
	Ilpriv *priv;
	char *p, *e;
	int i;

	priv = il->priv;
	p = buf;
	e = p+len;
	for(i = 0; i < Nstats; i++)
		p = seprint(p, e, "%s: %lud\n", statnames[i], priv->stats[i]);
	return p - buf;
}

void
ilackq(Ilcb *ic, Block *bp)
{
	Block *np;
	int n;

	n = blocklen(bp);

	/* Enqueue a copy on the unacked queue in case this one gets lost */
	np = copyblock(bp, n);
	if(ic->unacked)
		ic->unackedtail->list = np;
	else
		ic->unacked = np;
	ic->unackedtail = np;
	np->list = nil;
	ic->unackedbytes += n;
}

static
void
ilrttcalc(Ilcb *ic, Block *bp)
{
	int rtt, tt, pt, delay, rate;

	rtt = fastticks(nil) - ic->rttstart;
	rtt = (rtt*scalemul)/scalediv;
	delay = ic->delay;
	rate = ic->rate;

	/* Guard against zero wrap */
	if(rtt > 120000 || rtt < 0)
		return;

	/* this block had to be transmitted after the one acked so count its size */
	ic->rttlen += blocklen(bp)  + IL_IPSIZE + IL_HDRSIZE;

	if(ic->rttlen < 256){
		/* guess fixed delay as rtt of small packets */
		delay += rtt - (delay>>LogAGain);
		if(delay < AGain)
			delay = AGain;
		ic->delay = delay;
	} else {
		/* if packet took longer than avg rtt delay, recalc rate */
		tt = rtt - (delay>>LogAGain);
		if(tt > 0){
			rate += ic->rttlen/tt - (rate>>LogAGain);
			if(rate < AGain)
				rate = AGain;
			ic->rate = rate;
		}
	}

	/* mdev */
	pt = ic->rttlen/(rate>>LogAGain) + (delay>>LogAGain);
	ic->mdev += abs(rtt-pt) - (ic->mdev>>LogDGain);

	if(rtt > ic->maxrtt)
		ic->maxrtt = rtt;
}

void
ilackto(Ilcb *ic, ulong ackto, Block *bp)
{
	Ilhdr *h;
	ulong id;

	if(ic->rttack == ackto)
		ilrttcalc(ic, bp);

	/* Cancel if we've passed the packet we were interested in */
	if(ic->rttack <= ackto)
		ic->rttack = 0;

	qlock(&ic->ackq);
	while(ic->unacked) {
		h = (Ilhdr *)ic->unacked->rp;
		id = nhgetl(h->ilid);
		if(ackto < id)
			break;

		bp = ic->unacked;
		ic->unacked = bp->list;
		bp->list = nil;
		ic->unackedbytes -= blocklen(bp);
		freeblist(bp);
		ic->rexmit = 0;
		ilsettimeout(ic);
	}
	qunlock(&ic->ackq);
}

void
iliput(Proto *il, Ipifc*, Block *bp)
{
	char *st;
	Ilcb *ic;
	Ilhdr *ih;
	uchar raddr[IPaddrlen];
	uchar laddr[IPaddrlen];
	ushort sp, dp, csum;
	int plen, illen;
	Conv *new, *s;
	Ilpriv *ipriv;

	ipriv = il->priv;

	ih = (Ilhdr *)bp->rp;
	plen = blocklen(bp);
	if(plen < IL_IPSIZE+IL_HDRSIZE){
		netlog(il->f, Logil, "il: hlenerr\n");
		ipriv->stats[HlenErrs]++;
		goto raise;
	}

	illen = nhgets(ih->illen);
	if(illen+IL_IPSIZE > plen){
		netlog(il->f, Logil, "il: lenerr\n");
		ipriv->stats[LenErrs]++;
		goto raise;
	}

	sp = nhgets(ih->ildst);
	dp = nhgets(ih->ilsrc);
	v4tov6(raddr, ih->src);
	v4tov6(laddr, ih->dst);

	if((csum = ptclcsum(bp, IL_IPSIZE, illen)) != 0) {
		if(ih->iltype > Ilclose)
			st = "?";
		else
			st = iltype[ih->iltype];
		ipriv->stats[CsumErrs]++;
		netlog(il->f, Logil, "il: cksum %ux %ux, pkt(%s id %lud ack %lud %I/%d->%d)\n",
			csum, st, nhgetl(ih->ilid), nhgetl(ih->ilack), raddr, sp, dp);
		goto raise;
	}

	qlock(il);
	s = iphtlook(&ipriv->ht, raddr, dp, laddr, sp);
	if(s == nil){
		if(ih->iltype == Ilsync)
			ilreject(il->f, ih);		/* no listener */
		qunlock(il);
		goto raise;
	}

	ic = (Ilcb*)s->ptcl;
	if(ic->state == Illistening){
		if(ih->iltype != Ilsync){
			qunlock(il);
			if(ih->iltype > Ilclose)
				st = "?";
			else
				st = iltype[ih->iltype];
			ilreject(il->f, ih);		/* no channel and not sync */
			netlog(il->f, Logil, "il: no channel, pkt(%s id %lud ack %lud %I/%ud->%ud)\n",
				st, nhgetl(ih->ilid), nhgetl(ih->ilack), raddr, sp, dp); 
			goto raise;
		}

		new = Fsnewcall(s, raddr, dp, laddr, sp, V4);
		if(new == nil){
			qunlock(il);
			netlog(il->f, Logil, "il: bad newcall %I/%ud->%ud\n", raddr, sp, dp);
			ilsendctl(s, ih, Ilclose, 0, nhgetl(ih->ilid), 0);
			goto raise;
		}
		s = new;

		ic = (Ilcb*)s->ptcl;
	
		ic->conv = s;
		ic->state = Ilsyncee;
		ilcbinit(ic);
		ic->rstart = nhgetl(ih->ilid);
		iphtadd(&ipriv->ht, s);
	}

	qlock(s);
	qunlock(il);
	if(waserror()){
		qunlock(s);
		nexterror();
	}
	ilprocess(s, ih, bp);
	qunlock(s);
	poperror();
	return;
raise:
	freeblist(bp);
}

void
_ilprocess(Conv *s, Ilhdr *h, Block *bp)
{
	Ilcb *ic;
	ulong id, ack;
	Ilpriv *priv;

	id = nhgetl(h->ilid);
	ack = nhgetl(h->ilack);

	ic = (Ilcb*)s->ptcl;

	ic->lastrecv = NOW;
	ic->querytime = NOW + QueryTime;
	priv = s->p->priv;
	priv->stats[InMsgs]++;

	switch(ic->state) {
	default:
		netlog(s->p->f, Logil, "il: unknown state %d\n", ic->state);
	case Ilclosed:
		freeblist(bp);
		break;
	case Ilsyncer:
		switch(h->iltype) {
		default:
			break;
		case Ilsync:
			if(ack != ic->start)
				ilhangup(s, "connection rejected");
			else {
				ic->recvd = id;
				ic->rstart = id;
				ilsendctl(s, nil, Ilack, ic->next, ic->recvd, 0);
				ic->state = Ilestablished;
				ic->fasttimeout = 0;
				ic->rexmit = 0;
				Fsconnected(s, nil);
				ilpullup(s);
			}
			break;
		case Ilclose:
			if(ack == ic->start)
				ilhangup(s, "connection rejected");
			break;
		}
		freeblist(bp);
		break;
	case Ilsyncee:
		switch(h->iltype) {
		default:
			break;
		case Ilsync:
			if(id != ic->rstart || ack != 0){
				illocalclose(s);
			} else {
				ic->recvd = id;
				ilsendctl(s, nil, Ilsync, ic->start, ic->recvd, 0);
			}
			break;
		case Ilack:
			if(ack == ic->start) {
				ic->state = Ilestablished;
				ic->fasttimeout = 0;
				ic->rexmit = 0;
				ilpullup(s);
			}
			break;
		case Ildata:
			if(ack == ic->start) {
				ic->state = Ilestablished;