use GossipSender for GossipBroadcast

This makes GossipBroadcast properly async.

router/gossip.go

@@ -78,24 +78,27 @@ func (sender *GossipSender) Stop() {
 }
 
 type connectionSenders map[Connection]*GossipSender
+type peerSenders map[PeerName]*GossipSender
 
 type GossipChannel struct {
 	sync.Mutex
-	ourself  *LocalPeer
-	name     string
-	hash     uint32
-	gossiper Gossiper
-	senders  connectionSenders
+	ourself      *LocalPeer
+	name         string
+	hash         uint32
+	gossiper     Gossiper
+	senders      connectionSenders
+	broadcasters peerSenders
 }
 
 func (router *Router) NewGossip(channelName string, g Gossiper) Gossip {
 	channelHash := hash(channelName)
 	channel := &GossipChannel{
-		ourself:  router.Ourself,
-		name:     channelName,
-		hash:     channelHash,
-		gossiper: g,
-		senders:  make(connectionSenders)}
+		ourself:      router.Ourself,
+		name:         channelName,
+		hash:         channelHash,
+		gossiper:     g,
+		senders:      make(connectionSenders),
+		broadcasters: make(peerSenders)}
 	router.GossipChannels[channelHash] = channel
 	return channel
 }
@@ -233,16 +236,48 @@ func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error {
 }
 
 func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error {
+	names := c.ourself.Router.Peers.Names() // do this outside the lock so they don't nest
+	c.Lock()
+	defer c.Unlock()
+	// GC - randomly (courtesy of go's map iterator) pick some
+	// existing broadcasters and stop&remove them if their source peer
+	// is unknown. 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)[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 Peers.Fetch(name) below, we
+	// have that Peers.Names() invocation above. That is O(n_peers) at
+	// best.
+	for name, broadcaster := range c.broadcasters {
+		if _, found := names[name]; !found {
+			delete(c.broadcasters, name)
+			broadcaster.Stop()
+		} else {
+			break
+		}
+	}
+	broadcaster, found := c.broadcasters[srcName]
+	if !found {
+		broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) })
+		c.broadcasters[srcName] = broadcaster
+		broadcaster.Start()
+	}
+	broadcaster.Send(update)
+	return nil
+}
+
+func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) {
 	nextHops := c.ourself.Router.Routes.BroadcastAll(srcName)
 	if len(nextHops) == 0 {
-		return nil
+		return
 	}
 	protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.hash, srcName, update.Encode())}
 	// FIXME a single blocked connection can stall us
 	for _, conn := range c.ourself.ConnectionsTo(nextHops) {
 		conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
 	}
-	return nil
 }
 
 func (c *GossipChannel) log(args ...interface{}) {
@@ -259,6 +294,9 @@ func (router *Router) sendPendingGossip() {
 		for _, sender := range channel.senders {
 			sender.flush()
 		}
+		for _, sender := range channel.broadcasters {
+			sender.flush()
+		}
 	}
 }