Merge pull request openziti/fabric#291 from openziti/router-perf-follow-up

Router perf follow up

controller/handler_mgmt/list_links.go

@@ -54,7 +54,7 @@ func (h *listLinksHandler) HandleReceive(msg *channel2.Message, ch channel2.Chan
 			Src:        l.Src.Id,
 			Dst:        l.Dst.Id,
 			State:      l.CurrentState().Mode.String(),
-			Down:       l.Down,
+			Down:       l.IsDown(),
 			Cost:       l.GetStaticCost(),
 			SrcLatency: l.GetSrcLatency(),
 			DstLatency: l.GetDstLatency(),

controller/handler_mgmt/set_link_down.go

@@ -45,7 +45,7 @@ func (h *setLinkDownHandler) HandleReceive(msg *channel2.Message, ch channel2.Ch
 	set := &mgmt_pb.SetLinkDownRequest{}
 	if err := proto.Unmarshal(msg.Body, set); err == nil {
 		if l, found := h.network.GetLink(&identity.TokenId{Token: set.LinkId}); found {
-			l.Down = set.Down
+			l.SetDown(set.Down)
 			h.network.LinkChanged(l)
 			log.Infof("set down state of link [l/%s] to [%t]", set.LinkId, set.Down)
 			handler_common.SendSuccess(msg, ch, "")

controller/network/link.go

@@ -18,7 +18,9 @@ package network
 
 import (
 	"github.com/openziti/foundation/identity/identity"
+	"github.com/openziti/foundation/util/concurrenz"
 	"github.com/openziti/foundation/util/info"
+	"sync"
 	"sync/atomic"
 )
 
@@ -27,18 +29,20 @@ type Link struct {
 	Src        *Router
 	Dst        *Router
 	state      []*LinkState
-	Down       bool
+	down       bool
 	StaticCost int32
 	SrcLatency int64
 	DstLatency int64
 	Cost       int64
+	usable     concurrenz.AtomicBoolean
+	lock       sync.Mutex
 }
 
 func newLink(id *identity.TokenId) *Link {
 	l := &Link{
 		Id:         id,
 		state:      make([]*LinkState, 0),
-		Down:       false,
+		down:       false,
 		StaticCost: 1,
 		Cost:       1,
 	}
@@ -47,6 +51,8 @@ func newLink(id *identity.TokenId) *Link {
 }
 
 func (link *Link) CurrentState() *LinkState {
+	link.lock.Lock()
+	defer link.lock.Unlock()
 	if link.state == nil || len(link.state) < 1 {
 		return nil
 	}
@@ -54,10 +60,41 @@ func (link *Link) CurrentState() *LinkState {
 }
 
 func (link *Link) addState(s *LinkState) {
+	link.lock.Lock()
+	defer link.lock.Unlock()
+
 	if link.state == nil {
 		link.state = make([]*LinkState, 0)
 	}
 	link.state = append([]*LinkState{s}, link.state...)
+	link.recalculateUsable()
+}
+
+func (link *Link) SetDown(down bool) {
+	link.lock.Lock()
+	defer link.lock.Unlock()
+	link.down = down
+	link.recalculateUsable()
+}
+
+func (link *Link) IsDown() bool {
+	link.lock.Lock()
+	defer link.lock.Unlock()
+	return link.down
+}
+
+func (link *Link) recalculateUsable() {
+	if link.down {
+		link.usable.Set(false)
+	} else if len(link.state) < 1 || link.state[0].Mode != Connected {
+		link.usable.Set(false)
+	} else {
+		link.usable.Set(true)
+	}
+}
+
+func (link *Link) IsUsable() bool {
+	return link.usable.Get()
 }
 
 func (link *Link) GetStaticCost() int32 {

controller/network/link_controller.go

@@ -39,8 +39,8 @@ func newLinkController() *linkController {
 
 func (linkController *linkController) add(link *Link) {
 	linkController.linkTable.add(link)
-	link.Src.routerLinks.Add(link)
-	link.Dst.routerLinks.Add(link)
+	link.Src.routerLinks.Add(link, link.Dst)
+	link.Dst.routerLinks.Add(link, link.Src)
 }
 
 func (linkController *linkController) has(link *Link) bool {
@@ -58,8 +58,8 @@ func (linkController *linkController) all() []*Link {
 
 func (linkController *linkController) remove(link *Link) {
 	linkController.linkTable.remove(link)
-	link.Src.routerLinks.Remove(link)
-	link.Dst.routerLinks.Remove(link)
+	link.Src.routerLinks.Remove(link, link.Dst)
+	link.Dst.routerLinks.Remove(link, link.Src)
 }
 
 func (linkController *linkController) connectedNeighborsOfRouter(router *Router) []*Router {
@@ -67,8 +67,7 @@ func (linkController *linkController) connectedNeighborsOfRouter(router *Router)
 
 	links := router.routerLinks.GetLinks()
 	for _, link := range links {
-		currentState := link.CurrentState()
-		if currentState != nil && currentState.Mode == Connected && !link.Down {
+		if link.IsUsable() {
 			if link.Src != router {
 				neighborMap[link.Src.Id] = link.Src
 			}
@@ -89,18 +88,17 @@ func (linkController *linkController) leastExpensiveLink(a, b *Router) (*Link, b
 	var selected *Link
 	var cost int64 = math.MaxInt64
 
-	links := a.routerLinks.GetLinks()
+	linksByRouter := a.routerLinks.GetLinksByRouter()
+	links := linksByRouter[b.Id]
 	for _, link := range links {
-		currentState := link.CurrentState()
-		if currentState != nil && currentState.Mode == Connected && !link.Down {
+		if link.IsUsable() {
 			linkCost := link.GetCost()
-			if link.Src == a && link.Dst == b {
+			if link.Dst == b {
 				if linkCost < cost {
 					selected = link
 					cost = linkCost
 				}
-			}
-			if link.Dst == a && link.Src == b {
+			} else if link.Src == b {
 				if linkCost < cost {
 					selected = link
 					cost = linkCost

controller/network/route_perf_test.go

@@ -26,11 +26,10 @@ import (
 	"testing"
 )
 
-func BenchmarkShortestPathPerf(b *testing.B) {
-	b.StopTimer()
+func TestShortestPathAgainstEstablished(t *testing.T) {
 	pfxlog.GlobalInit(logrus.WarnLevel, pfxlog.DefaultOptions())
 
-	ctx := db.NewTestContext(b)
+	ctx := db.NewTestContext(t)
 	defer ctx.Cleanup()
 
 	closeNotify := make(chan struct{})
@@ -129,7 +128,6 @@ func BenchmarkShortestPathPerf(b *testing.B) {
 		{cost: 141, path: []string{"router-000", "router-050"}},
 	}
 
-	b.StartTimer()
 	srcRouter := routers[0]
 	replaceIdx := len(routers) - 1
 	for i := 1; i < len(routers); i++ {
@@ -153,7 +151,261 @@ func BenchmarkShortestPathPerf(b *testing.B) {
 	}
 }
 
+func BenchmarkShortestPathPerfWithRouterChanges(b *testing.B) {
+	b.StopTimer()
+	pfxlog.GlobalInit(logrus.WarnLevel, pfxlog.DefaultOptions())
+
+	ctx := db.NewTestContext(b)
+	defer ctx.Cleanup()
+
+	closeNotify := make(chan struct{})
+	defer close(closeNotify)
+
+	network, err := NewNetwork("test", nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
+	ctx.NoError(err)
+
+	entityHelper := newTestEntityHelper(ctx, network)
+
+	var routers []*Router
+
+	for i := 0; i < 50; i++ {
+		router := entityHelper.addTestRouter()
+		routers = append(routers, router)
+	}
+
+	linkIdx := 0
+
+	r := rand.New(rand.NewSource(1))
+
+	nextCost := func() int64 {
+		v := r.Uint32()
+		return int64(v % 1000)
+	}
+
+	addLink := func(srcRouter, dstRouter *Router) {
+		if srcRouter != dstRouter {
+			link := newLink(&identity.TokenId{Token: fmt.Sprintf("link-%04d", linkIdx)})
+			link.SetStaticCost(int32(nextCost()))
+			link.SetDstLatency(nextCost() * 100_000)
+			link.SetSrcLatency(nextCost() * 100_000)
+			link.Src = srcRouter
+			link.Dst = dstRouter
+			link.addState(newLinkState(Connected))
+			network.linkController.add(link)
+			linkIdx++
+		}
+	}
+
+	for _, srcRouter := range routers {
+		for _, dstRouter := range routers {
+			addLink(srcRouter, dstRouter)
+		}
+	}
+
+	b.StartTimer()
+	replaceIdx := len(routers) - 1
+	srcIndex := 0
+	dstIndex := 1
+	for i := 0; i < b.N; i++ {
+		srcRouter := routers[srcIndex]
+		dstRouter := routers[dstIndex]
+		_, _, err := network.shortestPath(srcRouter, dstRouter)
+		ctx.NoError(err)
+
+		network.DisconnectRouter(routers[replaceIdx])
+		newRouter := entityHelper.addTestRouter()
+		routers[replaceIdx] = newRouter
+		for _, r := range routers {
+			addLink(newRouter, r)
+			addLink(r, newRouter)
+		}
+		replaceIdx--
+		if replaceIdx < 0 {
+			replaceIdx = len(routers) - 1
+		}
+		entityHelper.discardControllerEvents()
+
+		dstIndex++
+		for dstIndex >= len(routers) {
+			srcIndex++
+			if srcIndex >= len(routers) {
+				srcIndex = 0
+			}
+			dstIndex = 0
+			if dstIndex == srcIndex {
+				dstIndex++
+			}
+		}
+
+	}
+}
+
 type expectedRoute struct {
 	cost int64
 	path []string
 }
+
+func BenchmarkShortestPathPerf(b *testing.B) {
+	b.StopTimer()
+	pfxlog.GlobalInit(logrus.WarnLevel, pfxlog.DefaultOptions())
+
+	ctx := db.NewTestContext(b)
+	defer ctx.Cleanup()
+
+	closeNotify := make(chan struct{})
+	defer close(closeNotify)
+
+	network, err := NewNetwork("test", nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
+	ctx.NoError(err)
+
+	entityHelper := newTestEntityHelper(ctx, network)
+
+	var routers []*Router
+
+	for i := 0; i < 400; i++ {
+		router := entityHelper.addTestRouter()
+		routers = append(routers, router)
+	}
+
+	linkIdx := 0
+
+	r := rand.New(rand.NewSource(1))
+
+	nextCost := func() int64 {
+		v := r.Uint32()
+		return int64(v % 1000)
+	}
+
+	addLink := func(srcRouter, dstRouter *Router) {
+		if srcRouter != dstRouter {
+			link := newLink(&identity.TokenId{Token: fmt.Sprintf("link-%04d", linkIdx)})
+			link.SetStaticCost(int32(nextCost()))
+			link.SetDstLatency(nextCost() * 100_000)
+			link.SetSrcLatency(nextCost() * 100_000)
+			link.Src = srcRouter
+			link.Dst = dstRouter
+			link.addState(newLinkState(Connected))
+			network.linkController.add(link)
+			linkIdx++
+		}
+	}
+
+	for _, srcRouter := range routers {
+		for _, dstRouter := range routers {
+			addLink(srcRouter, dstRouter)
+		}
+	}
+
+	b.StartTimer()
+	srcIndex := 0
+	dstIndex := 1
+	for i := 0; i < b.N; i++ {
+		srcRouter := routers[srcIndex]
+		dstRouter := routers[dstIndex]
+		_, _, err := network.shortestPath(srcRouter, dstRouter)
+		ctx.NoError(err)
+
+		dstIndex++
+		for dstIndex >= len(routers) {
+			srcIndex++
+			if srcIndex >= len(routers) {
+				srcIndex = 0
+			}
+			dstIndex = 0
+			if dstIndex == srcIndex {
+				dstIndex++
+			}
+		}
+	}
+}
+
+func BenchmarkMoreRealisticShortestPathPerf(b *testing.B) {
+	//b.StopTimer()
+	pfxlog.GlobalInit(logrus.WarnLevel, pfxlog.DefaultOptions())
+
+	ctx := db.NewTestContext(b)
+	defer ctx.Cleanup()
+
+	closeNotify := make(chan struct{})
+	defer close(closeNotify)
+
+	network, err := NewNetwork("test", nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
+	ctx.NoError(err)
+
+	entityHelper := newTestEntityHelper(ctx, network)
+
+	var routers []*Router
+
+	for i := 0; i < 200; i++ {
+		router := entityHelper.addTestRouter()
+		routers = append(routers, router)
+	}
+
+	linkIdx := 0
+
+	r := rand.New(rand.NewSource(1))
+
+	nextCost := func() int64 {
+		v := r.Uint32()
+		return int64(v % 1000)
+	}
+
+	addLink := func(srcRouter, dstRouter *Router) {
+		if srcRouter != dstRouter {
+			link := newLink(&identity.TokenId{Token: fmt.Sprintf("link-%04d", linkIdx)})
+			link.SetStaticCost(int32(nextCost()))
+			link.SetDstLatency(nextCost() * 100_000)
+			link.SetSrcLatency(nextCost() * 100_000)
+			link.Src = srcRouter
+			link.Dst = dstRouter
+			link.addState(newLinkState(Connected))
+			network.linkController.add(link)
+			linkIdx++
+		}
+	}
+
+	// make half the routers private routers
+	var privateRouters []*Router
+	var publicRouters []*Router
+	for idx, router := range routers {
+		if idx <= len(routers)/2 {
+			privateRouters = append(privateRouters, router)
+		} else {
+			publicRouters = append(publicRouters, router)
+		}
+	}
+
+	for _, srcRouter := range privateRouters {
+		for _, dstRouter := range publicRouters {
+			addLink(srcRouter, dstRouter)
+		}
+	}
+
+	for _, srcRouter := range publicRouters {
+		for _, dstRouter := range publicRouters {
+			addLink(srcRouter, dstRouter)
+		}
+	}
+
+	b.StartTimer()
+	srcIndex := 0
+	dstIndex := 1
+	for i := 0; i < b.N; i++ {
+		srcRouter := privateRouters[srcIndex]
+		dstRouter := privateRouters[dstIndex]
+		_, _, err := network.shortestPath(srcRouter, dstRouter)
+		ctx.NoError(err)
+
+		dstIndex++
+		for dstIndex >= len(privateRouters) {
+			srcIndex++
+			if srcIndex >= len(privateRouters) {
+				srcIndex = 0
+			}
+			dstIndex = 0
+			if dstIndex == srcIndex {
+				dstIndex++
+			}
+		}
+	}
+}

controller/network/router.go

@@ -71,7 +71,8 @@ func NewRouter(id, name, fingerprint string) *Router {
 		Name:        name,
 		Fingerprint: &fingerprint,
 	}
-	result.routerLinks.Store([]*Link{})
+	result.routerLinks.allLinks.Store([]*Link{})
+	result.routerLinks.linkByRouter.Store(map[string][]*Link{})
 	return result
 }
 
@@ -242,18 +243,27 @@ func (ctrl *RouterController) UpdateCachedFingerprint(id, fingerprint string) {
 
 type RouterLinks struct {
 	sync.Mutex
-	atomic.Value
+	allLinks     atomic.Value
+	linkByRouter atomic.Value
 }
 
 func (self *RouterLinks) GetLinks() []*Link {
-	result := self.Load()
+	result := self.allLinks.Load()
 	if result == nil {
 		return nil
 	}
 	return result.([]*Link)
 }
 
-func (self *RouterLinks) Add(link *Link) {
+func (self *RouterLinks) GetLinksByRouter() map[string][]*Link {
+	result := self.linkByRouter.Load()
+	if result == nil {
+		return nil
+	}
+	return result.(map[string][]*Link)
+}
+
+func (self *RouterLinks) Add(link *Link, other *Router) {
 	self.Lock()
 	defer self.Unlock()
 	links := self.GetLinks()
@@ -262,10 +272,20 @@ func (self *RouterLinks) Add(link *Link) {
 		newLinks = append(newLinks, l)
 	}
 	newLinks = append(newLinks, link)
-	self.Store(newLinks)
+	self.allLinks.Store(newLinks)
+
+	byRouter := self.GetLinksByRouter()
+	newLinksByRouter := map[string][]*Link{}
+	for k, v := range byRouter {
+		newLinksByRouter[k] = v
+	}
+	forRouterList := newLinksByRouter[other.Id]
+	newForRouterList := append([]*Link{link}, forRouterList...)
+	newLinksByRouter[other.Id] = newForRouterList
+	self.linkByRouter.Store(newLinksByRouter)
 }
 
-func (self *RouterLinks) Remove(link *Link) {
+func (self *RouterLinks) Remove(link *Link, other *Router) {
 	self.Lock()
 	defer self.Unlock()
 	links := self.GetLinks()
@@ -275,9 +295,31 @@ func (self *RouterLinks) Remove(link *Link) {
 			newLinks = append(newLinks, l)
 		}
 	}
-	self.Store(newLinks)
+	self.allLinks.Store(newLinks)
+
+	byRouter := self.GetLinksByRouter()
+	newLinksByRouter := map[string][]*Link{}
+	for k, v := range byRouter {
+		newLinksByRouter[k] = v
+	}
+	forRouterList := newLinksByRouter[other.Id]
+	var newForRouterList []*Link
+	for _, l := range forRouterList {
+		if l != link {
+			newForRouterList = append(newForRouterList, l)
+		}
+	}
+	if len(newForRouterList) == 0 {
+		delete(newLinksByRouter, other.Id)
+	} else {
+		newLinksByRouter[other.Id] = newForRouterList
+	}
+
+	self.linkByRouter.Store(newLinksByRouter)
+
 }
 
 func (self *RouterLinks) Clear() {
-	self.Store([]*Link{})
+	self.allLinks.Store([]*Link{})
+	self.linkByRouter.Store(map[string][]*Link{})
 }