From 569fd784fa58fabe47bf3b3fe1d4118899fd6ccd Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Jun 2015 20:40:23 +0100 Subject: [PATCH 1/5] 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 | 37 ++++++++++++++++++++++++++----------- router/gossip_test.go | 3 ++- 2 files changed, 28 insertions(+), 12 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index 4a2b18a2..0e345587 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,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 } } 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) { From bd5353024f51610072955066549f3682e894ee83 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Jun 2015 11:05:04 +0100 Subject: [PATCH 2/5] Helpful suggestions from the linter --- router/gossip.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index 0e345587..684bb21d 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -60,9 +60,8 @@ func (sender *GossipSender) run() { case pending := <-sender.cell: if pending == nil { // receive zero value when chan is closed return - } else { - sender.send(pending) } + sender.send(pending) case ch := <-sender.flushch: sender.flush() close(ch) @@ -324,7 +323,7 @@ func (c *GossipChannel) log(args ...interface{}) { func (router *Router) sendPendingGossip() { // copy the senders so we don't hold the lock for long - allSenders := make([]*GossipSender, 0) + var allSenders []*GossipSender for _, channel := range router.GossipChannels { channel.Lock() for _, sender := range channel.senders { From 35b3a2d6a67ebfd7886f4554282a00506a8390ad Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Jun 2015 12:18:09 +0100 Subject: [PATCH 3/5] Inline GossipSender.flush() and don't try to send nil --- router/gossip.go | 20 ++++++++------------ 1 file changed, 8 insertions(+), 12 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index 684bb21d..e119e661 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -63,7 +63,14 @@ func (sender *GossipSender) run() { } sender.send(pending) case ch := <-sender.flushch: - sender.flush() + // 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) } } @@ -341,14 +348,3 @@ func (router *Router) sendPendingGossip() { <-ch } } - -func (sender *GossipSender) flush() { - for { - select { - case pending := <-sender.cell: - sender.send(pending) - default: - return - } - } -} From 5ebfc9027bc8c1114dc8b4098b223964d9ddb4e7 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Jun 2015 12:22:38 +0100 Subject: [PATCH 4/5] Flush gossip inside the channel lock so we know sender hasn't been Stop()-ed --- router/gossip.go | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index e119e661..81ec7f1c 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -329,22 +329,18 @@ func (c *GossipChannel) log(args ...interface{}) { // for testing func (router *Router) sendPendingGossip() { - // copy the senders so we don't hold the lock for long - var allSenders []*GossipSender for _, channel := range router.GossipChannels { channel.Lock() for _, sender := range channel.senders { - allSenders = append(allSenders, sender) + ch := make(chan struct{}) + sender.flushch <- ch + <-ch } for _, sender := range channel.broadcasters { - allSenders = append(allSenders, sender) + ch := make(chan struct{}) + sender.flushch <- ch + <-ch } channel.Unlock() } - - for _, sender := range allSenders { - ch := make(chan struct{}) - sender.flushch <- ch - <-ch - } } From 0cd41270871b725e93fa1cd64e473fc5c2999f02 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Jun 2015 12:24:19 +0100 Subject: [PATCH 5/5] Factor out GossipSender.flush() --- router/gossip.go | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index 81ec7f1c..7326afa9 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -332,15 +332,17 @@ func (router *Router) sendPendingGossip() { for _, channel := range router.GossipChannels { channel.Lock() for _, sender := range channel.senders { - ch := make(chan struct{}) - sender.flushch <- ch - <-ch + sender.flush() } for _, sender := range channel.broadcasters { - ch := make(chan struct{}) - sender.flushch <- ch - <-ch + sender.flush() } channel.Unlock() } } + +func (sender *GossipSender) flush() { + ch := make(chan struct{}) + sender.flushch <- ch + <-ch +}