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

@@ -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 {

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) {

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 {

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) {

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()...)
 }

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
 }