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.
2016-01-05 17:18:19 +00:00
1 parent 262f8915a8
commit c6a73b8a75
@@ -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
@@ -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{}) {