From aa1dd47bd2299366fd25d28bacd79d1f9198398c Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 23 Mar 2015 23:23:15 +0000 Subject: [PATCH] accumulate gossip data until it can be sent Introduce an intermediary - GossipSender - between GossipChannel and Connection. This accumulates gossip data (now represented by the new GossipData interface) from the former until it can be passed onto the latter, allowing the former to proceed when the latter may be blocked on i/o. This 1) prevents deadlocks that arise from cycles in the communication topology 2) improves performance by allowing GossipChannel and its calling code to proceed when connection i/o is blocked 3) improves performance by accumulating GossipData - the accumulated data often is considerably more compact than the sum of all the accumulated bits, and it is transmitted in a single communication event. In order to support accumulation, the GossipData interface has a Merge method. Furthermore, encoding is deferred until the data can actually be sent, since accumulation would be hard/impossible to encoded data. This required changes on some interfaces, most notably Gossiper. The implementation of GossipData for topology gossip is TopologyGossipData. It carries a reference to Peers and a set of PeerNames, indexing into Peers and referencing the Peer entries which have changed. Previously updates were represented as a list of Peer entries. The indirection via PeerNames allows Encode to omit entries which have been removed. The change in representation required changes to the signature of some methods on Peers. Note that all the above only applies to the "periodic gossip" portion of the Gossiper API, which topology gossip in fact uses for *all* gossip. GossipUnicast and GossipBroadcast are unchanged, but could conceivably receive the same treatment in the future. Fixes #445. --- router/gossip.go | 114 ++++++++++++++++++++++++++++++++++++------ router/gossip_test.go | 25 +++++++++ router/local_peer.go | 2 +- router/peers.go | 31 +++++++----- router/peers_test.go | 2 +- router/router.go | 36 ++++++++++--- 6 files changed, 175 insertions(+), 35 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index 21e1d373..b6aabb17 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -5,11 +5,17 @@ import ( "encoding/gob" "fmt" "log" + "sync" "time" ) const GossipInterval = 30 * time.Second +type GossipData interface { + Encode() []byte + Merge(GossipData) +} + type Gossip interface { // specific message from one peer to another // intermediate peers relay it using unicast topology. @@ -21,37 +27,91 @@ type Gossip interface { type Gossiper interface { OnGossipUnicast(sender PeerName, msg []byte) error OnGossipBroadcast(msg []byte) error - // Return state of everything we know; gets called periodically - Gossip() []byte + // return state of everything we know; gets called periodically + Gossip() GossipData // merge in state and return "everything new I've just learnt", // or nil if nothing in the received message was new - OnGossip(buf []byte) ([]byte, error) + OnGossip(buf []byte) (GossipData, error) } +// Accumulates GossipData that needs to be sent to one destination, +// and sends it when possible. +type GossipSender struct { + channel *GossipChannel + sender ProtocolSender + cell chan GossipData +} + +func NewGossipSender(c *GossipChannel, ps ProtocolSender) *GossipSender { + return &GossipSender{channel: c, sender: ps} +} + +func (sender *GossipSender) Start() { + sender.cell = make(chan GossipData, 1) + go sender.run() +} + +func (sender *GossipSender) run() { + for { + if pending := <-sender.cell; pending == nil { // receive zero value when chan is closed + break + } else { + sender.sendPending(pending) + } + } +} + +func (sender *GossipSender) sendPending(pending GossipData) { + sender.sender.SendProtocolMsg(sender.channel.gossipMsg(pending.Encode())) +} + +func (sender *GossipSender) Send(data GossipData) { + // NB: this must not be invoked concurrently + select { + case pending := <-sender.cell: + pending.Merge(data) + sender.cell <- pending + default: + sender.cell <- data + } +} + +func (sender *GossipSender) Stop() { + close(sender.cell) +} + +type senderMap map[Connection]*GossipSender + type GossipChannel struct { + sync.Mutex ourself *LocalPeer name string hash uint32 gossiper Gossiper + senders senderMap } func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { channelHash := hash(channelName) - channel := &GossipChannel{router.Ourself, channelName, channelHash, g} + channel := &GossipChannel{ + ourself: router.Ourself, + name: channelName, + hash: channelHash, + gossiper: g, + senders: make(senderMap)} router.GossipChannels[channelHash] = channel return channel } func (router *Router) SendAllGossip() { for _, channel := range router.GossipChannels { - channel.SendGossipMsg(channel.gossiper.Gossip()) + channel.SendGossip(channel.gossiper.Gossip()) } } func (router *Router) SendAllGossipDown(conn Connection) { for _, channel := range router.GossipChannels { - protocolMsg := channel.gossipMsg(channel.gossiper.Gossip()) - conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + channel.SendGossipDown(conn, channel.gossiper.Gossip()) } } @@ -107,19 +167,45 @@ func deliverGossip(channel *GossipChannel, srcName PeerName, _ []byte, dec *gob. if err := dec.Decode(&payload); err != nil { return err } - if newBuf, err := channel.gossiper.OnGossip(payload); err != nil { + if data, err := channel.gossiper.OnGossip(payload); err != nil { return err - } else if newBuf != nil { - channel.SendGossipMsg(newBuf) + } else if data != nil { + channel.SendGossip(data) } return nil } -func (c *GossipChannel) SendGossipMsg(buf []byte) { - protocolMsg := c.gossipMsg(buf) - for _, conn := range c.ourself.Connections() { - conn.(ProtocolSender).SendProtocolMsg(protocolMsg) +func (c *GossipChannel) SendGossip(data GossipData) { + connections := c.ourself.Connections() // do this outside the lock so they don't nest + retainedSenders := make(senderMap) + c.Lock() + defer c.Unlock() + for _, conn := range connections { + c.sendGossipDown(conn, data) + retainedSenders[conn] = c.senders[conn] + delete(c.senders, conn) } + // stop any senders for connections that are gone + for _, sender := range c.senders { + sender.Stop() + } + c.senders = retainedSenders +} + +func (c *GossipChannel) SendGossipDown(conn Connection, data GossipData) { + c.Lock() + c.sendGossipDown(conn, data) + c.Unlock() +} + +func (c *GossipChannel) sendGossipDown(conn Connection, data GossipData) { + sender, found := c.senders[conn] + if !found { + sender = NewGossipSender(c, conn.(ProtocolSender)) + c.senders[conn] = sender + sender.Start() + } + sender.Send(data) } func (c *GossipChannel) gossipMsg(buf []byte) ProtocolMsg { diff --git a/router/gossip_test.go b/router/gossip_test.go index f93ea069..d72a517e 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -29,6 +29,29 @@ func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) { if err := conn.dest.handleGossip(protocolMsg.msg, deliverGossip); err != nil { panic(err) } + conn.dest.sendPendingGossip() +} + +// 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() { + for _, channel := range router.GossipChannels { + for _, sender := range channel.senders { + sender.flush() + } + } +} + +func (sender *GossipSender) flush() { + for { + select { + case pending := <-sender.cell: + sender.sendPending(pending) + default: + return + } + } } func (router *Router) AddTestChannelConnection(r *Router) { @@ -44,6 +67,7 @@ func (router *Router) AddTestChannelConnection(r *Router) { conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r} router.Ourself.handleAddConnection(conn) router.Ourself.handleConnectionEstablished(conn) + router.sendPendingGossip() } func (router *Router) DeleteTestChannelConnection(r *Router) { @@ -58,6 +82,7 @@ func (router *Router) DeleteTestChannelConnection(r *Router) { conn, _ := router.Ourself.ConnectionTo(toName) router.Ourself.handleDeleteConnection(conn) + router.sendPendingGossip() } func TestGossipTopology(t *testing.T) { diff --git a/router/local_peer.go b/router/local_peer.go index d719dd20..3e499aad 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -261,7 +261,7 @@ func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { // abstraction (hence the cast) and send a regular update. This is // less efficient though since it will almost certainly reach // peers more than once. - peer.Router.TopologyGossip.(*GossipChannel).SendGossipMsg(EncodePeers(append(peers, peer.Peer)...)) + peer.Router.TopologyGossip.(*GossipChannel).SendGossip(NewTopologyGossipData(peer.Router.Peers, append(peers, peer.Peer)...)) } func (peer *LocalPeer) checkConnectionLimit() error { diff --git a/router/peers.go b/router/peers.go index b22940b7..433446d0 100644 --- a/router/peers.go +++ b/router/peers.go @@ -69,7 +69,7 @@ func (peers *Peers) ForEach(fun func(PeerName, *Peer)) { // update contains a more recent version than known to us. The return // value is an "improved" update containing just these new/updated // elements. -func (peers *Peers) ApplyUpdate(update []byte) ([]byte, error) { +func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, error) { peers.Lock() newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update) @@ -95,19 +95,27 @@ func (peers *Peers) ApplyUpdate(update []byte) ([]byte, error) { // Don't need to hold peers lock any longer peers.Unlock() - return encodePeersMap(newUpdate), nil + return setFromPeersMap(newUpdate), nil } -func (peers *Peers) EncodeAllPeers() []byte { +func (peers *Peers) Names() PeerNameSet { peers.RLock() defer peers.RUnlock() - return encodePeersMap(peers.table) + return setFromPeersMap(peers.table) } -func EncodePeers(peers ...*Peer) []byte { +func (peers *Peers) EncodePeers(names PeerNameSet) []byte { + peers.RLock() + peerList := make([]*Peer, 0, len(names)) + for name, _ := range names { + if peer, found := peers.table[name]; found { + peerList = append(peerList, peer) + } + } + peers.RUnlock() // release lock so we don't hold it while encoding buf := new(bytes.Buffer) enc := gob.NewEncoder(buf) - for _, peer := range peers { + for _, peer := range peerList { peer.encode(enc) } return buf.Bytes() @@ -159,13 +167,12 @@ func (peers *Peers) garbageCollect() []*Peer { return removed } -func encodePeersMap(peers map[PeerName]*Peer) []byte { - buf := new(bytes.Buffer) - enc := gob.NewEncoder(buf) - for _, peer := range peers { - peer.encode(enc) +func setFromPeersMap(peers map[PeerName]*Peer) PeerNameSet { + names := make(PeerNameSet) + for name, _ := range peers { + names[name] = true } - return buf.Bytes() + return names } func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]byte, err error) { diff --git a/router/peers_test.go b/router/peers_test.go index b3257789..d81e4d7b 100644 --- a/router/peers_test.go +++ b/router/peers_test.go @@ -30,7 +30,7 @@ func checkApplyUpdate(t *testing.T, peers *Peers) { // into it. _, testBedPeers := newNode(dummyName) testBedPeers.AddTestConnection(peers.ourself) - testBedPeers.ApplyUpdate(peers.EncodeAllPeers()) + testBedPeers.ApplyUpdate(peers.EncodePeers(peers.Names())) checkTopologyPeers(t, true, testBedPeers.allPeersExcept(dummyName), peers.allPeers()...) } diff --git a/router/router.go b/router/router.go index 16a29eb3..f4e3a48a 100644 --- a/router/router.go +++ b/router/router.go @@ -361,14 +361,36 @@ func (router *Router) OnGossipBroadcast(msg []byte) error { return fmt.Errorf("unexpected topology gossip broadcast: %v", msg) } -// Return state of everything we know; intended to be called periodically -func (router *Router) Gossip() []byte { - return router.Peers.EncodeAllPeers() +type PeerNameSet map[PeerName]bool + +type TopologyGossipData struct { + peers *Peers + update PeerNameSet } -// merge in state and return "everything new I've just learnt", -// or nil if nothing in the received message was new -func (router *Router) OnGossip(buf []byte) ([]byte, error) { +func NewTopologyGossipData(peers *Peers, update ...*Peer) *TopologyGossipData { + names := make(PeerNameSet) + for _, p := range update { + names[p.Name] = true + } + return &TopologyGossipData{peers: peers, update: names} +} + +func (d *TopologyGossipData) Merge(other GossipData) { + for name, _ := range other.(*TopologyGossipData).update { + d.update[name] = true + } +} + +func (d *TopologyGossipData) Encode() []byte { + return d.peers.EncodePeers(d.update) +} + +func (router *Router) Gossip() GossipData { + return &TopologyGossipData{peers: router.Peers, update: router.Peers.Names()} +} + +func (router *Router) OnGossip(buf []byte) (GossipData, error) { newUpdate, err := router.Peers.ApplyUpdate(buf) if _, ok := err.(UnknownPeerError); err != nil && ok { // That update contained a reference to a peer which wasn't @@ -386,5 +408,5 @@ func (router *Router) OnGossip(buf []byte) ([]byte, error) { } router.ConnectionMaker.Refresh() router.Routes.Recalculate() - return newUpdate, nil + return &TopologyGossipData{peers: router.Peers, update: newUpdate}, nil }