Notify the Overlay when routes change
This commit is contained in:
1 parent
c2ad4c5633
commit
46d454730f
5 files changed
+88
-14
No files matched your search
@@ -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) {
|
||||
}
|
||||
|
||||
|
||||
+1
-1
@@ -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 {
|
||||
|
||||
+1
-1
@@ -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)
|
||||
|
||||
+75
-12
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user