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..09ac2d9e 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) { @@ -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 diff --git a/router/peers.go b/router/peers.go index 254849ff..0217d8b0 100644 --- a/router/peers.go +++ b/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 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