Skip to content

Commit 28882c5

Browse files
authored
feat(scheduler): add gpujoborder plugin for configurable JobOrderFn tiebreak (#1995)
Signed-off-by: CoolingCube <CoolingCubeInfo@proton.me>
1 parent 2914d32 commit 28882c5

9 files changed

Lines changed: 348 additions & 14 deletions

File tree

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
kind: Added
2+
body: |-
3+
Add gpujoborder plugin for configurable GPU-based JobOrderFn tiebreak on priority ties

pkg/scheduler/actions/utils/job_order_by_queue.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -255,7 +255,7 @@ func (jo *JobsOrderByQueues) createLeafNode(queue *queue_info.QueueInfo) *queueN
255255
queue: queue,
256256
children: scheduler_util.NewPriorityQueue(func(l, r interface{}) bool {
257257
if jo.options.VictimQueue {
258-
return !jo.ssn.JobOrderFn(l, r)
258+
return jo.ssn.VictimOrderFn(l, r)
259259
}
260260
return jo.ssn.JobOrderFn(l, r)
261261
}, jo.options.MaxJobsQueueDepth),

pkg/scheduler/api/podgroup_info/job_info.go

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ type PodGroupInfo struct {
9393
tasksToAllocateInitResourceVector resource_info.ResourceVector
9494
PodStatusIndex map[pod_status.PodStatus]pod_info.PodsMap
9595
activeAllocatedCount *int
96+
aliveTasksRequestedGPUs *float64
9697
}
9798

9899
func NewPodGroupInfo(uid common_info.PodGroupID, tasks ...*pod_info.PodInfo) *PodGroupInfo {
@@ -365,6 +366,7 @@ func (pgi *PodGroupInfo) invalidateTasksCache() {
365366
pgi.allPodsMap = nil
366367
pgi.tasksToAllocate = nil
367368
pgi.tasksToAllocateInitResourceVector = nil
369+
pgi.aliveTasksRequestedGPUs = nil
368370
}
369371

370372
func (pgi *PodGroupInfo) GetActiveAllocatedTasksCount() int {
@@ -456,14 +458,16 @@ func (pgi *PodGroupInfo) GetNumGatedTasks() int {
456458
}
457459

458460
func (pgi *PodGroupInfo) GetAliveTasksRequestedGPUs() float64 {
459-
tasksTotalRequestedGPUs := float64(0)
460-
for _, task := range pgi.GetAllPodsMap() {
461-
if pod_status.IsAliveStatus(task.Status) {
462-
tasksTotalRequestedGPUs += task.ResReqVector.Get(resource_info.GPUIndex)
461+
if pgi.aliveTasksRequestedGPUs == nil {
462+
tasksTotalRequestedGPUs := float64(0)
463+
for _, task := range pgi.GetAllPodsMap() {
464+
if pod_status.IsAliveStatus(task.Status) {
465+
tasksTotalRequestedGPUs += task.ResReqVector.Get(resource_info.GPUIndex)
466+
}
463467
}
468+
pgi.aliveTasksRequestedGPUs = ptr.To(tasksTotalRequestedGPUs)
464469
}
465-
466-
return tasksTotalRequestedGPUs
470+
return *pgi.aliveTasksRequestedGPUs
467471
}
468472

469473
func (pgi *PodGroupInfo) GetTasksActiveAllocatedReqResourceVector() resource_info.ResourceVector {

pkg/scheduler/framework/session.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ type Session struct {
7676
NodePreOrderFns []api.NodePreOrderFn
7777
NodeOrderFns []api.NodeOrderFn
7878
JobOrderFns []common_info.CompareFn
79+
VictimOrderFns []common_info.CompareFn
7980
SubGroupOrderFns []common_info.CompareFn
8081
TaskOrderFns []common_info.CompareFn
8182
QueueOrderFns []api.CompareQueueFn
@@ -434,6 +435,7 @@ func (ssn *Session) clear() {
434435
ssn.NodePreOrderFns = nil
435436
ssn.NodeOrderFns = nil
436437
ssn.JobOrderFns = nil
438+
ssn.VictimOrderFns = nil
437439
ssn.SubGroupOrderFns = nil
438440
ssn.TaskOrderFns = nil
439441
ssn.QueueOrderFns = nil

pkg/scheduler/framework/session_plugins.go

Lines changed: 44 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,22 @@ func (ssn *Session) AddJobOrderFn(jof common_info.CompareFn) {
6868
ssn.JobOrderFns = append(ssn.JobOrderFns, jof)
6969
}
7070

71+
// AddVictimOrderFn registers a comparator used only to rank candidate
72+
// eviction victims, without affecting pending-job allocation order.
73+
//
74+
// The registered JobOrderFns chain is always checked first (inverted,
75+
// matching existing allocation-order semantics); comparators registered
76+
// here apply only as a tiebreak, when every JobOrderFn treats l and r as
77+
// equal. This ensures job-level protections (e.g. elastic at-min/above-min
78+
// status) take precedence over any victim-specific comparator.
79+
//
80+
// Sign convention: vof(l, r) < 0 means l should be evicted before r. This
81+
// is the direct, non-inverted sense -- unlike the JobOrderFn chain, which
82+
// VictimOrderFn inverts internally when using it for the eviction path.
83+
func (ssn *Session) AddVictimOrderFn(vof common_info.CompareFn) {
84+
ssn.VictimOrderFns = append(ssn.VictimOrderFns, vof)
85+
}
86+
7187
func (ssn *Session) AddTaskOrderFn(tof common_info.CompareFn) {
7288
ssn.TaskOrderFns = append(ssn.TaskOrderFns, tof)
7389
}
@@ -268,21 +284,42 @@ func (ssn *Session) QueueAllocatedResources(queue *queue_info.QueueInfo) *resour
268284
return nil
269285
}
270286

287+
func jobOrderCreationFallback(l, r interface{}) bool {
288+
lv := l.(*podgroup_info.PodGroupInfo)
289+
rv := r.(*podgroup_info.PodGroupInfo)
290+
if lv.CreationTimestamp.Equal(&rv.CreationTimestamp) {
291+
return lv.UID < rv.UID
292+
}
293+
return lv.CreationTimestamp.Before(&rv.CreationTimestamp)
294+
}
295+
271296
func (ssn *Session) JobOrderFn(l, r interface{}) bool {
272297
for _, jof := range ssn.JobOrderFns {
273298
if j := jof(l, r); j != 0 {
274299
return j < 0
275300
}
276301
}
302+
return jobOrderCreationFallback(l, r)
303+
}
277304

278-
// If no job order funcs, order job by CreationTimestamp first, then by UID.
279-
lv := l.(*podgroup_info.PodGroupInfo)
280-
rv := r.(*podgroup_info.PodGroupInfo)
281-
if lv.CreationTimestamp.Equal(&rv.CreationTimestamp) {
282-
return lv.UID < rv.UID
283-
} else {
284-
return lv.CreationTimestamp.Before(&rv.CreationTimestamp)
305+
// VictimOrderFn reports whether l should be evicted before r.
306+
//
307+
// The JobOrderFns chain is consulted first, inverted; VictimOrderFns
308+
// registered via AddVictimOrderFn apply only when every JobOrderFn treats
309+
// l and r as equal. With no VictimOrderFns registered, this reduces to
310+
// the original !JobOrderFn(l, r) behavior.
311+
func (ssn *Session) VictimOrderFn(l, r interface{}) bool {
312+
for _, jof := range ssn.JobOrderFns {
313+
if j := jof(l, r); j != 0 {
314+
return j > 0
315+
}
316+
}
317+
for _, vof := range ssn.VictimOrderFns {
318+
if v := vof(l, r); v != 0 {
319+
return v < 0
320+
}
285321
}
322+
return !jobOrderCreationFallback(l, r)
286323
}
287324

288325
func (ssn *Session) TaskOrderFn(l, r interface{}) bool {

pkg/scheduler/framework/session_plugins_test.go

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,12 @@ import (
1818
kaiv1 "github.com/kai-scheduler/KAI-scheduler/pkg/apis/kai/v1"
1919
"github.com/kai-scheduler/KAI-scheduler/pkg/common/constants"
2020
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api"
21+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/common_info"
2122
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/node_info"
2223
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/pod_info"
2324
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/podgroup_info"
2425
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/podgroup_info/subgroup_info"
26+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/resource_info"
2527
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/conf"
2628
)
2729

@@ -305,3 +307,43 @@ func scenarioSearchDurationForTest(value string) metav1.Duration {
305307
}
306308
return metav1.Duration{Duration: duration}
307309
}
310+
311+
func TestVictimOrderFn_PreservesExistingJobOrderFnBehavior(t *testing.T) {
312+
priorityLikeJobOrderFn := func(l, r interface{}) int {
313+
lv := l.(*podgroup_info.PodGroupInfo)
314+
rv := r.(*podgroup_info.PodGroupInfo)
315+
if lv.Priority > rv.Priority {
316+
return -1
317+
}
318+
if lv.Priority < rv.Priority {
319+
return 1
320+
}
321+
return 0
322+
}
323+
324+
vm := resource_info.NewResourceVectorMap()
325+
priorities := []int32{1, 1, 5, 5, 10, 20, 20, 100}
326+
327+
for i, lp := range priorities {
328+
for j, rp := range priorities {
329+
if i == j {
330+
continue
331+
}
332+
ssn := &Session{JobOrderFns: []common_info.CompareFn{priorityLikeJobOrderFn}}
333+
lJob := podgroup_info.NewPodGroupInfoWithVectorMap(
334+
common_info.PodGroupID("l"), vm)
335+
lJob.Priority = lp
336+
rJob := podgroup_info.NewPodGroupInfoWithVectorMap(
337+
common_info.PodGroupID("r"), vm)
338+
rJob.Priority = rp
339+
340+
oldBehavior := !ssn.JobOrderFn(lJob, rJob)
341+
newBehavior := ssn.VictimOrderFn(lJob, rJob)
342+
343+
if oldBehavior != newBehavior {
344+
t.Errorf("mismatch for priorities l=%d r=%d: old !JobOrderFn=%v, new VictimOrderFn=%v",
345+
lp, rp, oldBehavior, newBehavior)
346+
}
347+
}
348+
}
349+
}

pkg/scheduler/plugins/factory.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/framework"
2424
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/dynamicresources"
2525
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/elastic"
26+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/gpujoborder"
2627
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/gpupack"
2728
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/gpusharingorder"
2829
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/plugins/gpuspread"
@@ -51,6 +52,7 @@ func InitDefaultPlugins() {
5152
// Plugins for PodGroupInfos
5253
framework.RegisterPluginBuilder("predicates", predicates.New)
5354
framework.RegisterPluginBuilder("priority", priority.New)
55+
framework.RegisterPluginBuilder("gpujoborder", gpujoborder.New)
5456
framework.RegisterPluginBuilder("nodeplacement", nodeplacement.New)
5557
framework.RegisterPluginBuilder("nominatednode", nominatednode.New)
5658
framework.RegisterPluginBuilder("numa", numa.New)
Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
// Copyright 2026 NVIDIA CORPORATION
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package gpujoborder
5+
6+
import (
7+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/api/podgroup_info"
8+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/framework"
9+
"github.com/kai-scheduler/KAI-scheduler/pkg/scheduler/log"
10+
)
11+
12+
const (
13+
ModeEvictLargerFirst = "evict-larger-first"
14+
ModeEvictSmallerFirst = "evict-smaller-first"
15+
)
16+
17+
type gpuJobOrderPlugin struct {
18+
mode string
19+
}
20+
21+
func New(arguments framework.PluginArguments) framework.Plugin {
22+
mode := arguments.GetString("mode", ModeEvictLargerFirst)
23+
if mode != ModeEvictLargerFirst && mode != ModeEvictSmallerFirst {
24+
log.InfraLogger.Warningf("gpujoborder: unrecognized mode %q, defaulting to %s", mode, ModeEvictLargerFirst)
25+
mode = ModeEvictLargerFirst
26+
}
27+
return &gpuJobOrderPlugin{mode: mode}
28+
}
29+
30+
func (rp *gpuJobOrderPlugin) Name() string {
31+
return "gpujoborder"
32+
}
33+
34+
func (rp *gpuJobOrderPlugin) OnSessionOpen(ssn *framework.Session) {
35+
ssn.AddVictimOrderFn(rp.VictimOrderFn)
36+
}
37+
38+
func (rp *gpuJobOrderPlugin) VictimOrderFn(l, r interface{}) int {
39+
lv := l.(*podgroup_info.PodGroupInfo)
40+
rv := r.(*podgroup_info.PodGroupInfo)
41+
42+
if lv.Priority != rv.Priority {
43+
return 0
44+
}
45+
46+
lGPU := lv.GetAliveTasksRequestedGPUs()
47+
rGPU := rv.GetAliveTasksRequestedGPUs()
48+
49+
switch rp.mode {
50+
case ModeEvictSmallerFirst:
51+
if lGPU < rGPU {
52+
return -1
53+
}
54+
if lGPU > rGPU {
55+
return 1
56+
}
57+
default:
58+
if lGPU > rGPU {
59+
return -1
60+
}
61+
if lGPU < rGPU {
62+
return 1
63+
}
64+
}
65+
return 0
66+
}
67+
68+
func (rp *gpuJobOrderPlugin) OnSessionClose(_ *framework.Session) {}

0 commit comments

Comments
 (0)