From 27c9e70ea9cb900339282cad6527b1e6f29dcb35 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Thu, 15 Jan 2015 14:16:16 +0000 Subject: [PATCH 1/3] better handling of unidirectional udp connectivity Currently we cannot distinguish bi-directional and uni-directional connectivity between peers. This is what currently happens for a connection from A to B... A: remoteUDPAddr=X B: remoteUDPAddr=nil A: send fast heartbeats over UDP B: receive heartbeat; remoteUDPAddr is nil remoteUDPAddr=Y send 'established' over TCP established=true send slow heartbeats over UDP A: receive 'established' over TCP established=true send slow heartbeats over UDP If udp packets make it through from A to B but not B to A then both ends mark the connection as 'established'. Conversely, if udp packets make it through from B to A but not A to B then the connection is not marked as 'established' at either end. In the new scheme, this is what happens... A: receivedHeartbeat=false, remoteUDPAddr=X B: receivedHeartbeat=false, remoteUDPAddr=nil A: send fast heartbeats over UDP B: receive heartbeat; receivedHeartbeat is false remoteUDPAddr=Y receivedHeartbeat=true send 'established' over TCP send fast heartbeats over UDP A: receive 'established' over TCP (*) established=true send slow heartbeats over UDP A: receive heartbeat; receivedHeartbeat is false (*) receivedHeartbeat=true send 'established' over TCP B: receive 'established' over TCP established=true send slow heartbeats over UDP (*) either of these may occur first Now, if udp packets make it through from A to B not not B to A then A marks the connection as 'established' but B doesn't. Conversely, if udp packets make it through from B to A but not A to B then B marks the connection as 'established' but A doesn't. --- router/connection.go | 60 ++++++++++++++++++++++++++------------------ router/consts.go | 2 +- router/router.go | 2 +- router/types.go | 39 ++++++++++++++-------------- 4 files changed, 58 insertions(+), 45 deletions(-) diff --git a/router/connection.go b/router/connection.go index eb340726..7107e272 100644 --- a/router/connection.go +++ b/router/connection.go @@ -123,7 +123,7 @@ func (conn *LocalConnection) log(args ...interface{}) { const ( CSendTCP = iota CSetEstablished - CSetRemoteUDPAddr + CReceivedHeartbeat CShutdown ) @@ -135,12 +135,12 @@ func (conn *LocalConnection) Shutdown(err error) { } // Async -func (conn *LocalConnection) SetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) { +func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) { if remoteUDPAddr == nil { return } conn.queryChan <- &ConnectionInteraction{ - Interaction: Interaction{code: CSetRemoteUDPAddr}, + Interaction: Interaction{code: CReceivedHeartbeat}, payload: remoteUDPAddr} } @@ -176,10 +176,7 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, accept conn.Router.Ourself.AddConnection(conn) if conn.remoteUDPAddr != nil { - if err := conn.ensureForwarders(); err == nil { - conn.heartbeat = time.NewTicker(FastHeartbeat) - conn.forwardHeartbeatFrame() // avoid initial wait - } else { + if err := conn.sendFastHeartbeats(); err != nil { conn.log("connection shutting down due to error:", err) return } @@ -207,10 +204,10 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) case CShutdown: err = query.payload.(error) terminate = true + case CReceivedHeartbeat: + err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr)) case CSetEstablished: err = conn.handleSetEstablished() - case CSetRemoteUDPAddr: - err = conn.handleSetRemoteUDPAddr(query.payload.(*net.UDPAddr)) case CSendTCP: err = conn.handleSendTCP(query.payload.([]byte)) } @@ -226,25 +223,36 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) return } -func (conn *LocalConnection) handleSetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) error { +// Handlers +// +// NB: The conn.* fields are only written by the connection actor +// process, which is the caller of the handlers. Hence we do not need +// locks for reading, and only need write locks for fields read by +// other processes. + +func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) error { + oldRemoteUDPAddr := conn.remoteUDPAddr + old := conn.receivedHeartbeat conn.Lock() - old := conn.remoteUDPAddr conn.remoteUDPAddr = remoteUDPAddr + conn.receivedHeartbeat = true conn.Unlock() - if old == nil { + if !old { if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil { return err } - return conn.handleSetEstablished() - } else if old.String() != remoteUDPAddr.String() { + } + if oldRemoteUDPAddr == nil { + return conn.sendFastHeartbeats() + } else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() { log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) } return nil } func (conn *LocalConnection) handleSetEstablished() error { - conn.Lock() old := conn.established + conn.Lock() conn.established = true conn.Unlock() if old { @@ -275,14 +283,6 @@ func (conn *LocalConnection) handleSendTCP(msg []byte) error { } func (conn *LocalConnection) handleShutdown() { - // Whilst some of these elements may have been written to whilst - // holding locks, they were only written to by the connection - // actor process. handleShutdown is only called by the connection - // actor process. So there is no need to take locks to read these - // (or write elements which are only read by the same - // process). Taking locks is only done for elements which are read - // by other processes. - if conn.TCPConn != nil { checkWarn(conn.TCPConn.Close()) } @@ -303,8 +303,10 @@ func (conn *LocalConnection) handleShutdown() { conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr) } +// Helpers + func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, acceptNewPeer bool) error { - // We do not need to worry about locking in here as at this point, + // We do not need to worry about locking in here as at this point // the connection is not reachable by any go-routine other than // ourself. Only when we add this connection to the conn.local // peer will it be visible from multiple go-routines. @@ -526,6 +528,16 @@ func (conn *LocalConnection) extendReadDeadline() { conn.TCPConn.SetReadDeadline(time.Now().Add(ReadTimeout)) } + +func (conn *LocalConnection) sendFastHeartbeats() error { + err := conn.ensureForwarders() + if err == nil { + conn.heartbeat = time.NewTicker(FastHeartbeat) + conn.forwardHeartbeatFrame() // avoid initial wait + } + return err +} + // 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). diff --git a/router/consts.go b/router/consts.go index 1b94a73f..5bdd14f8 100644 --- a/router/consts.go +++ b/router/consts.go @@ -7,7 +7,7 @@ import ( const ( Protocol = "weave" - ProtocolVersion = 8 + ProtocolVersion = 9 EthernetOverhead = 14 UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP Port = 6783 diff --git a/router/router.go b/router/router.go index a8b833d3..7e05ec04 100644 --- a/router/router.go +++ b/router/router.go @@ -265,7 +265,7 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F if relayConn.Remote().Name == srcPeer.Name { if frameLen == 0 { - relayConn.SetRemoteUDPAddr(sender) + relayConn.ReceivedHeartbeat(sender) return nil } else if frameLen == FragTestSize && bytes.Equal(frame, FragTest) { relayConn.SendTCP(ProtocolFragmentationReceivedByte) diff --git a/router/types.go b/router/types.go index 850767ac..d32ccaf5 100644 --- a/router/types.go +++ b/router/types.go @@ -42,25 +42,26 @@ type RemoteConnection struct { type LocalConnection struct { sync.RWMutex RemoteConnection - TCPConn *net.TCPConn - tcpSender TCPSender - remoteUDPAddr *net.UDPAddr - established bool - stackFrag bool - effectivePMTU int - SessionKey *[32]byte - heartbeat *time.Ticker - fetchAll *time.Ticker - fragTest *time.Ticker - forwardChan chan<- *ForwardedFrame - forwardChanDF chan<- *ForwardedFrame - stopForward chan<- interface{} - stopForwardDF chan<- interface{} - verifyPMTU chan<- int - Decryptor Decryptor - Router *Router - UID uint64 - queryChan chan<- *ConnectionInteraction + TCPConn *net.TCPConn + tcpSender TCPSender + remoteUDPAddr *net.UDPAddr + established bool + receivedHeartbeat bool + stackFrag bool + effectivePMTU int + SessionKey *[32]byte + heartbeat *time.Ticker + fetchAll *time.Ticker + fragTest *time.Ticker + forwardChan chan<- *ForwardedFrame + forwardChanDF chan<- *ForwardedFrame + stopForward chan<- interface{} + stopForwardDF chan<- interface{} + verifyPMTU chan<- int + Decryptor Decryptor + Router *Router + UID uint64 + queryChan chan<- *ConnectionInteraction } type ConnectionInteraction struct { From f2d6b5fdb0bdfbbecc4616211a30c40dca96fad8 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Thu, 15 Jan 2015 14:58:24 +0000 Subject: [PATCH 2/3] update docs --- docs/architecture.txt | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/docs/architecture.txt b/docs/architecture.txt index 1461192a..4dba5c45 100644 --- a/docs/architecture.txt +++ b/docs/architecture.txt @@ -83,13 +83,17 @@ Router: local peer. It will tell the local connection (communicating to the actor thread) about the UDP address of the remote peer. The local connection will then start its forwarder threads as - described in 6a, and start sending slow heartbeats. The local connection - is marked as established which means it is included in network updates - broadcast to our peers. We also send to the remote peer via TCP a - ConnectionEstablished message. The remote peer receives this (on - the TCP receiver process), tells the connection actor process, - which then replaces the fast heartbeater with a slow heartbeater - and similarly marks the connection as established. + described in 6a, and start sending fast heartbeats. We send to the + remote peer via TCP a ConnectionEstablished message. The remote + peer receives this (on the TCP receiver process), tells the + connection actor process, which then replaces the fast heartbeater + with a slow heartbeater and marks the connection as established + (which means it is included in network updates broadcast to our + peers). + 6c. When the connection initiator receives the fast heartbeat from + the remote peer it sends to the remote peer via a TCP a + ConnectionEstablished message. This is handled by the remote peer + as described in 6b. 7. Whenever a connection is established or terminated, the local peer's version is incremented. Whenever this happens, the peer generates a network update message which is broadcast to its From 9428b7c11664a82819a27d2aeaa3b39f58c486be Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Thu, 15 Jan 2015 15:10:13 +0000 Subject: [PATCH 3/3] report 'unestablished' in status --- router/peers.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/router/peers.go b/router/peers.go index fd5d5782..071fb5e0 100644 --- a/router/peers.go +++ b/router/peers.go @@ -112,7 +112,11 @@ func (peers *Peers) String() string { peers.ForEach(func(name PeerName, peer *Peer) { buf.WriteString(fmt.Sprint(peer, "\n")) peer.ForEachConnection(func(remoteName PeerName, conn Connection) { - buf.WriteString(fmt.Sprintf(" -> %v [%v]\n", remoteName, conn.RemoteTCPAddr())) + established := "" + if !conn.Established() { + established = " (unestablished)" + } + buf.WriteString(fmt.Sprintf(" -> %v [%v%s]\n", remoteName, conn.RemoteTCPAddr(), established)) }) }) return buf.String()