Merge pull request #349 from rade/342_unidirectional_udp

LGTM.  Fixes #342

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

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/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()

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 {