From c6a73b8a75d66cbd23db3555127875002eb5b1ed Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Thu, 24 Dec 2015 12:36:11 +0000 Subject: [PATCH] don't stall broadcasting when there are blocked connections Previously broadcasts were being handled by one GossipSender per broadcast source (and channel), which sends the broadcast to all next hops, and hence can stall when a single destination is stalled. Here we get rid of these per broadcast source GossipSenders. Instead broadcasts are sent to per-connection GossipSenders for all next hops. For this we use the existing GossipSenders we have set up for ordinary gossip. In order to deal with broadcasts they need one GossipData cell per broadcast source, since only broadcasts from the same source can be Merge()ed. So we add a PeerName->GossipData map of cells, in addition to the existing cell for ordinary gossip. The GossipSender goroutine picks one of the cells at a time, Encode()s the contents and sends it. It prefers the ordinary gossip cell over the broadcast cells since typically ordinary gossip is more important. To reduce coupling, the GossipSenders don't actually know how to construct protocol messages. They just invoke a couple of functions for that - one for ordinary gossip and one for broadcast - which are supplied by the GossipChannel. There are two downsides to this change: 1. broadcasts get encoded per connection, rather than just once 2. we can potentially end up with O(n_peers^2) cells, each containing accumulated (via GossipData.Merge()) broadcasts. For this to happen, the peer must - deal with gossip from most nodes. This doesn't happen in (near)complete connection topologies. Hypercube topologies are probably the worst case uniform topology. And a star topology is the worst case for a single (the centre) peer. - have backlogged connections to most of its neighbours, without those connections being completely stalled (since that would cause heartbeat timeouts to terminate them). Furthermore, for this to matter in practise, the accumulated broadcast GossipData must be sizeable: - For topology gossip, GossipData is just a set of PeerNames. Which takes very little space and is bounded, since there is a finite number of peers. And, if we could get rid of workaround for #1793 (cbaa92dcb), then topology broadcasts would only ever contain information about the source peer, so the GossipData would contain just one PeerName. - IPAM only employs broadcast during initialisation and shutdown. - DNS broadcasts contain DNS entries for containers on the source peer. Each entry will typically be 100-200 bytes. The accumulated broadcast GossipData from a peer will contain entries for all the peer's DNS entries, worst case. This includes tombstones, i.e. entries for containers that have died. If there is churn, i.e. DNS entries being added and removed continuously, and the churn rate exceeds the rate at which we can forward those entries, then the accumulated broadcast GossipData can grow unbounded. Note that this is the case on master too; the difference here is that we can have up to n_peers copies of that GossipData. --- mesh/gossip.go | 110 +++++++++++++++++++++++++++++++---------- mesh/gossip_channel.go | 92 ++++++++-------------------------- 2 files changed, 105 insertions(+), 97 deletions(-) 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{}) {