diff --git a/router/overlay.go b/router/overlay.go index c8c8d0c7..0be39ae9 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -11,6 +11,10 @@ type Overlay interface { // Form a packet-forwarding connection. MakeForwarder(ForwarderParams) (OverlayForwarder, error) + + // The routes have changed, so any cached information should + // be discarded. + InvalidateRoutes() } type ForwarderParams struct { @@ -81,6 +85,9 @@ func (NullOverlay) MakeForwarder(ForwarderParams) (OverlayForwarder, error) { return NullOverlay{}, nil } +func (NullOverlay) InvalidateRoutes() { +} + func (NullOverlay) SetListener(OverlayForwarderListener) { } diff --git a/router/peer.go b/router/peer.go index f01eece7..d55c4d27 100644 --- a/router/peer.go +++ b/router/peer.go @@ -66,7 +66,7 @@ func (peer *Peer) String() string { // NB: This function should generally be invoked while holding a read // lock on Peers and LocalPeer. func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) { - routes := make(map[PeerName]PeerName) + routes := make(unicastRoutes) routes[peer.Name] = UnknownPeerName nextWorklist := []*Peer{peer} for len(nextWorklist) > 0 { diff --git a/router/router.go b/router/router.go index 85140055..481efd32 100644 --- a/router/router.go +++ b/router/router.go @@ -74,7 +74,7 @@ func NewRouter(config Config, name PeerName, nickName string) *Router { router.Peers = NewPeers(router.Ourself) router.Peers.OnGC(onPeerGC) router.Peers.FetchWithDefault(router.Ourself.Peer) - router.Routes = NewRoutes(router.Ourself, router.Peers) + router.Routes = NewRoutes(router.Ourself, router.Peers, router.Overlay.InvalidateRoutes) router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port, router.PeerDiscovery) router.TopologyGossip = router.NewGossip("topology", router) router.acceptLimiter = NewTokenBucket(acceptMaxTokens, acceptTokenDelay) diff --git a/router/routes.go b/router/routes.go index bd8c3531..b463b658 100644 --- a/router/routes.go +++ b/router/routes.go @@ -5,30 +5,35 @@ import ( "sync" ) +type unicastRoutes map[PeerName]PeerName +type broadcastRoutes map[PeerName][]PeerName + type Routes struct { sync.RWMutex ourself *LocalPeer peers *Peers - unicast map[PeerName]PeerName - unicastAll map[PeerName]PeerName // [1] - broadcast map[PeerName][]PeerName - broadcastAll map[PeerName][]PeerName // [1] + onChange func() + unicast unicastRoutes + unicastAll unicastRoutes // [1] + broadcast broadcastRoutes + broadcastAll broadcastRoutes // [1] recalculate chan<- *struct{} wait chan<- chan struct{} // [1] based on *all* connections, not just established & // symmetric ones } -func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes { +func NewRoutes(ourself *LocalPeer, peers *Peers, onChange func()) *Routes { recalculate := make(chan *struct{}, 1) wait := make(chan chan struct{}) routes := &Routes{ ourself: ourself, peers: peers, - unicast: make(map[PeerName]PeerName), - unicastAll: make(map[PeerName]PeerName), - broadcast: make(map[PeerName][]PeerName), - broadcastAll: make(map[PeerName][]PeerName), + onChange: onChange, + unicast: make(unicastRoutes), + unicastAll: make(unicastRoutes), + broadcast: make(broadcastRoutes), + broadcastAll: make(broadcastRoutes), recalculate: recalculate, wait: wait} routes.unicast[ourself.Name] = UnknownPeerName @@ -147,6 +152,8 @@ func (routes *Routes) calculate() { routes.peers.RLock() routes.ourself.RLock() var ( + oldUnicast = routes.unicast + oldBroadcast = routes.broadcast unicast = routes.calculateUnicast(true) unicastAll = routes.calculateUnicast(false) broadcast = routes.calculateBroadcast(true) @@ -161,6 +168,10 @@ func (routes *Routes) calculate() { routes.broadcast = broadcast routes.broadcastAll = broadcastAll routes.Unlock() + + if !unicast.equals(oldUnicast) || !broadcast.equals(oldBroadcast) { + routes.onChange() + } } // Calculate all the routes for the question: if *we* want to send a @@ -172,7 +183,7 @@ func (routes *Routes) calculate() { // any knowledge of the MAC address at all. Thus there's no need // to exchange knowledge of MAC addresses, nor any constraints on // the routes that we construct. -func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) map[PeerName]PeerName { +func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRoutes { _, unicast := routes.ourself.Routes(nil, establishedAndSymmetric) return unicast } @@ -195,8 +206,8 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) map[PeerNam // Y =/= Z /\ X.Routes(Y) <= X.Routes(Z) => // X.Routes(Y) u [P | Y.HasSymmetricConnectionTo(P)] <= X.Routes(Z) // where <= is the subset relationship on keys of the returned map. -func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerName][]PeerName { - broadcast := make(map[PeerName][]PeerName) +func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) broadcastRoutes { + broadcast := make(broadcastRoutes) for _, peer := range routes.peers.table { hops := []PeerName{} if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found { @@ -207,3 +218,55 @@ func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerN } return broadcast } + +func (a unicastRoutes) equals(b unicastRoutes) bool { + for key, aval := range a { + if bval, ok := b[key]; !ok || bval != aval { + return false + } + } + + for key := range b { + if _, ok := a[key]; !ok { + return false + } + } + + return true +} + +func (a broadcastRoutes) equals(b broadcastRoutes) bool { + set := make(map[PeerName]struct{}) + + for key, aval := range a { + bval, ok := b[key] + if !ok { + return false + } + + for _, peer := range aval { + set[peer] = struct{}{} + } + + for _, peer := range bval { + if _, ok := set[peer]; !ok { + return false + } + + delete(set, peer) + } + + if len(set) != 0 { + return false + } + + } + + for key := range b { + if _, ok := a[key]; !ok { + return false + } + } + + return true +} diff --git a/router/sleeve.go b/router/sleeve.go index f844f4c1..8ab2b119 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -125,6 +125,10 @@ func (sleeve *SleeveOverlay) StartConsumingPackets(localPeer *Peer, peers *Peers return nil } +func (*SleeveOverlay) InvalidateRoutes() { + // no cached information, so nothing to do +} + func (sleeve *SleeveOverlay) lookupForwarder(peer PeerName) *sleeveForwarder { sleeve.lock.Lock() defer sleeve.lock.Unlock()