Keep flushing gossip until it's all gone

A better (i.e. working?) fix for #885.

router/gossip.go

@@ -39,9 +39,11 @@ 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
-	flushch chan chan struct{}
+	send func(GossipData)
+	cell chan GossipData
+	// for testing
+	sent    bool
+	flushch chan chan bool
 }
 
 func NewGossipSender(send func(GossipData)) *GossipSender {
@@ -50,7 +52,7 @@ func NewGossipSender(send func(GossipData)) *GossipSender {
 
 func (sender *GossipSender) Start() {
 	sender.cell = make(chan GossipData, 1)
-	sender.flushch = make(chan chan struct{})
+	sender.flushch = make(chan chan bool)
 	go sender.run()
 }
 
@@ -62,16 +64,18 @@ func (sender *GossipSender) run() {
 				return
 			}
 			sender.send(pending)
+			sender.sent = true
 		case ch := <-sender.flushch:
-			// ensure anything pending is sent, then close the channel we were given
+			// send anything pending, then reply back whether we sent
+			// anything since previous flush
 			select {
 			case pending := <-sender.cell:
-				if pending != nil {
-					sender.send(pending)
-				}
+				sender.send(pending)
+				sender.sent = true
 			default:
 			}
-			close(ch)
+			ch <- sender.sent
+			sender.sent = false
 		}
 	}
 }
@@ -328,21 +332,23 @@ func (c *GossipChannel) log(args ...interface{}) {
 
 // for testing
 
-func (router *Router) sendPendingGossip() {
+func (router *Router) sendPendingGossip() bool {
+	sentSomething := false
 	for _, channel := range router.GossipChannels {
 		channel.Lock()
 		for _, sender := range channel.senders {
-			sender.flush()
+			sentSomething = sender.flush() || sentSomething
 		}
 		for _, sender := range channel.broadcasters {
-			sender.flush()
+			sentSomething = sender.flush() || sentSomething
 		}
 		channel.Unlock()
 	}
+	return sentSomething
 }
 
-func (sender *GossipSender) flush() {
-	ch := make(chan struct{})
+func (sender *GossipSender) flush() bool {
+	ch := make(chan bool)
 	sender.flushch <- ch
-	<-ch
+	return <-ch
 }

router/gossip_test.go

@@ -32,6 +32,16 @@ func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) {
 	}
 }
 
+func sendPendingGossip(routers ...*Router) {
+	// Loop until all routers report they didn't send anything
+	for sentSomething := true; sentSomething; {
+		sentSomething = false
+		for _, router := range routers {
+			sentSomething = router.sendPendingGossip() || sentSomething
+		}
+	}
+}
+
 func (router *Router) AddTestChannelConnection(r *Router) {
 	fromName := router.Ourself.Peer.Name
 	toName := r.Ourself.Peer.Name
@@ -45,7 +55,6 @@ func (router *Router) AddTestChannelConnection(r *Router) {
 	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r}
 	router.Ourself.handleAddConnection(conn)
 	router.Ourself.handleConnectionEstablished(conn)
-	router.sendPendingGossip()
 }
 
 func (router *Router) DeleteTestChannelConnection(r *Router) {
@@ -60,11 +69,10 @@ func (router *Router) DeleteTestChannelConnection(r *Router) {
 
 	conn, _ := router.Ourself.ConnectionTo(toName)
 	router.Ourself.handleDeleteConnection(conn)
-	router.sendPendingGossip()
 }
 
 func TestGossipTopology(t *testing.T) {
-	wt.RunWithTimeout(t, 1*time.Second, func() {
+	wt.RunWithTimeout(t, 5*time.Second, func() {
 		implTestGossipTopology(t)
 	})
 }
@@ -106,14 +114,17 @@ func implTestGossipTopology(t *testing.T) {
 
 	// Now try adding some connections
 	r1.AddTestChannelConnection(r2)
+	sendPendingGossip(r1, r2)
 	checkTopology(t, r1, r1.tp(r2), r2.tp())
 	checkTopology(t, r2, r1.tp(r2), r2.tp())
 	r2.AddTestChannelConnection(r1)
+	sendPendingGossip(r1, r2)
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1))
 	checkTopology(t, r2, r1.tp(r2), r2.tp(r1))
 
 	// Currently, the connection from 2 to 3 is one-way only
 	r2.AddTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp())
 	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp())
 	// When r2 gossiped to r3, 1 was unreachable from r3 so it got removed from the
@@ -122,23 +133,27 @@ func implTestGossipTopology(t *testing.T) {
 
 	// Add a connection from 3 to 1 and now r1 is reachable.
 	r3.AddTestChannelConnection(r1)
+	sendPendingGossip(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1))
 	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1))
 	checkTopology(t, r3, r1.tp(), r2.tp(r1, r3), r3.tp(r1))
 
 	r1.AddTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
 	checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
 	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
 
 	// Drop the connection from 2 to 3
 	r2.DeleteTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
 	checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1))
 	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
 
 	// Drop the connection from 1 to 3
 	r1.DeleteTestChannelConnection(r3)
+	sendPendingGossip(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
 
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
@@ -148,5 +163,6 @@ func implTestGossipTopology(t *testing.T) {
 
 	// On a timer, r3 will gossip to r1
 	r3.SendAllGossip()
+	sendPendingGossip(r1, r2, r3)
 	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
 }