reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1773 from weaveworks/1761-lazy-route-calculation-1

Browse Source 
lazy route calculation; LGTM
This commit is contained in:
Bryan Boreham committed 2015-12-23 12:27:02 +00:00
2 parents 4214a09392 b1cb0c298d
commit 1796582032
1 file changed
+54 -78
Download Patch File Download Diff File Expand all files Collapse all files
mesh/routes.go
+54 -78
View File
@@ -19,6 +19,7 @@ type Routes struct {
broadcastAll broadcastRoutes // [1]
recalculate chan<- *struct{}
wait chan<- chan struct{}
action chan<- func()
// [1] based on *all* connections, not just established &
// symmetric ones
}
@@ -26,6 +27,7 @@ type Routes struct {
func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes {
recalculate := make(chan *struct{}, 1)
wait := make(chan chan struct{})
action := make(chan func())
routes := &Routes{
ourself: ourself,
peers: peers,
@@ -34,12 +36,13 @@ func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes {
broadcast: make(broadcastRoutes),
broadcastAll: make(broadcastRoutes),
recalculate: recalculate,
wait: wait}
wait: wait,
action: action}
routes.unicast[ourself.Name] = UnknownPeerName
routes.unicastAll[ourself.Name] = UnknownPeerName
routes.broadcast[ourself.Name] = []PeerName{}
routes.broadcastAll[ourself.Name] = []PeerName{}
go routes.run(recalculate, wait)
go routes.run(recalculate, wait, action)
return routes
}
@@ -68,15 +71,39 @@ func (routes *Routes) UnicastAll(name PeerName) (PeerName, bool) {
}
func (routes *Routes) Broadcast(name PeerName) []PeerName {
routes.RLock()
defer routes.RUnlock()
return routes.broadcast[name]
return routes.lookupOrCalculate(name, &routes.broadcast, true)
}
func (routes *Routes) BroadcastAll(name PeerName) []PeerName {
return routes.lookupOrCalculate(name, &routes.broadcastAll, false)
}
func (routes *Routes) lookupOrCalculate(name PeerName, broadcast *broadcastRoutes, establishedAndSymmetric bool) []PeerName {
routes.RLock()
defer routes.RUnlock()
return routes.broadcastAll[name]
hops, found := (*broadcast)[name]
routes.RUnlock()
if found {
return hops
}
res := make(chan []PeerName)
routes.action <- func() {
routes.RLock()
hops, found := (*broadcast)[name]
routes.RUnlock()
if found {
res <- hops
}
routes.peers.RLock()
routes.ourself.RLock()
hops = routes.calculateBroadcast(name, establishedAndSymmetric)
routes.ourself.RUnlock()
routes.peers.RUnlock()
res <- hops
routes.Lock()
(*broadcast)[name] = hops
routes.Unlock()
}
return <-res
}
// Choose min(log2(n_peers), n_neighbouring_peers) neighbours, with a
@@ -123,6 +150,7 @@ func (routes *Routes) Recalculate() {
select {
case routes.recalculate <- nil:
default:
log.Println("route recalculation coalesced")
}
}
@@ -133,7 +161,7 @@ func (routes *Routes) EnsureRecalculated() {
<-done
}
func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}) {
func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}, action <-chan func()) {
for {
select {
case <-recalculate:
@@ -145,6 +173,8 @@ func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{
default:
}
close(done)
case f := <-action:
f()
}
}
}
@@ -153,13 +183,13 @@ 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)
broadcastAll = routes.calculateBroadcast(false)
broadcast = make(broadcastRoutes)
broadcastAll = make(broadcastRoutes)
)
broadcast[routes.ourself.Name] = routes.calculateBroadcast(routes.ourself.Name, true)
broadcastAll[routes.ourself.Name] = routes.calculateBroadcast(routes.ourself.Name, false)
routes.ourself.RUnlock()
routes.peers.RUnlock()
@@ -171,10 +201,8 @@ func (routes *Routes) calculate() {
onChange := routes.onChange
routes.Unlock()
if !unicast.equals(oldUnicast) || !broadcast.equals(oldBroadcast) {
for _, callback := range onChange {
callback()
}
for _, callback := range onChange {
callback()
}
}
@@ -192,7 +220,7 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRout
return unicast
}
// Calculate all the routes for the question: if we receive a
// Calculate the route to answer the question: if we receive a
// broadcast originally from Peer X, which peers should we pass the
// frames on to?
//
@@ -210,67 +238,15 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRout
// 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) broadcastRoutes {
broadcast := make(broadcastRoutes)
for _, peer := range routes.peers.byName {
hops := []PeerName{}
if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found {
routes.ourself.ForEachConnectedPeer(establishedAndSymmetric, reached,
func(remotePeer *Peer) { hops = append(hops, remotePeer.Name) })
}
broadcast[peer.Name] = hops
func (routes *Routes) calculateBroadcast(name PeerName, establishedAndSymmetric bool) []PeerName {
hops := []PeerName{}
peer, found := routes.peers.byName[name]
if !found {
return hops
}
return broadcast
}
func (a unicastRoutes) equals(b unicastRoutes) bool {
for key, aval := range a {
if bval, ok := b[key]; !ok || bval != aval {
return false
}
if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found {
routes.ourself.ForEachConnectedPeer(establishedAndSymmetric, reached,
func(remotePeer *Peer) { hops = append(hops, remotePeer.Name) })
}
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
return hops
}