diff --git a/mesh/gossip.go b/mesh/gossip.go index 0aba953a..f4b94fec 100644 --- a/mesh/gossip.go +++ b/mesh/gossip.go @@ -34,18 +34,28 @@ type Gossiper interface { // and sends it when possible. type GossipSender struct { sync.Mutex - send func(GossipData) - cell GossipData - stop chan<- struct{} - more chan<- struct{} - flush chan<- chan<- bool // for testing + makeMsg func(msg []byte) ProtocolMsg + makeBroadcastMsg func(srcName PeerName, msg []byte) ProtocolMsg + sender ProtocolSender + gossip GossipData + broadcasts map[PeerName]GossipData + stop chan<- struct{} + more chan<- struct{} + flush chan<- chan<- bool // for testing } -func NewGossipSender(send func(GossipData)) *GossipSender { +func NewGossipSender(makeMsg func(msg []byte) ProtocolMsg, makeBroadcastMsg func(srcName PeerName, msg []byte) ProtocolMsg, sender ProtocolSender) *GossipSender { stop := make(chan struct{}) more := make(chan struct{}, 1) flush := make(chan chan<- bool) - s := &GossipSender{send: send, stop: stop, more: more, flush: flush} + s := &GossipSender{ + makeMsg: makeMsg, + makeBroadcastMsg: makeBroadcastMsg, + sender: sender, + broadcasts: make(map[PeerName]GossipData), + stop: stop, + more: more, + flush: flush} go s.run(stop, more, flush) return s } @@ -57,15 +67,13 @@ func (s *GossipSender) run(stop <-chan struct{}, more <-chan struct{}, flush <-c case <-stop: return case <-more: - s.deliver() - sent = true + sent = s.deliver() || sent case ch := <-flush: // for testing // send anything pending, then reply back whether we sent // anything since previous flush select { case <-more: - s.deliver() - sent = true + sent = s.deliver() || sent default: } ch <- sent @@ -74,25 +82,78 @@ func (s *GossipSender) run(stop <-chan struct{}, more <-chan struct{}, flush <-c } } -func (s *GossipSender) deliver() { +func (s *GossipSender) deliver() bool { + sent := false + // We must not hold our lock when sending, since that would block + // the callers of Send/Broadcast while we are stuck waiting for + // network congestion to clear. So we pick and send one piece of + // data at a time, only holding the lock during the picking. + for { + data, makeProtocolMsg := s.pick() + if data == nil { + return sent + } + for _, msg := range data.Encode() { + if s.sender.SendProtocolMsg(makeProtocolMsg(msg)) != nil { + break + } + } + sent = true + } +} + +func (s *GossipSender) pick() (data GossipData, makeProtocolMsg func(msg []byte) ProtocolMsg) { s.Lock() - data := s.cell - s.cell = nil - s.Unlock() - s.send(data) + defer s.Unlock() + switch { + case s.gossip != nil: // usually more important than broadcasts + data = s.gossip + makeProtocolMsg = s.makeMsg + s.gossip = nil + case len(s.broadcasts) > 0: + for srcName, d := range s.broadcasts { + data = d + makeProtocolMsg = func(msg []byte) ProtocolMsg { return s.makeBroadcastMsg(srcName, msg) } + delete(s.broadcasts, srcName) + break + } + } + return } func (s *GossipSender) Send(data GossipData) { s.Lock() defer s.Unlock() - if s.cell == nil { - s.cell = data - select { - case s.more <- void: - default: - } + if s.empty() { + defer s.prod() + } + if s.gossip == nil { + s.gossip = data } else { - s.cell = s.cell.Merge(data) + s.gossip = s.gossip.Merge(data) + } +} + +func (s *GossipSender) Broadcast(srcName PeerName, data GossipData) { + s.Lock() + defer s.Unlock() + if s.empty() { + defer s.prod() + } + d, found := s.broadcasts[srcName] + if !found { + s.broadcasts[srcName] = data + } else { + s.broadcasts[srcName] = d.Merge(data) + } +} + +func (s *GossipSender) empty() bool { return s.gossip == nil && len(s.broadcasts) == 0 } + +func (s *GossipSender) prod() { + select { + case s.more <- void: + default: } } @@ -173,9 +234,6 @@ func (router *Router) sendPendingGossip() bool { for _, sender := range channel.senders { sentSomething = sender.Flush() || sentSomething } - for _, sender := range channel.broadcasters { - sentSomething = sender.Flush() || sentSomething - } channel.Unlock() } return sentSomething diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go index 1ecc364e..83431872 100644 --- a/mesh/gossip_channel.go +++ b/mesh/gossip_channel.go @@ -9,25 +9,23 @@ import ( type GossipChannel struct { sync.Mutex - name string - ourself *LocalPeer - routes *Routes - gossiper Gossiper - senders connectionSenders - broadcasters peerSenders + name string + ourself *LocalPeer + routes *Routes + gossiper Gossiper + senders connectionSenders } type connectionSenders map[Connection]*GossipSender -type peerSenders map[PeerName]*GossipSender func NewGossipChannel(channelName string, ourself *LocalPeer, routes *Routes, g Gossiper) *GossipChannel { return &GossipChannel{ - name: channelName, - ourself: ourself, - routes: routes, - gossiper: g, - senders: make(connectionSenders), - broadcasters: make(peerSenders)} + name: channelName, + ourself: ourself, + routes: routes, + gossiper: g, + senders: make(connectionSenders), + } } func (router *Router) handleGossip(tag ProtocolTag, payload []byte) error { @@ -128,56 +126,11 @@ func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) (err erro } func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error { - names := c.routes.PeerNames() // do this outside the lock so they don't nest - c.Lock() - // 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 - } - c.Unlock() - broadcaster.Send(update) - return nil -} - -func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) { c.routes.EnsureRecalculated() - connections := c.ourself.ConnectionsTo(c.routes.BroadcastAll(srcName)) - if len(connections) == 0 { - return - } - msgs := update.Encode() - protocolMsgs := make([]ProtocolMsg, len(msgs), len(msgs)) - for i, msg := range msgs { - protocolMsgs[i] = ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, msg)} - } - // FIXME a single blocked connection can stall us - for _, conn := range connections { - for _, protocolMsg := range protocolMsgs { - if conn.(ProtocolSender).SendProtocolMsg(protocolMsg) != nil { - break - } - } + for _, sender := range c.sendersFor(c.ourself.ConnectionsTo(c.routes.BroadcastAll(srcName))) { + sender.Broadcast(srcName, update) } + return nil } func (c *GossipChannel) relay(srcName PeerName, data GossipData) { @@ -218,7 +171,7 @@ func (c *GossipChannel) sendersFor(conns []Connection) []*GossipSender { for i, conn := range conns { sender, found := c.senders[conn] if !found { - sender = c.makeSender(conn) + sender = NewGossipSender(c.makeMsg, c.makeBroadcastMsg, conn.(ProtocolSender)) c.senders[conn] = sender } senders[i] = sender @@ -226,15 +179,12 @@ func (c *GossipChannel) sendersFor(conns []Connection) []*GossipSender { return senders } -func (c *GossipChannel) makeSender(conn Connection) *GossipSender { - return NewGossipSender(func(pending GossipData) { - for _, msg := range pending.Encode() { - protocolMsg := ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)} - if conn.(ProtocolSender).SendProtocolMsg(protocolMsg) != nil { - break - } - } - }) +func (c *GossipChannel) makeMsg(msg []byte) ProtocolMsg { + return ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)} +} + +func (c *GossipChannel) makeBroadcastMsg(srcName PeerName, msg []byte) ProtocolMsg { + return ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, msg)} } func (c *GossipChannel) log(args ...interface{}) {