il.c

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

#include "all.h"
#include "mem.h"

#include "../ip/ip.h"

#define	DEBUG	if(cons.flags&ilflag)print
#define	msec	(MACHP(0)->ticks * (1000/HZ))

enum
{
	Ilsync		= 0,		/* Packet types */
	Ildata,
	Ildataquery,
	Ilack,
	Ilquery,
	Ilstate,
	Ilclose,

	Ilclosed	= 0,		/* Connection state */
	Ilsyncer,
	Ilsyncee,
	Ilestablished,
	Illistening,
	Ilclosing,
	Ilopening,

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

	MaxRexmit 	= 16,		/* max retransmissions before hangup */
	DefWin		= 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 */
};

static
struct
{
	Lock;
	Queue*	reply;
	Chan*	chan;
} il;

static	void	ilout(void);
static	void	ilprocess(Chan*, Msgbuf*);
static	Chan*	getchan(Ifc*, Ilpkt*, Msgbuf*);
static	void	ilhangup(Chan*, char*, int);
static	void	ilsendctl(Chan*, Ilpkt*, int, ulong, ulong, int);
static	void	ilpullup(Chan*);
static	void	ilackto(Chan*, ulong, Msgbuf*);
static	void	ilrexmit(Ilp*);
static	void	iloutoforder(Chan*, Ilpkt*, Msgbuf*);
static	void	ilfreeq(Chan*);
static	void	ilackq(Chan*, Msgbuf*);
static	void	ilbackoff(Ilp*);
static	void	ilsettimeout(Ilp*);
static	int	ilnextqt(Ilp*);
static	void	iltimer(void);
static void	ilgoaway(Msgbuf*, Ifc*);

static
char*	ilstate[] =
{
	"Closed",
	"Syncer",
	"Syncee",
	"Established",
	"Listening",
	"Closing",
	"Opening",
};
static
char*	iltype[] =
{
	"sync",
	"data",
	"dataquery",
	"ack",
	"query",
	"state",
	"close",
};
static	Rendez	ild;
static	int	ilflag;

static void
ilwhoprint(Chan* cp)
{
	Ilp *ilp;
	Timet t;

	if(cp->type != Devil)
		return;

	ilp = cp->pdata;
	t = MACHP(0)->ticks * (1000/HZ);
	print(" (%d,%d)", ilp->alloc, ilp->state);
	print(" (%ld,%ld,%ld)",
		ilp->timeout-t, ilp->querytime-t,
		ilp->lastrecv-t);
	print(" (%ld,%ld,%ld,%ld)", ilp->rate, ilp->delay,
		ilp->mdev, ilp->unackedbytes);
}

static
void
ilpinit(Ilp *ilp)
{
	ilp->start = (toytime() * 80021) & 0x3fffffffUL;
	ilp->next = ilp->start + 1;
	ilp->rstart = 0;
	ilp->recvd = 0;
	ilp->window = DefWin;
	ilp->unackedbytes = 0;
	ilp->unacked = nil;
	ilp->unackedtail = nil;
	ilp->outoforder = nil;
	ilp->rexmit = 0;

	/* timers */
	ilp->delay = DefRtt<<LogAGain;
	ilp->mdev = DefRtt<<LogDGain;
	ilp->rate = DefByteRate<<LogAGain;
	ilp->querytime = msec + QueryTime;
	ilp->lastrecv = msec;			/* to avoid immediate timeout */
	ilsettimeout(ilp);
}

static
Chan*
getchan(Ifc *ifc, Ilpkt *p, Msgbuf *mb)
{
	Ilp *ilp;
	Chan *cp, *xcp;
	int srcp, dstp;

	srcp = nhgets(p->ilsrc);
	dstp = nhgets(p->ildst);

	lock(&il);
	xcp = 0;
	for(cp = il.chan; cp; cp = ilp->chan) {
		ilp = cp->pdata;
		if(ilp->alloc == 0) {
			xcp = cp;
			continue;
		}
		if(srcp == ilp->srcp)
		if(dstp == ilp->dstp)
		if(memcmp(p->src, ilp->iphis, Pasize) == 0)
		if(memcmp(p->dst, ifc->ipa, Pasize) == 0){
			unlock(&il);
			return cp;
		}
	}

	if(il.reply == 0) {
		il.reply = newqueue(Nqueue);
		userinit(ilout, &il, "ilo");
		userinit(iltimer, &il, "ilt");
		ilflag = flag_install("il", "-- on errors");
	}

	if(dstp != Ilfsport) {
		ilgoaway(mb, ifc);
		unlock(&il);
		DEBUG("open not fsport %I.%d -> %I.%d\n", p->src, srcp, p->dst, dstp);
		return nil;
	}

	if(p->iltype != Ilsync) {
		ilgoaway(mb, ifc);
		unlock(&il);
		DEBUG("open not sync %I.%d -> %I.%d\n", p->src, srcp, p->dst, dstp);
		return nil;
	}

	cp = xcp;
	if(cp == 0) {
		cp = chaninit(Devil, 1, sizeof(Ilp));
		ilp = cp->pdata;
		ilp->chan = il.chan;
		il.chan = cp;
	}


	cp->ifc = ifc;
	ilp = cp->pdata;
	memmove(ilp->iphis, p->src, Pasize);
	memmove(ifc->ipa, p->dst, Pasize);
	ilp->srcp = srcp;
	ilp->dstp = dstp;
	ilp->state = Ilopening;

	ilpinit(ilp);

	memmove(ilp->ipgate, ilp->iphis, Pasize);
	if((nhgetl(ifc->ipa)&ifc->mask) != (nhgetl(p->src)&ifc->mask))
		iproute(ilp->ipgate, p->src, ifc->netgate);

	cp->send = serveq;
	cp->reply = il.reply;
	ilp->reply = ifc->reply;
	cp->protocol = nil;
	cp->msize = 0;
	cp->whotime = 0;
	sprint(cp->whochan, "il!%I!%d", p->src, srcp);
	cp->whoprint = ilwhoprint;
	ilp->alloc = 1;

	unlock(&il);
	return cp;
}

void
ilrecv(Msgbuf *mb, Ifc *ifc)
{
	Ilpkt *ih;
	Chan *cp;
	Ilp *ilp;
	int illen, plen;

	ih = (Ilpkt*)mb->data;

	plen = mb->count;
	if(plen < Ensize+Ipsize+Ilsize)
		goto drop;

	illen = nhgets(ih->illen);
	if(illen+Ilsize > plen)
		goto drop;

	if(ptclcsum((uchar*)ih+(Ensize+Ipsize), illen) != 0) {
		print("il: cksum error %E %I\n", ih->s, ih->src);
		ifc->sumerr++;
		goto drop;
	}

	cp = getchan(ifc, ih, mb);
	if(cp == nil)
		goto drop;
	mb->chan = cp;
	ilp = cp->pdata;

	if(ilp->state == Ilopening) {
		ilp->state = Ilsyncee;
		ilpinit(ilp);
		ilp->rstart = nhgetl(ih->ilid);
		print("il: allocating %s\n", cp->whochan);
	}

	ilprocess(cp, mb);
	return;

drop:
	mbfree(mb);
}

/*
 * process to convert p9 to il/ip
 */
static
void
ilout(void)
{
	Ifc *ifc;
	Msgbuf *mb;
	Ilp *ilp;
	Ilpkt *ih;
	Chan *cp;
	int dlen;
	ulong id, ack;

	for (;;) {
		while ((mb = recv(il.reply, 0)) == nil)
			continue;

		cp = mb->chan;
		ilp = cp->pdata;

		switch(ilp->state) {
		case Ilclosed:
		case Illistening:
		case Ilclosing:
			print("ilout: error\n");
			mbfree(mb);
			continue;
		}

		dlen = mb->count;
		mb->data -= Ensize+Ipsize+Ilsize;    /* make room for header */
		mb->count += Ensize+Ipsize+Ilsize;
		if(mb->data < mb->xdata)
			panic("ilout: no room for header");
		ih = (Ilpkt*)mb->data;

		/*
		 * Ip fields
		 */
		ifc = cp->ifc;
		memmove(ih->src, ifc->ipa, Pasize);
		memmove(ih->dst, ilp->iphis, Pasize);
		ih->proto = Ilproto;

		/*
		 * Il fields
		 */
		hnputs(ih->illen, Ilsize+dlen);
		hnputs(ih->ilsrc, ilp->dstp);
		hnputs(ih->ildst, ilp->srcp);
		id = ilp->next++;
		hnputl(ih->ilid, id);
		ack = ilp->recvd;
		hnputl(ih->ilack, ack);
		ilp->acksent = ack;
		ilp->acktime = msec + AckDelay;
		ih->iltype = Ildata;
		ih->ilspec = 0;
		ih->ilsum[0] = 0;
		ih->ilsum[1] = 0;

		/*
		 * checksum
		 */
		hnputs(ih->ilsum, ptclcsum((uchar*)ih+(Ensize+Ipsize),
			dlen+Ilsize));

		ilackq(cp, mb);

		/*
		 * Start the round trip timer for this packet if the timer
		 * is free.
		 */
		if(ilp->rttack == 0) {
			ilp->rttack = id;
			ilp->rttstart = msec;
			ilp->rttlen = dlen+Ipsize+Ilsize;
		}

		if(ilp->timeout <= msec)
			ilsettimeout(ilp);
		ipsend(mb);
	}
}

static
void
ilackq(Chan *cp, Msgbuf *mb)
{
	Msgbuf *nmb;
	Ilp *ilp;

	/*
	 * Enqueue a copy on the unacked queue in case this one gets lost
	 */
/* botch -- a reference count will save this copy */
	nmb = mballoc(mb->count, cp, Mbil2);
	memmove(nmb->data, mb->data, mb->count);
	nmb->next = 0;

	ilp = cp->pdata;
	lock(ilp);
	if(ilp->unacked)
		ilp->unackedtail->next = nmb;
	else
		ilp->unacked = nmb;
	ilp->unackedtail = nmb;
	ilp->unackedbytes += nmb->count;
	unlock(ilp);
}

static
void
ilprocess(Chan *cp, Msgbuf *mb)
{
	ulong id, ack;
	Ilp* ilp;
	Ilpkt *h;

	ilp = cp->pdata;
	h = (Ilpkt*)mb->data;

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

	ilp->lastrecv = msec;

	switch(ilp->state) {
	default:
		print("il unknown state\n");
	case Ilclosed:
		mbfree(mb);
		break;
	case Ilsyncer:
		switch(h->iltype) {
		default:
			break;
		case Ilsync:
			if(ack != ilp->start) {
				ilp->state = Ilclosed;
				ilhangup(cp, "connection rejected", 1);
			} else {
				ilp->recvd = id;
				ilp->rstart = id;
				ilsendctl(cp, 0, Ilack, ilp->next, ilp->recvd, 0);
				ilp->state = Ilestablished;
				wakeup(&ilp->syn);
				ilpullup(cp);
			}
			break;
		case Ilclose:
			if(ack == ilp->start) {
				ilp->state = Ilclosed;
				ilhangup(cp, "remote close-1", 1);
			}
			break;
		}
		mbfree(mb);
		break;

	case Ilsyncee:
		switch(h->iltype) {
		default:
			break;
		case Ilsync:
			if(id != ilp->rstart || ack != 0)
				ilp->state = Ilclosed;
			else {
				ilp->recvd = id;
				ilsendctl(cp, 0, Ilsync, ilp->start, ilp->recvd, 0);
			}
			break;
		case Ilack:
			if(ack == ilp->start) {
				ilp->state = Ilestablished;
				ilpullup(cp);
			}
			break;
		case Ildata:
			if(ack == ilp->start) {
				ilp->state = Ilestablished;
				goto established;
			}
			break;
		case Ilclose:
			if(id == ilp->next) {
				ilp->state = Ilclosed;
				ilhangup(cp, "remote close-2", 1);
			}
			break;
		}
		mbfree(mb);
		break;

	case Ilestablished:
	established:
		switch(h->iltype) {
		default:
			mbfree(mb);
			break;
		case Ilsync:
			if(id != ilp->rstart) {
				ilp->state = Ilclosed;
				ilhangup(cp, "remote close-3", 1);
			} else
				ilsendctl(cp, 0, Ilack, ilp->next, ilp->rstart, 0);
			mbfree(mb);
			break;
		case Ildata:
			ilackto(cp, ack, mb);
			iloutoforder(cp, h, mb);
			ilpullup(cp);
			break;
		case Ildataquery:
			ilackto(cp, ack, mb);
			iloutoforder(cp, h, mb);
			ilpullup(cp);
			ilsendctl(cp, 0, Ilstate, ilp->next, ilp->recvd, h->ilspec);
			break;
		case Ilack:
			ilackto(cp, ack, mb);
			mbfree(mb);
			break;
		case Ilquery:
			ilackto(cp, ack, mb);
			ilsendctl(cp, 0, Ilstate, ilp->next, ilp->recvd, h->ilspec);
			mbfree(mb);
			break;
		case Ilstate:
			if(ack >= ilp->rttack)
				ilp->rttack = 0;
			ilackto(cp, ack, mb);
			if(h->ilspec > Nqt)
				h->ilspec = 0;
			if(ilp->qt[h->ilspec] > ack){
				ilrexmit(ilp);
				ilsettimeout(ilp);
			}
			mbfree(mb);
			break;
		case Ilclose:
			mbfree(mb);
			if(ack < ilp->start || ack > ilp->next)
				break;
			ilp->recvd = id;
			ilsendctl(cp, 0, Ilclose, ilp->next, ilp->recvd, 0);
			ilp->state = Ilclosing;
			ilfreeq(cp);
			break;
		}
		break;

	case Illistening:
		mbfree(mb);
		break;

	case Ilclosing:
		switch(h->iltype) {
		case Ilclose:
			ilp->recvd = id;
			ilsendctl(cp, 0, Ilclose, ilp->next, ilp->recvd, 0);
			if(ack == ilp->next) {
				ilp->state = Ilclosed;
				ilhangup(cp, "closed", 1);
			}