topology-sensitive random selection of gossip neighbours

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)
 	}
 }

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()