Make flushing of GossipSender wait till done in tests,

and take extra locks to avoid race conditions.
Also removed the sendPendingGossip() from mockChannelConnection.SendProtocolMsg(),
because waiting in flush now creates a deadlock.  Tests will have to
call sendPendingGossip() explicitly if they need it.

router/gossip.go

@@ -39,8 +39,9 @@ type Gossiper interface {
 // Accumulates GossipData that needs to be sent to one destination,
 // and sends it when possible.
 type GossipSender struct {
-	send func(GossipData)
-	cell chan GossipData
+	send    func(GossipData)
+	cell    chan GossipData
+	flushch chan chan struct{}
 }
 
 func NewGossipSender(send func(GossipData)) *GossipSender {
@@ -49,15 +50,22 @@ func NewGossipSender(send func(GossipData)) *GossipSender {
 
 func (sender *GossipSender) Start() {
 	sender.cell = make(chan GossipData, 1)
+	sender.flushch = make(chan chan struct{})
 	go sender.run()
 }
 
 func (sender *GossipSender) run() {
 	for {
-		if pending := <-sender.cell; pending == nil { // receive zero value when chan is closed
-			break
-		} else {
-			sender.send(pending)
+		select {
+		case pending := <-sender.cell:
+			if pending == nil { // receive zero value when chan is closed
+				return
+			} else {
+				sender.send(pending)
+			}
+		case ch := <-sender.flushch:
+			sender.flush()
+			close(ch)
 		}
 	}
 }
@@ -314,17 +322,24 @@ func (c *GossipChannel) log(args ...interface{}) {
 
 // for testing
 
-// FIXME this doesn't actually guarantee everything has been sent
-// since a GossipSender may be in the process of sending and there is
-// no easy way for us to know when that has completed.
 func (router *Router) sendPendingGossip() {
+	// copy the senders so we don't hold the lock for long
+	allSenders := make([]*GossipSender, 0)
 	for _, channel := range router.GossipChannels {
+		channel.Lock()
 		for _, sender := range channel.senders {
-			sender.flush()
+			allSenders = append(allSenders, sender)
 		}
 		for _, sender := range channel.broadcasters {
-			sender.flush()
+			allSenders = append(allSenders, sender)
 		}
+		channel.Unlock()
+	}
+
+	for _, sender := range allSenders {
+		ch := make(chan struct{})
+		sender.flushch <- ch
+		<-ch
 	}
 }
 

router/gossip_test.go

@@ -30,7 +30,6 @@ func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) {
 	if err := conn.dest.handleGossip(protocolMsg.tag, protocolMsg.msg); err != nil {
 		panic(err)
 	}
-	conn.dest.sendPendingGossip()
 }
 
 func (router *Router) AddTestChannelConnection(r *Router) {
@@ -87,7 +86,9 @@ func (router *Router) tp(routers ...*Router) *Peer {
 
 // Check that the topology of router matches the peers and all of their connections
 func checkTopology(t *testing.T, router *Router, wantedPeers ...*Peer) {
+	router.Peers.RLock()
 	checkTopologyPeers(t, true, router.Peers.allPeers(), wantedPeers...)
+	router.Peers.RUnlock()
 }
 
 func implTestGossipTopology(t *testing.T) {