Calculate scores in parallel on spreading benchmarks

This is closer to what happens in the core scheduler

Signed-off-by: Aldo Culquicondor <[email protected]>

Author: alculquicondor

pkg/scheduler/framework/plugins/defaultpodtopologyspread/BUILD

@@ -27,6 +27,7 @@ go_test(
     deps = [
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/internal/cache:go_default_library",
+        "//pkg/scheduler/internal/parallelize:go_default_library",
         "//pkg/scheduler/testing:go_default_library",
         "//staging/src/k8s.io/api/apps/v1:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",

pkg/scheduler/framework/plugins/defaultpodtopologyspread/default_pod_topology_spread_perf_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	"k8s.io/kubernetes/pkg/scheduler/internal/cache"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 )
 
@@ -76,15 +77,14 @@ func BenchmarkTestSelectorSpreadPriority(b *testing.B) {
 				if !status.IsSuccess() {
 					b.Fatalf("unexpected error: %v", status)
 				}
-				var gotList framework.NodeScoreList
-				for _, node := range filteredNodes {
-					score, status := plugin.Score(ctx, state, pod, node.Name)
-					if !status.IsSuccess() {
-						b.Errorf("unexpected error: %v", status)
-					}
-					gotList = append(gotList, framework.NodeScore{Name: node.Name, Score: score})
+				gotList := make(framework.NodeScoreList, len(filteredNodes))
+				scoreNode := func(i int) {
+					n := filteredNodes[i]
+					score, _ := plugin.Score(ctx, state, pod, n.Name)
+					gotList[i] = framework.NodeScore{Name: n.Name, Score: score}
 				}
-				status = plugin.NormalizeScore(context.Background(), state, pod, gotList)
+				parallelize.Until(ctx, len(filteredNodes), scoreNode)
+				status = plugin.NormalizeScore(ctx, state, pod, gotList)
 				if !status.IsSuccess() {
 					b.Fatal(status)
 				}

pkg/scheduler/framework/plugins/podtopologyspread/BUILD

@@ -41,6 +41,7 @@ go_test(
     deps = [
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/internal/cache:go_default_library",
+        "//pkg/scheduler/internal/parallelize:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
         "//pkg/scheduler/testing:go_default_library",
         "//staging/src/k8s.io/api/apps/v1:go_default_library",

pkg/scheduler/framework/plugins/podtopologyspread/scoring_test.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	"k8s.io/kubernetes/pkg/scheduler/internal/cache"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 	"k8s.io/utils/pointer"
 )
@@ -746,19 +747,18 @@ func BenchmarkTestDefaultEvenPodsSpreadPriority(b *testing.B) {
 			b.ResetTimer()
 
 			for i := 0; i < b.N; i++ {
-				var gotList framework.NodeScoreList
 				status := p.PreScore(ctx, state, pod, filteredNodes)
 				if !status.IsSuccess() {
 					b.Fatalf("unexpected error: %v", status)
 				}
-				for _, n := range filteredNodes {
-					score, status := p.Score(context.Background(), state, pod, n.Name)
-					if !status.IsSuccess() {
-						b.Fatalf("unexpected error: %v", status)
-					}
-					gotList = append(gotList, framework.NodeScore{Name: n.Name, Score: score})
+				gotList := make(framework.NodeScoreList, len(filteredNodes))
+				scoreNode := func(i int) {
+					n := filteredNodes[i]
+					score, _ := p.Score(ctx, state, pod, n.Name)
+					gotList[i] = framework.NodeScore{Name: n.Name, Score: score}
 				}
-				status = p.NormalizeScore(context.Background(), state, pod, gotList)
+				parallelize.Until(ctx, len(filteredNodes), scoreNode)
+				status = p.NormalizeScore(ctx, state, pod, gotList)
 				if !status.IsSuccess() {
 					b.Fatal(status)
 				}