diff --git a/router/gossip.go b/router/gossip.go index 4a2b18a2..7326afa9 100644 --- a/router/gossip.go +++ b/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,28 @@ 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 { + select { + case pending := <-sender.cell: + if pending == nil { // receive zero value when chan is closed + return + } sender.send(pending) + case ch := <-sender.flushch: + // ensure anything pending is sent, then close the channel we were given + select { + case pending := <-sender.cell: + if pending != nil { + sender.send(pending) + } + default: + } + close(ch) } } } @@ -314,27 +328,21 @@ 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() { for _, channel := range router.GossipChannels { + channel.Lock() for _, sender := range channel.senders { sender.flush() } for _, sender := range channel.broadcasters { sender.flush() } + channel.Unlock() } } func (sender *GossipSender) flush() { - for { - select { - case pending := <-sender.cell: - sender.send(pending) - default: - return - } - } + ch := make(chan struct{}) + sender.flushch <- ch + <-ch } diff --git a/router/gossip_test.go b/router/gossip_test.go index 9633e92d..047d2ab5 100644 --- a/router/gossip_test.go +++ b/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) {