From 52575915d0785094c837ddb961f7b24ab7e258a4 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 23 Dec 2015 11:27:58 +0000 Subject: [PATCH 1/4] clean up GossipSender and make it more self-contained In particular don't require Send() to be serialised by the caller. We split up the triple(!) purpose 'cell' channel into: 1) a 'cell' member, holding pending GossipData or nil 2) a 'more' channel, for signalling that 'cell' has been filled 3) a 'stop' channel, for requesting termination --- mesh/gossip.go | 54 ++++++++++++++++++++++++++++++++------------------ 1 file changed, 35 insertions(+), 19 deletions(-) diff --git a/mesh/gossip.go b/mesh/gossip.go index 3754aeb5..0aba953a 100644 --- a/mesh/gossip.go +++ b/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 From a2b192c472eeca8b50d045951cb1e93d47b5726e Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 23 Dec 2015 18:48:51 +0000 Subject: [PATCH 2/4] we no longer need the GossipChannel lock around GossipSender.Send --- mesh/gossip_channel.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go index f8343e67..478f6880 100644 --- a/mesh/gossip_channel.go +++ b/mesh/gossip_channel.go @@ -128,7 +128,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 +152,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 } @@ -189,7 +189,6 @@ func (c *GossipChannel) sendDown(conns []Connection, data GossipData) { } ourConnections := c.ourself.Connections() c.Lock() - defer c.Unlock() // GC - randomly (courtesy of go's map iterator) pick some // existing senders and stop&remove them if the associated // connection is no longer active. We stop as soon as we @@ -209,13 +208,19 @@ 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 } + senders[i] = sender + } + c.Unlock() + // send + for _, sender := range senders { sender.Send(data) } } From e9520215a09de9b2b8bf0e3fdb6d9f4cbd2c1d6b Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 23 Dec 2015 20:10:52 +0000 Subject: [PATCH 3/4] refactor: introduce a helper which has a very narrow purpose, and can use 'defer' to release the lock, which is nice. --- mesh/gossip_channel.go | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go index 478f6880..17e570c7 100644 --- a/mesh/gossip_channel.go +++ b/mesh/gossip_channel.go @@ -184,11 +184,18 @@ func (c *GossipChannel) relay(srcName PeerName, data GossipData) { } func (c *GossipChannel) sendDown(conns []Connection, data GossipData) { + for _, sender := range c.sendersFor(conns) { + sender.Send(data) + } +} + +func (c *GossipChannel) sendersFor(conns []Connection) []*GossipSender { if len(conns) == 0 { - return + return nil } ourConnections := c.ourself.Connections() c.Lock() + defer c.Unlock() // GC - randomly (courtesy of go's map iterator) pick some // existing senders and stop&remove them if the associated // connection is no longer active. We stop as soon as we @@ -218,11 +225,7 @@ func (c *GossipChannel) sendDown(conns []Connection, data GossipData) { } senders[i] = sender } - c.Unlock() - // send - for _, sender := range senders { - sender.Send(data) - } + return senders } func (c *GossipChannel) makeSender(conn Connection) *GossipSender { From 4d304146f378cd566e9ef3170b5a22bbaff71a99 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 30 Dec 2015 12:29:23 +0000 Subject: [PATCH 4/4] refactor: inlining --- mesh/gossip_channel.go | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go index 17e570c7..1ecc364e 100644 --- a/mesh/gossip_channel.go +++ b/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) { @@ -180,11 +182,7 @@ 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) -} - -func (c *GossipChannel) sendDown(conns []Connection, data GossipData) { - for _, sender := range c.sendersFor(conns) { + for _, sender := range c.sendersFor(c.ourself.ConnectionsTo(c.routes.RandomNeighbours(srcName))) { sender.Send(data) } }