diff --git a/mesh/connection.go b/mesh/connection.go index 436183f2..0adfc793 100644 --- a/mesh/connection.go +++ b/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 diff --git a/mesh/gossip.go b/mesh/gossip.go index b803b6da..d757e996 100644 --- a/mesh/gossip.go +++ b/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 } diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go index 83431872..e0e049ef 100644 --- a/mesh/gossip_channel.go +++ b/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)} } diff --git a/mesh/gossip_test.go b/mesh/gossip_test.go index a2bf4497..aea73453 100644 --- a/mesh/gossip_test.go +++ b/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