From 1efd4ebb3d0c419865eccb2cdb6909c0d4718015 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 22 Nov 2021 15:01:25 -0500 Subject: [PATCH 1/3] Fix routing benchmark --- controller/network/route_perf_test.go | 261 +++++++++++++++++++++++++- 1 file changed, 257 insertions(+), 4 deletions(-) diff --git a/controller/network/route_perf_test.go b/controller/network/route_perf_test.go index 8ca571fd5..ee3d472d1 100644 --- a/controller/network/route_perf_test.go +++ b/controller/network/route_perf_test.go @@ -23,14 +23,16 @@ import ( "github.com/openziti/foundation/identity/identity" "github.com/sirupsen/logrus" "math/rand" + "os" + "runtime/pprof" "testing" + "time" ) -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 +131,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 +154,259 @@ 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(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 < 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(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(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) + } + } + + f, err := os.Create("/home/plorenz/tmp/routing.pprof") + ctx.NoError(err) + defer func() { _ = f.Close() }() + + if err := pprof.StartCPUProfile(f); err != nil { + ctx.NoError(err) + } + + b.StartTimer() + srcRouter := privateRouters[0] + start := time.Now() + for i := 0; i < 50; i++ { + _, _, err := network.shortestPath(srcRouter, privateRouters[1+i]) + ctx.NoError(err) + } + fmt.Printf("took: %v\n", time.Now().Sub(start)) + + pprof.StopCPUProfile() +} From 3968c268b949b9c0794958c2d35d5a3ae46c2b30 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 22 Nov 2021 16:02:40 -0500 Subject: [PATCH 2/3] Fix benchmark. Add some additional tuning --- controller/handler_mgmt/list_links.go | 2 +- controller/handler_mgmt/set_link_down.go | 2 +- controller/network/link.go | 41 +++++++++++++++++++-- controller/network/link_controller.go | 11 +++--- controller/network/route_perf_test.go | 45 ++++++++++++------------ 5 files changed, 67 insertions(+), 34 deletions(-) diff --git a/controller/handler_mgmt/list_links.go b/controller/handler_mgmt/list_links.go index 686a3ef1e..efc341cd3 100644 --- a/controller/handler_mgmt/list_links.go +++ b/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(), diff --git a/controller/handler_mgmt/set_link_down.go b/controller/handler_mgmt/set_link_down.go index 6dbc8bc2e..1fcd166f3 100644 --- a/controller/handler_mgmt/set_link_down.go +++ b/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, "") diff --git a/controller/network/link.go b/controller/network/link.go index 9f665429a..f7a1e5e67 100644 --- a/controller/network/link.go +++ b/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 { diff --git a/controller/network/link_controller.go b/controller/network/link_controller.go index 945aef73c..e3b3b81ba 100644 --- a/controller/network/link_controller.go +++ b/controller/network/link_controller.go @@ -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 } @@ -91,16 +90,14 @@ func (linkController *linkController) leastExpensiveLink(a, b *Router) (*Link, b links := a.routerLinks.GetLinks() 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 diff --git a/controller/network/route_perf_test.go b/controller/network/route_perf_test.go index ee3d472d1..fe12d2ce2 100644 --- a/controller/network/route_perf_test.go +++ b/controller/network/route_perf_test.go @@ -23,10 +23,7 @@ import ( "github.com/openziti/foundation/identity/identity" "github.com/sirupsen/logrus" "math/rand" - "os" - "runtime/pprof" "testing" - "time" ) func TestShortestPathAgainstEstablished(t *testing.T) { @@ -187,7 +184,7 @@ func BenchmarkShortestPathPerfWithRouterChanges(b *testing.B) { addLink := func(srcRouter, dstRouter *Router) { if srcRouter != dstRouter { - link := newLink(fmt.Sprintf("link-%04d", linkIdx)) + link := newLink(&identity.TokenId{Token: fmt.Sprintf("link-%04d", linkIdx)}) link.SetStaticCost(int32(nextCost())) link.SetDstLatency(nextCost() * 100_000) link.SetSrcLatency(nextCost() * 100_000) @@ -265,7 +262,7 @@ func BenchmarkShortestPathPerf(b *testing.B) { var routers []*Router - for i := 0; i < 50; i++ { + for i := 0; i < 100; i++ { router := entityHelper.addTestRouter() routers = append(routers, router) } @@ -281,7 +278,7 @@ func BenchmarkShortestPathPerf(b *testing.B) { addLink := func(srcRouter, dstRouter *Router) { if srcRouter != dstRouter { - link := newLink(fmt.Sprintf("link-%04d", linkIdx)) + link := newLink(&identity.TokenId{Token: fmt.Sprintf("link-%04d", linkIdx)}) link.SetStaticCost(int32(nextCost())) link.SetDstLatency(nextCost() * 100_000) link.SetSrcLatency(nextCost() * 100_000) @@ -319,7 +316,6 @@ func BenchmarkShortestPathPerf(b *testing.B) { dstIndex++ } } - } } @@ -356,7 +352,7 @@ func BenchmarkMoreRealisticShortestPathPerf(b *testing.B) { addLink := func(srcRouter, dstRouter *Router) { if srcRouter != dstRouter { - link := newLink(fmt.Sprintf("link-%04d", linkIdx)) + link := newLink(&identity.TokenId{Token: fmt.Sprintf("link-%04d", linkIdx)}) link.SetStaticCost(int32(nextCost())) link.SetDstLatency(nextCost() * 100_000) link.SetSrcLatency(nextCost() * 100_000) @@ -391,22 +387,25 @@ func BenchmarkMoreRealisticShortestPathPerf(b *testing.B) { } } - f, err := os.Create("/home/plorenz/tmp/routing.pprof") - ctx.NoError(err) - defer func() { _ = f.Close() }() - - if err := pprof.StartCPUProfile(f); err != nil { - ctx.NoError(err) - } - b.StartTimer() - srcRouter := privateRouters[0] - start := time.Now() - for i := 0; i < 50; i++ { - _, _, err := network.shortestPath(srcRouter, privateRouters[1+i]) + 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) - } - fmt.Printf("took: %v\n", time.Now().Sub(start)) - pprof.StopCPUProfile() + dstIndex++ + for dstIndex >= len(privateRouters) { + srcIndex++ + if srcIndex >= len(privateRouters) { + srcIndex = 0 + } + dstIndex = 0 + if dstIndex == srcIndex { + dstIndex++ + } + } + } } From 3a5099936590933f0be250a1179ce050b40c16c8 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 23 Nov 2021 08:24:24 -0500 Subject: [PATCH 3/3] Add cache of links by router --- controller/network/link_controller.go | 11 ++--- controller/network/route_perf_test.go | 2 +- controller/network/router.go | 58 +++++++++++++++++++++++---- 3 files changed, 57 insertions(+), 14 deletions(-) diff --git a/controller/network/link_controller.go b/controller/network/link_controller.go index e3b3b81ba..dc879e5c8 100644 --- a/controller/network/link_controller.go +++ b/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 { @@ -88,7 +88,8 @@ 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 { if link.IsUsable() { linkCost := link.GetCost() diff --git a/controller/network/route_perf_test.go b/controller/network/route_perf_test.go index fe12d2ce2..d80b298a5 100644 --- a/controller/network/route_perf_test.go +++ b/controller/network/route_perf_test.go @@ -262,7 +262,7 @@ func BenchmarkShortestPathPerf(b *testing.B) { var routers []*Router - for i := 0; i < 100; i++ { + for i := 0; i < 400; i++ { router := entityHelper.addTestRouter() routers = append(routers, router) } diff --git a/controller/network/router.go b/controller/network/router.go index 309485723..190aa00e5 100644 --- a/controller/network/router.go +++ b/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{}) }