From f25696c2458993a273b30069edb27eb4d255957e Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 1 Feb 2015 16:52:39 +0000 Subject: [PATCH] prevent premature gc of peers from topology Topology gossip can be received on a connection before that connection has been marked as 'established'. If we do not include unestablished connections in topology gossip, then parts of the topology can be removed by gc, since they are deemed unreachable, which can cause subsequent incremental topology gossips to be rejected due to them containing references to peers that have been removed by the gc. Including unestablished connections in topology gossip requires: - removing the check for 'established' connections in the encoding of peers for topology gossip, - adding the 'established' flag to the internal representation of remote connections, - adding the 'established' flag to to encoding of connections in topology gossip, - treating the *addition* of a connection as an update to the local peer. The latter entails - bumping the peer version on connection addition (as well as on marking 'established' as previously), - bumping the peer version *unconditionally* on connection deletion (rather than only when the connection was 'established', as previously), - moving the initial sending of complete topology on the connection, and the broadcasting of a topology update containing the two peers at the ends of the connection, from the connection 'established' logic to the connection addition logic, - broadcasting just an topology update for the local peer when a connection gets marked as 'established', - broadcasting a topology update for the local peer *unconditionally* when a connection is removed (rather than only doing this when the connection was 'established', as previously). --- router/connection.go | 9 +++++---- router/gossip_test.go | 2 +- router/local_peer.go | 11 +++++------ router/mocks_test.go | 4 ++-- router/peer.go | 6 ++---- router/peers.go | 25 ++++++++++++++----------- router/protocol.go | 2 +- router/router.go | 2 +- site/how-it-works.md | 14 +++++++++----- 9 files changed, 40 insertions(+), 35 deletions(-) diff --git a/router/connection.go b/router/connection.go index 450996da..2cbf48cc 100644 --- a/router/connection.go +++ b/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) { diff --git a/router/gossip_test.go b/router/gossip_test.go index d2423090..050d48e7 100644 --- a/router/gossip_test.go +++ b/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) } diff --git a/router/local_peer.go b/router/local_peer.go index 170265a6..3d532378 100644 --- a/router/local_peer.go +++ b/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 } diff --git a/router/mocks_test.go b/router/mocks_test.go index da252a38..50e75d7e 100644 --- a/router/mocks_test.go +++ b/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) { diff --git a/router/peer.go b/router/peer.go index faa5be01..9462a189 100644 --- a/router/peer.go +++ b/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) { diff --git a/router/peers.go b/router/peers.go index 43f328b5..bff8c789 100644 --- a/router/peers.go +++ b/router/peers.go @@ -108,7 +108,7 @@ func EncodePeers(peers ...*Peer) []byte { buf := new(bytes.Buffer) enc := gob.NewEncoder(buf) for _, peer := range peers { - peer.encodePeer(enc) + peer.encode(enc) } return buf.Bytes() } @@ -163,7 +163,7 @@ func encodePeersMap(peers map[PeerName]*Peer) []byte { buf := new(bytes.Buffer) enc := gob.NewEncoder(buf) for _, peer := range peers { - peer.encodePeer(enc) + peer.encode(enc) } return buf.Bytes() } @@ -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 @@ -252,7 +252,7 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]byte) ma return newUpdate } -func (peer *Peer) encodePeer(enc *gob.Encoder) { +func (peer *Peer) encode(enc *gob.Encoder) { peer.RLock() defer peer.RUnlock() @@ -264,11 +264,10 @@ func (peer *Peer) encodePeer(enc *gob.Encoder) { 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 +288,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 +301,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 diff --git a/router/protocol.go b/router/protocol.go index a2475822..7a6fff36 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -2,7 +2,7 @@ package router const ( Protocol = "weave" - ProtocolVersion = 11 + ProtocolVersion = 12 ) type ProtocolTag byte diff --git a/router/router.go b/router/router.go index 445f89cb..180a2aea 100644 --- a/router/router.go +++ b/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) } diff --git a/site/how-it-works.md b/site/how-it-works.md index 1f7c5e5a..d5d7c57b 100644 --- a/site/how-it-works.md +++ b/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