rudp.c

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

	qunlock(rudp);

	if(reliput(c, bp, raddr, rport) < 0){
		qunlock(ucb);
		freeb(bp);
		return;
	}

	/*
	 * Trim the packet down to data size
	 */

	len -= (UDP_RHDRSIZE-UDP_PHDRSIZE);
	bp = trimblock(bp, UDP_IPHDR+UDP_RHDRSIZE, len);
	if(bp == nil) {
		netlog(f, Logrudp, "rudp: len err %I.%d -> %I.%d\n", 
			raddr, rport, laddr, lport);
		DPRINT("rudp: len err %I.%d -> %I.%d\n", 
			raddr, rport, laddr, lport);
		upriv->lenerr++;
		return;
	}

	netlog(f, Logrudpmsg, "rudp: %I.%d -> %I.%d l %d\n", 
		raddr, rport, laddr, lport, len);

	switch(ucb->headers){
	case 7:
		/* pass the src address */
		bp = padblock(bp, UDP_USEAD7);
		p = bp->rp;
		ipmove(p, raddr); p += IPaddrlen;
		ipmove(p, laddr); p += IPaddrlen;
		ipmove(p, ifc->lifc->local); p += IPaddrlen;
		hnputs(p, rport); p += 2;
		hnputs(p, lport);
		break;
	case 6:
		/* pass the src address */
		bp = padblock(bp, UDP_USEAD6);
		p = bp->rp;
		ipmove(p, raddr); p += IPaddrlen;
		ipmove(p, ipforme(f, laddr)==Runi ? laddr : ifc->lifc->local); p += IPaddrlen;
		hnputs(p, rport); p += 2;
		hnputs(p, lport);
		break;
	default:
		/* connection oriented rudp */
		if(ipcmp(c->raddr, IPnoaddr) == 0){
			/* save the src address in the conversation */
		 	ipmove(c->raddr, raddr);
			c->rport = rport;

			/* reply with the same ip address (if not broadcast) */
			if(ipforme(f, laddr) == Runi)
				ipmove(c->laddr, laddr);
			else
				v4tov6(c->laddr, ifc->lifc->local);
		}
		break;
	}
	if(bp->next)
		bp = concatblock(bp);

	if(qfull(c->rq)) {
		netlog(f, Logrudp, "rudp: qfull %I.%d -> %I.%d\n", raddr, rport,
			laddr, lport);
		freeblist(bp);
	}
	else
		qpass(c->rq, bp);
	
	qunlock(ucb);
}

static char *rudpunknown = "unknown rudp ctl request";

char*
rudpctl(Conv *c, char **f, int n)
{
	Rudpcb *ucb;
	uchar ip[IPaddrlen];
	int x;

	ucb = (Rudpcb*)c->ptcl;
	if(n < 1)
		return rudpunknown;

	if(strcmp(f[0], "headers++4") == 0){
		ucb->headers = 7;
		return nil;
	} else if(strcmp(f[0], "headers") == 0){
		ucb->headers = 6;
		return nil;
	} else if(strcmp(f[0], "hangup") == 0){
		if(n < 3)
			return "bad syntax";
		parseip(ip, f[1]);
		x = atoi(f[2]);
		qlock(ucb);
		relforget(c, ip, x, 1);
		qunlock(ucb);
		return nil;
	} else if(strcmp(f[0], "randdrop") == 0){
		x = 10;		/* default is 10% */
		if(n > 1)
			x = atoi(f[1]);
		if(x > 100 || x < 0)
			return "illegal rudp drop rate";
		ucb->randdrop = x;
		return nil;
	}
	return rudpunknown;
}

void
rudpadvise(Proto *rudp, Block *bp, char *msg)
{
	Udphdr *h;
	uchar source[IPaddrlen], dest[IPaddrlen];
	ushort psource, pdest;
	Conv *s, **p;

	h = (Udphdr*)(bp->rp);

	v4tov6(dest, h->udpdst);
	v4tov6(source, h->udpsrc);
	psource = nhgets(h->udpsport);
	pdest = nhgets(h->udpdport);

	/* Look for a connection */
	for(p = rudp->conv; *p; p++) {
		s = *p;
		if(s->rport == pdest)
		if(s->lport == psource)
		if(ipcmp(s->raddr, dest) == 0)
		if(ipcmp(s->laddr, source) == 0){
			qhangup(s->rq, msg);
			qhangup(s->wq, msg);
			break;
		}
	}
	freeblist(bp);
}

int
rudpstats(Proto *rudp, char *buf, int len)
{
	Rudppriv *upriv;

	upriv = rudp->priv;
	return snprint(buf, len, "%lud %lud %lud %lud %lud %lud\n",
		upriv->ustats.rudpInDatagrams,
		upriv->ustats.rudpNoPorts,
		upriv->ustats.rudpInErrors,
		upriv->ustats.rudpOutDatagrams,
		upriv->rxmits,
		upriv->orders);
}

void
rudpinit(Fs *fs)
{

	Proto *rudp;

	rudp = smalloc(sizeof(Proto));
	rudp->priv = smalloc(sizeof(Rudppriv));
	rudp->name = "rudp";
	rudp->connect = rudpconnect;
	rudp->announce = rudpannounce;
	rudp->ctl = rudpctl;
	rudp->state = rudpstate;
	rudp->create = rudpcreate;
	rudp->close = rudpclose;
	rudp->rcv = rudpiput;
	rudp->advise = rudpadvise;
	rudp->stats = rudpstats;
	rudp->ipproto = IP_UDPPROTO;
	rudp->nc = 16;
	rudp->ptclsize = sizeof(Rudpcb);

	Fsproto(fs, rudp);
}

/*********************************************/
/* Here starts the reliable helper functions */
/*********************************************/
/*
 *  Enqueue a copy of an unacked block for possible retransmissions
 */
void
relackq(Reliable *r, Block *bp)
{
	Block *np;

	np = copyblock(bp, blocklen(bp));
	if(r->unacked)
		r->unackedtail->list = np;
	else {
		/* restart timer */
		r->timeout = 0;
		r->xmits = 1;
		r->unacked = np;
	}
	r->unackedtail = np;
	np->list = nil;
}

/*
 *  retransmit unacked blocks
 */
void
relackproc(void *a)
{
	Rudpcb *ucb;
	Proto *rudp;
	Reliable *r;
	Conv **s, *c;

	rudp = (Proto *)a;

loop:
	tsleep(&up->sleep, return0, 0, Rudptickms);

	for(s = rudp->conv; *s; s++) {
		c = *s;
		ucb = (Rudpcb*)c->ptcl;
		qlock(ucb);

		for(r = ucb->r; r; r = r->next) {
			if(r->unacked != nil){
				r->timeout += Rudptickms;
				if(r->timeout > Rudprxms*r->xmits)
					relrexmit(c, r);
			}
			if(r->acksent != r->rcvseq)
				relsendack(c, r, 0);
		}
		qunlock(ucb);
	}
	goto loop;
}

/*
 *  get the state record for a conversation
 */
Reliable*
relstate(Rudpcb *ucb, uchar *addr, ushort port, char *from)
{
	Reliable *r, **l;

	l = &ucb->r;
	for(r = *l; r; r = *l){
		if(memcmp(addr, r->addr, IPaddrlen) == 0 && 
		    port == r->port)
			break;
		l = &r->next;
	}

	/* no state for this addr/port, create some */
	if(r == nil){
		while(generation == 0)
			generation = rand();

		DPRINT("from %s new state %lud for %I!%ud\n", 
		        from, generation, addr, port);

		r = smalloc(sizeof(Reliable));
		memmove(r->addr, addr, IPaddrlen);
		r->port = port;
		r->unacked = 0;
		if(generation == Hangupgen)
			generation++;
		r->sndgen = generation++;
		r->sndseq = 0;
		r->ackrcvd = 0;
		r->rcvgen = 0;
		r->rcvseq = 0;
		r->acksent = 0;
		r->xmits = 0;
		r->timeout = 0;
		r->ref = 0;
		incref(r);	/* one reference for being in the list */

		*l = r;
	}

	incref(r);
	return r;
}

void
relput(Reliable *r)
{
	if(decref(r) == 0)
		free(r);
}

/*
 *  forget a Reliable state
 */
void
relforget(Conv *c, uchar *ip, int port, int originator)
{
	Rudpcb *ucb;
	Reliable *r, **l;

	ucb = (Rudpcb*)c->ptcl;

	l = &ucb->r;
	for(r = *l; r; r = *l){
		if(ipcmp(ip, r->addr) == 0 && port == r->port){
			*l = r->next;
			if(originator)
				relsendack(c, r, 1);
			relhangup(c, r);
			relput(r);	/* remove from the list */
			break;
		}
		l = &r->next;
	}
}

/* 
 *  process a rcvd reliable packet. return -1 if not to be passed to user process,
 *  0 therwise.
 *
 *  called with ucb locked.
 */
int
reliput(Conv *c, Block *bp, uchar *addr, ushort port)
{
	Block *nbp;
	Rudpcb *ucb;
	Rudppriv *upriv;
	Udphdr *uh;
	Reliable *r;
	Rudphdr *rh;
	ulong seq, ack, sgen, agen, ackreal;
	int rv = -1;

	/* get fields */
	uh = (Udphdr*)(bp->rp);
	rh = (Rudphdr*)uh;
	seq = nhgetl(rh->relseq);
	sgen = nhgetl(rh->relsgen);
	ack = nhgetl(rh->relack);
	agen = nhgetl(rh->relagen);

	upriv = c->p->priv;
	ucb = (Rudpcb*)c->ptcl;
	r = relstate(ucb, addr, port, "input");

	DPRINT("rcvd %lud/%lud, %lud/%lud, r->sndgen = %lud\n", 
		seq, sgen, ack, agen, r->sndgen);

	/* if acking an incorrect generation, ignore */
	if(ack && agen != r->sndgen)
		goto out;

	/* Look for a hangup */
	if(sgen == Hangupgen) {
		if(agen == r->sndgen)
			relforget(c, addr, port, 0);
		goto out;
	}

	/* make sure we're not talking to a new remote side */
	if(r->rcvgen != sgen){
		if(seq != 0 && seq != 1)
			goto out;

		/* new connection */
		if(r->rcvgen != 0){
			DPRINT("new con r->rcvgen = %lud, sgen = %lud\n", r->rcvgen, sgen);
			relhangup(c, r);
		}
		r->rcvgen = sgen;
	}

	/* dequeue acked packets */
	if(ack && agen == r->sndgen){
		ackreal = 0;
		while(r->unacked != nil && INSEQ(ack, r->ackrcvd, r->sndseq)){
			nbp = r->unacked;
			r->unacked = nbp->list;
			DPRINT("%lud/%lud acked, r->sndgen = %lud\n", 
			       ack, agen, r->sndgen);
			freeb(nbp);
			r->ackrcvd = NEXTSEQ(r->ackrcvd);
			ackreal = 1;
		}

		/* flow control */
		if(UNACKED(r) < Maxunacked/8 && r->blocked)
			wakeup(&r->vous);

		/*
		 *  retransmit next packet if the acked packet
		 *  was transmitted more than once
		 */
		if(ackreal && r->unacked != nil){
			r->timeout = 0;
			if(r->xmits > 1){
				r->xmits = 1;
				relrexmit(c, r);
			}
		}
		
	}

	/* no message or input queue full */
	if(seq == 0 || qfull(c->rq))
		goto out;

	/* refuse out of order delivery */
	if(seq != NEXTSEQ(r->rcvseq)){
		relsendack(c, r, 0);	/* tell him we got it already */
		upriv->orders++;
		DPRINT("out of sequence %lud not %lud\n", seq, NEXTSEQ(r->rcvseq));
		goto out;
	}
	r->rcvseq = seq;

	rv = 0;
out:
	relput(r);
	return rv;
}

void
relsendack(Conv *c, Reliable *r, int hangup)
{
	Udphdr *uh;
	Block *bp;
	Rudphdr *rh;
	int ptcllen;
	Fs *f;

	bp = allocb(UDP_IPHDR + UDP_RHDRSIZE);
	if(bp == nil)
		return;
	bp->wp += UDP_IPHDR + UDP_RHDRSIZE;
	f = c->p->f;
	uh = (Udphdr *)(bp->rp);
	uh->vihl = IP_VER4;
	rh = (Rudphdr*)uh;

	ptcllen = (UDP_RHDRSIZE-UDP_PHDRSIZE);
	uh->Unused = 0;
	uh->udpproto = IP_UDPPROTO;
	uh->frag[0] = 0;
	uh->frag[1] = 0;
	hnputs(uh->udpplen, ptcllen);

	v6tov4(uh->udpdst, r->addr);
	hnputs(uh->udpdport, r->port);
	hnputs(uh->udpsport, c->lport);
	if(ipcmp(c->laddr, IPnoaddr) == 0)
		findlocalip(f, c->laddr, c->raddr);
	v6tov4(uh->udpsrc, c->laddr);
	hnputs(uh->udplen, ptcllen);

	if(hangup)
		hnputl(rh->relsgen, Hangupgen);
	else
		hnputl(rh->relsgen, r->sndgen);
	hnputl(rh->relseq, 0);
	hnputl(rh->relagen, r->rcvgen);
	hnputl(rh->relack, r->rcvseq);

	if(r->acksent < r->rcvseq)
		r->acksent = r->rcvseq;

	uh->udpcksum[0] = 0;
	uh->udpcksum[1] = 0;
	hnputs(uh->udpcksum, ptclcsum(bp, UDP_IPHDR, UDP_RHDRSIZE));

	DPRINT("sendack: %lud/%lud, %lud/%lud\n", 0L, r->sndgen, r->rcvseq, r->rcvgen);
	doipoput(c, f, bp, 0, c->ttl, c->tos);
}


/*
 *  called with ucb locked (and c locked if user initiated close)
 */
void
relhangup(Conv *c, Reliable *r)
{
	int n;
	Block *bp;
	char hup[ERRMAX];

	n = snprint(hup, sizeof(hup), "hangup %I!%d", r->addr, r->port);
	qproduce(c->eq, hup, n);

	/*
	 *  dump any unacked outgoing messages
	 */
	for(bp = r->unacked; bp != nil; bp = r->unacked){
		r->unacked = bp->list;
		bp->list = nil;
		freeb(bp);
	}

	r->rcvgen = 0;
	r->rcvseq = 0;
	r->acksent = 0;
	if(generation == Hangupgen)
		generation++;
	r->sndgen = generation++;
	r->sndseq = 0;
	r->ackrcvd = 0;
	r->xmits = 0;
	r->timeout = 0;
	wakeup(&r->vous);
}

/*
 *  called with ucb locked
 */
void
relrexmit(Conv *c, Reliable *r)
{
	Rudppriv *upriv;
	Block *np;
	Fs *f;

	upriv = c->p->priv;
	f = c->p->f;
	r->timeout = 0;
	if(r->xmits++ > Rudpmaxxmit){
		relhangup(c, r);
		return;
	}

	upriv->rxmits++;
	np = copyblock(r->unacked, blocklen(r->unacked));
	DPRINT("rxmit r->ackrvcd+1 = %lud\n", r->ackrcvd+1);
	doipoput(c, f, np, 0, c->ttl, c->tos);
}