From 4e41f649fb76a069f0847c99a77832643d81c61d Mon Sep 17 00:00:00 2001 From: David Wragg Date: Mon, 8 Jun 2015 17:49:15 +0100 Subject: [PATCH] Introduce InterHost / sleeve --- router/connection.go | 252 +++------- router/forwarder.go | 411 ---------------- router/handshake.go | 2 - router/interhost.go | 79 ++++ router/local_peer.go | 13 +- router/protocol.go | 5 +- router/router.go | 163 ++----- router/sleeve.go | 1058 ++++++++++++++++++++++++++++++++++++++++++ router/udp_sender.go | 138 ------ router/utils.go | 8 - weaver/main.go | 1 + 11 files changed, 1239 insertions(+), 891 deletions(-) delete mode 100644 router/forwarder.go create mode 100644 router/interhost.go create mode 100644 router/sleeve.go delete mode 100644 router/udp_sender.go diff --git a/router/connection.go b/router/connection.go index cbd1b71e..6614d0a9 100644 --- a/router/connection.go +++ b/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 { diff --git a/router/forwarder.go b/router/forwarder.go deleted file mode 100644 index 3af9f81a..00000000 --- a/router/forwarder.go +++ /dev/null @@ -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, ð, &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 -} diff --git a/router/handshake.go b/router/handshake.go index 8e6d9467..850f9a75 100644 --- a/router/handshake.go +++ b/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)) diff --git a/router/interhost.go b/router/interhost.go new file mode 100644 index 00000000..f3ec29ee --- /dev/null +++ b/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) { +} diff --git a/router/local_peer.go b/router/local_peer.go index 930fbf1b..5afc6d48 100644 --- a/router/local_peer.go +++ b/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) diff --git a/router/protocol.go b/router/protocol.go index 34373c47..de4d0fde 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -9,10 +9,7 @@ type ProtocolTag byte const ( ProtocolHeartbeat ProtocolTag = iota - ProtocolConnectionEstablished - ProtocolFragmentationReceived - ProtocolStartFragmentationTest - ProtocolPMTUVerified + ProtocolInterHostControlMsg ProtocolGossip ProtocolGossipUnicast ProtocolGossipBroadcast diff --git a/router/router.go b/router/router.go index 4edbf253..fc9c9bd8 100644 --- a/router/router.go +++ b/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) } } diff --git a/router/sleeve.go b/router/sleeve.go new file mode 100644 index 00000000..7507294e --- /dev/null +++ b/router/sleeve.go @@ -0,0 +1,1058 @@ +// This contains the InterHost implementation for weave's own UDP +// encapsulation protocol ("sleeve" because a sleeve encapsulates +// something, it's often woven, it rhymes with "weave", make up your +// own cheesy reason). + +package router + +import ( + "bytes" + "encoding/binary" + "fmt" + "io" + "log" + "net" + "sync" + "syscall" + "time" + + "code.google.com/p/gopacket" + "code.google.com/p/gopacket/layers" +) + +type SleeveInterHost struct { + localPort int + + // These fields are set in ConsumePackets, and not + // subsequently modified + localPeer *Peer + localPeerBin []byte + consumer InterHostConsumer + peers *Peers + conn *net.UDPConn + + lock sync.Mutex + forwarders map[PeerName]*sleeveForwarder +} + +func NewSleeveInterHost(localPort int) InterHost { + return &SleeveInterHost{localPort: localPort} +} + +func (sleeve *SleeveInterHost) ConsumePackets(localPeer *Peer, peers *Peers, + consumer InterHostConsumer) error { + localAddr, err := net.ResolveUDPAddr("udp4", + fmt.Sprint(":", sleeve.localPort)) + if err != nil { + return err + } + + conn, err := net.ListenUDP("udp4", localAddr) + if err != nil { + return err + } + + f, err := conn.File() + if err != nil { + return err + } + + defer f.Close() + fd := int(f.Fd()) + + // This 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) + if err != nil { + return err + } + + sleeve.lock.Lock() + defer sleeve.lock.Unlock() + + if sleeve.localPeer != nil { + conn.Close() + return fmt.Errorf("ConsumePackets already called") + } + + sleeve.localPeer = localPeer + sleeve.localPeerBin = localPeer.NameByte + sleeve.consumer = consumer + sleeve.peers = peers + sleeve.conn = conn + sleeve.forwarders = make(map[PeerName]*sleeveForwarder) + go sleeve.readUDP() + return nil +} + +func (sleeve *SleeveInterHost) logFrame(string, []byte, *layers.Ethernet) { + // XXX +} + +func (sleeve *SleeveInterHost) lookupForwarder(peer PeerName) *sleeveForwarder { + sleeve.lock.Lock() + defer sleeve.lock.Unlock() + return sleeve.forwarders[peer] +} + +func (sleeve *SleeveInterHost) readUDP() { + defer sleeve.conn.Close() + dec := NewEthernetDecoder() + buf := make([]byte, MaxUDPPacketSize) + + for { + n, sender, err := sleeve.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 + } + + fwdName := PeerNameFromBin(buf[:NameSize]) + fwd := sleeve.lookupForwarder(fwdName) + if fwd == nil { + continue + } + + packet := make([]byte, n-NameSize) + copy(packet, buf[NameSize:n]) + + err = fwd.crypto.Dec.IterateFrames(packet, + func(src []byte, dst []byte, frame []byte) { + sleeve.handleFrame(sender, fwd, + src, dst, frame, dec) + }) + if 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. + fwd.logFrom(sender, err) + } + } +} + +func (sleeve *SleeveInterHost) handleFrame(sender *net.UDPAddr, + fwd *sleeveForwarder, src []byte, dst []byte, frame []byte, + dec *EthernetDecoder) { + dec.DecodeLayers(frame) + decodedLen := len(dec.decoded) + if decodedLen == 0 { + return + } + + srcPeer, found := sleeve.peers.Fetch(PeerNameFromBin(src)) + if !found { + return + } + + dstPeer, found := sleeve.peers.Fetch(PeerNameFromBin(dst)) + if !found { + 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 == fwd.remotePeer && dstPeer == fwd.sleeve.localPeer { + select { + case fwd.specialChan <- specialFrame{sender, frame}: + case <-fwd.finishedChan: + } + } + + return + } + + if sleeve.consumer == nil { + return + } + + sleeve.consumer(srcPeer, dstPeer, frame, dec) +} + +type udpSender interface { + send([]byte, *net.UDPAddr) error +} + +func (sleeve *SleeveInterHost) send(msg []byte, raddr *net.UDPAddr) error { + sleeve.lock.Lock() + conn := sleeve.conn + sleeve.lock.Unlock() + + if conn == nil { + // Consume wasn't called yet + return nil + } + + _, err := conn.WriteToUDP(msg, raddr) + return err +} + +type sleeveForwarder struct { + // Immutable + sleeve *SleeveInterHost + remotePeer *Peer + remotePeerBin []byte + sendControlMsg func([]byte) error + uid uint64 + + // Channels to communicate with the aggregator goroutine + aggregatorChan chan<- aggregatorFrame + aggregatorDFChan chan<- aggregatorFrame + specialChan chan<- specialFrame + confirmedChan chan<- struct{} + finishedChan <-chan struct{} + + // Explicitly locked state + lock sync.RWMutex + listener InterHostForwarderListener + remoteAddr *net.UDPAddr + + // These fields are accessed and updated independently, so no + // locking needed + effectivePMTU int + stackFrag bool + + // dec is only used from the readUDP goroutine, enc and encDF + // are only used in the forwarder goroutine + crypto InterHostCrypto + + // State only used within the forwarder goroutine + senderDF *udpSenderDF + maxPayload int + heartbeatInterval time.Duration + heartbeatTimer *time.Timer + heartbeatTimeout *time.Timer + fragTestTicker *time.Ticker + ackedHeartbeat bool + + pmtuTestTimeout *time.Timer + pmtuTestsSent uint + epmtuHighestGood int + epmtuLowestBad int + epmtuCandidate int +} + +type aggregatorFrame struct { + src []byte + dst []byte + frame []byte +} + +// A "special" message over UDP, or a control message. The sender is +// nil for control messages. +type specialFrame struct { + sender *net.UDPAddr + frame []byte +} + +func (sleeve *SleeveInterHost) MakeForwarder(remotePeer *Peer, + localIP net.IP, remote *net.UDPAddr, + connUid uint64, crypto InterHostCrypto, + sendControlMsg func([]byte) error) (InterHostForwarder, error) { + aggChan := make(chan aggregatorFrame, ChannelSize) + aggDFChan := make(chan aggregatorFrame, ChannelSize) + specialChan := make(chan specialFrame, 1) + confirmedChan := make(chan struct{}) + finishedChan := make(chan struct{}) + remotePeerBin := remotePeer.NameByte + + fwd := &sleeveForwarder{ + sleeve: sleeve, + remotePeer: remotePeer, + remotePeerBin: remotePeerBin, + sendControlMsg: sendControlMsg, + uid: connUid, + aggregatorChan: aggChan, + aggregatorDFChan: aggDFChan, + specialChan: specialChan, + confirmedChan: confirmedChan, + finishedChan: finishedChan, + remoteAddr: remote, + effectivePMTU: DefaultPMTU, + crypto: crypto, + senderDF: newUDPSenderDF(localIP, sleeve.localPort), + maxPayload: DefaultPMTU - UDPOverhead, + } + + // XXX handle case where remotePeer already registered + sleeve.lock.Lock() + sleeve.forwarders[remotePeer.Name] = fwd + sleeve.lock.Unlock() + + go fwd.run(aggChan, aggDFChan, specialChan, confirmedChan, + finishedChan) + return fwd, nil +} + +func (fwd *sleeveForwarder) logFrom(sender *net.UDPAddr, args ...interface{}) { + parts := make([]interface{}, len(args)+1) + parts[0] = fmt.Sprintf("->[%s|%s]:", sender, fwd.remotePeer) + copy(parts[1:], args) + log.Println(parts...) +} + +func (fwd *sleeveForwarder) log(args ...interface{}) { + fwd.lock.RLock() + remoteAddr := fwd.remoteAddr + fwd.lock.RUnlock() + fwd.logFrom(remoteAddr, args...) +} + +func (fwd *sleeveForwarder) debug(args ...interface{}) { + fwd.lock.RLock() + remoteAddr := fwd.remoteAddr + fwd.lock.RUnlock() + fwd.logFrom(remoteAddr, args...) +} + +func (fwd *sleeveForwarder) SetListener(listener InterHostForwarderListener) { + fwd.debug("SetListener", listener) + + fwd.lock.Lock() + fwd.listener = listener + fwd.lock.Unlock() + + // Setting the listener confirms that the forwarder is really + // wanted + if listener != nil { + select { + case fwd.confirmedChan <- struct{}{}: + case <-fwd.finishedChan: + } + } +} + +func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, + dec *EthernetDecoder) error { + var df bool + if dec != nil { + df = dec.DF() + } + + fwd.lock.RLock() + haveContact := (fwd.remoteAddr != nil) + effectivePMTU := fwd.effectivePMTU + stackFrag := fwd.stackFrag + fwd.lock.RUnlock() + + if !haveContact { + fwd.log("Cannot forward frame yet - awaiting contact") + return nil + } + + srcName := src.NameByte + dstName := dst.NameByte + + // 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) { + return FrameTooBigError{EPMTU: effectivePMTU} + } + + fwd.aggregate(fwd.aggregatorDFChan, srcName, dstName, frame) + return nil + } + + if stackFrag || dec == nil || len(dec.decoded) < 2 { + fwd.aggregate(fwd.aggregatorChan, srcName, dstName, 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) { + fwd.aggregate(fwd.aggregatorDFChan, srcName, dstName, frame) + return nil + } + + fwd.sleeve.logFrame("Fragmenting", frame, &dec.Eth) + + // 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, + func(segFrame []byte) { + fwd.aggregate(fwd.aggregatorDFChan, srcName, dstName, + segFrame) + }) +} + +func (fwd *sleeveForwarder) aggregate(ch chan<- aggregatorFrame, src []byte, + dst []byte, frame []byte) { + select { + case ch <- aggregatorFrame{src, dst, frame}: + case <-fwd.finishedChan: + } +} + +type FrameTooBigError struct { + EPMTU int // effective pmtu, i.e. what we tell packet senders +} + +func (ftbe FrameTooBigError) Error() string { + return fmt.Sprint("Frame too big error. Effective PMTU is ", ftbe.EPMTU) +} + +func fragment(eth layers.Ethernet, ip layers.IPv4, pmtu int, + forward func([]byte)) 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, ð, &ip, + &segPayload) + if err != nil { + return err + } + + forward(buf.Bytes()) + } + return nil +} + +func frameTooBig(frame []byte, 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) > effectivePMTU+EthernetOverhead +} + +func (fwd *sleeveForwarder) ControlMessage(msg []byte) { + select { + case fwd.specialChan <- specialFrame{nil, msg}: + case <-fwd.finishedChan: + } +} + +func (fwd *sleeveForwarder) Close() { + sleeve := fwd.sleeve + sleeve.lock.Lock() + if sleeve.forwarders[fwd.remotePeer.Name] == fwd { + delete(sleeve.forwarders, fwd.remotePeer.Name) + } + sleeve.lock.Unlock() + fwd.SetListener(nil) + + // Tell the forwarder goroutine to finish. We don't need to + // wait for it. + close(fwd.confirmedChan) +} + +func (fwd *sleeveForwarder) run(aggChan <-chan aggregatorFrame, + aggDFChan <-chan aggregatorFrame, + specialChan <-chan specialFrame, + confirmedChan <-chan struct{}, + finishedChan chan<- struct{}) { + defer close(finishedChan) + + var err error +loop: + for err == nil { + select { + case frame := <-aggChan: + err = fwd.aggregateAndSend(frame, aggChan, + fwd.crypto.Enc, fwd.sleeve, + MaxUDPPacketSize-UDPOverhead) + + case frame := <-aggDFChan: + err = fwd.aggregateAndSend(frame, aggDFChan, + fwd.crypto.EncDF, fwd.senderDF, fwd.maxPayload) + + case special := <-specialChan: + err = fwd.handleSpecialFrame(special) + + case _, ok := <-confirmedChan: + if !ok { + // specialChan is closed to indicate + // the forwarder is being closed + break loop + } + + err = fwd.confirmed() + + case <-timerChan(fwd.heartbeatTimer): + err = fwd.sendHeartbeat() + + case <-timerChan(fwd.heartbeatTimeout): + err = fmt.Errorf("timed out waiting for UDP heartbeat") + + case <-tickerChan(fwd.fragTestTicker): + err = fwd.sendFragTest() + + case <-timerChan(fwd.pmtuTestTimeout): + err = fwd.handlePMTUTestFailure() + } + } + + if fwd.heartbeatTimer != nil { + fwd.heartbeatTimer.Stop() + } + if fwd.heartbeatTimeout != nil { + fwd.heartbeatTimeout.Stop() + } + if fwd.fragTestTicker != nil { + fwd.fragTestTicker.Stop() + } + if fwd.pmtuTestTimeout != nil { + fwd.pmtuTestTimeout.Stop() + } + + checkWarn(fwd.senderDF.close()) + + fwd.lock.RLock() + defer fwd.lock.RUnlock() + if fwd.listener != nil { + fwd.listener.Error(err) + } +} + +func (fwd *sleeveForwarder) aggregateAndSend(frame aggregatorFrame, + aggChan <-chan aggregatorFrame, enc Encryptor, sender udpSender, + limit int) error { + // Give up after processing N frames, to avoid starving the + // other activities of the forwarder goroutine. + i := 0 + + for { + // Adding the first frame to an empty buffer + if !fits(frame, enc, limit) { + fwd.log("Dropping too big frame during forwarding: frame len", len(frame.frame), ", limit ", limit) + return nil + } + + for { + enc.AppendFrame(frame.src, frame.dst, frame.frame) + i++ + + gotOne := false + if i < 100 { + select { + case frame = <-aggChan: + gotOne = true + default: + } + } + + if !gotOne { + return fwd.flushEncryptor(enc, sender) + } + + // Accumulate frames until doing so would + // exceed the PMTU. Even in the non-DF case, + // it doesn't seem worth adding a frame where + // that would lead to fragmentation, + // potentially delaying or risking other + // frames. + if !fits(frame, enc, fwd.maxPayload) { + break + } + } + + if err := fwd.flushEncryptor(enc, sender); err != nil { + return err + } + } +} + +func fits(frame aggregatorFrame, enc Encryptor, limit int) bool { + return enc.TotalLen()+enc.FrameOverhead()+len(frame.frame) <= limit +} + +func (fwd *sleeveForwarder) flushEncryptor(enc Encryptor, + sender udpSender) error { + msg, err := enc.Bytes() + if err != nil { + return err + } + + return fwd.processSendError(sender.send(msg, fwd.remoteAddr)) +} + +func (fwd *sleeveForwarder) sendSpecial(enc Encryptor, sender udpSender, + data []byte) error { + enc.AppendFrame(fwd.sleeve.localPeerBin, fwd.remotePeerBin, data) + return fwd.flushEncryptor(enc, sender) +} + +func (fwd *sleeveForwarder) handleSpecialFrame(special specialFrame) error { + if special.sender == nil { + return fwd.handleControlMsg(special.frame) + } + + // The special frame types are distinguished by length + switch len(special.frame) { + case EthernetOverhead + 8: + return fwd.handleHeartbeat(special) + + case FragTestSize: + return fwd.handleFragTest(special.frame) + + default: + return fwd.handlePMTUTest(special.frame) + } +} + +const ( + HeartbeatAck = iota + FragTestAck + PMTUTestAck +) + +func (fwd *sleeveForwarder) handleControlMsg(msg []byte) error { + if len(msg) == 0 { + fwd.log("Received zero-length control message") + return nil + } + + switch msg[0] { + case HeartbeatAck: + return fwd.handleHeartbeatAck() + + case FragTestAck: + return fwd.handleFragTestAck() + + case PMTUTestAck: + return fwd.handlePMTUTestAck(msg) + + default: + fwd.log("Ignoring unknown control message:", msg[0]) + return nil + } +} + +func (fwd *sleeveForwarder) confirmed() error { + fwd.debug("confirmed") + + if fwd.heartbeatInterval != 0 { + // already confirmed + return nil + } + + // heartbeatInterval flags that we want to send heartbeats, + // even if we don't do sendHeartbeat() yet due to lacking the + // remote address. + fwd.heartbeatInterval = time.Duration(FastHeartbeat) + if fwd.remoteAddr != nil { + if err := fwd.sendHeartbeat(); err != nil { + return err + } + } + + fwd.heartbeatTimeout = time.NewTimer(HeartbeatTimeout) + return nil +} + +func (fwd *sleeveForwarder) sendHeartbeat() error { + fwd.debug("sendHeartbeat") + + // Prime the timer for the next heartbeat. We don't use a + // ticker because the interval is not constant. + fwd.heartbeatTimer = setTimer(fwd.heartbeatTimer, fwd.heartbeatInterval) + + buf := make([]byte, EthernetOverhead+8) + binary.BigEndian.PutUint64(buf[EthernetOverhead:], fwd.uid) + return fwd.sendSpecial(fwd.crypto.EncDF, fwd.senderDF, buf) +} + +func (fwd *sleeveForwarder) handleHeartbeat(special specialFrame) error { + uid := binary.BigEndian.Uint64(special.frame[EthernetOverhead:]) + if uid != fwd.uid { + return nil + } + + fwd.debug("handleHeartbeat") + + if fwd.remoteAddr == nil { + fwd.setRemoteAddr(special.sender) + if fwd.heartbeatInterval != time.Duration(0) { + if err := fwd.sendHeartbeat(); err != nil { + return err + } + } + } else if !udpAddrsEqual(fwd.remoteAddr, special.sender) { + fwd.log("Peer UDP address changed to", special.sender) + fwd.setRemoteAddr(special.sender) + } + + if !fwd.ackedHeartbeat { + fwd.ackedHeartbeat = true + if err := fwd.sendControlMsg([]byte{HeartbeatAck}); err != nil { + return err + } + } + + fwd.heartbeatTimeout.Reset(HeartbeatTimeout) + return nil +} + +func (fwd *sleeveForwarder) setRemoteAddr(addr *net.UDPAddr) { + // Although we don't need to lock when reading remoteAddr, + // because this thread is the only one that modifies + // remoteAddr, we do need to lock when writing it, because + // memory models. + fwd.lock.Lock() + fwd.remoteAddr = addr + fwd.lock.Unlock() +} + +func (fwd *sleeveForwarder) handleHeartbeatAck() error { + // The connection is nowregarded as established + fwd.notifyEstablished() + + if fwd.heartbeatInterval != SlowHeartbeat { + fwd.heartbeatInterval = SlowHeartbeat + if fwd.heartbeatTimer != nil { + fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) + } + } + + fwd.fragTestTicker = time.NewTicker(FragTestInterval) + if err := fwd.sendFragTest(); err != nil { + return err + } + + // Send a large frame down the DF channel in order to prompt + // PMTU discovery to start. + return fwd.sendSpecial(fwd.crypto.EncDF, fwd.senderDF, + make([]byte, PMTUDiscoverySize)) +} + +func (fwd *sleeveForwarder) notifyEstablished() { + fwd.lock.RLock() + defer fwd.lock.RUnlock() + if fwd.listener != nil { + fwd.listener.Established() + } +} + +func (fwd *sleeveForwarder) sendFragTest() error { + fwd.debug("sendFragTest") + fwd.stackFrag = false + return fwd.sendSpecial(fwd.crypto.Enc, fwd.sleeve, + make([]byte, FragTestSize)) +} + +func (fwd *sleeveForwarder) handleFragTest(frame []byte) error { + if !allZeros(frame) { + return nil + } + + return fwd.sendControlMsg([]byte{FragTestAck}) +} + +func (fwd *sleeveForwarder) handleFragTestAck() error { + fwd.debug("handleFragTestAck") + fwd.stackFrag = true + return nil +} + +func (fwd *sleeveForwarder) processSendError(err error) error { + if mtbe, ok := err.(msgTooBigError); ok { + epmtu := mtbe.PMTU - fwd.overhead() + if fwd.epmtuCandidate != 0 && epmtu >= fwd.epmtuCandidate { + return nil + } + + fwd.epmtuHighestGood = 8 + fwd.epmtuLowestBad = epmtu + 1 + fwd.epmtuCandidate = epmtu + fwd.pmtuTestsSent = 0 + fwd.maxPayload = mtbe.PMTU - UDPOverhead + fwd.effectivePMTU = epmtu + return fwd.sendPMTUTest() + } + + return err +} + +// How many bytes of overhead it takes to turn an IP packet on the +// overlay network into an encapsulated packet on the underlay network +func (fwd *sleeveForwarder) overhead() int { + return UDPOverhead + fwd.crypto.EncDF.PacketOverhead() + + fwd.crypto.EncDF.FrameOverhead() + EthernetOverhead +} + +func (fwd *sleeveForwarder) sendPMTUTest() error { + fwd.debug("sendPMTUTest: epmtu candidate", fwd.epmtuCandidate) + err := fwd.sendSpecial(fwd.crypto.EncDF, fwd.senderDF, + make([]byte, fwd.epmtuCandidate+EthernetOverhead)) + if err != nil { + return err + } + + fwd.pmtuTestTimeout = setTimer(fwd.pmtuTestTimeout, + PMTUVerifyTimeout<= fwd.epmtuLowestBad { + epmtu := fwd.epmtuHighestGood + fwd.log("Effective PMTU verified at", epmtu) + + if fwd.pmtuTestTimeout != nil { + fwd.pmtuTestTimeout.Stop() + fwd.pmtuTestTimeout = nil + } + + fwd.epmtuCandidate = 0 + fwd.maxPayload = epmtu + fwd.overhead() - UDPOverhead + fwd.effectivePMTU = epmtu + return nil + } + + fwd.epmtuCandidate = (fwd.epmtuHighestGood + fwd.epmtuLowestBad) / 2 + fwd.pmtuTestsSent = 0 + return fwd.sendPMTUTest() +} + +type udpSenderDF struct { + ipBuf gopacket.SerializeBuffer + opts gopacket.SerializeOptions + udpHeader *layers.UDP + localIP net.IP + remoteIP net.IP + socket *net.IPConn +} + +func newUDPSenderDF(localIP net.IP, localPort int) *udpSenderDF { + return &udpSenderDF{ + 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, + }, + udpHeader: &layers.UDP{SrcPort: layers.UDPPort(localPort)}, + localIP: localIP, + } +} + +func (sender *udpSenderDF) dial() error { + if sender.socket != nil { + if err := sender.socket.Close(); err != nil { + return err + } + + sender.socket = nil + } + + laddr := &net.IPAddr{IP: sender.localIP} + raddr := &net.IPAddr{IP: sender.remoteIP} + s, err := net.DialIP("ip4:UDP", laddr, raddr) + + f, err := s.File() + if err != nil { + return err + } + + defer f.Close() + + // This makes sure all packets we send out have DF set on them. + err = syscall.SetsockoptInt(int(f.Fd()), syscall.IPPROTO_IP, + syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DO) + if err != nil { + return err + } + + sender.socket = s + return nil +} + +func (sender *udpSenderDF) send(msg []byte, raddr *net.UDPAddr) error { + // Ensure we have a socket sending to the right IP address + if sender.socket == nil || !bytes.Equal(sender.remoteIP, raddr.IP) { + sender.remoteIP = raddr.IP + if err := sender.dial(); err != nil { + return err + } + } + + sender.udpHeader.DstPort = layers.UDPPort(raddr.Port) + payload := gopacket.Payload(msg) + 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() + + log.Println("EMSGSIZE on send, expecting PMTU update (IP packet was", + len(packet), "bytes, payload was", len(msg), "bytes)") + pmtu, err := syscall.GetsockoptInt(int(f.Fd()), syscall.IPPROTO_IP, + syscall.IP_MTU) + if err != nil { + return err + } + + return msgTooBigError{PMTU: pmtu} +} + +type msgTooBigError struct { + PMTU int // actual pmtu, i.e. what the kernel told us +} + +func (mtbe msgTooBigError) Error() string { + return fmt.Sprint("Msg too big error. PMTU is ", mtbe.PMTU) +} + +func (sender *udpSenderDF) close() error { + if sender.socket == nil { + return nil + } + + return sender.socket.Close() +} + +func udpAddrsEqual(a *net.UDPAddr, b *net.UDPAddr) bool { + return bytes.Equal(a.IP, b.IP) && a.Port == b.Port && a.Zone == b.Zone +} + +func allZeros(s []byte) bool { + for _, b := range s { + if b != byte(0) { + return false + } + } + + return true +} + +func setTimer(timer *time.Timer, d time.Duration) *time.Timer { + if timer == nil { + return time.NewTimer(d) + } + + timer.Reset(d) + return timer + +} + +func timerChan(timer *time.Timer) <-chan time.Time { + if timer != nil { + return timer.C + } + return nil +} diff --git a/router/udp_sender.go b/router/udp_sender.go deleted file mode 100644 index 9e91ee0b..00000000 --- a/router/udp_sender.go +++ /dev/null @@ -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 -} diff --git a/router/utils.go b/router/utils.go index 9cf4f1bd..c9174cea 100644 --- a/router/utils.go +++ b/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) } diff --git a/weaver/main.go b/weaver/main.go index ea0239b9..3f1c94d2 100644 --- a/weaver/main.go +++ b/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)