From 12c97b4fc5c39f3e652e6075ab366cb881f9f7d1 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 6 Apr 2015 11:06:04 +0100 Subject: [PATCH] topology-sensitive random selection of gossip neighbours --- router/gossip.go | 34 +++++++++++++++++++++++++--------- router/routes.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 9 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index f1338ddd..c0cfcde6 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -105,7 +105,7 @@ func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { func (router *Router) SendAllGossip() { for _, channel := range router.GossipChannels { - channel.Send(channel.gossiper.Gossip()) + channel.Send(router.Ourself.Name, channel.gossiper.Gossip()) } } @@ -167,7 +167,7 @@ func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.De return c.relayBroadcast(srcName, data) } -func (c *GossipChannel) deliver(_ PeerName, _ []byte, dec *gob.Decoder) error { +func (c *GossipChannel) deliver(srcName PeerName, _ []byte, dec *gob.Decoder) error { var payload []byte if err := dec.Decode(&payload); err != nil { return err @@ -175,30 +175,46 @@ func (c *GossipChannel) deliver(_ PeerName, _ []byte, dec *gob.Decoder) error { if data, err := c.gossiper.OnGossip(payload); err != nil { return err } else if data != nil { - c.Send(data) + c.Send(srcName, data) } return nil } -func (c *GossipChannel) Send(data GossipData) { - connections := c.ourself.Connections() // do this outside the lock so they don't nest +func (c *GossipChannel) Send(srcName PeerName, data GossipData) { + // do this outside the lock below so we avoid lock nesting + c.ourself.Router.Routes.EnsureRecalculated() + selectedConnections := make(ConnectionSet) + for name := range c.ourself.Router.Routes.RandomNeighbours(srcName) { + if conn, found := c.ourself.ConnectionTo(name); found { + selectedConnections[conn] = void + } + } + if len(selectedConnections) == 0 { + return + } + connections := c.ourself.Connections() c.Lock() defer c.Unlock() // GC - randomly (courtesy of go's map iterator) pick some // existing entries and stop&remove them if the associated // connection is no longer active. We stop as soon as we // encounter a valid entry; the idea being that when there is - // little or no garbage then this executes close to O(1), whereas - // when there is lots of garbage we remove it quickly. + // little or no garbage then this executes close to O(1)[1], + // whereas when there is lots of garbage we remove it quickly. + // + // [1] TODO Unfortunately, due to the desire to avoid nested + // locks, instead of simply invoking Peer.ConnectionTo(name) + // below, we have that Peer.Connections() invocation above. That + // is O(n_our_connections) at best. for conn, sender := range c.senders { - if foundConn, found := c.ourself.ConnectionTo(conn.Remote().Name); !found || foundConn != conn { + if _, found := connections[conn]; !found { delete(c.senders, conn) sender.Stop() } else { break } } - for conn := range connections { + for conn := range selectedConnections { c.sendDown(conn, data) } } diff --git a/router/routes.go b/router/routes.go index 3da3d1f2..1e854223 100644 --- a/router/routes.go +++ b/router/routes.go @@ -3,6 +3,7 @@ package router import ( "bytes" "fmt" + "math" "sync" ) @@ -77,6 +78,36 @@ func (routes *Routes) BroadcastAll(name PeerName) []PeerName { return hops } +// Choose min(log2(n_peers), n_neighbouring_peers) neighbours, with a +// random distribution that is topology-sensitive, favouring +// neighbours at the end of "bottleneck links". We determine the +// latter based on the unicast routing table. If a neighbour appears +// as the value more frequently than others - meaning that we reach a +// higher proportion of peers via that neighbour than other neighbours +// - then it is chosen with a higher probability. +// +// Note that we choose log2(n_peers) *neighbours*, not +// peers. Consequently, on sparsely connected peers this function +// returns a higher proportion of neighbours than elsewhere. In +// extremis, on peers with fewer than log2(n_peers) neighbours, all +// neighbours are returned. +func (routes *Routes) RandomNeighbours(except PeerName) PeerNameSet { + res := make(PeerNameSet) + routes.RLock() + defer routes.RUnlock() + count := int(math.Log2(float64(len(routes.unicastAll)))) + // depends on go's random map iteration + for _, dst := range routes.unicastAll { + if dst != UnknownPeerName && dst != except { + res[dst] = void + if len(res) >= count { + break + } + } + } + return res +} + func (routes *Routes) String() string { var buf bytes.Buffer routes.RLock()