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 diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go index f8343e67..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) { @@ -128,7 +130,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 +154,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 } @@ -180,12 +182,14 @@ 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) + for _, sender := range c.sendersFor(c.ourself.ConnectionsTo(c.routes.RandomNeighbours(srcName))) { + sender.Send(data) + } } -func (c *GossipChannel) sendDown(conns []Connection, data GossipData) { +func (c *GossipChannel) sendersFor(conns []Connection) []*GossipSender { if len(conns) == 0 { - return + return nil } ourConnections := c.ourself.Connections() c.Lock() @@ -209,15 +213,17 @@ 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 } - sender.Send(data) + senders[i] = sender } + return senders } func (c *GossipChannel) makeSender(conn Connection) *GossipSender {