diff --git a/router/gossip.go b/router/gossip.go index 78c2706b..df6ad30a 100644 --- a/router/gossip.go +++ b/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() + } } }