From f25696c2458993a273b30069edb27eb4d255957e Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 1 Feb 2015 16:52:39 +0000 Subject: [PATCH 1/3] 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 From 7834057b2ea5ebb81e9fa9151ea9a1a0e963b81c Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 1 Feb 2015 17:52:07 +0000 Subject: [PATCH 2/3] cosmetic: remove duplicate comment --- router/peers.go | 1 - 1 file changed, 1 deletion(-) diff --git a/router/peers.go b/router/peers.go index bff8c789..0217d8b0 100644 --- a/router/peers.go +++ b/router/peers.go @@ -263,7 +263,6 @@ 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 checkFatal(connsEnc.Encode(conn.Remote().NameByte)) checkFatal(connsEnc.Encode(conn.RemoteTCPAddr())) // DANGER holding rlock on peer, going to take rlock on conn From ed650230213b88cc294efdd83d421959260f4d4a Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 1 Feb 2015 19:13:51 +0000 Subject: [PATCH 3/3] only include established connections in route calculation except for reachability analysis in gc --- router/peer.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/router/peer.go b/router/peer.go index 9462a189..09ac2d9e 100644 --- a/router/peer.go +++ b/router/peer.go @@ -119,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 @@ -131,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) @@ -147,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