Support topology aware in the preempt action #4279

bibibox · 2025-05-13T01:19:40Z

What type of PR is this?

/kind feature
improve the preempt action to support topology aware

What this PR does / why we need it:

For more details, please refer to the proposal content.

Monokaix · 2025-05-13T01:57:35Z

Please attach the doc.

JesseStutler · 2025-05-13T06:48:05Z

Please fix the CI

Copilot

Pull Request Overview

This PR enhances the preempt action with topology awareness by adding simulation functions and new cycle state management to various plugins and session functions. Key changes include:

Enhancements in proportion, predicates, and capacity plugins with new simulation and parallel functions.
Updates to session framework for managing the new cycle state functions.
Addition of new topology-aware configuration constants and design documentation.

Reviewed Changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
pkg/scheduler/plugins/proportion/proportion.go	Added capability fields, new state key, simulation, and parallel allocatable functions.
pkg/scheduler/plugins/predicates/predicates.go	Introduced predicate state management and simulation functions.
pkg/scheduler/plugins/capacity/capacity.go	Enhanced simulation functions and parallel allocatable checks with hierarchical support.
pkg/scheduler/framework/session_plugins*.go & session.go	Added new session functions for cycle state initialization and simulation operations.
pkg/scheduler/conf/constants.go	Introduced new topology-aware preemption configuration constants.
pkg/scheduler/api/types.go	Updated Status methods (changed String() receiver to pointer) and added helper functions.
pkg/scheduler/actions/reclaim/reclaim.go	Updated retrieval of candidate nodes using new filtering methods.
docs/design/preempt-action-support-topology.md	Added design documentation for topology-aware preemption.

Comments suppressed due to low confidence (1)

pkg/scheduler/api/types.go:170

Changing the receiver from value to pointer in Status.String() may affect existing usages; ensure compatibility or update the API documentation accordingly.

func (s *Status) String() string {

Copilot · 2025-05-13T07:50:57Z

pkg/scheduler/plugins/proportion/proportion.go

+
+	s, ok := c.(*proportionState)
+	if !ok {
+		return nil, fmt.Errorf("%+v  convert to capacity.state error", c)


Error message in getProportionState incorrectly refers to 'capacity.state' instead of 'proportionState'; update the error text for clarity.

Suggested change

return nil, fmt.Errorf("%+v convert to capacity.state error", c)

return nil, fmt.Errorf("%+v convert to proportionState error", c)

JesseStutler · 2025-05-13T11:57:51Z

pkg/scheduler/actions/preempt/preempt.go

 	return &Action{
-		enablePredicateErrorCache: true,
+		enablePredicateErrorCache:     true,
+		enableTopologyAwarePreemption: true,


As a newly introduced feature, I think this version should be set to false by default, to avoid the impact of user upgrades, and then set it to true after a few iterations of the version is stable

the default value is set to false

JesseStutler · 2025-05-13T12:36:21Z

pkg/scheduler/actions/preempt/preempt.go

+	return true, nil
+}
+
+func (pmpt *Action) taskEligibleToPreemptOthers(preemptor *api.TaskInfo) (bool, string) {


There is already a taskEligibleToPreempt function, which can be unified together

JesseStutler · 2025-05-13T12:45:49Z

pkg/scheduler/actions/preempt/preempt.go

+			return true, ""
+		}
+
+		podPriority := PodPriority(preemptor.Pod)


Better to rename as preemptorPodPriority for better clarification

JesseStutler · 2025-05-13T12:51:40Z

pkg/scheduler/actions/preempt/preempt.go

+	filter func(*api.TaskInfo) bool,
+	predicateHelper util.PredicateHelper,
+) (bool, error) {
+	if err := ssn.PrePredicateFn(preemptor); err != nil {


L397-L410 is the same as in normalPreempt L277-L285(Swapping the order of PrePredicate and taskEligibleToPreempt should not affect？), I think we can extract a public func

JesseStutler · 2025-05-13T13:02:01Z

pkg/scheduler/actions/preempt/preempt.go

+}
+
+func (pmpt *Action) findCandidates(preemptor *api.TaskInfo, filter func(*api.TaskInfo) bool, predicateHelper util.PredicateHelper, stmt *framework.Statement) ([]Candidate, map[string]api.Status, error) {
+	// we should filter out those nodes that are UnschedulableAndUnresolvable status got in allocate action


L486-L499 also is duplicated with L287-L290

JesseStutler · 2025-05-14T02:04:59Z

pkg/scheduler/actions/preempt/preempt.go

+// GetOffsetAndNumCandidates chooses a random offset and calculates the number
+// of candidates that should be shortlisted for dry running preemption.
+func (pmpt *Action) GetOffsetAndNumCandidates(numNodes int) (int, int) {
+	return rand.Intn(numNodes), pmpt.calculateNumCandidates(numNodes)


What is the purpose of offset here?

offset is used to randomly select a starting point in the potentialNodes array. This helps distribute the preemption checks across different nodes and avoid always starting from the beginning of the node list, which could lead to uneven distribution of preemption attempts.

JesseStutler · 2025-05-14T02:07:33Z

pkg/scheduler/framework/session_plugins.go

 	return nil
 }

+// InitCycleStateFn 调用插件的初始化CycleState函数


Please translate all the Chinese comments into English

JesseStutler · 2025-05-14T02:14:40Z

pkg/scheduler/actions/preempt/preempt.go

+		statusesLock.Unlock()
+	}
+	klog.Infof("the worker number is %d, the potentialNodes number is %d", 160, len(potentialNodes))
+	workqueue.ParallelizeUntil(ctx, 160, len(potentialNodes), checkNode)


Default worker size is 160 or 16?

the default value should be 16, and have been updated

JesseStutler · 2025-05-14T02:42:02Z

pkg/scheduler/plugins/capacity/capacity.go

+	return newState
+}
+
+func updateQueueAttrShare(attr *queueAttr) {


Duplicate with cp.updateShare, only different at capacity's updateShared method will do one more step to update metrics: metrics.UpdateQueueShare(attr.name, attr.share), which we can unify the logic together

JesseStutler · 2025-05-14T02:53:19Z

pkg/scheduler/plugins/predicates/predicates.go

+		k8sNodeInfo := k8sframework.NewNodeInfo(nodeInfo.Pods()...)
+		k8sNodeInfo.SetNode(nodeInfo.Node)
+
+		if predicate.podAffinityEnable {


Only contains podAffinity is enough?

other plugin can be added in the future

JesseStutler · 2025-05-14T02:58:22Z

pkg/scheduler/plugins/capacity/capacity.go

 		return !overused
 	})

+	// queueAllocatable := func(queue *api.QueueInfo, candidate *api.TaskInfo) bool {


useless comment

JesseStutler · 2025-05-14T03:01:38Z

pkg/scheduler/plugins/capacity/capacity.go

+	// 	return allocatable
+	// }
+
+	queueConcurrentAllocatable := func(state *capacityState, queue *api.QueueInfo, candidate *api.TaskInfo) bool {


I don't see the difference with capacity's own queueAllocatable, it doesn't reflect concurrent, I think it's not necessary to add a new function to pass one more cycleState,

volcano/pkg/scheduler/plugins/capacity/capacity.go

Lines 724 to 734 in d2c9211

func (cp *capacityPlugin) queueAllocatable(queue *api.QueueInfo, candidate *api.TaskInfo) bool {

attr := cp.queueOpts[queue.UID]

futureUsed := attr.allocated.Clone().Add(candidate.Resreq)

allocatable := futureUsed.LessEqualWithDimension(attr.realCapability, candidate.Resreq)

if !allocatable {

klog.V(3).Infof("Queue <%v>: realCapability <%v>, allocated <%v>; Candidate <%v>: resource request <%v>",

queue.Name, attr.realCapability, attr.allocated, candidate.Name, candidate.Resreq)

}

return allocatable

}

This function is still retained. Can it not be reused? The only difference is the state parameter. In addition, can the name Concurrent be changed to be like Simulate?

JesseStutler · 2025-05-14T03:04:39Z

pkg/scheduler/framework/session_plugins.go

+}
+
+// ConcurrentAllocatableFn 调用插件的并发分配函数
+func (ssn *Session) ConcurrentAllocatableFn(ctx context.Context, state *k8sframework.CycleState, queue *api.QueueInfo, task *api.TaskInfo) bool {


Where does ConcurrentAllocatableFn reflect Concurrent? I didn't see it. And the name will be easy to confuse users, there are both AllocatableFn and ConcurrentAllocatableFn

JesseStutler · 2025-05-14T03:06:03Z

pkg/scheduler/framework/session_plugins.go

+	return true
+}
+
+// ParallelPredicateFn 调用插件的并发断言函数


Same as ParallelPredicateFn, didn't see codes reflect the concept of Parallel

JesseStutler · 2025-05-14T03:13:40Z

pkg/scheduler/framework/session_plugins.go

 	ssn.jobStarvingFns[name] = fn
 }

+func (ssn *Session) AddInitCycleStateFn(name string, fn api.InitCycleStateFn) {


https://github.com/JesseStutler/volcano/blob/6078b0aabd4d8a617efca1a38430e8c2cf757166/pkg/scheduler/framework/session.go#L112-L117

#4152, In this PR, I added a cycleStateMap to the session to store the cycleState of different tasks. Since cycleState may not only be used in predicates plugins. I think there is no need to add a new extension point called InitCycleStateFn? I think it is very likely that this extension point will only be used for simulation in preempt action. Therefore we can do it this way: Before simulation starts, we can clear the cycleStateMap again, and then simulateXXXFn can directly get the cycleState from this cycleStateMap. There is no need to add so many new extension points, which will confuse users. What do you think?

have removed this function and use the cache in the session

JesseStutler · 2025-05-14T03:18:29Z

pkg/scheduler/plugins/capacity/capacity.go

+	queueAttrs map[api.QueueID]*queueAttr
+}
+
+// Clone 深拷贝 queueAttr 对象，包括其所有子节点


Remove useless Chinese comments or translate it into Engligh

JesseStutler · 2025-05-14T03:21:04Z

pkg/scheduler/plugins/proportion/proportion.go

+	return s, nil
+}
+
+func updateQueueAttrShare(attr *queueAttr) {


Also has duplicate codes with proportion's own updateShare method

Where was it changed? Isn't the updateshare function of proportion still kept on line 400?

JesseStutler · 2025-05-14T06:31:36Z

docs/design/preempt-action-support-topology.md

+
+### Key Function Modifications
+
+- `GetBestNodeByPreemptCost`: A function that finds the best node for preemption by calculating and comparing preemption costs. It takes a list of candidate nodes and their corresponding victim pods, iterates through them to compute the cost of preempting victims on each node using the provided cost function, and returns the node with the minimum preemption cost. This helps select the most suitable node that minimizes the impact of preemption.


I didn't see GetBestNodeByPreemptCost and PreemptCostNodeOrderFn in codes?

JesseStutler · 2025-05-14T06:33:47Z

pkg/scheduler/api/types.go

 // NodeOrderReduceFn is the func declaration used to reduce priority score of all nodes for a plugin for a particular task.
 type NodeOrderReduceFn func(*TaskInfo, map[string]k8sframework.NodeScoreList) (map[string]float64, error)

+// PreemptOrderMapFn is the func declaration used to calculate the score of a node for evicting victims


I didn't see PreemptOrderMapFn being used in the code.

JesseStutler · 2025-05-14T06:34:21Z

pkg/scheduler/api/types.go

+// ParallelAllocatableFn is the function declaration used to determine if a task can run on a node in concurrent scenarios.
+type ParallelAllocatableFn func(ctx context.Context, state *k8sframework.CycleState, queue *QueueInfo, task *TaskInfo) bool
+
+type EvictCostFn func(ctx context.Context, state *k8sframework.CycleState, task *TaskInfo, nodeInfo *NodeInfo) (int, error)


Didn't see EvictCostFn being used in the code

JesseStutler · 2025-05-14T06:42:33Z

pkg/scheduler/actions/preempt/preempt.go

+
+// Candidate represents a nominated node on which the preemptor can be scheduled,
+// along with the list of victims that should be evicted for the preemptor to fit the node.
+type Candidate interface {


Is it necessary to define a interface? Seems that it only returns its victims list and name

yes, its for candidateList to manage candidates

JesseStutler · 2025-05-14T06:54:30Z

pkg/scheduler/actions/preempt/preempt.go

+		}
+
+		if !fits {
+			if err := removeTask(pi); err != nil {


Is is necessary to do removeTask again?

if we dont remove again, it will affect next predicate

JesseStutler · 2025-05-14T06:56:47Z

pkg/scheduler/actions/preempt/preempt.go

+
+// pickOneNodeForPreemption chooses one node among the given nodes.
+// It assumes pods in each map entry are ordered by decreasing priority.
+// If the scoreFuns is not empty, It picks a node based on score scoreFuns returns.


typo scoreFuns ----> scoreFuncs

JesseStutler · 2025-05-14T07:00:07Z

pkg/scheduler/actions/preempt/preempt.go

+}
+
+func OrderedScoreFuncs(nodesToVictims map[string][]*api.TaskInfo) []func(node string) int64 {
+	return nil


return nil? Does it need to be modified later?

JesseStutler · 2025-05-14T07:08:46Z

I think this PR should add e2e test cases, and it would be better to make the test scenarios clear in the docs

Signed-off-by: Box Zhang <[email protected]>

bibibox · 2025-06-04T09:28:36Z

I think this PR should add e2e test cases, and it would be better to make the test scenarios clear in the docs

have add the predicate e2e test case

bibibox · 2025-06-05T02:20:43Z

Please attach the doc.
the doc has been updated.

wangyang0616 · 2025-06-05T07:57:30Z

/lgtm

JesseStutler · 2025-06-05T13:15:18Z

pkg/scheduler/api/types.go

+// SimulateAddTaskFn is the func declaration used to simulate the result of adding a task to a node.
+type SimulateAddTaskFn func(ctx context.Context, state *k8sframework.CycleState, taskToSchedule *TaskInfo, taskInfoToAdd *TaskInfo, nodeInfo *NodeInfo) error
+
+// SimulatePredicateFn is the function declaration used to determine if a task can run on a node in parallel scenarios.


The comments here need to be consistent with the documentation: Simulate the predicate check for a task on a node, plugins implement this function to verify if the task can be scheduled to the node while maintaining topology constraints

JesseStutler · 2025-06-05T13:16:06Z

pkg/scheduler/api/types.go

+// It ensures thread safety and state consistency in parallel environments.
+type SimulatePredicateFn func(ctx context.Context, state *k8sframework.CycleState, task *TaskInfo, nodeInfo *NodeInfo) error
+
+// SimulateAllocatableFn is the function declaration used to determine if a task can run on a node in concurrent scenarios.


Same as SimulateAllocatableFn, the comments here is inconsistent with the doc

JesseStutler · 2025-06-05T13:28:01Z

pkg/scheduler/plugins/capacity/capacity.go

+	// 	return allocatable
+	// }
+
+	queueConcurrentAllocatable := func(state *capacityState, queue *api.QueueInfo, candidate *api.TaskInfo) bool {


This function is still retained. Can it not be reused? The only difference is the state parameter. In addition, can the name Concurrent be changed to be like Simulate?

JesseStutler · 2025-06-05T13:32:52Z

pkg/scheduler/plugins/proportion/proportion.go

+	return s, nil
+}
+
+func updateQueueAttrShare(attr *queueAttr) {


Where was it changed? Isn't the updateshare function of proportion still kept on line 400?

JesseStutler · 2025-06-05T13:34:16Z

pkg/scheduler/plugins/capacity/capacity.go

+	}
+
+	attr.share = res
+	metrics.UpdateQueueShare(attr.name, attr.share)


I think we should not update metrics when simulate?

Signed-off-by: Box Zhang <[email protected]>

wangyang0616 · 2025-06-06T08:03:31Z

/approve

volcano-sh-bot · 2025-06-06T08:03:40Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: wangyang0616

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [wangyang0616]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

JesseStutler · 2025-06-06T08:17:58Z

/lgtm

volcano-sh-bot requested review from alcorj-mizar, jiangkaihua and lowang-bh May 13, 2025 01:19

volcano-sh-bot added the size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. label May 13, 2025

bibibox force-pushed the add_preempt_proposal branch from ad7f931 to 90731b3 Compare May 13, 2025 01:27

bibibox force-pushed the add_preempt_proposal branch 2 times, most recently from 7adfca5 to 0816c63 Compare May 13, 2025 07:43

Monokaix requested a review from Copilot May 13, 2025 07:50

Copilot AI reviewed May 13, 2025

View reviewed changes

volcano-sh-bot added the kind/feature Categorizes issue or PR as related to a new feature. label May 13, 2025

bibibox force-pushed the add_preempt_proposal branch 2 times, most recently from 29d6408 to 436e1a3 Compare May 13, 2025 08:49

JesseStutler reviewed May 14, 2025

View reviewed changes

bibibox force-pushed the add_preempt_proposal branch 8 times, most recently from 66872d1 to 38579b2 Compare May 23, 2025 09:42

bibibox force-pushed the add_preempt_proposal branch from 38579b2 to 0f5150c Compare May 26, 2025 03:12

volcano-sh-bot added the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label May 27, 2025

bibibox force-pushed the add_preempt_proposal branch from 0f5150c to 563f3b6 Compare May 27, 2025 08:36

bibibox force-pushed the add_preempt_proposal branch from 563f3b6 to 9aa38be Compare May 27, 2025 09:36

volcano-sh-bot removed the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label May 27, 2025

bibibox force-pushed the add_preempt_proposal branch 3 times, most recently from 7b6a9bb to c909caa Compare June 2, 2025 18:39

Proposal: Preempt action support topology

3506a70

Signed-off-by: Box Zhang <[email protected]>

bibibox force-pushed the add_preempt_proposal branch from c909caa to 9a18058 Compare June 4, 2025 03:33

volcano-sh-bot assigned wangyang0616 Jun 5, 2025

volcano-sh-bot added the lgtm Indicates that a PR is ready to be merged. label Jun 5, 2025

JesseStutler reviewed Jun 5, 2025

View reviewed changes

Preempt action support topology

e9040d3

Signed-off-by: Box Zhang <[email protected]>

bibibox force-pushed the add_preempt_proposal branch from 9a18058 to e9040d3 Compare June 6, 2025 07:20

volcano-sh-bot removed the lgtm Indicates that a PR is ready to be merged. label Jun 6, 2025

volcano-sh-bot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Jun 6, 2025

volcano-sh-bot assigned JesseStutler Jun 6, 2025

volcano-sh-bot added the lgtm Indicates that a PR is ready to be merged. label Jun 6, 2025

volcano-sh-bot merged commit 3aa260c into volcano-sh:master Jun 6, 2025
18 checks passed

	return nil, fmt.Errorf("%+v convert to capacity.state error", c)
	return nil, fmt.Errorf("%+v convert to proportionState error", c)

	func (cp capacityPlugin) queueAllocatable(queue api.QueueInfo, candidate *api.TaskInfo) bool {
	attr := cp.queueOpts[queue.UID]
	futureUsed := attr.allocated.Clone().Add(candidate.Resreq)
	allocatable := futureUsed.LessEqualWithDimension(attr.realCapability, candidate.Resreq)
	if !allocatable {
	klog.V(3).Infof("Queue <%v>: realCapability <%v>, allocated <%v>; Candidate <%v>: resource request <%v>",
	queue.Name, attr.realCapability, attr.allocated, candidate.Name, candidate.Resreq)
	}

	return allocatable
	}


		### Key Function Modifications

		- `GetBestNodeByPreemptCost`: A function that finds the best node for preemption by calculating and comparing preemption costs. It takes a list of candidate nodes and their corresponding victim pods, iterates through them to compute the cost of preempting victims on each node using the provided cost function, and returns the node with the minimum preemption cost. This helps select the most suitable node that minimizes the impact of preemption.

Support topology aware in the preempt action #4279

Support topology aware in the preempt action #4279

Uh oh!

Conversation

bibibox commented May 13, 2025 • edited by Monokaix Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What type of PR is this?

What this PR does / why we need it:

Uh oh!

Monokaix commented May 13, 2025

Uh oh!

JesseStutler commented May 13, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI May 13, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

bibibox commented May 13, 2025 •

edited by Monokaix

Loading