Merge pull request kubernetes#89070 from alculquicondor/static-partitioning

k8s-ci-robot · web-flow · commit f898f45b2081 · 2020-03-24T15:27:02.000-07:00
Add chunk size option to ParallelizeUntil
diff --git a/pkg/scheduler/BUILD b/pkg/scheduler/BUILD
@@ -123,6 +123,7 @@ filegroup(
         "//pkg/scheduler/framework:all-srcs",
         "//pkg/scheduler/internal/cache:all-srcs",
         "//pkg/scheduler/internal/heap:all-srcs",
+        "//pkg/scheduler/internal/parallelize:all-srcs",
         "//pkg/scheduler/internal/queue:all-srcs",
         "//pkg/scheduler/listers:all-srcs",
         "//pkg/scheduler/metrics:all-srcs",
diff --git a/pkg/scheduler/core/BUILD b/pkg/scheduler/core/BUILD
@@ -13,6 +13,7 @@ go_library(
         "//pkg/scheduler/apis/config:go_default_library",
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//pkg/scheduler/internal/cache:go_default_library",
+        "//pkg/scheduler/internal/parallelize:go_default_library",
         "//pkg/scheduler/internal/queue:go_default_library",
         "//pkg/scheduler/listers:go_default_library",
         "//pkg/scheduler/metrics:go_default_library",
@@ -28,7 +29,6 @@ go_library(
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/listers/policy/v1beta1:go_default_library",
         "//staging/src/k8s.io/client-go/rest:go_default_library",
-        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
         "//staging/src/k8s.io/kube-scheduler/extender/v1:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
         "//vendor/k8s.io/utils/trace:go_default_library",
diff --git a/pkg/scheduler/core/generic_scheduler.go b/pkg/scheduler/core/generic_scheduler.go
@@ -28,14 +28,14 @@ import (
 	"time"
 
 	"k8s.io/klog"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 
 	v1 "k8s.io/api/core/v1"
 	policy "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	policylisters "k8s.io/client-go/listers/policy/v1beta1"
-	"k8s.io/client-go/util/workqueue"
 	extenderv1 "k8s.io/kube-scheduler/extender/v1"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
@@ -479,7 +479,7 @@ func (g *genericScheduler) findNodesThatPassFilters(ctx context.Context, prof *p
 
 	// Stops searching for more nodes once the configured number of feasible nodes
 	// are found.
-	workqueue.ParallelizeUntil(ctx, 16, len(allNodes), checkNode)
+	parallelize.Until(ctx, len(allNodes), checkNode)
 	processedNodes := int(filteredLen) + len(statuses)
 	g.nextStartNodeIndex = (g.nextStartNodeIndex + processedNodes) % len(allNodes)
 
@@ -870,7 +870,7 @@ func (g *genericScheduler) selectNodesForPreemption(
 			resultLock.Unlock()
 		}
 	}
-	workqueue.ParallelizeUntil(context.TODO(), 16, len(potentialNodes), checkNode)
+	parallelize.Until(ctx, len(potentialNodes), checkNode)
 	return nodeToVictims, nil
 }
 
diff --git a/pkg/scheduler/framework/plugins/interpodaffinity/BUILD b/pkg/scheduler/framework/plugins/interpodaffinity/BUILD
@@ -11,6 +11,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
+        "//pkg/scheduler/internal/parallelize:go_default_library",
         "//pkg/scheduler/listers:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
         "//pkg/scheduler/util:go_default_library",
@@ -20,7 +21,6 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/validation/field:go_default_library",
-        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
         "//vendor/k8s.io/utils/pointer:go_default_library",
     ],
diff --git a/pkg/scheduler/framework/plugins/interpodaffinity/filtering.go b/pkg/scheduler/framework/plugins/interpodaffinity/filtering.go
@@ -25,9 +25,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 	"k8s.io/kubernetes/pkg/scheduler/nodeinfo"
 	schedutil "k8s.io/kubernetes/pkg/scheduler/util"
 )
@@ -240,7 +240,7 @@ func getTPMapMatchingExistingAntiAffinity(pod *v1.Pod, allNodes []*nodeinfo.Node
 			}
 		}
 	}
-	workqueue.ParallelizeUntil(ctx, 16, len(allNodes), processNode)
+	parallelize.Until(ctx, len(allNodes), processNode)
 
 	if err := errCh.ReceiveError(); err != nil {
 		return nil, err
@@ -304,7 +304,7 @@ func getTPMapMatchingIncomingAffinityAntiAffinity(pod *v1.Pod, allNodes []*nodei
 			appendResult(node.Name, nodeTopologyPairsAffinityPodsMap, nodeTopologyPairsAntiAffinityPodsMap)
 		}
 	}
-	workqueue.ParallelizeUntil(context.Background(), 16, len(allNodes), processNode)
+	parallelize.Until(context.Background(), len(allNodes), processNode)
 
 	return topologyPairsAffinityPodsMap, topologyToMatchedExistingAntiAffinityTerms, nil
 }
diff --git a/pkg/scheduler/framework/plugins/interpodaffinity/scoring.go b/pkg/scheduler/framework/plugins/interpodaffinity/scoring.go
@@ -22,9 +22,9 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 	"k8s.io/kubernetes/pkg/scheduler/nodeinfo"
 	schedutil "k8s.io/kubernetes/pkg/scheduler/util"
 )
@@ -259,7 +259,7 @@ func (pl *InterPodAffinity) PreScore(
 			pl.Unlock()
 		}
 	}
-	workqueue.ParallelizeUntil(ctx, 16, len(allNodes), processNode)
+	parallelize.Until(ctx, len(allNodes), processNode)
 	if err := errCh.ReceiveError(); err != nil {
 		return framework.NewStatus(framework.Error, err.Error())
 	}
diff --git a/pkg/scheduler/framework/plugins/podtopologyspread/BUILD b/pkg/scheduler/framework/plugins/podtopologyspread/BUILD
@@ -13,6 +13,7 @@ go_library(
     deps = [
         "//pkg/scheduler/framework/plugins/helper:go_default_library",
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
+        "//pkg/scheduler/internal/parallelize:go_default_library",
         "//pkg/scheduler/listers:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
@@ -25,7 +26,6 @@ go_library(
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/listers/apps/v1:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
-        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )
diff --git a/pkg/scheduler/framework/plugins/podtopologyspread/filtering.go b/pkg/scheduler/framework/plugins/podtopologyspread/filtering.go
@@ -24,10 +24,10 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 	pluginhelper "k8s.io/kubernetes/pkg/scheduler/framework/plugins/helper"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 	"k8s.io/kubernetes/pkg/scheduler/nodeinfo"
 )
 
@@ -267,7 +267,7 @@ func (pl *PodTopologySpread) calPreFilterState(pod *v1.Pod) (*preFilterState, er
 			addTopologyPairMatchNum(pair, matchTotal)
 		}
 	}
-	workqueue.ParallelizeUntil(context.Background(), 16, len(allNodes), processNode)
+	parallelize.Until(context.Background(), len(allNodes), processNode)
 
 	// calculate min match for each topology pair
 	for i := 0; i < len(constraints); i++ {
diff --git a/pkg/scheduler/framework/plugins/podtopologyspread/scoring.go b/pkg/scheduler/framework/plugins/podtopologyspread/scoring.go
@@ -25,10 +25,10 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 	pluginhelper "k8s.io/kubernetes/pkg/scheduler/framework/plugins/helper"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 )
 
 const preScoreStateKey = "PreScore" + Name
@@ -153,7 +153,7 @@ func (pl *PodTopologySpread) PreScore(
 			atomic.AddInt64(state.TopologyPairToPodCounts[pair], matchSum)
 		}
 	}
-	workqueue.ParallelizeUntil(ctx, 16, len(allNodes), processAllNode)
+	parallelize.Until(ctx, len(allNodes), processAllNode)
 
 	cycleState.Write(preScoreStateKey, state)
 	return nil
diff --git a/pkg/scheduler/framework/v1alpha1/BUILD b/pkg/scheduler/framework/v1alpha1/BUILD
@@ -15,6 +15,7 @@ go_library(
     deps = [
         "//pkg/controller/volume/scheduling:go_default_library",
         "//pkg/scheduler/apis/config:go_default_library",
+        "//pkg/scheduler/internal/parallelize:go_default_library",
         "//pkg/scheduler/listers:go_default_library",
         "//pkg/scheduler/metrics:go_default_library",
         "//pkg/scheduler/nodeinfo:go_default_library",
@@ -26,7 +27,6 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
-        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
         "//staging/src/k8s.io/component-base/metrics:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
         "//vendor/sigs.k8s.io/yaml:go_default_library",
diff --git a/pkg/scheduler/framework/v1alpha1/framework.go b/pkg/scheduler/framework/v1alpha1/framework.go
@@ -28,10 +28,10 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/controller/volume/scheduling"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config"
+	"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
 	schedulerlisters "k8s.io/kubernetes/pkg/scheduler/listers"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
 	schedulernodeinfo "k8s.io/kubernetes/pkg/scheduler/nodeinfo"
@@ -513,7 +513,7 @@ func (f *framework) RunScorePlugins(ctx context.Context, state *CycleState, pod
 	errCh := schedutil.NewErrorChannel()
 
 	// Run Score method for each node in parallel.
-	workqueue.ParallelizeUntil(ctx, 16, len(nodes), func(index int) {
+	parallelize.Until(ctx, len(nodes), func(index int) {
 		for _, pl := range f.scorePlugins {
 			nodeName := nodes[index].Name
 			s, status := f.runScorePlugin(ctx, pl, state, pod, nodeName)
@@ -534,7 +534,7 @@ func (f *framework) RunScorePlugins(ctx context.Context, state *CycleState, pod
 	}
 
 	// Run NormalizeScore method for each ScorePlugin in parallel.
-	workqueue.ParallelizeUntil(ctx, 16, len(f.scorePlugins), func(index int) {
+	parallelize.Until(ctx, len(f.scorePlugins), func(index int) {
 		pl := f.scorePlugins[index]
 		nodeScoreList := pluginToNodeScores[pl.Name()]
 		if pl.ScoreExtensions() == nil {
@@ -554,7 +554,7 @@ func (f *framework) RunScorePlugins(ctx context.Context, state *CycleState, pod
 	}
 
 	// Apply score defaultWeights for each ScorePlugin in parallel.
-	workqueue.ParallelizeUntil(ctx, 16, len(f.scorePlugins), func(index int) {
+	parallelize.Until(ctx, len(f.scorePlugins), func(index int) {
 		pl := f.scorePlugins[index]
 		// Score plugins' weight has been checked when they are initialized.
 		weight := f.pluginNameToWeightMap[pl.Name()]
diff --git a/pkg/scheduler/internal/parallelize/BUILD b/pkg/scheduler/internal/parallelize/BUILD
@@ -0,0 +1,23 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["parallelism.go"],
+    importpath = "k8s.io/kubernetes/pkg/scheduler/internal/parallelize",
+    visibility = ["//pkg/scheduler:__subpackages__"],
+    deps = ["//staging/src/k8s.io/client-go/util/workqueue:go_default_library"],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)
diff --git a/pkg/scheduler/internal/parallelize/parallelism.go b/pkg/scheduler/internal/parallelize/parallelism.go
@@ -0,0 +1,43 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package parallelize
+
+import (
+	"context"
+	"math"
+
+	"k8s.io/client-go/util/workqueue"
+)
+
+const parallelism = 16
+
+// chunkSizeFor returns a chunk size for the given number of items to use for
+// parallel work. The size aims to produce good CPU utilization.
+func chunkSizeFor(n int) workqueue.Options {
+	s := int(math.Sqrt(float64(n)))
+	if r := n/parallelism + 1; s > r {
+		s = r
+	} else if s < 1 {
+		s = 1
+	}
+	return workqueue.WithChunkSize(s)
+}
+
+// Until is a wrapper around workqueue.ParallelizeUntil to use in scheduling algorithms.
+func Until(ctx context.Context, pieces int, doWorkPiece workqueue.DoWorkPieceFunc) {
+	workqueue.ParallelizeUntil(ctx, parallelism, pieces, doWorkPiece, chunkSizeFor(pieces))
+}
diff --git a/staging/src/k8s.io/client-go/go.mod b/staging/src/k8s.io/client-go/go.mod
@@ -14,6 +14,7 @@ require (
 	github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903
 	github.com/golang/protobuf v1.3.2
 	github.com/google/btree v1.0.0 // indirect
+	github.com/google/go-cmp v0.3.1
 	github.com/google/gofuzz v1.1.0
 	github.com/google/uuid v1.1.1
 	github.com/googleapis/gnostic v0.1.0
diff --git a/staging/src/k8s.io/client-go/util/workqueue/BUILD b/staging/src/k8s.io/client-go/util/workqueue/BUILD
@@ -13,13 +13,15 @@ go_test(
         "delaying_queue_test.go",
         "main_test.go",
         "metrics_test.go",
+        "parallelizer_test.go",
         "queue_test.go",
         "rate_limiting_queue_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp:go_default_library",
     ],
 )
 
diff --git a/staging/src/k8s.io/client-go/util/workqueue/parallelizer.go b/staging/src/k8s.io/client-go/util/workqueue/parallelizer.go
diff --git a/staging/src/k8s.io/client-go/util/workqueue/parallelizer_test.go b/staging/src/k8s.io/client-go/util/workqueue/parallelizer_test.go
diff --git a/test/e2e/framework/.import-restrictions b/test/e2e/framework/.import-restrictions
diff --git a/test/integration/scheduler_perf/scheduler_bench_test.go b/test/integration/scheduler_perf/scheduler_bench_test.go

Original file line number	Diff line number	Diff line change
`@@ -25,9 +25,9 @@ import (`
`25`	`25`	`metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"`
`26`	`26`	`"k8s.io/apimachinery/pkg/labels"`
`27`	`27`	`"k8s.io/apimachinery/pkg/util/sets"`
`28`		`- "k8s.io/client-go/util/workqueue"`
`29`	`28`	`"k8s.io/klog"`
`30`	`29`	`framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"`
	`30`	`+ "k8s.io/kubernetes/pkg/scheduler/internal/parallelize"`
`31`	`31`	`"k8s.io/kubernetes/pkg/scheduler/nodeinfo"`
`32`	`32`	`schedutil "k8s.io/kubernetes/pkg/scheduler/util"`
`33`	`33`	`)`
`@@ -240,7 +240,7 @@ func getTPMapMatchingExistingAntiAffinity(pod v1.Pod, allNodes []nodeinfo.Node`
`240`	`240`	`}`
`241`	`241`	`}`
`242`	`242`	`}`
`243`		`- workqueue.ParallelizeUntil(ctx, 16, len(allNodes), processNode)`
	`243`	`+ parallelize.Until(ctx, len(allNodes), processNode)`
`244`	`244`
`245`	`245`	`if err := errCh.ReceiveError(); err != nil {`
`246`	`246`	`return nil, err`
`@@ -304,7 +304,7 @@ func getTPMapMatchingIncomingAffinityAntiAffinity(pod v1.Pod, allNodes []nodei`
`304`	`304`	`appendResult(node.Name, nodeTopologyPairsAffinityPodsMap, nodeTopologyPairsAntiAffinityPodsMap)`
`305`	`305`	`}`
`306`	`306`	`}`
`307`		`- workqueue.ParallelizeUntil(context.Background(), 16, len(allNodes), processNode)`
	`307`	`+ parallelize.Until(context.Background(), len(allNodes), processNode)`
`308`	`308`
`309`	`309`	`return topologyPairsAffinityPodsMap, topologyToMatchedExistingAntiAffinityTerms, nil`
`310`	`310`	`}`
Original file line number	Diff line number	Diff line change
`@@ -22,9 +22,9 @@ import (`
`22`	`22`
`23`	`23`	`v1 "k8s.io/api/core/v1"`
`24`	`24`	`metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"`
`25`		`- "k8s.io/client-go/util/workqueue"`
`26`	`25`	`"k8s.io/klog"`
`27`	`26`	`framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"`
	`27`	`+ "k8s.io/kubernetes/pkg/scheduler/internal/parallelize"`
`28`	`28`	`"k8s.io/kubernetes/pkg/scheduler/nodeinfo"`
`29`	`29`	`schedutil "k8s.io/kubernetes/pkg/scheduler/util"`
`30`	`30`	`)`
`@@ -259,7 +259,7 @@ func (pl *InterPodAffinity) PreScore(`
`259`	`259`	`pl.Unlock()`
`260`	`260`	`}`
`261`	`261`	`}`
`262`		`- workqueue.ParallelizeUntil(ctx, 16, len(allNodes), processNode)`
	`262`	`+ parallelize.Until(ctx, len(allNodes), processNode)`
`263`	`263`	`if err := errCh.ReceiveError(); err != nil {`
`264`	`264`	`return framework.NewStatus(framework.Error, err.Error())`
`265`	`265`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,10 +24,10 @@ import (`
`24`	`24`
`25`	`25`	`v1 "k8s.io/api/core/v1"`
`26`	`26`	`"k8s.io/apimachinery/pkg/labels"`
`27`		`- "k8s.io/client-go/util/workqueue"`
`28`	`27`	`"k8s.io/klog"`
`29`	`28`	`pluginhelper "k8s.io/kubernetes/pkg/scheduler/framework/plugins/helper"`
`30`	`29`	`framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"`
	`30`	`+ "k8s.io/kubernetes/pkg/scheduler/internal/parallelize"`
`31`	`31`	`"k8s.io/kubernetes/pkg/scheduler/nodeinfo"`
`32`	`32`	`)`
`33`	`33`
`@@ -267,7 +267,7 @@ func (pl PodTopologySpread) calPreFilterState(pod v1.Pod) (*preFilterState, er`
`267`	`267`	`addTopologyPairMatchNum(pair, matchTotal)`
`268`	`268`	`}`
`269`	`269`	`}`
`270`		`- workqueue.ParallelizeUntil(context.Background(), 16, len(allNodes), processNode)`
	`270`	`+ parallelize.Until(context.Background(), len(allNodes), processNode)`
`271`	`271`
`272`	`272`	`// calculate min match for each topology pair`
`273`	`273`	`for i := 0; i < len(constraints); i++ {`
Original file line number	Diff line number	Diff line change
`@@ -25,10 +25,10 @@ import (`
`25`	`25`	`v1 "k8s.io/api/core/v1"`
`26`	`26`	`"k8s.io/apimachinery/pkg/labels"`
`27`	`27`	`"k8s.io/apimachinery/pkg/util/sets"`
`28`		`- "k8s.io/client-go/util/workqueue"`
`29`	`28`	`"k8s.io/klog"`
`30`	`29`	`pluginhelper "k8s.io/kubernetes/pkg/scheduler/framework/plugins/helper"`
`31`	`30`	`framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"`
	`31`	`+ "k8s.io/kubernetes/pkg/scheduler/internal/parallelize"`
`32`	`32`	`)`
`33`	`33`
`34`	`34`	`const preScoreStateKey = "PreScore" + Name`
`@@ -153,7 +153,7 @@ func (pl *PodTopologySpread) PreScore(`
`153`	`153`	`atomic.AddInt64(state.TopologyPairToPodCounts[pair], matchSum)`
`154`	`154`	`}`
`155`	`155`	`}`
`156`		`- workqueue.ParallelizeUntil(ctx, 16, len(allNodes), processAllNode)`
	`156`	`+ parallelize.Until(ctx, len(allNodes), processAllNode)`
`157`	`157`
`158`	`158`	`cycleState.Write(preScoreStateKey, state)`
`159`	`159`	`return nil`