git: 9front

Info  •  Files  •  Log  •  Branches

merge

--- a/sys/src/9/ip/devip.c

+++ b/sys/src/9/ip/devip.c

@@ -53,9 +53,8 @@

 static char network[] = "network";

-QLock	fslock;

-Fs	*ipfs[Nfs];	/* attached fs's */

-Queue	*qlog;

+static Fs	*ipfs[Nfs];	/* attached fs's */

+static Queue	*qlog;

 extern	void nullmediumlink(void);

 extern	void pktmediumlink(void);

@@ -290,6 +289,7 @@

 static Chan*

 ipattach(char* spec)

 {

+	static QLock lk;

 	Chan *c;

 	ulong dev;

@@ -297,7 +297,7 @@

 	if(dev >= Nfs)

 		error(Enodev);

-	qlock(&fslock);

+	qlock(&lk);

 	if(ipfs[dev] == nil){

 		extern void (*ipprotoinit[])(Fs*);

 		Fs *f;

@@ -310,9 +310,10 @@

 		for(i = 0; ipprotoinit[i]; i++)

 			ipprotoinit[i](f);

 		f->dev = dev;

+		coherence();

 		ipfs[dev] = f;

 	}

-	qunlock(&fslock);

+	qunlock(&lk);

 	c = devattach('I', spec);

 	mkqid(&c->qid, QID(0, 0, Qtopdir), 0, QTDIR);

--- a/sys/src/9/ip/ip.c

+++ b/sys/src/9/ip/ip.c

@@ -146,6 +146,23 @@

 	} else {

 		eh->vihl = IP_VER4|IP_HLEN4;

 		eh->tos = tos;

+

+		/* bypass for loopback */

+		if(r->type & Runi){

+			eh->ttl = ttl;

+			hnputs(eh->length, len);

+			hnputs(eh->id, incref(&ip->id4));

+			eh->frag[0] = 0;

+			eh->frag[1] = 0;

+			eh->cksum[0] = 0;

+			eh->cksum[1] = 0;

+			hnputs(eh->cksum, ipcsum(&eh->vihl));

+			bp->flag |= Bipck;

+			(*loopbackmedium.bwrite)(ifc, concatblock(bp), V4, gate, rh);

+			runlock(ifc);

+			poperror();

+			return 0;

+		}

 	}

 	eh->ttl = ttl;

@@ -160,7 +177,6 @@

 		eh->cksum[0] = 0;

 		eh->cksum[1] = 0;

 		hnputs(eh->cksum, ipcsum(&eh->vihl));

-

 		ipifcoput(ifc, bp, V4, gate, rh);

 		runlock(ifc);

 		poperror();

@@ -244,7 +260,6 @@

 		feh->cksum[0] = 0;

 		feh->cksum[1] = 0;

 		hnputs(feh->cksum, ipcsum(&feh->vihl));

-

 		ipifcoput(ifc, nb, V4, gate, rh);

 		ip->stats[FragCreates]++;

 	}

--- a/sys/src/9/ip/ip.h

+++ b/sys/src/9/ip/ip.h

@@ -391,6 +391,7 @@

 	uchar	mac[MAClen];	/* MAC address */

 	Iplifc	*lifc;		/* logical interfaces on this physical one */

+	Queue	*loopback;

 	ulong	in, out;	/* message statistics */

 	ulong	inerr, outerr;	/* ... */

@@ -719,6 +720,8 @@

  */

 extern Medium	ethermedium;

 extern Medium	nullmedium;

+extern Medium	unboundmedium;	/* same as nullmedium, to prevent unbind while bind or unbind is in progress */

+extern Medium	loopbackmedium;

 extern Medium	pktmedium;

 /*

--- a/sys/src/9/ip/ipifc.c

+++ b/sys/src/9/ip/ipifc.c

@@ -122,9 +122,6 @@

 	return *mp;

 }

-/* same as nullmedium, to prevent unbind while bind or unbind is in progress */

-extern Medium unboundmedium;

-

 /*

  *  attach a device (or pkt driver) to the interface.

  *  called with c locked

@@ -174,6 +171,8 @@

 		wunlock(ifc);

 		nexterror();

 	}

+	if(m != &loopbackmedium)

+		(*loopbackmedium.bind)(ifc, argc, argv);

 	(*m->bind)(ifc, argc, argv);

 	poperror();

@@ -466,6 +465,7 @@

 	c->sq = qopen(QMAX, 0, 0, 0);

 	if(c->rq == nil || c->wq == nil || c->sq == nil)

 		error(Enomem);

+

 	ifc = (Ipifc*)c->ptcl;

 	ifc->conv = c;

 	ifc->m = nil;

--- a/sys/src/9/ip/ipv6.c

+++ b/sys/src/9/ip/ipv6.c

@@ -101,6 +101,16 @@

 		eh->vcf[0] = IP_VER6;

 		eh->vcf[0] |= tos >> 4;

 		eh->vcf[1]  = tos << 4;

+

+		/* bypass for loopback */

+		if(r->type & Runi){

+			eh->ttl = ttl;

+			hnputs(eh->ploadlen, len - IP6HDR);

+			(*loopbackmedium.bwrite)(ifc, concatblock(bp), V6, gate, rh);

+			runlock(ifc);

+			poperror();

+			return 0;

+		}

 	}

 	eh->ttl = ttl;

--- a/sys/src/9/ip/loopbackmedium.c

+++ b/sys/src/9/ip/loopbackmedium.c

@@ -7,112 +7,62 @@

 #include "ip.h"

-enum

-{

-	Maxtu=	64*1024,

-};

-

-typedef struct LB LB;

-struct LB

-{

-	Proc	*readp;

-	Queue	*q;

-	Fs	*f;

-};

-

-static void loopbackread(void *a);

-

 static void

-loopbackbind(Ipifc *ifc, int, char**)

+readloopback(void *x)

 {

-	LB *lb;

+	Ipifc *ifc = (Ipifc*)x;

+	Fs *f = ifc->conv->p->f;

+	Block *bp;

-	lb = smalloc(sizeof(*lb));

-	lb->readp = (void*)-1;

-	lb->f = ifc->conv->p->f;

-	lb->q = qopen(1024*1024, Qmsg, nil, nil);

-	ifc->arg = lb;

-

-	kproc("loopbackread", loopbackread, ifc);

-

-}

-

-static void

-loopbackunbind(Ipifc *ifc)

-{

-	LB *lb = ifc->arg;

-

 	while(waserror())

 		;

-	/* wat for reader to start */

-	while(lb->readp == (void*)-1)

-		tsleep(&up->sleep, return0, 0, 300);

-		

-	if(lb->readp != nil)

-		postnote(lb->readp, 1, "unbind", 0);

-	poperror();

-

-	while(waserror())

-		;

-	/* wait for reader to die */

-	while(lb->readp != nil)

-		tsleep(&up->sleep, return0, 0, 300);

-	poperror();

-

-	/* clean up */

-	qfree(lb->q);

-	free(lb);

-}

-

-static void

-loopbackbwrite(Ipifc *ifc, Block *bp, int, uchar*, Routehint*)

-{

-	LB *lb;

-

-	lb = ifc->arg;

-	if(qpass(lb->q, bp) < 0)

-		ifc->outerr++;

-	ifc->out++;

-}

-

-static void

-loopbackread(void *a)

-{

-	Ipifc *ifc;

-	Block *bp;

-	LB *lb;

-

-	ifc = a;

-	lb = ifc->arg;

-	lb->readp = up;	/* hide identity under a rock for unbind */

-	if(!waserror())

-	while((bp = qbread(lb->q, Maxtu)) != nil){

+	while((bp = qbread(ifc->loopback, IP_MAX)) != nil){

 		rlock(ifc);

 		if(waserror()){

 			runlock(ifc);

 			continue;

 		}

-		ifc->in++;

 		if(ifc->lifc == nil)

 			freeb(bp);

-		else

-			ipiput4(lb->f, ifc, bp);

+		else {

+			ifc->in++;

+			ipiput4(f, ifc, bp);

+		}

 		runlock(ifc);

 		poperror();

 	}

-	lb->readp = nil;

 	pexit("hangup", 1);

 }

+static void

+loopbackbind(Ipifc *ifc, int, char**)

+{

+	if(ifc->loopback != nil)

+		return;

+

+	ifc->loopback = qopen(1024*1024, Qmsg, 0, 0);

+	if(ifc->loopback == nil)

+		error(Enomem);

+	kproc("loopbackread", readloopback, ifc);

+

+}

+

+static void

+loopbackbwrite(Ipifc *ifc, Block *bp, int, uchar*, Routehint*)

+{

+	if(qpass(ifc->loopback, bp) < 0)

+		ifc->outerr++;

+	ifc->out++;

+}

+

 Medium loopbackmedium =

 {

 .hsize=		0,

 .mintu=		0,

-.maxtu=		Maxtu,

+.maxtu=		IP_MAX,

 .maclen=	0,

 .name=		"loopback",

 .bind=		loopbackbind,

-.unbind=	loopbackunbind,

 .bwrite=	loopbackbwrite,

 };

--- a/sys/src/9/ip/tcp.c

+++ b/sys/src/9/ip/tcp.c

@@ -200,6 +200,7 @@

 typedef struct Tcpctl Tcpctl;

 struct Tcpctl

 {

+	Conv	*bypass;		/* The other when bypassing */

 	uchar	state;			/* Connection state */

 	uchar	type;			/* Listening or active connection */

 	uchar	code;			/* Icmp code */

@@ -251,6 +252,7 @@

 	ulong	rttseq;			/* Round trip sequence */

 	int	srtt;			/* Smoothed round trip */

 	int	mdev;			/* Mean deviation of round trip */

+

 	union {

 		Tcp4hdr	tcp4hdr;

 		Tcp6hdr	tcp6hdr;

@@ -412,6 +414,10 @@

 		qclose(s->rq);

 		qclose(s->wq);

 		qclose(s->eq);

+		if(tcb->bypass != nil){

+			tcb->bypass = nil;

+			qsetbypass(s->wq, nil);

+		}

 		break;

 	case Close_wait:		/* Remote closes */

@@ -523,8 +529,14 @@

 	case Syn_sent:

 		localclose(c, nil);

 		break;

-	case Syn_received:

 	case Established:

+		if(tcb->bypass != nil){

+			qhangup(tcb->bypass->rq, nil);

+			localclose(c, nil);

+			break;

+		}

+		/* wet floor */

+	case Syn_received:

 		tcb->flgcnt++;

 		tcb->snd.nxt++;

 		tcpsetstate(c, Finwait1);

@@ -547,9 +559,12 @@

 	qlock(s);

 	switch(tcb->state) {

+	case Established:

+		if(tcb->bypass != nil)

+			break;

+		/* wet floor */

 	case Syn_sent:

 	case Syn_received:

-	case Established:

 	case Close_wait:

 		/*

 		 * Push data

@@ -623,7 +638,7 @@

 	Tcpctl *tcb = (Tcpctl*)s->ptcl;

 	qlock(s);

-	if(tcb->state != Closed){

+	if(tcb->state != Closed && tcb->bypass == nil){

 		tcb->flags |= FORCE;

 		tcpoutput(s);

 	}

@@ -1325,7 +1340,7 @@

 {

 	Tcpctl *tcb = (Tcpctl*)s->ptcl;

-	if(ipcmp(s->raddr, IPnoaddr) != 0 && tcb->state != Closed) {

+	if(ipcmp(s->raddr, IPnoaddr) != 0 && tcb->state != Closed && tcb->bypass == nil) {

 		Block *bp;

 		Tcp seg;

@@ -1565,7 +1580,60 @@

 	tcb->mdev = rtt<<(LOGDGAIN-1);

 }

+static void

+tcpbypass(void *arg, Block *b)

+{

+	Conv *c = (Conv*)arg;

+	Tcpctl *tcb = (Tcpctl*)c->ptcl;

+	Conv *o = tcb->bypass;

+

+	if(o == nil || ((Tcpctl*)o->ptcl)->bypass != c){

+		freeblist(b);

+		return;

+	}

+	qbwrite(o->rq, b);

+}

+

 /*

+ *  splice two local tcp conversations together,

+ *  having the wq directly bypass into the other

+ *  ends rq.

+ *

+ *  both new, old and proto are locked.

+ */

+static void

+tcpsplice(Conv *new, Conv *old)

+{

+	Tcppriv *tpriv;

+	Tcpctl *ntcb, *otcb;

+

+	ntcb = (Tcpctl*)new->ptcl;

+	otcb = (Tcpctl*)old->ptcl;

+

+	if(otcb->state != Established)

+		return;

+

+	ntcb->bypass = old;

+	otcb->bypass = new;

+

+	qsetlimit(new->rq, conf.pipeqsize);

+	qsetlimit(old->rq, conf.pipeqsize);

+	qsetbypass(new->wq, tcpbypass);

+	qsetbypass(old->wq, tcpbypass);

+

+	/* stop timers and drump resequencing queue */

+	tpriv = (Tcppriv*)new->p->priv;

+	tcphalt(tpriv, &ntcb->timer);

+	tcphalt(tpriv, &otcb->timer);

+	tcphalt(tpriv, &ntcb->acktimer);

+	tcphalt(tpriv, &otcb->acktimer);

+	tcphalt(tpriv, &ntcb->katimer);

+	tcphalt(tpriv, &otcb->katimer);

+	dumpreseq(ntcb);

+	dumpreseq(otcb);

+}

+

+/*

  *  come here when we finally get an ACK to our SYN-ACK.

  *  lookup call in limbo.  if found, create a new conversation

  *

@@ -1575,6 +1643,7 @@

 tcpincoming(Conv *s, Tcp *seg, uchar *src, uchar *dst, int version)

 {

 	Tcppriv *tpriv;

+	Route *rp;

 	Limbo *lp, **l;

 	Conv *new;

 	Tcpctl *tcb;

@@ -1630,7 +1699,8 @@

 	tcpsynackrtt(tcb);

 	/* the same as what we sent in SYN,ACK */

-	tcb->mss = tcpmtu(v6lookup(s->p->f, src, dst, new), &tcb->scale, version);

+	rp = v6lookup(s->p->f, src, dst, new);

+	tcb->mss = tcpmtu(rp, &tcb->scale, version);

 	tcpsetmss(new, lp->mss);

 	tcpsetscale(new, lp->ws);

@@ -1666,8 +1736,23 @@

 		panic("tcpincoming: version %d", new->ipversion);

 	}

+	qlock(new);

 	tcpsetstate(new, Established);

 	iphtadd(&tpriv->ht, new);

+	if(rp != nil && (rp->type & Runi) != 0){

+		/* see if we can find the local conversation and splice them together */

+		Iphash *iph = iphtlook(&tpriv->ht, new->laddr, new->lport, new->raddr, new->rport);

+		if(iph != nil && iph->trans == 0){

+			Conv *old = iphconv(iph);

+			if(old != new){

+				qlock(old);

+				tcpsplice(new, old);

+				qunlock(old);

+			}

+		}

+	}

+	qunlock(new);

+

 	return new;

 }

@@ -2144,6 +2229,9 @@

 			tcpsetstate(s, Established);

 			break;

 		case Established:

+			if(tcb->bypass != nil)

+				goto raise;

+			/* wet floor */

 		case Close_wait:

 		case Finwait2:

 			update(s, &seg);

@@ -2513,7 +2601,7 @@

 	Tcpctl *tcb = (Tcpctl*)s->ptcl;

 	qlock(s);

-	if(tcb->state != Closed){

+	if(tcb->state != Closed && tcb->bypass == nil){

 		if(++(tcb->kato) > MAX_KAT) {

 			localclose(s, Etimedout);

 		} else if(tcpsendka(s) < 0) {

@@ -2536,6 +2624,9 @@

 	if(tcb->state != Established)

 		return "connection must be in Establised state";

+

+	if(tcb->bypass != nil)

+		return nil;

 	x = DEF_KAT;

 	if(n > 1){

--- a/sys/src/9/port/portfns.h

+++ b/sys/src/9/port/portfns.h

@@ -288,6 +288,7 @@

 Block*		qbread(Queue*, int);

 long		qbwrite(Queue*, Block*);

 Queue*		qbypass(void (*)(void*, Block*), void*);

+void		qsetbypass(Queue*, void (*)(void*, Block*));

 int		qcanread(Queue*);

 void		qclose(Queue*);

 int		qconsume(Queue*, void*, int);

--- a/sys/src/9/port/qio.c

+++ b/sys/src/9/port/qio.c

@@ -759,6 +759,24 @@

 	return q;

 }

+void

+qsetbypass(Queue *q, void (*bypass)(void*, Block*))

+{

+	Block *b;

+

+	ilock(q);

+	if(bypass != nil){

+		while((b = qremove(q)) != nil){

+			iunlock(q);

+			(*bypass)(q->arg, b);

+			ilock(q);

+		}

+		q->state |= Qstarve;

+	}

+	q->bypass = bypass;

+	iunlock_consumer(q);

+}

+

 static int

 notempty(void *a)

 {

-- 

⑨