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

@@ -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).

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

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)

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 {