Merge pull request #1826 from weaveworks/cleaner-gossip-sender

clean up GossipSender

mesh/gossip.go

@@ -2,6 +2,7 @@ package mesh
 
 import (
 	"fmt"
+	"sync"
 )
 
 type GossipData interface {
@@ -32,35 +33,38 @@ type Gossiper interface {
 // Accumulates GossipData that needs to be sent to one destination,
 // and sends it when possible.
 type GossipSender struct {
+	sync.Mutex
 	send  func(GossipData)
-	cell  chan GossipData
+	cell  GossipData
+	stop  chan<- struct{}
+	more  chan<- struct{}
 	flush chan<- chan<- bool // for testing
 }
 
 func NewGossipSender(send func(GossipData)) *GossipSender {
-	cell := make(chan GossipData, 1)
+	stop := make(chan struct{})
+	more := make(chan struct{}, 1)
 	flush := make(chan chan<- bool)
-	s := &GossipSender{send: send, cell: cell, flush: flush}
-	go s.run(flush)
+	s := &GossipSender{send: send, stop: stop, more: more, flush: flush}
+	go s.run(stop, more, flush)
 	return s
 }
 
-func (s *GossipSender) run(flush <-chan chan<- bool) {
+func (s *GossipSender) run(stop <-chan struct{}, more <-chan struct{}, flush <-chan chan<- bool) {
 	sent := false
 	for {
 		select {
-		case pending := <-s.cell:
-			if pending == nil { // receive zero value when chan is closed
-				return
-			}
-			s.send(pending)
+		case <-stop:
+			return
+		case <-more:
+			s.deliver()
 			sent = true
 		case ch := <-flush: // for testing
 			// send anything pending, then reply back whether we sent
 			// anything since previous flush
 			select {
-			case pending := <-s.cell:
-				s.send(pending)
+			case <-more:
+				s.deliver()
 				sent = true
 			default:
 			}
@@ -70,13 +74,25 @@ func (s *GossipSender) run(flush <-chan chan<- bool) {
 	}
 }
 
+func (s *GossipSender) deliver() {
+	s.Lock()
+	data := s.cell
+	s.cell = nil
+	s.Unlock()
+	s.send(data)
+}
+
 func (s *GossipSender) Send(data GossipData) {
-	// NB: this must not be invoked concurrently
-	select {
-	case pending := <-s.cell:
-		s.cell <- pending.Merge(data)
-	default:
-		s.cell <- data
+	s.Lock()
+	defer s.Unlock()
+	if s.cell == nil {
+		s.cell = data
+		select {
+		case s.more <- void:
+		default:
+		}
+	} else {
+		s.cell = s.cell.Merge(data)
 	}
 }
 
@@ -88,7 +104,7 @@ func (s *GossipSender) Flush() bool {
 }
 
 func (s *GossipSender) Stop() {
-	close(s.cell)
+	close(s.stop)
 }
 
 type GossipChannels map[string]*GossipChannel

mesh/gossip_channel.go

@@ -111,7 +111,9 @@ func (c *GossipChannel) Send(data GossipData) {
 }
 
 func (c *GossipChannel) SendDown(conn Connection, data GossipData) {
-	c.sendDown([]Connection{conn}, data)
+	for _, sender := range c.sendersFor([]Connection{conn}) {
+		sender.Send(data)
+	}
 }
 
 func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) (err error) {
@@ -128,7 +130,6 @@ 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()
-	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
@@ -153,6 +154,7 @@ func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) erro
 		broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) })
 		c.broadcasters[srcName] = broadcaster
 	}
+	c.Unlock()
 	broadcaster.Send(update)
 	return nil
 }
@@ -180,12 +182,14 @@ func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) {
 
 func (c *GossipChannel) relay(srcName PeerName, data GossipData) {
 	c.routes.EnsureRecalculated()
-	c.sendDown(c.ourself.ConnectionsTo(c.routes.RandomNeighbours(srcName)), data)
+	for _, sender := range c.sendersFor(c.ourself.ConnectionsTo(c.routes.RandomNeighbours(srcName))) {
+		sender.Send(data)
+	}
 }
 
-func (c *GossipChannel) sendDown(conns []Connection, data GossipData) {
+func (c *GossipChannel) sendersFor(conns []Connection) []*GossipSender {
 	if len(conns) == 0 {
-		return
+		return nil
 	}
 	ourConnections := c.ourself.Connections()
 	c.Lock()
@@ -209,15 +213,17 @@ func (c *GossipChannel) sendDown(conns []Connection, data GossipData) {
 			break
 		}
 	}
-	// start senders, if necessary, and send.
-	for _, conn := range conns {
+	// start senders, if necessary
+	senders := make([]*GossipSender, len(conns), len(conns))
+	for i, conn := range conns {
 		sender, found := c.senders[conn]
 		if !found {
 			sender = c.makeSender(conn)
 			c.senders[conn] = sender
 		}
-		sender.Send(data)
+		senders[i] = sender
 	}
+	return senders
 }
 
 func (c *GossipChannel) makeSender(conn Connection) *GossipSender {