Merge pull request #381 from rade/380_prevent_premature_peer_topology_gc

LGTM.  Fixes #380

router/connection.go

@@ -33,6 +33,7 @@ type RemoteConnection struct {
 	local         *Peer
 	remote        *Peer
 	remoteTCPAddr string
+	established   bool
 }
 
 type LocalConnection struct {
@@ -41,7 +42,6 @@ type LocalConnection struct {
 	TCPConn            *net.TCPConn
 	tcpSender          TCPSender
 	remoteUDPAddr      *net.UDPAddr
-	established        bool
 	receivedHeartbeat  bool
 	stackFrag          bool
 	effectivePMTU      int
@@ -66,11 +66,12 @@ type ConnectionInteraction struct {
 	payload interface{}
 }
 
-func NewRemoteConnection(from, to *Peer, tcpAddr string) *RemoteConnection {
+func NewRemoteConnection(from, to *Peer, tcpAddr string, established bool) *RemoteConnection {
 	return &RemoteConnection{
 		local:         from,
 		remote:        to,
-		remoteTCPAddr: tcpAddr}
+		remoteTCPAddr: tcpAddr,
+		established:   established}
 }
 
 func (conn *RemoteConnection) Local() *Peer {
@@ -90,7 +91,7 @@ func (conn *RemoteConnection) RemoteTCPAddr() string {
 }
 
 func (conn *RemoteConnection) Established() bool {
-	return true
+	return conn.established
 }
 
 func (conn *RemoteConnection) Shutdown(error) {

router/gossip_test.go

@@ -41,7 +41,7 @@ func (router *Router) AddTestChannelConnection(r *Router) {
 	r.Peers.FetchWithDefault(fromPeer)    // Has side-effect of incrementing refcount
 	router.Peers.FetchWithDefault(toPeer) //
 
-	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, ""}, r}
+	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false}, r}
 	router.Ourself.handleAddConnection(conn)
 	router.Ourself.handleConnectionEstablished(conn)
 }

router/local_peer.go

@@ -108,7 +108,7 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
 	if err != nil {
 		return err
 	}
-	connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String())
+	connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), false)
 	connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.Router)
 	connLocal.Start(acceptNewPeer)
 	return nil
@@ -226,6 +226,8 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) bool {
 		return false
 	}
 	peer.addConnection(conn)
+	peer.Router.SendAllGossipDown(conn)
+	peer.broadcastPeerUpdate(conn.Remote())
 	return true
 }
 
@@ -238,8 +240,7 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) bool {
 		return false
 	}
 	peer.connectionEstablished(conn)
-	peer.Router.SendAllGossipDown(conn)
-	peer.broadcastPeerUpdate(conn.Remote())
+	peer.broadcastPeerUpdate()
 	return true
 }
 
@@ -258,9 +259,7 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) bool {
 	// Must do garbage collection first to ensure we don't send out an
 	// update with unreachable peers (can cause looping)
 	peer.Router.Peers.GarbageCollect()
-	if conn.Established() {
-		peer.broadcastPeerUpdate()
-	}
+	peer.broadcastPeerUpdate()
 	return true
 }
 

router/mocks_test.go

@@ -27,7 +27,7 @@ func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) {
 	toName := p2.Name
 	toPeer := NewPeer(toName, p2.UID, 0)
 	toPeer = peers.FetchWithDefault(toPeer)
-	peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, ""})
+	peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false})
 }
 
 func (peers *Peers) DeleteTestConnection(p *Peer) {
@@ -44,7 +44,7 @@ func (peers *Peers) DeleteTestConnection(p *Peer) {
 // from what is created by the real code.
 func newMockConnection(from, to *Peer) Connection {
 	type mockConnection struct{ RemoteConnection }
-	return &mockConnection{RemoteConnection{from, to, ""}}
+	return &mockConnection{RemoteConnection{from, to, "", false}}
 }
 
 func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Connection) {

router/peer.go

@@ -90,16 +90,14 @@ func (peer *Peer) addConnection(conn Connection) {
 	peer.Lock()
 	defer peer.Unlock()
 	peer.connections[conn.Remote().Name] = conn
+	peer.version += 1
 }
 
 func (peer *Peer) deleteConnection(conn Connection) {
-	established := conn.Established()
 	peer.Lock()
 	defer peer.Unlock()
 	delete(peer.connections, conn.Remote().Name)
-	if established {
-		peer.version += 1
-	}
+	peer.version += 1
 }
 
 func (peer *Peer) connectionEstablished(conn Connection) {
@@ -121,9 +119,9 @@ func (peer *Peer) connectionEstablished(conn Connection) {
 // same result. This is important since otherwise we risk message loss
 // or routing cycles.
 //
-// When the 'symmetric' flag is set, only symmetric connections are
-// considered, i.e. where both sides indicate they have a connection
-// to the other.
+// When the 'establishedAndSymmetric' flag is set, only connections
+// that are marked as 'established' and are symmetric (i.e. where both
+// sides indicate they have a connection to the other) are considered.
 //
 // When a non-nil stopAt peer is supplied, the widening stops when it
 // reaches that peer. The boolean return indicates whether that has
@@ -133,7 +131,7 @@ func (peer *Peer) connectionEstablished(conn Connection) {
 // traversal. This prevents the connectivity graph from changing
 // underneath us in ways that would invalidate the result. Thus the
 // answer returned may be out of date, but never inconsistent.
-func (peer *Peer) Routes(stopAt *Peer, symmetric bool) (bool, map[PeerName]PeerName) {
+func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) {
 	peer.RLock()
 	defer peer.RUnlock()
 	routes := make(map[PeerName]PeerName)
@@ -149,12 +147,17 @@ func (peer *Peer) Routes(stopAt *Peer, symmetric bool) (bool, map[PeerName]PeerN
 			}
 			curName := curPeer.Name
 			for remoteName, conn := range curPeer.connections {
+				if establishedAndSymmetric && !conn.Established() {
+					continue
+				}
 				if _, found := routes[remoteName]; found {
 					continue
 				}
 				remote := conn.Remote()
 				remote.RLock()
-				if _, found := remote.connections[curName]; !symmetric || found {
+				// DANGER holding rlock on remote, possibly taking
+				// rlock on remoteConn
+				if remoteConn, found := remote.connections[curName]; !establishedAndSymmetric || (found && remoteConn.Established()) {
 					defer remote.RUnlock()
 					nextWorklist = append(nextWorklist, remote)
 					// We now know how to get to remoteName: the same

router/peers.go

@@ -199,7 +199,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
 	}
 
 	for _, connsBuf := range decodedConns {
-		decErr := connsIterator(connsBuf, func(remoteNameByte []byte, _ string) {
+		decErr := connsIterator(connsBuf, func(remoteNameByte []byte, _ string, _ bool) {
 			remoteName := PeerNameFromBin(remoteNameByte)
 			if _, found := newPeers[remoteName]; found {
 				return
@@ -263,12 +263,10 @@ func (peer *Peer) encode(enc *gob.Encoder) {
 	connsBuf := new(bytes.Buffer)
 	connsEnc := gob.NewEncoder(connsBuf)
 	for _, conn := range peer.connections {
-		// DANGER holding rlock on peer, going to take rlock on conn
-		if !conn.Established() {
-			continue
-		}
 		checkFatal(connsEnc.Encode(conn.Remote().NameByte))
 		checkFatal(connsEnc.Encode(conn.RemoteTCPAddr()))
+		// DANGER holding rlock on peer, going to take rlock on conn
+		checkFatal(connsEnc.Encode(conn.Established()))
 	}
 	checkFatal(enc.Encode(connsBuf.Bytes()))
 }
@@ -289,7 +287,7 @@ func decodePeerNoConns(dec *gob.Decoder) (nameByte []byte, uid uint64, version u
 	return
 }
 
-func connsIterator(input []byte, fun func([]byte, string)) error {
+func connsIterator(input []byte, fun func([]byte, string, bool)) error {
 	buf := new(bytes.Buffer)
 	buf.Write(input)
 	dec := gob.NewDecoder(buf)
@@ -302,16 +300,20 @@ func connsIterator(input []byte, fun func([]byte, string)) error {
 		if err := dec.Decode(&foundAt); err != nil {
 			return err
 		}
-		fun(nameByte, string(foundAt))
+		var established bool
+		if err := dec.Decode(&established); err != nil {
+			return err
+		}
+		fun(nameByte, foundAt, established)
 	}
 }
 
 func readConnsMap(peer *Peer, buf []byte, table map[PeerName]*Peer) map[PeerName]Connection {
 	conns := make(map[PeerName]Connection)
-	if err := connsIterator(buf, func(nameByte []byte, remoteTCPAddr string) {
+	if err := connsIterator(buf, func(nameByte []byte, remoteTCPAddr string, established bool) {
 		name := PeerNameFromBin(nameByte)
 		remotePeer := table[name]
-		conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr)
+		conn := NewRemoteConnection(peer, remotePeer, remoteTCPAddr, established)
 		conns[name] = conn
 	}); err != io.EOF {
 		// this should never happen since we've already successfully

router/protocol.go

@@ -2,7 +2,7 @@ package router
 
 const (
 	Protocol        = "weave"
-	ProtocolVersion = 11
+	ProtocolVersion = 12
 )
 
 type ProtocolTag byte

router/router.go

@@ -192,7 +192,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
 	// start.
 	remoteAddrStr := tcpConn.RemoteAddr().String()
 	log.Printf("->[%s] connection accepted\n", remoteAddrStr)
-	connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr)
+	connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false)
 	connLocal := NewLocalConnection(connRemote, tcpConn, nil, router)
 	connLocal.Start(true)
 }

site/how-it-works.md

@@ -108,12 +108,16 @@ network knows about any change.
 Topology is communicated over the TCP links between peers, using a
 Gossip mechanism.  Topology messages are sent by a peer...
 
-- when a connection has been established; the entire topology is sent
-  to the remote peer, and an incremental update, containing
-  information on just the two peers at the ends of the connection, is
-  sent to all neighbours
+- when a connection has been added; the entire topology is sent to the
+  remote peer, and an incremental update, containing information on
+  just the two peers at the ends of the connection, is sent to all
+  neighbours,
+- when a connection has been marked as 'established', indicating that
+  the remote peer can receive UDP traffic from the peer; an update
+  containing just information about the local peer is sent to all
+  neighbours,
 - when a connection has been torn down; an update containing just
-  information about the local peer is sent to all neighbours
+  information about the local peer is sent to all neighbours,
 - periodically, on a timer, in case someone has missed an update.
 
 The receiver of a topology update merges that update with its own