get rid of GossipSender gc

With the per-broadcast-source GossipSenders gone, we can get rid of
GossipSender gc...

Instead of maintaining a Connection->GossipSender map in the
GossipChannels - which we then need to gc when connections get closed
- we keep gossip_channel_name->GossipSender maps in LocalConnections.

Access to the map is guarded by a separate lock, rather than re-using
the LocalConnection lock. This reduces lock contention and risk of
deadlock.

In the GossipSenders, we replace the Stop() method with a signalling
channel, i.e. senders stop when that channel gets closed. Handily, the
LocalConnection.finished channel fits that description. So when a
LocalConnection's run loop terminates, any existing and future
associated GossipSenders will end up terminating too.

mesh/connection.go

@@ -53,6 +53,11 @@ type LocalConnection struct {
 	errorChan       chan<- error
 	finished        <-chan struct{} // closed to signal that actorLoop has finished
 	OverlayConn     OverlayConnection
+	gossipSenders   *GossipSenders
+}
+
+type GossipConnection interface {
+	GossipSenders() *GossipSenders
 }
 
 type ConnectionAction func() error
@@ -101,6 +106,7 @@ func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, ro
 		actionChan:       actionChan,
 		errorChan:        errorChan,
 		finished:         finished}
+	conn.gossipSenders = NewGossipSenders(conn, finished)
 	go conn.run(actionChan, errorChan, finished, acceptNewPeer)
 }
 
@@ -129,6 +135,10 @@ func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) error {
 	return nil
 }
 
+func (conn *LocalConnection) GossipSenders() *GossipSenders {
+	return conn.gossipSenders
+}
+
 // ACTOR methods
 
 // NB: The conn.* fields are only written by the connection actor

mesh/gossip.go

@@ -39,13 +39,11 @@ type GossipSender struct {
 	sender           ProtocolSender
 	gossip           GossipData
 	broadcasts       map[PeerName]GossipData
-	stop             chan<- struct{}
 	more             chan<- struct{}
 	flush            chan<- chan<- bool // for testing
 }
 
-func NewGossipSender(makeMsg func(msg []byte) ProtocolMsg, makeBroadcastMsg func(srcName PeerName, msg []byte) ProtocolMsg, sender ProtocolSender) *GossipSender {
-	stop := make(chan struct{})
+func NewGossipSender(makeMsg func(msg []byte) ProtocolMsg, makeBroadcastMsg func(srcName PeerName, msg []byte) ProtocolMsg, sender ProtocolSender, stop <-chan struct{}) *GossipSender {
 	more := make(chan struct{}, 1)
 	flush := make(chan chan<- bool)
 	s := &GossipSender{
@@ -53,7 +51,6 @@ func NewGossipSender(makeMsg func(msg []byte) ProtocolMsg, makeBroadcastMsg func
 		makeBroadcastMsg: makeBroadcastMsg,
 		sender:           sender,
 		broadcasts:       make(map[PeerName]GossipData),
-		stop:             stop,
 		more:             more,
 		flush:            flush}
 	go s.run(stop, more, flush)
@@ -177,8 +174,37 @@ func (s *GossipSender) Flush() bool {
 	return <-ch
 }
 
-func (s *GossipSender) Stop() {
-	close(s.stop)
+type GossipSenders struct {
+	sync.Mutex
+	sender  ProtocolSender
+	stop    <-chan struct{}
+	senders map[string]*GossipSender
+}
+
+func NewGossipSenders(sender ProtocolSender, stop <-chan struct{}) *GossipSenders {
+	return &GossipSenders{sender: sender, stop: stop, senders: make(map[string]*GossipSender)}
+}
+
+func (gs *GossipSenders) Sender(channelName string, makeGossipSender func(sender ProtocolSender, stop <-chan struct{}) *GossipSender) *GossipSender {
+	gs.Lock()
+	defer gs.Unlock()
+	s, found := gs.senders[channelName]
+	if !found {
+		s = makeGossipSender(gs.sender, gs.stop)
+		gs.senders[channelName] = s
+	}
+	return s
+}
+
+// for testing
+func (gs *GossipSenders) Flush() bool {
+	sent := false
+	gs.Lock()
+	defer gs.Unlock()
+	for _, sender := range gs.senders {
+		sent = sender.Flush() || sent
+	}
+	return sent
 }
 
 type GossipChannels map[string]*GossipChannel
@@ -242,12 +268,8 @@ func (router *Router) SendAllGossipDown(conn Connection) {
 
 func (router *Router) sendPendingGossip() bool {
 	sentSomething := false
-	for channel := range router.gossipChannelSet() {
-		channel.Lock()
-		for _, sender := range channel.senders {
-			sentSomething = sender.Flush() || sentSomething
-		}
-		channel.Unlock()
+	for conn := range router.Ourself.Connections() {
+		sentSomething = conn.(GossipConnection).GossipSenders().Flush() || sentSomething
 	}
 	return sentSomething
 }

mesh/gossip_channel.go

@@ -4,27 +4,21 @@ import (
 	"bytes"
 	"encoding/gob"
 	"fmt"
-	"sync"
 )
 
 type GossipChannel struct {
-	sync.Mutex
 	name     string
 	ourself  *LocalPeer
 	routes   *Routes
 	gossiper Gossiper
-	senders  connectionSenders
 }
 
-type connectionSenders map[Connection]*GossipSender
-
 func NewGossipChannel(channelName string, ourself *LocalPeer, routes *Routes, g Gossiper) *GossipChannel {
 	return &GossipChannel{
 		name:     channelName,
 		ourself:  ourself,
 		routes:   routes,
 		gossiper: g,
-		senders:  make(connectionSenders),
 	}
 }
 
@@ -141,44 +135,17 @@ func (c *GossipChannel) relay(srcName PeerName, data GossipData) {
 }
 
 func (c *GossipChannel) sendersFor(conns []Connection) []*GossipSender {
-	if len(conns) == 0 {
-		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
-	// encounter a valid entry; the idea being that when there is
-	// little or no garbage then this executes close to O(1)[1],
-	// whereas when there is lots of garbage we remove it quickly.
-	//
-	// [1] TODO Unfortunately, due to the desire to avoid nested
-	// locks, instead of simply invoking LocalPeer.ConnectionTo(name),
-	// we operate on LocalPeer.Connections(). That is
-	// O(n_our_connections) at best.
-	for conn, sender := range c.senders {
-		if _, found := ourConnections[conn]; !found {
-			delete(c.senders, conn)
-			sender.Stop()
-		} else {
-			break
-		}
-	}
-	// start senders, if necessary
 	senders := make([]*GossipSender, len(conns), len(conns))
 	for i, conn := range conns {
-		sender, found := c.senders[conn]
-		if !found {
-			sender = NewGossipSender(c.makeMsg, c.makeBroadcastMsg, conn.(ProtocolSender))
-			c.senders[conn] = sender
-		}
-		senders[i] = sender
+		senders[i] = conn.(GossipConnection).GossipSenders().Sender(c.name, c.makeGossipSender)
 	}
 	return senders
 }
 
+func (c *GossipChannel) makeGossipSender(sender ProtocolSender, stop <-chan struct{}) *GossipSender {
+	return NewGossipSender(c.makeMsg, c.makeBroadcastMsg, sender, stop)
+}
+
 func (c *GossipChannel) makeMsg(msg []byte) ProtocolMsg {
 	return ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)}
 }

mesh/gossip_test.go

@@ -13,8 +13,9 @@ import (
 
 type MockGossipConnection struct {
 	RemoteConnection
-	dest  *Router
-	start chan struct{}
+	dest          *Router
+	gossipSenders *GossipSenders
+	start         chan struct{}
 }
 
 func NewTestRouter(name string) *Router {
@@ -29,6 +30,10 @@ func (conn *MockGossipConnection) SendProtocolMsg(protocolMsg ProtocolMsg) error
 	return conn.dest.handleGossip(protocolMsg.tag, protocolMsg.msg)
 }
 
+func (conn *MockGossipConnection) GossipSenders() *GossipSenders {
+	return conn.gossipSenders
+}
+
 func (conn *MockGossipConnection) Start() {
 	close(conn.start)
 }
@@ -56,7 +61,11 @@ func (router *Router) NewTestGossipConnection(r *Router) *MockGossipConnection {
 	toPeer = router.Peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount
 
 	conn := &MockGossipConnection{
-		RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r, make(chan struct{})}
+		RemoteConnection: RemoteConnection{router.Ourself.Peer, toPeer, "", false, true},
+		dest:             r,
+		start:            make(chan struct{}),
+	}
+	conn.gossipSenders = NewGossipSenders(conn, make(chan struct{}))
 	router.Ourself.handleAddConnection(conn)
 	router.Ourself.handleConnectionEstablished(conn)
 	return conn