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 }