Merge pull request #476 from rade/445-accumulate-gossip

LGTM.  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
 }