From 18df01a4bf2915ffb4c2634f84a1b0693747072c Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 27 Jan 2015 12:38:19 +0000 Subject: [PATCH 1/3] Basic gossip implementation --- router/connection.go | 6 ++ router/gossip.go | 162 +++++++++++++++++++++++++++++++++++++++++++ router/local_peer.go | 49 +++++++------ router/protocol.go | 5 +- router/router.go | 10 +-- router/utils.go | 18 +++++ 6 files changed, 224 insertions(+), 26 deletions(-) create mode 100644 router/gossip.go diff --git a/router/connection.go b/router/connection.go index 994874b7..0469c06c 100644 --- a/router/connection.go +++ b/router/connection.go @@ -603,6 +603,12 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) } case ProtocolPMTUVerified: conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload)) + case ProtocolGossipUnicast: + return conn.Router.handleGossip(payload, deliverGossipUnicast) + case ProtocolGossipBroadcast: + return conn.Router.handleGossip(payload, deliverGossipBroadcast) + case ProtocolGossip: + return conn.Router.handleGossip(payload, deliverGossip) default: conn.log("ignoring unknown protocol tag:", tag) } diff --git a/router/gossip.go b/router/gossip.go new file mode 100644 index 00000000..5c374221 --- /dev/null +++ b/router/gossip.go @@ -0,0 +1,162 @@ +package router + +import ( + "bytes" + "encoding/gob" + "fmt" + "log" + "time" +) + +const GossipInterval = 30 * time.Second + +type Gossip interface { + // specific message from one peer to another + // intermediate peers relay it using unicast topology. + GossipUnicast(dstPeerName PeerName, buf []byte) error + // send a message to every peer, relayed using broadcast topology. + GossipBroadcast(buf []byte) error +} + +type Gossiper interface { + OnGossipUnicast(sender PeerName, msg []byte) error + OnGossipBroadcast(msg []byte) error + // Return state of everything we know; gets called periodically + Gossip() []byte + // merge in state and return "everything new I've just learnt", + // or nil if nothing in the received message was new + OnGossip(buf []byte) ([]byte, error) +} + +type GossipChannel struct { + ourself *LocalPeer + name string + hash uint32 + gossiper Gossiper +} + +func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { + channelHash := hash(channelName) + channel := &GossipChannel{router.Ourself, channelName, channelHash, g} + router.GossipChannels[channelHash] = channel + return channel +} + +func (router *Router) SendAllGossip() { + for _, channel := range router.GossipChannels { + channel.SendGossipMsg(channel.gossiper.Gossip()) + } +} + +func (router *Router) SendAllGossipDown(conn Connection) { + for _, channel := range router.GossipChannels { + protocolMsg := channel.gossipMsg(channel.gossiper.Gossip()) + conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + } +} + +func (router *Router) handleGossip(payload []byte, onok func(*GossipChannel, PeerName, []byte, *gob.Decoder) error) error { + decoder := gob.NewDecoder(bytes.NewReader(payload)) + var channelHash uint32 + if err := decoder.Decode(&channelHash); err != nil { + return err + } + channel, found := router.GossipChannels[channelHash] + if !found { + return fmt.Errorf("[gossip] received unknown channel with hash %v", channelHash) + } + var srcName PeerName + if err := decoder.Decode(&srcName); err != nil { + return err + } + if err := onok(channel, srcName, payload, decoder); err != nil { + return err + } + return nil +} + +func deliverGossipUnicast(channel *GossipChannel, srcName PeerName, origPayload []byte, dec *gob.Decoder) error { + var destName PeerName + if err := dec.Decode(&destName); err != nil { + return err + } + if channel.ourself.Name == destName { + var payload []byte + if err := dec.Decode(&payload); err != nil { + return err + } + return channel.gossiper.OnGossipUnicast(srcName, payload) + } else { + return channel.relayGossipUnicast(destName, origPayload) + } +} + +func deliverGossipBroadcast(channel *GossipChannel, srcName PeerName, origPayload []byte, dec *gob.Decoder) error { + var payload []byte + if err := dec.Decode(&payload); err != nil { + return err + } + if err := channel.gossiper.OnGossipBroadcast(payload); err != nil { + return err + } + return channel.relayGossipBroadcast(srcName, origPayload) +} + +func deliverGossip(channel *GossipChannel, srcName PeerName, _ []byte, dec *gob.Decoder) error { + var payload []byte + if err := dec.Decode(&payload); err != nil { + return err + } + if newBuf, err := channel.gossiper.OnGossip(payload); err != nil { + return err + } else if newBuf != nil { + channel.SendGossipMsg(newBuf) + } + return nil +} + +func (c *GossipChannel) SendGossipMsg(buf []byte) { + protocolMsg := c.gossipMsg(buf) + c.ourself.ForEachConnection(func(_ PeerName, conn Connection) { + conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + }) +} + +func (c *GossipChannel) gossipMsg(buf []byte) ProtocolMsg { + return ProtocolMsg{ProtocolGossip, GobEncode(c.hash, c.ourself.Name, buf)} +} + +func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, buf []byte) error { + return c.relayGossipUnicast(dstPeerName, GobEncode(c.hash, c.ourself.Name, dstPeerName, buf)) +} + +func (c *GossipChannel) GossipBroadcast(buf []byte) error { + return c.relayGossipBroadcast(c.ourself.Name, GobEncode(c.hash, c.ourself.Name, buf)) +} + +func (c *GossipChannel) relayGossipUnicast(dstPeerName PeerName, msg []byte) error { + if relayPeerName, found := c.ourself.Router.Routes.Unicast(dstPeerName); !found { + c.log("unknown relay destination:", dstPeerName) + } else if conn, found := c.ourself.ConnectionTo(relayPeerName); !found { + c.log("unable to find connection to relay peer", relayPeerName) + } else { + conn.(ProtocolSender).SendProtocolMsg(ProtocolMsg{ProtocolGossipUnicast, msg}) + } + return nil +} + +func (c *GossipChannel) relayGossipBroadcast(srcName PeerName, msg []byte) error { + if srcPeer, found := c.ourself.Router.Peers.Fetch(srcName); !found { + c.log("unable to relay broadcast from unknown peer", srcName) + } else { + protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, msg} + for _, conn := range c.ourself.NextBroadcastHops(srcPeer) { + conn.SendProtocolMsg(protocolMsg) + } + } + return nil +} + +func (c *GossipChannel) log(args ...interface{}) { + log.Println(append(append([]interface{}{}, "[gossip "+c.name+"]:"), args...)...) +} diff --git a/router/local_peer.go b/router/local_peer.go index 93f13ca9..6df816aa 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -4,6 +4,7 @@ import ( "fmt" "log" "net" + "time" ) type LocalPeer struct { @@ -156,30 +157,35 @@ func (peer *LocalPeer) SendProtocolMsg(m ProtocolMsg) { // ACTOR server func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) { + gossipTimer := time.Tick(GossipInterval) for { - query, ok := <-queryChan - if !ok { - return - } - switch query.code { - case PAddConnection: - conn := query.payload.(*LocalConnection) - if peer.handleAddConnection(conn) { - conn.log("connection added") + select { + case query, ok := <-queryChan: + if !ok { + return } - case PDeleteConnection: - conn := query.payload.(*LocalConnection) - if peer.handleDeleteConnection(conn) { - conn.log("connection deleted") + switch query.code { + case PAddConnection: + conn := query.payload.(*LocalConnection) + if peer.handleAddConnection(conn) { + conn.log("connection added") + } + case PDeleteConnection: + conn := query.payload.(*LocalConnection) + if peer.handleDeleteConnection(conn) { + conn.log("connection deleted") + } + query.resultChan <- nil + case PConnectionEstablished: + conn := query.payload.(*LocalConnection) + if peer.handleConnectionEstablished(conn) { + conn.log("connection fully established") + } + case PSendProtocolMsg: + peer.handleSendProtocolMsg(query.payload.(ProtocolMsg)) } - query.resultChan <- nil - case PConnectionEstablished: - conn := query.payload.(*LocalConnection) - if peer.handleConnectionEstablished(conn) { - conn.log("connection fully established") - } - case PSendProtocolMsg: - peer.handleSendProtocolMsg(query.payload.(ProtocolMsg)) + case <-gossipTimer: + peer.Router.SendAllGossip() } } } @@ -251,6 +257,7 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) bool { return false } peer.connectionEstablished(conn) + peer.Router.SendAllGossipDown(conn) peer.broadcastPeerUpdate(conn.Remote()) return true } diff --git a/router/protocol.go b/router/protocol.go index 2218d813..7291f35c 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -2,7 +2,7 @@ package router const ( Protocol = "weave" - ProtocolVersion = 10 + ProtocolVersion = 11 ) type ProtocolTag byte @@ -15,6 +15,9 @@ const ( ProtocolFetchAll ProtocolUpdate ProtocolPMTUVerified + ProtocolGossip + ProtocolGossipUnicast + ProtocolGossipBroadcast ) type ProtocolMsg struct { diff --git a/router/router.go b/router/router.go index 8ed59c28..d54a24d0 100644 --- a/router/router.go +++ b/router/router.go @@ -24,6 +24,7 @@ type Router struct { Peers *Peers Routes *Routes ConnectionMaker *ConnectionMaker + GossipChannels map[uint32]*GossipChannel UDPListener *net.UDPConn Password *[]byte ConnLimit int @@ -46,10 +47,11 @@ type PacketSourceSink interface { func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit int, bufSz int, logFrame func(string, []byte, *layers.Ethernet)) *Router { router := &Router{ - Iface: iface, - ConnLimit: connLimit, - BufSz: bufSz, - LogFrame: logFrame} + Iface: iface, + GossipChannels: make(map[uint32]*GossipChannel), + ConnLimit: connLimit, + BufSz: bufSz, + LogFrame: logFrame} if len(password) > 0 { router.Password = &password } diff --git a/router/utils.go b/router/utils.go index 7f0a2b3b..0a6fe57c 100644 --- a/router/utils.go +++ b/router/utils.go @@ -1,8 +1,11 @@ package router import ( + "bytes" "crypto/rand" + "encoding/gob" "fmt" + "hash/fnv" "log" "net" ) @@ -74,6 +77,21 @@ func randUint64() (r uint64) { return } +func hash(s string) uint32 { + h := fnv.New32a() + h.Write([]byte(s)) + return h.Sum32() +} + +func GobEncode(items ...interface{}) []byte { + buf := new(bytes.Buffer) + enc := gob.NewEncoder(buf) + for _, i := range items { + checkFatal(enc.Encode(i)) + } + return buf.Bytes() +} + func macint(mac net.HardwareAddr) (r uint64) { for _, b := range mac { r <<= 8 From 5dc9b085a018e639cfc05a6d63694517784b3c07 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 27 Jan 2015 12:41:15 +0000 Subject: [PATCH 2/3] Replace previous topology communication with gossip-based mechanism --- router/connection.go | 37 ----------------------------- router/consts.go | 3 +-- router/local_peer.go | 9 ++++++- router/protocol.go | 2 -- router/router.go | 38 ++++++++++++++++++++++++++++++ site/how-it-works.md | 56 +++++++++++++++++++++++--------------------- 6 files changed, 76 insertions(+), 69 deletions(-) diff --git a/router/connection.go b/router/connection.go index 0469c06c..d268eb0b 100644 --- a/router/connection.go +++ b/router/connection.go @@ -48,7 +48,6 @@ type LocalConnection struct { SessionKey *[32]byte heartbeatFrame *ForwardedFrame heartbeat *time.Ticker - fetchAll *time.Ticker fragTest *time.Ticker forwardChan chan<- *ForwardedFrame forwardChanDF chan<- *ForwardedFrame @@ -285,8 +284,6 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) } case <-tickerChan(conn.heartbeat): conn.Forward(true, conn.heartbeatFrame, nil) - case <-tickerChan(conn.fetchAll): - err = conn.handleSendSimpleProtocolMsg(ProtocolFetchAll) case <-tickerChan(conn.fragTest): conn.setStackFrag(false) err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest) @@ -343,13 +340,9 @@ func (conn *LocalConnection) handleSetEstablished() error { frame: PMTUDiscovery}, nil) conn.heartbeat = time.NewTicker(SlowHeartbeat) - conn.fetchAll = time.NewTicker(FetchAllInterval) conn.fragTest = time.NewTicker(FragTestInterval) // avoid initial waits for timers to fire conn.Forward(true, conn.heartbeatFrame, nil) - if err := conn.handleSendSimpleProtocolMsg(ProtocolFetchAll); err != nil { - return err - } conn.setStackFrag(false) if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { return err @@ -376,7 +369,6 @@ func (conn *LocalConnection) handleShutdown() { } stopTicker(conn.heartbeat) - stopTicker(conn.fetchAll) stopTicker(conn.fragTest) // blank out the forwardChan so that the router processes don't @@ -572,35 +564,6 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) return fmt.Errorf("unexpected nonce on unencrypted connection") } conn.Decryptor.ReceiveNonce(payload) - case ProtocolFetchAll: - // There are exactly two messages that relate to topology - // updates. - // - // 1. FetchAll. This carries no payload. The receiver responds - // with the entire topology model as the receiver has it. - // - // 2. Update. This carries a topology payload. The receiver - // merges it with its own topology model. If the payload is a - // subset of the receiver's topology, no further action is - // taken. Otherwise, the receiver sends out to all its - // connections an "improved" update. - conn.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, conn.Router.Peers.EncodeAllPeers()}) - case ProtocolUpdate: - newUpdate, err := conn.Router.Peers.ApplyUpdate(payload) - if _, ok := err.(UnknownPeersError); err != nil && ok { - // That update contained a peer we didn't know about; - // request full update - conn.SendProtocolMsg(ProtocolMsg{ProtocolFetchAll, nil}) - break - } - if err != nil { - return err - } - if len(newUpdate) != 0 { - conn.Router.ConnectionMaker.Refresh() - conn.Router.Routes.Recalculate() - conn.Router.Ourself.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, newUpdate}) - } case ProtocolPMTUVerified: conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload)) case ProtocolGossipUnicast: diff --git a/router/consts.go b/router/consts.go index b15e24a4..70b558c2 100644 --- a/router/consts.go +++ b/router/consts.go @@ -18,9 +18,8 @@ const ( PMTUDiscoverySize = 60000 FastHeartbeat = 500 * time.Millisecond SlowHeartbeat = 10 * time.Second - FetchAllInterval = 30 * time.Second FragTestInterval = 5 * time.Minute - ReadTimeout = 2 * FetchAllInterval + ReadTimeout = 1 * time.Minute PMTUVerifyAttempts = 8 PMTUVerifyTimeout = 10 * time.Millisecond // gets doubled with every attempt MaxDuration = time.Duration(math.MaxInt64) diff --git a/router/local_peer.go b/router/local_peer.go index 6df816aa..72d2d799 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -270,7 +270,14 @@ func (peer *LocalPeer) handleSendProtocolMsg(m ProtocolMsg) { func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { peer.Router.Routes.Recalculate() - peer.handleSendProtocolMsg(ProtocolMsg{ProtocolUpdate, EncodePeers(append(peers, peer.Peer)...)}) + // TODO We should just be invoking TopologyGossip.GossipBroadcast + // here, but route calculation is asynchronous and in this + // particular case would likely result in the broadcast not + // reaching all peers. So instead we slightly break the Gossip + // abstraction (hence the cast) and send a regular update. This is + // less efficient though since it will almost certainly reach + // peers more than once. + peer.Router.TopologyGossip.(*GossipChannel).SendGossipMsg(EncodePeers(append(peers, peer.Peer)...)) } func (peer *LocalPeer) checkConnectionLimit() error { diff --git a/router/protocol.go b/router/protocol.go index 7291f35c..a2475822 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -12,8 +12,6 @@ const ( ProtocolFragmentationReceived ProtocolStartFragmentationTest ProtocolNonce - ProtocolFetchAll - ProtocolUpdate ProtocolPMTUVerified ProtocolGossip ProtocolGossipUnicast diff --git a/router/router.go b/router/router.go index d54a24d0..8bceeff9 100644 --- a/router/router.go +++ b/router/router.go @@ -25,6 +25,7 @@ type Router struct { Routes *Routes ConnectionMaker *ConnectionMaker GossipChannels map[uint32]*GossipChannel + TopologyGossip Gossip UDPListener *net.UDPConn Password *[]byte ConnLimit int @@ -68,6 +69,7 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i router.Peers.FetchWithDefault(router.Ourself.Peer) router.Routes = NewRoutes(router.Ourself.Peer, router.Peers) router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers) + router.TopologyGossip = router.NewGossip("topology", router) return router } @@ -350,3 +352,39 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F return nil } } + +// Gossiper methods - the Router is the topology Gossiper + +func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error { + return fmt.Errorf("unexpected topology gossip unicast: %v", msg) +} + +func (router *Router) OnGossipBroadcast(msg []byte) error { + return fmt.Errorf("unexpected topology gossip broadcast: %v", msg) +} + +// Return state of everything we know; intended to be called periodically +func (router *Router) Gossip() []byte { + return router.Peers.EncodeAllPeers() +} + +// merge in state and return "everything new I've just learnt", +// or nil if nothing in the received message was new +func (router *Router) OnGossip(buf []byte) ([]byte, error) { + newUpdate, err := router.Peers.ApplyUpdate(buf) + if _, ok := err.(UnknownPeersError); err != nil && ok { + // That update contained a peer we didn't know about; we + // ignore this; eventually we should receive an update + // containing a complete topology. + return nil, nil + } + if err != nil { + return nil, err + } + if len(newUpdate) == 0 { + return nil, nil + } + router.ConnectionMaker.Refresh() + router.Routes.Recalculate() + return newUpdate, nil +} diff --git a/site/how-it-works.md b/site/how-it-works.md index 5e9560cf..1f7c5e5a 100644 --- a/site/how-it-works.md +++ b/site/how-it-works.md @@ -105,38 +105,40 @@ other peers. Weave peers communicate their knowledge of the topology information on to their own neighbours, and so on, until the entire network knows about any change. -Topology is communicated over the TCP links between peers. There are -two message types: +Topology is communicated over the TCP links between peers, using a +Gossip mechanism. Topology messages are sent by a peer... -##### FetchAll -This carries no payload. The receiver responds with the entire -topology model as the receiver has it. FetchAll is sent: +- 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 torn down; an update containing just + information about the local peer is sent to all neighbours +- periodically, on a timer, in case someone has missed an update. - * when an update is received that references a peer that the receiver - does not know - * periodically, on a timer, in case any updates have been missed - -##### Update -This carries a topology payload. This is sent: - - * upon receipt of a FetchAll message, as above, - * when a connection is established - the update contains the two -peers that have just connected - * when a connection is deleted - the update contains just the peer -that lost the connection - -The receiver merges the update with its own topology model, adding -peers hitherto unknown to it, and updating peers for which the update -contains a more recent version than known to it. If there were any -such new/updated peers, then an improved update containing them is -sent out on all connections. +The receiver of a topology update merges that update with its own +topology model, adding peers hitherto unknown to it, and updating +peers for which the update contains a more recent version than known +to it. If there were any such new/updated peers, then an improved +update containing them is sent out on all connections. If the update mentions a peer that the receiver does not know, then -the entire update is rejected and the receiver sends a FetchAll -message back to the sender. +the entire update is ignored. #### Message details -A topology update message is laid out like this: +Every gossip message is structured as follows: + + +-----------------------------------+ + | 1-byte message type - Gossip | + +-----------------------------------+ + | 4-byte Gossip channel - Topology | + +-----------------------------------+ + | Peer Name of source | + +-----------------------------------+ + | Gossip payload (topology update) | + +-----------------------------------+ + +The topology update payload is laid out like this: +-----------------------------------+ | Peer 1: Name | @@ -183,7 +185,7 @@ no longer has any connections within the network, it drops all knowledge of that second peer. #### Out-of-date topology -The peer-to-peer passing of updates is not instantaneous, so it is +The peer-to-peer gossiping of updates is not instantaneous, so it is very possible for a node elsewhere in the network to have an out-of-date view. From 4647c36a15b1c61670622e5498ab9a5c8d0e5836 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 27 Jan 2015 12:41:44 +0000 Subject: [PATCH 3/3] Unit tests for gossip-based topology communication --- router/gossip_test.go | 146 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) create mode 100644 router/gossip_test.go diff --git a/router/gossip_test.go b/router/gossip_test.go new file mode 100644 index 00000000..d2423090 --- /dev/null +++ b/router/gossip_test.go @@ -0,0 +1,146 @@ +package router + +import ( + wt "github.com/zettio/weave/testing" + "testing" + "time" +) + +// TODO test gossip unicast and broadcast; atm we only test topology +// gossip, which does not employ unicast or broadcast. + +type mockChannelConnection struct { + RemoteConnection + dest *Router +} + +// Construct a "passive" Router, i.e. without any goroutines. +// +// We need to create some dummy channels otherwise tests hang on nil +// channels when Router.OnGossip() calls async methods. +func NewTestRouter(name PeerName) *Router { + router := NewRouter(nil, name, nil, 10, 1024, nil) + router.ConnectionMaker.queryChan = make(chan *ConnectionMakerInteraction, ChannelSize) + router.Routes.queryChan = make(chan *Interaction, ChannelSize) + return router +} + +func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) { + if err := conn.dest.handleGossip(protocolMsg.msg, deliverGossip); err != nil { + panic(err) + } +} + +func (router *Router) AddTestChannelConnection(r *Router) { + fromName := router.Ourself.Peer.Name + toName := r.Ourself.Peer.Name + + fromPeer := NewPeer(fromName, router.Ourself.Peer.UID, 0) + toPeer := NewPeer(toName, r.Ourself.Peer.UID, 0) + + r.Peers.FetchWithDefault(fromPeer) // Has side-effect of incrementing refcount + router.Peers.FetchWithDefault(toPeer) // + + conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, ""}, r} + router.Ourself.handleAddConnection(conn) + router.Ourself.handleConnectionEstablished(conn) +} + +func (router *Router) DeleteTestChannelConnection(r *Router) { + fromName := router.Ourself.Peer.Name + toName := r.Ourself.Peer.Name + + fromPeer, _ := r.Peers.Fetch(fromName) + toPeer, _ := router.Peers.Fetch(toName) + + fromPeer.DecrementLocalRefCount() + toPeer.DecrementLocalRefCount() + + conn, _ := router.Ourself.ConnectionTo(toName) + router.Ourself.handleDeleteConnection(conn) +} + +func TestGossipTopology(t *testing.T) { + wt.RunWithTimeout(t, 1*time.Second, func() { + implTestGossipTopology(t) + }) +} + +// Create a Peer representing the receiver router, with connections to +// the routers supplied as arguments, carrying across all UID and +// version information. +func (router *Router) tp(routers ...*Router) *Peer { + peer := NewPeer(router.Ourself.Peer.Name, router.Ourself.Peer.UID, 0) + connections := make(map[PeerName]Connection) + for _, r := range routers { + p := NewPeer(r.Ourself.Peer.Name, r.Ourself.Peer.UID, r.Ourself.Peer.version) + connections[r.Ourself.Peer.Name] = newMockConnection(peer, p) + } + peer.SetVersionAndConnections(router.Ourself.Peer.version, connections) + return peer +} + +// Check that the topology of router matches the peers and all of their connections +func checkTopology(t *testing.T, router *Router, wantedPeers ...*Peer) { + checkTopologyPeers(t, true, router.Peers.allPeers(), wantedPeers...) +} + +func implTestGossipTopology(t *testing.T) { + // Create some peers that will talk to each other + peer1Name, _ := PeerNameFromString("01:00:00:01:00:00") + peer2Name, _ := PeerNameFromString("02:00:00:02:00:00") + peer3Name, _ := PeerNameFromString("03:00:00:03:00:00") + r1 := NewTestRouter(peer1Name) + r2 := NewTestRouter(peer2Name) + r3 := NewTestRouter(peer3Name) + + // Check state when they have no connections + checkTopology(t, r1, r1.tp()) + checkTopology(t, r2, r2.tp()) + + // Now try adding some connections + r1.AddTestChannelConnection(r2) + checkTopology(t, r1, r1.tp(r2), r2.tp()) + checkTopology(t, r2, r1.tp(r2), r2.tp()) + r2.AddTestChannelConnection(r1) + checkTopology(t, r1, r1.tp(r2), r2.tp(r1)) + checkTopology(t, r2, r1.tp(r2), r2.tp(r1)) + + // Currently, the connection from 2 to 3 is one-way only + r2.AddTestChannelConnection(r3) + checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp()) + checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp()) + // When r2 gossiped to r3, 1 was unreachable from r3 so it got removed from the + // list of peers, but remains referenced in the connection from 1 to 3. + checkTopology(t, r3, r2.tp(r1, r3), r3.tp()) + + // Add a connection from 3 to 1 and now r1 is reachable. + r3.AddTestChannelConnection(r1) + checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1)) + checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1)) + checkTopology(t, r3, r1.tp(), r2.tp(r1, r3), r3.tp(r1)) + + r1.AddTestChannelConnection(r3) + checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1)) + checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1)) + checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1)) + + // Drop the connection from 2 to 3 + r2.DeleteTestChannelConnection(r3) + checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1)) + checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1)) + checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1)) + + // Drop the connection from 1 to 3 + r1.DeleteTestChannelConnection(r3) + checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1)) + + checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1)) + checkTopology(t, r2, r1.tp(r2), r2.tp(r1)) + // r3 still thinks r1 has a connection to it + checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1)) + + // On a timer, r3 will gossip to r1 + r3.SendAllGossip() + checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1)) +}