diff --git a/router/gossip.go b/router/gossip.go index 9f1a30a4..96f4abe8 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -20,13 +20,15 @@ type Gossip interface { // specific message from one peer to another // intermediate peers relay it using unicast topology. GossipUnicast(dstPeerName PeerName, msg []byte) error - // send a message to every peer, relayed using broadcast topology. - GossipBroadcast(msg []byte) error + // send gossip to every peer, relayed using broadcast topology. + GossipBroadcast(update GossipData) error } type Gossiper interface { OnGossipUnicast(sender PeerName, msg []byte) error - OnGossipBroadcast(msg []byte) error + // merge received data into state and return a representation of + // the received data, for further propagation + OnGossipBroadcast(update []byte) (GossipData, error) // return state of everything we know; gets called periodically Gossip() GossipData // merge received data into state and return "everything new I've @@ -76,24 +78,27 @@ func (sender *GossipSender) Stop() { } type connectionSenders map[Connection]*GossipSender +type peerSenders map[PeerName]*GossipSender type GossipChannel struct { sync.Mutex - ourself *LocalPeer - name string - hash uint32 - gossiper Gossiper - senders connectionSenders + ourself *LocalPeer + name string + hash uint32 + gossiper Gossiper + senders connectionSenders + broadcasters peerSenders } func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { channelHash := hash(channelName) channel := &GossipChannel{ - ourself: router.Ourself, - name: channelName, - hash: channelHash, - gossiper: g, - senders: make(connectionSenders)} + ourself: router.Ourself, + name: channelName, + hash: channelHash, + gossiper: g, + senders: make(connectionSenders), + broadcasters: make(peerSenders)} router.GossipChannels[channelHash] = channel return channel } @@ -150,15 +155,16 @@ func (c *GossipChannel) deliverUnicast(srcName PeerName, origPayload []byte, dec return c.gossiper.OnGossipUnicast(srcName, payload) } -func (c *GossipChannel) deliverBroadcast(srcName PeerName, origPayload []byte, dec *gob.Decoder) error { +func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.Decoder) error { var payload []byte if err := dec.Decode(&payload); err != nil { return err } - if err := c.gossiper.OnGossipBroadcast(payload); err != nil { + data, err := c.gossiper.OnGossipBroadcast(payload) + if err != nil || data == nil { return err } - return c.relayBroadcast(srcName, origPayload) + return c.relayBroadcast(srcName, data) } func (c *GossipChannel) deliver(_ PeerName, _ []byte, dec *gob.Decoder) error { @@ -214,8 +220,8 @@ func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, msg []byte) error { return c.relayUnicast(dstPeerName, GobEncode(c.hash, c.ourself.Name, dstPeerName, msg)) } -func (c *GossipChannel) GossipBroadcast(msg []byte) error { - return c.relayBroadcast(c.ourself.Name, GobEncode(c.hash, c.ourself.Name, msg)) +func (c *GossipChannel) GossipBroadcast(update GossipData) error { + return c.relayBroadcast(c.ourself.Name, update) } func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error { @@ -229,17 +235,50 @@ func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error { return nil } -func (c *GossipChannel) relayBroadcast(srcName PeerName, buf []byte) error { +func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error { + names := c.ourself.Router.Peers.Names() // do this outside the lock so they don't nest + c.Lock() + defer c.Unlock() + // GC - randomly (courtesy of go's map iterator) pick some + // existing broadcasters and stop&remove them if their source peer + // is unknown. 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 Peers.Fetch(name) below, we + // have that Peers.Names() invocation above. That is O(n_peers) at + // best. + for name, broadcaster := range c.broadcasters { + if _, found := names[name]; !found { + delete(c.broadcasters, name) + broadcaster.Stop() + } else { + break + } + } + broadcaster, found := c.broadcasters[srcName] + if !found { + broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) }) + c.broadcasters[srcName] = broadcaster + broadcaster.Start() + } + broadcaster.Send(update) + return nil +} + +func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) { + c.ourself.Router.Routes.EnsureRecalculated() nextHops := c.ourself.Router.Routes.BroadcastAll(srcName) if len(nextHops) == 0 { - return nil + return } - protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, buf} + protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.hash, srcName, update.Encode())} // FIXME a single blocked connection can stall us for _, conn := range c.ourself.ConnectionsTo(nextHops) { conn.(ProtocolSender).SendProtocolMsg(protocolMsg) } - return nil } func (c *GossipChannel) log(args ...interface{}) { @@ -256,6 +295,9 @@ func (router *Router) sendPendingGossip() { for _, sender := range channel.senders { sender.flush() } + for _, sender := range channel.broadcasters { + sender.flush() + } } } diff --git a/router/local_peer.go b/router/local_peer.go index dfde0493..7346c594 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -241,14 +241,7 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) { func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { peer.Router.Routes.Recalculate() - // TODO We should just be invoking TopologyGossip.GossipBroadcast - // here, but route calculation is asynchronous and in this - // particular case would likely result in the broadcast not - // reaching all peers. So instead we slightly break the Gossip - // 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).Send(NewTopologyGossipData(peer.Router.Peers, append(peers, peer.Peer)...)) + peer.Router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(peer.Router.Peers, append(peers, peer.Peer)...)) } func (peer *LocalPeer) checkConnectionLimit() error { diff --git a/router/peers.go b/router/peers.go index 2ade26bd..059e41c3 100644 --- a/router/peers.go +++ b/router/peers.go @@ -69,15 +69,15 @@ func (peers *Peers) ForEach(fun func(*Peer)) { // // We add peers hitherto unknown to us, and update peers for which the // 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) (PeerNameSet, error) { +// value is a) a representation of the received update, and b) an +// "improved" update containing just these new/updated elements. +func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) { peers.Lock() newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update) if err != nil { peers.Unlock() - return nil, err + return nil, nil, err } // By this point, we know the update doesn't refer to any peers we @@ -97,7 +97,11 @@ func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, error) { // Don't need to hold peers lock any longer peers.Unlock() - return setFromPeersMap(newUpdate), nil + updateNames := make(PeerNameSet) + for _, peer := range decodedUpdate { + updateNames[peer.Name] = void + } + return updateNames, setFromPeersMap(newUpdate), nil } func (peers *Peers) Names() PeerNameSet { diff --git a/router/protocol.go b/router/protocol.go index 9184512c..b9bb9f29 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -2,7 +2,7 @@ package router const ( Protocol = "weave" - ProtocolVersion = 14 + ProtocolVersion = 15 ) type ProtocolTag byte diff --git a/router/router.go b/router/router.go index 7ef3a9a2..2eb9af36 100644 --- a/router/router.go +++ b/router/router.go @@ -377,8 +377,12 @@ func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error { return fmt.Errorf("unexpected topology gossip unicast: %v", msg) } -func (router *Router) OnGossipBroadcast(msg []byte) error { - return fmt.Errorf("unexpected topology gossip broadcast: %v", msg) +func (router *Router) OnGossipBroadcast(update []byte) (GossipData, error) { + origUpdate, _, err := router.applyTopologyUpdate(update) + if err != nil || len(origUpdate) == 0 { + return nil, err + } + return &TopologyGossipData{peers: router.Peers, update: origUpdate}, nil } func (router *Router) Gossip() GossipData { @@ -386,29 +390,29 @@ func (router *Router) Gossip() GossipData { } func (router *Router) OnGossip(update []byte) (GossipData, error) { - newUpdate, err := router.applyTopologyUpdate(update) + _, newUpdate, err := router.applyTopologyUpdate(update) if err != nil || len(newUpdate) == 0 { return nil, err } return &TopologyGossipData{peers: router.Peers, update: newUpdate}, nil } -func (router *Router) applyTopologyUpdate(update []byte) (PeerNameSet, error) { - newUpdate, err := router.Peers.ApplyUpdate(update) +func (router *Router) applyTopologyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) { + origUpdate, newUpdate, err := router.Peers.ApplyUpdate(update) if _, ok := err.(UnknownPeerError); err != nil && ok { // That update contained a reference to a peer which wasn't // itself included in the update, and we didn't know about // already. We ignore this; eventually we should receive an // update containing a complete topology. log.Println("Topology gossip:", err) - return nil, nil + return nil, nil, nil } if err != nil { - return nil, err + return nil, nil, err } if len(newUpdate) > 0 { router.ConnectionMaker.Refresh() router.Routes.Recalculate() } - return newUpdate, nil + return origUpdate, newUpdate, nil } diff --git a/router/routes.go b/router/routes.go index 4745972d..df2b890a 100644 --- a/router/routes.go +++ b/router/routes.go @@ -15,6 +15,7 @@ type Routes struct { broadcast map[PeerName][]PeerName broadcastAll map[PeerName][]PeerName // [1] recalculate chan<- *struct{} + wait chan<- chan struct{} // [1] based on *all* connections, not just established & // symmetric ones } @@ -36,8 +37,10 @@ func NewRoutes(ourself *Peer, peers *Peers) *Routes { func (routes *Routes) Start() { recalculate := make(chan *struct{}, 1) + wait := make(chan chan struct{}) routes.recalculate = recalculate - go routes.run(recalculate) + routes.wait = wait + go routes.run(recalculate, wait) } func (routes *Routes) Unicast(name PeerName) (PeerName, bool) { @@ -91,17 +94,39 @@ func (routes *Routes) String() string { return buf.String() } +// Request recalculation of the routing table. This is async but can +// effectively be made synchronous with a subsequent call to +// EnsureRecalculated. func (routes *Routes) Recalculate() { + // The use of a 1-capacity channel in combination with the + // non-blocking send is an optimisation that results in multiple + // requests being coalesced. select { case routes.recalculate <- nil: default: } } -func (routes *Routes) run(recalculate <-chan *struct{}) { +// Wait for any preceding Recalculate requests to be processed. +func (routes *Routes) EnsureRecalculated() { + done := make(chan struct{}) + routes.wait <- done + <-done +} + +func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}) { for { - <-recalculate - routes.calculate() + select { + case <-recalculate: + routes.calculate() + case done := <-wait: + select { + case <-recalculate: + routes.calculate() + default: + } + close(done) + } } }