Merge pull request #1855 from weaveworks/gossip-broadcast-stall-3

don't stall gossip broadcasting when there are blocked connections; LGTM

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,21 @@ func (s *GossipSender) run(stop <-chan struct{}, more <-chan struct{}, flush <-c
 		case <-stop:
 			return
 		case <-more:
-			s.deliver()
-			sent = true
+			sentSomething, err := s.deliver(stop)
+			if err != nil {
+				return
+			}
+			sent = sent || sentSomething
 		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
+				sentSomething, err := s.deliver(stop)
+				if err != nil {
+					return
+				}
+				sent = sent || sentSomething
 			default:
 			}
 			ch <- sent
@@ -74,25 +90,83 @@ func (s *GossipSender) run(stop <-chan struct{}, more <-chan struct{}, flush <-c
 	}
 }
 
-func (s *GossipSender) deliver() {
+func (s *GossipSender) deliver(stop <-chan struct{}) (bool, error) {
+	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 {
+		select {
+		case <-stop:
+			return sent, nil
+		default:
+		}
+		data, makeProtocolMsg := s.pick()
+		if data == nil {
+			return sent, nil
+		}
+		for _, msg := range data.Encode() {
+			if err := s.sender.SendProtocolMsg(makeProtocolMsg(msg)); err != nil {
+				return sent, err
+			}
+		}
+		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 +247,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

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{}) {