Introduce InterHost / sleeve

router/connection.go

@@ -1,7 +1,6 @@
 package router
 
 import (
-	"encoding/binary"
 	"encoding/gob"
 	"fmt"
 	"log"
@@ -40,25 +39,16 @@ type RemoteConnection struct {
 type LocalConnection struct {
 	sync.RWMutex
 	RemoteConnection
-	TCPConn           *net.TCPConn
-	tcpSender         TCPSender
-	remoteUDPAddr     *net.UDPAddr
-	receivedHeartbeat bool
-	stackFrag         bool
-	effectivePMTU     int
-	SessionKey        *[32]byte
-	heartbeatTCP      *time.Ticker
-	heartbeatTimeout  *time.Timer
-	heartbeatFrame    *ForwardedFrame
-	heartbeat         *time.Ticker
-	fragTest          *time.Ticker
-	forwarder         *Forwarder
-	forwarderDF       *ForwarderDF
-	Decryptor         Decryptor
-	Router            *Router
-	uid               uint64
-	actionChan        chan<- ConnectionAction
-	finished          <-chan struct{} // closed to signal that actorLoop has finished
+	TCPConn       *net.TCPConn
+	tcpSender     TCPSender
+	remoteUDPAddr *net.UDPAddr
+	SessionKey    *[32]byte
+	heartbeatTCP  *time.Ticker
+	Router        *Router
+	uid           uint64
+	actionChan    chan<- ConnectionAction
+	finished      <-chan struct{} // closed to signal that actorLoop has finished
+	forwarder     InterHostForwarder
 }
 
 type ConnectionAction func() error
@@ -111,7 +101,6 @@ func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, ud
 		Router:           router,
 		TCPConn:          tcpConn,
 		remoteUDPAddr:    udpAddr,
-		effectivePMTU:    DefaultPMTU,
 		actionChan:       actionChan,
 		finished:         finished}
 	go conn.run(actionChan, finished, acceptNewPeer)
@@ -130,49 +119,12 @@ func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak {
 	}
 }
 
-// Read by the forwarder processes when in the UDP senders
-func (conn *LocalConnection) RemoteUDPAddr() *net.UDPAddr {
-	conn.RLock()
-	defer conn.RUnlock()
-	return conn.remoteUDPAddr
-}
-
 func (conn *LocalConnection) Established() bool {
 	conn.RLock()
 	defer conn.RUnlock()
 	return conn.established
 }
 
-// Called by forwarder processes, read in Forward (by sniffer and udp
-// listener process in router).
-func (conn *LocalConnection) setEffectivePMTU(pmtu int) {
-	conn.Lock()
-	defer conn.Unlock()
-	if conn.effectivePMTU != pmtu {
-		conn.effectivePMTU = pmtu
-		conn.Log("Effective PMTU set to", pmtu)
-	}
-}
-
-// Called by the connection's actor process, and by the connection's
-// TCP received process. StackFrag is read in conn.Forward (called by
-// router udp listener and sniffer processes)
-func (conn *LocalConnection) setStackFrag(frag bool) {
-	conn.Lock()
-	defer conn.Unlock()
-	conn.stackFrag = frag
-}
-
-// Called by the connection's TCP receiver process.
-func (conn *LocalConnection) pmtuVerified(pmtu int) {
-	conn.RLock()
-	fwd := conn.forwarderDF
-	conn.RUnlock()
-	if fwd != nil {
-		fwd.PMTUVerified(pmtu)
-	}
-}
-
 // Send directly, not via the Actor.  If it goes via the Actor we can
 // get a deadlock where LocalConnection is blocked talking to
 // LocalPeer and LocalPeer is blocked trying send a ProtocolMsg via
@@ -202,71 +154,6 @@ func (conn *LocalConnection) Shutdown(err error) {
 	go func() { conn.sendAction(func() error { return err }) }()
 }
 
-// Async
-//
-// Heartbeating serves two purposes: a) keeping NAT paths alive, and
-// b) updating a remote peer's knowledge of our address, in the event
-// it changes (e.g. because NAT paths expired).
-func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr, connUID uint64) {
-	if remoteUDPAddr == nil || connUID != conn.uid {
-		return
-	}
-	conn.sendAction(func() error {
-		oldRemoteUDPAddr := conn.remoteUDPAddr
-		old := conn.receivedHeartbeat
-		conn.Lock()
-		conn.remoteUDPAddr = remoteUDPAddr
-		conn.receivedHeartbeat = true
-		conn.Unlock()
-		conn.heartbeatTimeout.Reset(HeartbeatTimeout)
-		if !old {
-			if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
-				return err
-			}
-		}
-		if oldRemoteUDPAddr == nil {
-			return conn.sendFastHeartbeats()
-		} else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() {
-			log.Println("Peer", conn.remote, "moved from", old, "to", remoteUDPAddr)
-		}
-		return nil
-	})
-}
-
-// Async
-func (conn *LocalConnection) SetEstablished() {
-	conn.sendAction(func() error {
-		stopTicker(conn.heartbeat)
-		old := conn.established
-		conn.Lock()
-		conn.established = true
-		conn.Unlock()
-		if old {
-			return nil
-		}
-		conn.Router.Ourself.ConnectionEstablished(conn)
-		if err := conn.ensureForwarders(); err != nil {
-			return err
-		}
-		// Send a large frame down the DF channel in order to prompt
-		// PMTU discovery to start.
-		conn.Forward(true, &ForwardedFrame{
-			srcPeer: conn.local,
-			dstPeer: conn.remote,
-			frame:   PMTUDiscovery},
-			nil)
-		conn.heartbeat = time.NewTicker(SlowHeartbeat)
-		conn.fragTest = time.NewTicker(FragTestInterval)
-		// avoid initial waits for timers to fire
-		conn.Forward(true, conn.heartbeatFrame, nil)
-		conn.setStackFrag(false)
-		if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
 // Send an actor request to the actorLoop, but don't block if
 // actorLoop has exited - see http://blog.golang.org/pipelines for
 // pattern
@@ -296,17 +183,19 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 
 	// The ordering of the following is very important. [1]
 
-	if conn.remoteUDPAddr != nil {
-		if err = conn.ensureForwarders(); err != nil {
-			return
-		}
+	localIP := tcpConn.LocalAddr().(*net.TCPAddr).IP
+	if conn.forwarder, err = conn.Router.InterHost.MakeForwarder(conn.remote, localIP, conn.remoteUDPAddr, conn.uid, conn.udpCrypto(), conn.sendInterHostControlMessage); err != nil {
+		return
 	}
+
 	if err = conn.Router.Ourself.AddConnection(conn); err != nil {
 		return
 	}
-	if err = conn.initHeartbeats(); err != nil {
-		return
-	}
+
+	// Tell the forwarder that the connection is confirmed
+	conn.forwarder.SetListener(ConnectionAsForwarderListener{conn})
+
+	conn.heartbeatTCP = time.NewTicker(TCPHeartbeat)
 	go conn.receiveTCP(dec)
 	err = conn.actorLoop(actionChan)
 }
@@ -356,21 +245,6 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 // prevent that completely, since, for example, forwarder can only be
 // created when we know the remote UDP address, but it helps to try.
 
-func (conn *LocalConnection) initHeartbeats() error {
-	conn.heartbeatTCP = time.NewTicker(TCPHeartbeat)
-	conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout)
-	heartbeatFrameBytes := make([]byte, EthernetOverhead+8)
-	binary.BigEndian.PutUint64(heartbeatFrameBytes[EthernetOverhead:], conn.uid)
-	conn.heartbeatFrame = &ForwardedFrame{
-		srcPeer: conn.local,
-		dstPeer: conn.remote,
-		frame:   heartbeatFrameBytes}
-	if conn.remoteUDPAddr == nil {
-		return nil
-	}
-	return conn.sendFastHeartbeats()
-}
-
 func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) {
 	for err == nil {
 		select {
@@ -378,13 +252,6 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err
 			err = action()
 		case <-conn.heartbeatTCP.C:
 			err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat)
-		case <-conn.heartbeatTimeout.C:
-			err = fmt.Errorf("timed out waiting for UDP heartbeat")
-		case <-tickerChan(conn.heartbeat):
-			conn.Forward(true, conn.heartbeatFrame, nil)
-		case <-tickerChan(conn.fragTest):
-			conn.setStackFrag(false)
-			err = conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest)
 		}
 	}
 	return
@@ -406,21 +273,55 @@ func (conn *LocalConnection) shutdown(err error) {
 		conn.Router.Ourself.DeleteConnection(conn)
 	}
 
-	if conn.heartbeatTimeout != nil {
-		conn.heartbeatTimeout.Stop()
+	stopTicker(conn.heartbeatTCP)
+
+	if conn.forwarder != nil {
+		conn.forwarder.Close()
 	}
 
-	stopTicker(conn.heartbeatTCP)
-	stopTicker(conn.heartbeat)
-	stopTicker(conn.fragTest)
-
-	// blank out the forwardChan so that the router processes don't
-	// try to send any more
-	conn.stopForwarders()
-
 	conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr, err)
 }
 
+func (conn *LocalConnection) udpCrypto() InterHostCrypto {
+	name := conn.local.NameByte
+	if conn.Router.UsingPassword() {
+		return InterHostCrypto{
+			Dec:   NewNaClDecryptor(conn.SessionKey, conn.outbound),
+			Enc:   NewNaClEncryptor(name, conn.SessionKey, conn.outbound, false),
+			EncDF: NewNaClEncryptor(name, conn.SessionKey, conn.outbound, true),
+		}
+	} else {
+		return InterHostCrypto{
+			Dec:   NewNonDecryptor(),
+			Enc:   NewNonEncryptor(name),
+			EncDF: NewNonEncryptor(name),
+		}
+	}
+}
+
+func (conn *LocalConnection) sendInterHostControlMessage(msg []byte) error {
+	return conn.sendProtocolMsg(ProtocolMsg{ProtocolInterHostControlMsg, msg})
+}
+
+type ConnectionAsForwarderListener struct{ conn *LocalConnection }
+
+func (l ConnectionAsForwarderListener) Established() {
+	l.conn.sendAction(func() error {
+		old := l.conn.established
+		l.conn.Lock()
+		l.conn.established = true
+		l.conn.Unlock()
+		if !old {
+			l.conn.Router.Ourself.ConnectionEstablished(l.conn)
+		}
+		return nil
+	})
+}
+
+func (l ConnectionAsForwarderListener) Error(err error) {
+	l.conn.sendAction(func() error { return err })
+}
+
 // Helpers
 
 func (conn *LocalConnection) sendSimpleProtocolMsg(tag ProtocolTag) error {
@@ -464,22 +365,8 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
 func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) error {
 	switch tag {
 	case ProtocolHeartbeat:
-	case ProtocolConnectionEstablished:
-		// We sent fast heartbeats to the remote peer, which has now
-		// received at least one of them and told us via this message.
-		// We can now consider the connection as established from our
-		// end.
-		conn.SetEstablished()
-	case ProtocolStartFragmentationTest:
-		conn.Forward(false, &ForwardedFrame{
-			srcPeer: conn.local,
-			dstPeer: conn.remote,
-			frame:   FragTest},
-			nil)
-	case ProtocolFragmentationReceived:
-		conn.setStackFrag(true)
-	case ProtocolPMTUVerified:
-		conn.pmtuVerified(int(binary.BigEndian.Uint16(payload)))
+	case ProtocolInterHostControlMsg:
+		conn.forwarder.ControlMessage(payload)
 	case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip:
 		return conn.Router.handleGossip(tag, payload)
 	default:
@@ -492,13 +379,8 @@ func (conn *LocalConnection) extendReadDeadline() {
 	conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2))
 }
 
-func (conn *LocalConnection) sendFastHeartbeats() error {
-	err := conn.ensureForwarders()
-	if err == nil {
-		conn.heartbeat = time.NewTicker(FastHeartbeat)
-		conn.Forward(true, conn.heartbeatFrame, nil) // avoid initial wait
-	}
-	return err
+func (conn *LocalConnection) Forward(srcPeer, dstPeer *Peer, frame []byte, dec *EthernetDecoder) error {
+	return conn.forwarder.Forward(srcPeer, dstPeer, frame, dec)
 }
 
 func tickerChan(ticker *time.Ticker) <-chan time.Time {

router/forwarder.go

@@ -1,411 +0,0 @@
-package router
-
-import (
-	"code.google.com/p/gopacket"
-	"code.google.com/p/gopacket/layers"
-	"syscall"
-	"time"
-)
-
-type ForwardedFrame struct {
-	srcPeer *Peer
-	dstPeer *Peer
-	frame   []byte
-}
-
-type FrameTooBigError struct {
-	EPMTU int // effective pmtu, i.e. what we tell packet senders
-}
-
-func (conn *LocalConnection) ensureForwarders() error {
-	if conn.forwarder != nil || conn.forwarderDF != nil {
-		return nil
-	}
-	udpSender := NewSimpleUDPSender(conn)
-	udpSenderDF, err := NewRawUDPSender(conn) // only thing that can error, so do it early
-	if err != nil {
-		return err
-	}
-
-	usingPassword := conn.SessionKey != nil
-	var encryptor, encryptorDF Encryptor
-	if usingPassword {
-		encryptor = NewNaClEncryptor(conn.local.NameByte, conn.SessionKey, conn.outbound, false)
-		encryptorDF = NewNaClEncryptor(conn.local.NameByte, conn.SessionKey, conn.outbound, true)
-	} else {
-		encryptor = NewNonEncryptor(conn.local.NameByte)
-		encryptorDF = NewNonEncryptor(conn.local.NameByte)
-	}
-
-	forwarder := NewForwarder(conn, encryptor, udpSender, DefaultPMTU)
-	forwarderDF := NewForwarderDF(conn, encryptorDF, udpSenderDF, DefaultPMTU)
-
-	// Various fields in the conn struct are read by other processes,
-	// so we have to use locks.
-	conn.Lock()
-	conn.forwarder = forwarder
-	conn.forwarderDF = forwarderDF
-	conn.Unlock()
-
-	return nil
-}
-
-func (conn *LocalConnection) stopForwarders() {
-	if conn.forwarder == nil || conn.forwarderDF == nil {
-		return
-	}
-	conn.forwarder.Shutdown()
-	conn.forwarderDF.Shutdown()
-}
-
-// Called from peer.Relay[Broadcast] which is itself invoked from
-// router (both UDP listener process and sniffer process). Also called
-// from connection's heartbeat process, and from the connection's TCP
-// receiver process.
-func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *EthernetDecoder) error {
-	conn.RLock()
-	var (
-		forwarder     = conn.forwarder
-		forwarderDF   = conn.forwarderDF
-		effectivePMTU = conn.effectivePMTU
-		stackFrag     = conn.stackFrag
-	)
-	conn.RUnlock()
-
-	if forwarder == nil || forwarderDF == nil {
-		conn.Log("Cannot forward frame yet - awaiting contact")
-		return nil
-	}
-
-	if dec != nil {
-		df = dec.DF()
-	}
-
-	// We could use non-blocking channel sends here, i.e. drop frames
-	// on the floor when the forwarder is busy. This would allow our
-	// caller - the capturing loop in the router - to read frames more
-	// quickly when under load, i.e. we'd drop fewer frames on the
-	// floor during capture. And we could maximise CPU utilisation
-	// since we aren't stalling a thread. However, a lot of work has
-	// already been done by the time we get here. Since any packet we
-	// drop will likely get re-transmitted we end up paying that cost
-	// multiple times. So it's better to drop things at the beginning
-	// of our pipeline.
-	if df {
-		if !frameTooBig(frame, effectivePMTU) {
-			forwarderDF.Forward(frame)
-			return nil
-		}
-		return FrameTooBigError{EPMTU: effectivePMTU}
-	}
-
-	if stackFrag || dec == nil || len(dec.decoded) < 2 {
-		forwarder.Forward(frame)
-		return nil
-	}
-	// Don't have trustworthy stack, so we're going to have to
-	// send it DF in any case.
-	if !frameTooBig(frame, effectivePMTU) {
-		forwarderDF.Forward(frame)
-		return nil
-	}
-	conn.Router.LogFrame("Fragmenting", frame.frame, dec)
-	// We can't trust the stack to fragment, we have IP, and we
-	// have a frame that's too big for the MTU, so we have to
-	// fragment it ourself.
-	return fragment(dec.Eth, dec.IP, effectivePMTU, frame, func(segFrame *ForwardedFrame) {
-		forwarderDF.Forward(segFrame)
-	})
-}
-
-func frameTooBig(frame *ForwardedFrame, effectivePMTU int) bool {
-	// We capture/forward complete ethernet frames. Therefore the
-	// frame length includes the ethernet header. However, MTUs
-	// operate at the IP layer and thus do not include the ethernet
-	// header. To put it another way, when a sender that was told an
-	// MTU of M sends an IP packet of exactly that length, we will
-	// capture/forward M + EthernetOverhead bytes of data.
-	return len(frame.frame) > effectivePMTU+EthernetOverhead
-}
-
-func fragment(eth layers.Ethernet, ip layers.IPv4, pmtu int, frame *ForwardedFrame, forward func(*ForwardedFrame)) error {
-	// We are not doing any sort of NAT, so we don't need to worry
-	// about checksums of IP payload (eg UDP checksum).
-	headerSize := int(ip.IHL) * 4
-	// &^ is bit clear (AND NOT). So here we're clearing the lowest 3
-	// bits.
-	maxSegmentSize := (pmtu - headerSize) &^ 7
-	opts := gopacket.SerializeOptions{
-		FixLengths:       false,
-		ComputeChecksums: true}
-	payloadSize := int(ip.Length) - headerSize
-	payload := ip.BaseLayer.Payload[:payloadSize]
-	offsetBase := int(ip.FragOffset) << 3
-	origFlags := ip.Flags
-	ip.Flags = ip.Flags | layers.IPv4MoreFragments
-	ip.Length = uint16(headerSize + maxSegmentSize)
-	if eth.EthernetType == layers.EthernetTypeLLC {
-		// using LLC, so must set eth length correctly. eth length
-		// is just the length of the payload
-		eth.Length = ip.Length
-	} else {
-		eth.Length = 0
-	}
-	for offset := 0; offset < payloadSize; offset += maxSegmentSize {
-		var segmentPayload []byte
-		if len(payload) <= maxSegmentSize {
-			// last one
-			segmentPayload = payload
-			ip.Length = uint16(len(payload) + headerSize)
-			ip.Flags = origFlags
-			if eth.EthernetType == layers.EthernetTypeLLC {
-				eth.Length = ip.Length
-			} else {
-				eth.Length = 0
-			}
-		} else {
-			segmentPayload = payload[:maxSegmentSize]
-			payload = payload[maxSegmentSize:]
-		}
-		ip.FragOffset = uint16((offset + offsetBase) >> 3)
-		buf := gopacket.NewSerializeBuffer()
-		segPayload := gopacket.Payload(segmentPayload)
-		err := gopacket.SerializeLayers(buf, opts, &eth, &ip, &segPayload)
-		if err != nil {
-			return err
-		}
-		// make copies of the frame we received
-		segFrame := *frame
-		segFrame.frame = buf.Bytes()
-		forward(&segFrame)
-	}
-	return nil
-}
-
-// Forwarder
-
-type Forwarder struct {
-	conn             *LocalConnection
-	ch               chan<- *ForwardedFrame
-	finished         <-chan struct{}
-	enc              Encryptor
-	udpSender        UDPSender
-	maxPayload       int
-	processSendError func(error) error
-}
-
-func NewForwarder(conn *LocalConnection, enc Encryptor, udpSender UDPSender, pmtu int) *Forwarder {
-	ch := make(chan *ForwardedFrame, ChannelSize)
-	finished := make(chan struct{})
-	fwd := &Forwarder{
-		conn:             conn,
-		ch:               ch,
-		finished:         finished,
-		enc:              enc,
-		udpSender:        udpSender,
-		maxPayload:       pmtu - UDPOverhead,
-		processSendError: func(err error) error { return err }}
-	go fwd.run(ch, finished)
-	return fwd
-}
-
-func (fwd *Forwarder) Shutdown() {
-	fwd.ch <- nil
-}
-
-func (fwd *Forwarder) Forward(frame *ForwardedFrame) {
-	select {
-	case fwd.ch <- frame:
-	case <-fwd.finished:
-	}
-}
-
-func (fwd *Forwarder) run(ch <-chan *ForwardedFrame, finished chan<- struct{}) {
-	defer fwd.udpSender.Shutdown()
-	for {
-		if !fwd.accumulateAndSendFrames(ch, <-ch) {
-			close(finished)
-			return
-		}
-	}
-}
-
-func (fwd *Forwarder) effectiveOverhead() int {
-	return UDPOverhead + fwd.enc.PacketOverhead() + fwd.enc.FrameOverhead() + EthernetOverhead
-}
-
-// Drain the inbound channel of frames, aggregating them into larger
-// packets for efficient transmission.
-//
-// FIXME Depending on the golang scheduler, and the rate at which
-// franes get sent to the forwarder, we can be going around this loop
-// forever. That is bad since there may be other stuff for us to do,
-// i.e. the other branches in the run loop.
-func (fwd *Forwarder) accumulateAndSendFrames(ch <-chan *ForwardedFrame, frame *ForwardedFrame) bool {
-	if frame == nil {
-		return false
-	}
-	if !fwd.appendFrame(frame) {
-		fwd.logDrop(frame)
-		// [1] The buffer is empty at this point and therefore we must
-		// not flush it. The easiest way to accomplish that is simply
-		// by returning to the surrounding run loop.
-		return true
-	}
-	for {
-		select {
-		case frame = <-ch:
-			if frame == nil {
-				return false
-			}
-			if !fwd.appendFrame(frame) {
-				fwd.flush()
-				if !fwd.appendFrame(frame) {
-					fwd.logDrop(frame)
-					return true // see [1]
-				}
-			}
-		default:
-			fwd.flush()
-			return true
-		}
-	}
-}
-
-func (fwd *Forwarder) logDrop(frame *ForwardedFrame) {
-	fwd.conn.Log("Dropping too big frame during forwarding: frame len:", len(frame.frame), "; effective PMTU:", fwd.maxPayload+UDPOverhead-fwd.effectiveOverhead())
-}
-
-func (fwd *Forwarder) appendFrame(frame *ForwardedFrame) bool {
-	frameLen := len(frame.frame)
-	if fwd.enc.TotalLen()+fwd.enc.FrameOverhead()+frameLen > fwd.maxPayload {
-		return false
-	}
-	fwd.enc.AppendFrame(frame.srcPeer.NameByte, frame.dstPeer.NameByte, frame.frame)
-	return true
-}
-
-func (fwd *Forwarder) flush() {
-	msg, err := fwd.enc.Bytes()
-	if err != nil {
-		fwd.conn.Shutdown(err)
-	}
-	err = fwd.processSendError(fwd.udpSender.Send(msg))
-	if err != nil && PosixError(err) != syscall.ENOBUFS {
-		fwd.conn.Shutdown(err)
-	}
-}
-
-type ForwarderDF struct {
-	Forwarder
-	verifyPMTUTick  <-chan time.Time
-	verifyPMTU      chan<- int
-	pmtuVerifyCount uint
-	pmtuVerified    bool
-	highestGoodPMTU int
-	unverifiedPMTU  int
-	lowestBadPMTU   int
-}
-
-func NewForwarderDF(conn *LocalConnection, enc Encryptor, udpSender UDPSender, pmtu int) *ForwarderDF {
-	ch := make(chan *ForwardedFrame, ChannelSize)
-	finished := make(chan struct{})
-	verifyPMTU := make(chan int, ChannelSize)
-	fwd := &ForwarderDF{
-		Forwarder: Forwarder{
-			conn:       conn,
-			ch:         ch,
-			finished:   finished,
-			enc:        enc,
-			udpSender:  udpSender,
-			maxPayload: pmtu - UDPOverhead},
-		verifyPMTU: verifyPMTU}
-	fwd.Forwarder.processSendError = fwd.processSendError
-	fwd.unverifiedPMTU = pmtu - fwd.effectiveOverhead()
-	conn.setEffectivePMTU(fwd.unverifiedPMTU)
-	go fwd.run(ch, finished, verifyPMTU)
-	return fwd
-}
-
-func (fwd *ForwarderDF) PMTUVerified(pmtu int) {
-	select {
-	case fwd.verifyPMTU <- pmtu:
-	case <-fwd.finished:
-	}
-}
-
-func (fwd *ForwarderDF) run(ch <-chan *ForwardedFrame, finished chan<- struct{}, verifyPMTU <-chan int) {
-	defer fwd.udpSender.Shutdown()
-	for {
-		select {
-		case <-fwd.verifyPMTUTick:
-			// We only do this case here when we know the
-			// buffers are all empty so that we don't risk
-			// appending verify-frames to other data.
-			fwd.verifyPMTUTick = nil
-			if fwd.pmtuVerified {
-				continue
-			}
-			if fwd.pmtuVerifyCount > 0 {
-				fwd.pmtuVerifyCount--
-				fwd.attemptVerifyEffectivePMTU()
-			} else {
-				// we've exceeded the verification
-				// attempts of the unverifiedPMTU
-				fwd.lowestBadPMTU = fwd.unverifiedPMTU
-				fwd.verifyEffectivePMTU((fwd.highestGoodPMTU + fwd.lowestBadPMTU) / 2)
-			}
-		case epmtu := <-verifyPMTU:
-			if fwd.pmtuVerified || epmtu != fwd.unverifiedPMTU {
-				continue
-			}
-			if epmtu+1 < fwd.lowestBadPMTU {
-				fwd.highestGoodPMTU = fwd.unverifiedPMTU // = epmtu
-				fwd.verifyEffectivePMTU((fwd.highestGoodPMTU + fwd.lowestBadPMTU) / 2)
-			} else {
-				fwd.pmtuVerified = true
-				fwd.maxPayload = epmtu + fwd.effectiveOverhead() - UDPOverhead
-				fwd.conn.setEffectivePMTU(epmtu)
-				fwd.conn.Log("Effective PMTU verified at", epmtu)
-			}
-		case frame := <-ch:
-			if !fwd.accumulateAndSendFrames(ch, frame) {
-				close(finished)
-				return
-			}
-		}
-	}
-}
-
-func (fwd *ForwarderDF) verifyEffectivePMTU(newUnverifiedPMTU int) {
-	fwd.unverifiedPMTU = newUnverifiedPMTU
-	fwd.pmtuVerifyCount = PMTUVerifyAttempts
-	fwd.attemptVerifyEffectivePMTU()
-}
-
-func (fwd *ForwarderDF) attemptVerifyEffectivePMTU() {
-	fwd.enc.AppendFrame(fwd.conn.local.NameByte, fwd.conn.remote.NameByte,
-		make([]byte, fwd.unverifiedPMTU+EthernetOverhead))
-	fwd.flush()
-	if fwd.verifyPMTUTick == nil {
-		fwd.verifyPMTUTick = time.After(PMTUVerifyTimeout << (PMTUVerifyAttempts - fwd.pmtuVerifyCount))
-	}
-}
-
-func (fwd *ForwarderDF) processSendError(err error) error {
-	if mtbe, ok := err.(MsgTooBigError); ok {
-		newUnverifiedPMTU := mtbe.PMTU - fwd.effectiveOverhead()
-		if newUnverifiedPMTU < fwd.unverifiedPMTU {
-			fwd.pmtuVerified = false
-			fwd.maxPayload = mtbe.PMTU - UDPOverhead
-			fwd.highestGoodPMTU = 8
-			fwd.lowestBadPMTU = newUnverifiedPMTU + 1
-			fwd.conn.setEffectivePMTU(newUnverifiedPMTU)
-			fwd.verifyEffectivePMTU(newUnverifiedPMTU)
-		}
-
-		return nil
-	}
-
-	return err
-}

router/handshake.go

@@ -103,13 +103,11 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep
 		}
 		conn.SessionKey = FormSessionKey(&remotePublic, private, conn.Router.Password)
 		conn.tcpSender = NewEncryptedTCPSender(enc, conn.SessionKey, conn.outbound)
-		conn.Decryptor = NewNaClDecryptor(conn.SessionKey, conn.outbound)
 	} else {
 		if rpErr == nil {
 			return fmt.Errorf("Remote network is encrypted. Password required.")
 		}
 		conn.tcpSender = NewSimpleTCPSender(enc)
-		conn.Decryptor = NewNonDecryptor()
 	}
 
 	return conn.setRemote(NewPeer(name, nickNameStr, uid, 0))

router/interhost.go

@@ -0,0 +1,79 @@
+package router
+
+import (
+	"net"
+)
+
+// Interface to inter-host (i.e. overlay network) packet handling
+type InterHost interface {
+	// Start consuming forwarded packets.
+	ConsumePackets(*Peer, *Peers, InterHostConsumer) error
+
+	// Form a packet-forwarding connection.  The remote UDPAddr
+	// can be nil if unknown (in which case the implementation
+	// needs to discover it).
+	MakeForwarder(remotePeer *Peer, localIP net.IP, remote *net.UDPAddr,
+		connUID uint64, crypto InterHostCrypto,
+		sendControlMessage func([]byte) error) (InterHostForwarder, error)
+}
+
+// When a consumer is called, the decoder will already have been used
+// to decode the frame.
+type InterHostConsumer func(src *Peer, dst *Peer, frame []byte,
+	dec *EthernetDecoder)
+
+// Crypto settings for a forwarder.
+type InterHostCrypto struct {
+	Dec   Decryptor
+	Enc   Encryptor
+	EncDF Encryptor
+}
+
+// All of the machinery to forward packets to a particular peer
+type InterHostForwarder interface {
+	// Register a callback for forwarder state changes.
+	// side-effect, calling this confirms that the connection is
+	// really wanted, and so the provider should activate it.
+	// However, Forward might be called before this is called
+	// (e.g. on another thread).
+	SetListener(InterHostForwarderListener)
+
+	// Forward a packet across the connection.  The caller must
+	// supply an EthernetDecoder specific to this thread, which
+	// has already been used to decode the frame.
+	Forward(src *Peer, dest *Peer, frame []byte, dec *EthernetDecoder) error
+
+	Close()
+
+	// Handle a message from the peer
+	ControlMessage([]byte)
+}
+
+type InterHostForwarderListener interface {
+	Established()
+	Error(error)
+}
+
+type NullInterHost struct{}
+
+func (NullInterHost) ConsumePackets(*Peer, *Peers, InterHostConsumer) error {
+	return nil
+}
+
+func (NullInterHost) MakeForwarder(*Peer, net.IP, *net.UDPAddr, uint64,
+	InterHostCrypto, func([]byte) error) (InterHostForwarder, error) {
+	return NullInterHost{}, nil
+}
+
+func (NullInterHost) SetListener(InterHostForwarderListener) {
+}
+
+func (NullInterHost) Forward(*Peer, *Peer, []byte, *EthernetDecoder) error {
+	return nil
+}
+
+func (NullInterHost) Close() {
+}
+
+func (NullInterHost) ControlMessage([]byte) {
+}

router/local_peer.go

@@ -50,11 +50,7 @@ func (peer *LocalPeer) Relay(srcPeer, dstPeer *Peer, frame []byte, dec *Ethernet
 		log.Println("Unable to find connection to relay peer", relayPeerName)
 		return nil
 	}
-	return conn.(*LocalConnection).Forward(false, &ForwardedFrame{
-		srcPeer: srcPeer,
-		dstPeer: dstPeer,
-		frame:   frame},
-		dec)
+	return conn.(*LocalConnection).Forward(srcPeer, dstPeer, frame, dec)
 }
 
 func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, frame []byte, dec *EthernetDecoder) {
@@ -63,11 +59,8 @@ func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, frame []byte, dec *Ethernet
 		return
 	}
 	for _, conn := range peer.ConnectionsTo(nextHops) {
-		err := conn.(*LocalConnection).Forward(false, &ForwardedFrame{
-			srcPeer: srcPeer,
-			dstPeer: conn.Remote(),
-			frame:   frame},
-			dec)
+		err := conn.(*LocalConnection).Forward(srcPeer, conn.Remote(),
+			frame, dec)
 		if err != nil {
 			if ftbe, ok := err.(FrameTooBigError); ok {
 				log.Printf("dropping too big DF broadcast frame (%v -> %v): PMTU= %v\n", dec.IP.DstIP, dec.IP.SrcIP, ftbe.EPMTU)

router/protocol.go

@@ -9,10 +9,7 @@ type ProtocolTag byte
 
 const (
 	ProtocolHeartbeat ProtocolTag = iota
-	ProtocolConnectionEstablished
-	ProtocolFragmentationReceived
-	ProtocolStartFragmentationTest
-	ProtocolPMTUVerified
+	ProtocolInterHostControlMsg
 	ProtocolGossip
 	ProtocolGossipUnicast
 	ProtocolGossipBroadcast

router/router.go

@@ -2,12 +2,9 @@ package router
 
 import (
 	"bytes"
-	"encoding/binary"
 	"fmt"
-	"io"
 	"log"
 	"net"
-	"syscall"
 	"time"
 )
 
@@ -24,6 +21,7 @@ type RouterConfig struct {
 	ConnLimit int
 	LogFrame  LogFrameFunc
 	IntraHost IntraHost
+	InterHost InterHost
 }
 
 type Router struct {
@@ -35,7 +33,6 @@ type Router struct {
 	ConnectionMaker *ConnectionMaker
 	GossipChannels  map[uint32]*GossipChannel
 	TopologyGossip  Gossip
-	UDPListener     *net.UDPConn
 }
 
 func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
@@ -47,6 +44,10 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
 		router.IntraHost = NullIntraHost{}
 	}
 
+	if router.InterHost == nil {
+		router.InterHost = NullInterHost{}
+	}
+
 	onMacExpiry := func(mac net.HardwareAddr, peer *Peer) {
 		log.Println("Expired MAC", mac, "at", peer)
 	}
@@ -67,8 +68,8 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
 // Start listening for packets from containers, TCP connections and
 // packets forwarded over UDP.
 func (router *Router) Start() {
+	checkFatal(router.InterHost.ConsumePackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket))
 	checkFatal(router.IntraHost.ConsumePackets(router))
-	router.UDPListener = router.listenUDP(router.Port)
 	router.listenTCP(router.Port)
 }
 
@@ -169,140 +170,36 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
 	StartLocalConnection(connRemote, tcpConn, nil, router, true)
 }
 
-func (router *Router) listenUDP(localPort int) *net.UDPConn {
-	localAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprint(":", localPort))
-	checkFatal(err)
-	conn, err := net.ListenUDP("udp4", localAddr)
-	checkFatal(err)
-	f, err := conn.File()
-	defer f.Close()
-	checkFatal(err)
-	fd := int(f.Fd())
-	// This one makes sure all packets we send out do not have DF set on them.
-	err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DONT)
-	checkFatal(err)
-	go router.udpReader(conn)
-	return conn
-}
+func (router *Router) handleForwardedPacket(srcPeer *Peer, dstPeer *Peer,
+	frame []byte, dec *EthernetDecoder) {
+	if dstPeer != router.Ourself.Peer {
+		// it's not for us, we're just relaying it
+		router.LogFrame("Relaying", frame, dec)
+		err := router.Ourself.Relay(srcPeer, dstPeer, frame, dec)
+		if ftbe, ok := err.(FrameTooBigError); ok {
+			err = dec.sendICMPFragNeeded(ftbe.EPMTU, func(icmpFrame []byte) error {
+				return router.Ourself.Forward(srcPeer, icmpFrame, nil)
+			})
+		}
 
-func (router *Router) udpReader(conn *net.UDPConn) {
-	defer conn.Close()
-	dec := NewEthernetDecoder()
-	buf := make([]byte, MaxUDPPacketSize)
-	for {
-		n, sender, err := conn.ReadFromUDP(buf)
-		if err == io.EOF {
-			return
-		} else if err != nil {
-			log.Println("ignoring UDP read error", err)
-			continue
-		} else if n < NameSize {
-			log.Println("ignoring too short UDP packet from", sender)
-			continue
-		}
-		name := PeerNameFromBin(buf[:NameSize])
-		packet := make([]byte, n-NameSize)
-		copy(packet, buf[NameSize:n])
-		peerConn, found := router.Ourself.ConnectionTo(name)
-		if !found {
-			continue
-		}
-		relayConn, ok := peerConn.(*LocalConnection)
-		if !ok {
-			continue
-		}
-		if err := relayConn.Decryptor.IterateFrames(packet, router.handleUDPPacketFunc(relayConn, dec, sender)); err != nil {
-			// Errors during UDP packet decoding / processing are
-			// non-fatal. One common cause is that we receive and
-			// attempt to decrypt a "stray" packet. This can actually
-			// happen quite easily if there is some connection churn
-			// between two peers. After all, UDP isn't a
-			// connection-oriented protocol, yet we pretend it is.
-			//
-			// If anything really is seriously, unrecoverably amiss
-			// with a connection, that will typically result in missed
-			// heartbeats and the connection getting shut down because
-			// of that.
-			relayConn.Log(err)
-		}
+		checkWarn(err)
+		return
 	}
-}
 
-func (router *Router) handleUDPPacketFunc(relayConn *LocalConnection, dec *EthernetDecoder, sender *net.UDPAddr) FrameConsumer {
-	return func(srcNameByte, dstNameByte []byte, frame []byte) {
-		srcPeer, found := router.Peers.Fetch(PeerNameFromBin(srcNameByte))
-		if !found {
-			return
-		}
-		dstPeer, found := router.Peers.Fetch(PeerNameFromBin(dstNameByte))
-		if !found {
-			return
-		}
+	srcMac := dec.Eth.SrcMAC
+	dstMac := dec.Eth.DstMAC
 
-		dec.DecodeLayers(frame)
-		decodedLen := len(dec.decoded)
-		if decodedLen == 0 {
-			return
-		}
-		// Handle special frames produced internally (rather than
-		// captured/forwarded) by the remote router.
-		//
-		// We really shouldn't be decoding these above, since they are
-		// not genuine Ethernet frames. However, it is actually more
-		// efficient to do so, as we want to optimise for the common
-		// (i.e. non-special) frames. These always need decoding, and
-		// detecting special frames is cheaper post decoding than pre.
-		if decodedLen == 1 && dec.IsSpecial() {
-			if srcPeer == relayConn.Remote() && dstPeer == router.Ourself.Peer {
-				handleSpecialFrame(relayConn, sender, frame)
-			}
-			return
-		}
-
-		if dstPeer != router.Ourself.Peer {
-			// it's not for us, we're just relaying it
-			router.LogFrame("Relaying", frame, dec)
-			err := router.Ourself.Relay(srcPeer, dstPeer, frame, dec)
-			if ftbe, ok := err.(FrameTooBigError); ok {
-				err = dec.sendICMPFragNeeded(ftbe.EPMTU, func(icmpFrame []byte) error {
-					return router.Ourself.Forward(srcPeer, icmpFrame, nil)
-				})
-			}
-
-			checkWarn(err)
-			return
-		}
-
-		srcMac := dec.Eth.SrcMAC
-		dstMac := dec.Eth.DstMAC
-
-		if router.Macs.Enter(srcMac, srcPeer) {
-			log.Println("Discovered remote MAC", srcMac, "at", srcPeer)
-		}
-
-		router.LogFrame("Injecting", frame, dec)
-		checkWarn(router.IntraHost.InjectPacket(frame))
-
-		dstPeer, found = router.Macs.Lookup(dstMac)
-		if !found || dstPeer != router.Ourself.Peer {
-			router.LogFrame("Relaying broadcast", frame, dec)
-			router.Ourself.RelayBroadcast(srcPeer, frame, dec)
-		}
+	if router.Macs.Enter(srcMac, srcPeer) {
+		log.Println("Discovered remote MAC", srcMac, "at", srcPeer)
 	}
-}
 
-func handleSpecialFrame(relayConn *LocalConnection, sender *net.UDPAddr, frame []byte) {
-	frameLen := len(frame)
-	switch {
-	case frameLen == EthernetOverhead+8:
-		relayConn.ReceivedHeartbeat(sender, binary.BigEndian.Uint64(frame[EthernetOverhead:]))
-	case frameLen == FragTestSize && bytes.Equal(frame, FragTest):
-		relayConn.SendProtocolMsg(ProtocolMsg{ProtocolFragmentationReceived, nil})
-	case frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery):
-	default:
-		frameLenBytes := []byte{0, 0}
-		binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead))
-		relayConn.SendProtocolMsg(ProtocolMsg{ProtocolPMTUVerified, frameLenBytes})
+	router.LogFrame("Injecting", frame, dec)
+	checkWarn(router.IntraHost.InjectPacket(frame))
+
+	dstPeer, found := router.Macs.Lookup(dstMac)
+	if !found || dstPeer != router.Ourself.Peer {
+		router.LogFrame("Relaying broadcast", frame, dec)
+		router.Ourself.RelayBroadcast(srcPeer, frame, dec)
 	}
 }
 

router/udp_sender.go

@@ -1,138 +0,0 @@
-package router
-
-import (
-	"code.google.com/p/gopacket"
-	"code.google.com/p/gopacket/layers"
-	"log"
-	"net"
-	"syscall"
-)
-
-type UDPSender interface {
-	Send([]byte) error
-	Shutdown() error
-}
-
-type SimpleUDPSender struct {
-	conn    *LocalConnection
-	udpConn *net.UDPConn
-}
-
-type RawUDPSender struct {
-	ipBuf     gopacket.SerializeBuffer
-	opts      gopacket.SerializeOptions
-	udpHeader *layers.UDP
-	socket    *net.IPConn
-	conn      *LocalConnection
-}
-
-type MsgTooBigError struct {
-	PMTU int // actual pmtu, i.e. what the kernel told us
-}
-
-func NewSimpleUDPSender(conn *LocalConnection) *SimpleUDPSender {
-	return &SimpleUDPSender{udpConn: conn.Router.UDPListener, conn: conn}
-}
-
-func (sender *SimpleUDPSender) Send(msg []byte) error {
-	_, err := sender.udpConn.WriteToUDP(msg, sender.conn.RemoteUDPAddr())
-	return err
-}
-
-func (sender *SimpleUDPSender) Shutdown() error {
-	return nil
-}
-
-func NewRawUDPSender(conn *LocalConnection) (*RawUDPSender, error) {
-	ipSocket, err := dialIP(conn)
-	if err != nil {
-		return nil, err
-	}
-	udpHeader := &layers.UDP{SrcPort: layers.UDPPort(conn.Router.Port)}
-	ipBuf := gopacket.NewSerializeBuffer()
-	opts := gopacket.SerializeOptions{
-		FixLengths: true,
-		// UDP header is calculated with a phantom IP
-		// header. Yes, it's totally nuts. Thankfully, for UDP
-		// over IPv4, the checksum is optional. It's not
-		// optional for IPv6, but we'll ignore that for
-		// now. TODO
-		ComputeChecksums: false}
-
-	return &RawUDPSender{
-		ipBuf:     ipBuf,
-		opts:      opts,
-		udpHeader: udpHeader,
-		socket:    ipSocket,
-		conn:      conn}, nil
-}
-
-func (sender *RawUDPSender) Send(msg []byte) error {
-	payload := gopacket.Payload(msg)
-	sender.udpHeader.DstPort = layers.UDPPort(sender.conn.RemoteUDPAddr().Port)
-
-	err := gopacket.SerializeLayers(sender.ipBuf, sender.opts, sender.udpHeader, &payload)
-	if err != nil {
-		return err
-	}
-	packet := sender.ipBuf.Bytes()
-	_, err = sender.socket.Write(packet)
-	if err == nil || PosixError(err) != syscall.EMSGSIZE {
-		return err
-	}
-	f, err := sender.socket.File()
-	if err != nil {
-		return err
-	}
-	defer f.Close()
-	fd := int(f.Fd())
-	log.Println("EMSGSIZE on send, expecting PMTU update (IP packet was",
-		len(packet), "bytes, payload was", len(msg), "bytes)")
-	pmtu, err := syscall.GetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU)
-	if err != nil {
-		return err
-	}
-	return MsgTooBigError{PMTU: pmtu}
-}
-
-func (sender *RawUDPSender) Shutdown() error {
-	defer func() { sender.socket = nil }()
-	return sender.socket.Close()
-}
-
-func dialIP(conn *LocalConnection) (*net.IPConn, error) {
-	ipLocalAddr, err := ipAddr(conn.TCPConn.LocalAddr())
-	if err != nil {
-		return nil, err
-	}
-	ipRemoteAddr, err := ipAddr(conn.TCPConn.RemoteAddr())
-	if err != nil {
-		return nil, err
-	}
-	ipSocket, err := net.DialIP("ip4:UDP", ipLocalAddr, ipRemoteAddr)
-	if err != nil {
-		return nil, err
-	}
-	f, err := ipSocket.File()
-	if err != nil {
-		return nil, err
-	}
-	defer f.Close()
-	fd := int(f.Fd())
-	// This Makes sure all packets we send out have DF set on them.
-	err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DO)
-	if err != nil {
-		return nil, err
-	}
-	return ipSocket, nil
-}
-
-func ipAddr(addr net.Addr) (*net.IPAddr, error) {
-	host, _, err := net.SplitHostPort(addr.String())
-	if err != nil {
-		return nil, err
-	}
-	return &net.IPAddr{
-		IP:   net.ParseIP(host),
-		Zone: ""}, nil
-}

router/utils.go

@@ -35,14 +35,6 @@ func PosixError(err error) error {
 	return operr.Err
 }
 
-func (mtbe MsgTooBigError) Error() string {
-	return fmt.Sprint("Msg too big error. PMTU is ", mtbe.PMTU)
-}
-
-func (ftbe FrameTooBigError) Error() string {
-	return fmt.Sprint("Frame too big error. Effective PMTU is ", ftbe.EPMTU)
-}
-
 func (upe UnknownPeerError) Error() string {
 	return fmt.Sprint("Reference to unknown peer ", upe.Name)
 }

weaver/main.go

@@ -133,6 +133,7 @@ func main() {
 		defer profile.Start(&p).Stop()
 	}
 
+	config.InterHost = weave.NewSleeveInterHost(config.Port)
 	config.LogFrame = logFrameFunc(pktdebug)
 
 	router := weave.NewRouter(config, name, nickName)