You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by ma...@apache.org on 2026/10/28 12:14:23 UTC
(yunikorn-core) branch master updated: [YUNIKORN-3514] Score the victim list sent to the shim in preemption node selection (#1195)
This is an automated email from the ASF dual-hosted git repository.
manirajv06 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new f4c3a8a3 [YUNIKORN-3514] Score the victim list sent to the shim in preemption node selection (#1195)
f4c3a8a3 is described below
commit f4c3a8a35e251f1f35522b99734e5303a084801a
Author: rjgoyln <pt...@gmail.com>
AuthorDate: Wed Oct 28 17:42:01 2026 +0530
[YUNIKORN-3514] Score the victim list sent to the shim in preemption node selection (#1195)
The index the shim returns from PreemptionPredicates points into the
victim list it was sent, which calculateVictimsByNode has already
filtered and reordered. checkPreemptionPredicates compared nodes with
betterThan and isSatisfactory against the sorted allocations of the node
instead, so whenever the two lists diverged the originator and opted-out
penalties were read off allocations that were not going to be preempted.
A node whose real victims include an originator could then win against a
node that only costs an opted-out pod, and the early exit between
predicate batches could fire on a score that did not describe the chosen
victims. populateVictims already used the victim list, so only the node
choice was wrong.
Closes: #1195
Signed-off-by: mani <ma...@gmail.com>
---
pkg/scheduler/objects/predicates.go | 22 +--
pkg/scheduler/objects/preemption.go | 22 ++-
pkg/scheduler/objects/preemption_test.go | 226 +++++++++++++++++++++++++++++++
3 files changed, 247 insertions(+), 23 deletions(-)
diff --git a/pkg/scheduler/objects/predicates.go b/pkg/scheduler/objects/predicates.go
index 1d485e4a..b7d96cc0 100644
--- a/pkg/scheduler/objects/predicates.go
+++ b/pkg/scheduler/objects/predicates.go
@@ -36,15 +36,17 @@ type predicateCheckResult struct {
predicateErrors map[string]int32
}
-func (pcr *predicateCheckResult) betterThan(other *predicateCheckResult, allocationsByNode map[string][]*Allocation) bool {
- return pcr.getSolutionScore(allocationsByNode) < other.getSolutionScore(allocationsByNode)
+func (pcr *predicateCheckResult) betterThan(other *predicateCheckResult, victimsByNode map[string][]*Allocation) bool {
+ return pcr.getSolutionScore(victimsByNode) < other.getSolutionScore(victimsByNode)
}
-func (pcr *predicateCheckResult) getSolutionScore(allocationsByNode map[string][]*Allocation) uint64 {
+// getSolutionScore scores the victims up to and including pcr.index. The index returned by the shim refers to the
+// victim list it was sent, so victimsByNode must hold that list and not the sorted allocations of the node.
+func (pcr *predicateCheckResult) getSolutionScore(victimsByNode map[string][]*Allocation) uint64 {
if pcr == nil || !pcr.success {
return scoreUnfit
}
- allocations, ok := allocationsByNode[pcr.nodeID]
+ victims, ok := victimsByNode[pcr.nodeID]
if !ok {
return scoreUnfit
}
@@ -53,16 +55,16 @@ func (pcr *predicateCheckResult) getSolutionScore(allocationsByNode map[string][
if pcr.index < 0 {
return score
}
- if pcr.index >= len(allocations) {
+ if pcr.index >= len(victims) {
// shouldn't happen
return scoreUnfit
}
for i := 0; i <= pcr.index; i++ {
- allocation := allocations[i]
- if allocation.IsOriginator() {
+ victim := victims[i]
+ if victim.IsOriginator() {
score |= scoreOriginator
}
- if !allocation.IsAllowPreemptSelf() {
+ if !victim.IsAllowPreemptSelf() {
score |= scoreNoPreempt
}
}
@@ -71,8 +73,8 @@ func (pcr *predicateCheckResult) getSolutionScore(allocationsByNode map[string][
return score
}
-func (pcr *predicateCheckResult) isSatisfactory(allocationsByNode map[string][]*Allocation) bool {
- return pcr.getSolutionScore(allocationsByNode) < scoreFitMax
+func (pcr *predicateCheckResult) isSatisfactory(victimsByNode map[string][]*Allocation) bool {
+ return pcr.getSolutionScore(victimsByNode) < scoreFitMax
}
func (pcr *predicateCheckResult) populateVictims(victimsByNode map[string][]*Allocation) {
diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go
index f863e470..405740c7 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -401,28 +401,24 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr
predicateErrors := make(map[string]int)
for _, batch := range batches {
var wg sync.WaitGroup
- ch := make(chan *predicateCheckResult, len(batch))
- expected := 0
- for _, args := range batch {
+ // keep the results in the sorted order of the checks, so that nodes with an equal score are resolved by that
+ // order and not by the order in which the shim answers
+ results := make([]*predicateCheckResult, len(batch))
+ for i, args := range batch {
// add goroutine for checking preemption
wg.Add(1)
- expected++
go func() {
defer wg.Done()
- ch <- runPredicateChecks(plugin, args)
+ results[i] = runPredicateChecks(plugin, args)
}()
}
- // wait for completion and close channel
- go func() {
- wg.Wait()
- close(ch)
- }()
- for result := range ch {
+ wg.Wait()
+ for _, result := range results {
// if resultType is successful, keep track of it
if result.success {
if bestResult == nil {
bestResult = result
- } else if result.betterThan(bestResult, p.allocationsByNode) {
+ } else if result.betterThan(bestResult, victimsByNode) {
bestResult = result
}
} else {
@@ -432,7 +428,7 @@ func (p *Preemptor) checkPreemptionPredicates(predicateChecks []*si.PreemptionPr
}
}
// if the best resultType we have from this batch meets all our criteria, don't run another batch
- if bestResult.isSatisfactory(p.allocationsByNode) {
+ if bestResult.isSatisfactory(victimsByNode) {
break
}
}
diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go
index cdc975fc..d2cc44bf 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -23,6 +23,7 @@ import (
"fmt"
"strconv"
"strings"
+ "sync/atomic"
"testing"
"time"
@@ -613,6 +614,230 @@ func TestTryPreemptionOnNode(t *testing.T) {
assert.Equal(t, len(ask3.GetAllocationLog()), 0)
}
+// TestTryPreemption_OnNode_ScoresReorderedVictims Node selection must score the victim list sent to the shim. On node1
+// regular-2 no longer reduces the shortfall once regular-1 is freed, so the shim receives [regular-1, originator,
+// regular-2] and answers index 1: node1 costs an originator, node2 only an opted-out pod, and node2 must win. Scored on
+// the sorted allocations, node1 would be charged for two regular pods and the originator would be preempted.
+func TestTryPreemption_OnNode_ScoresReorderedVictims(t *testing.T) {
+ appQueueMapping := NewAppQueueMapping()
+ node1 := newNode(nodeID1, map[string]resources.Quantity{"first": 2, "second": 1})
+ node2 := newNode(nodeID2, map[string]resources.Quantity{"first": 1, "second": 1})
+ iterator := getNodeIteratorFn(node1, node2)
+ rootQ, err := createRootQueue(map[string]string{"first": "10", "second": "10"})
+ assert.NilError(t, err)
+ parentQ, err := createManagedQueueGuaranteed(rootQ, "parent", true, nil, nil, appQueueMapping)
+ assert.NilError(t, err)
+ childQ1, err := createManagedQueueGuaranteed(parentQ, "child1", false, nil, nil, appQueueMapping)
+ assert.NilError(t, err)
+ childQ2, err := createManagedQueueGuaranteed(parentQ, "child2", false, nil, map[string]string{"first": "5", "second": "5"}, appQueueMapping)
+ assert.NilError(t, err)
+
+ app1 := newApplication(appID1, "default", "root.parent.child1")
+ app1.SetQueue(childQ1)
+ childQ1.AddApplication(app1)
+ appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
+
+ first := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1})
+ second := resources.NewResourceFromMap(map[string]resources.Quantity{"second": 1})
+ both := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1, "second": 1})
+ regular1 := createAllocation("regular-1", appID1, nodeID1, true, false, 0, false, first)
+ regular1.createTime = testBaseTime.Add(time.Minute)
+ regular2 := createAllocation("regular-2", appID1, nodeID1, true, false, 0, false, first)
+ regular2.createTime = testBaseTime
+ originator := createAllocation("originator", appID1, nodeID1, true, true, 0, false, second)
+ optedOut := createAllocation("opted-out", appID1, nodeID2, false, false, 0, false, both)
+ nodes := map[string]*Node{nodeID1: node1, nodeID2: node2}
+ for _, victim := range []*Allocation{regular1, regular2, originator, optedOut} {
+ assert.NilError(t, app1.AddAllocation(victim))
+ assert.Check(t, nodes[victim.GetNodeID()].TryAddAllocation(victim), "node %s failed", victim.GetAllocationKey())
+ assert.NilError(t, childQ1.TryIncAllocatedResource(victim.GetAllocatedResource()))
+ }
+
+ app2, ask, err := creatApp2(childQ2, map[string]resources.Quantity{"first": 1, "second": 1}, "ask", appQueueMapping)
+ assert.NilError(t, err)
+
+ headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5, "second": 5})
+ preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask, iterator(), false)
+
+ // register predicate handler
+ preemptions := []mock.Preemption{
+ mock.NewPreemption(true, "ask", nodeID1, []string{"regular-1", "originator", "regular-2"}, 1, 1),
+ mock.NewPreemption(true, "ask", nodeID2, []string{"opted-out"}, 0, 0),
+ }
+ plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ result, ok := preemptor.TryPreemption()
+ assert.NilError(t, plugin.GetPredicateError())
+ assert.Assert(t, result != nil, "no result")
+ assert.Assert(t, ok, "no victims found")
+ assert.Equal(t, nodeID2, result.NodeID, "wrong node")
+ assert.Check(t, optedOut.IsPreempted(), "opted-out not preempted")
+ assert.Check(t, !regular1.IsPreempted(), "regular-1 preempted")
+ assert.Check(t, !regular2.IsPreempted(), "regular-2 preempted")
+ assert.Check(t, !originator.IsPreempted(), "originator preempted")
+}
+
+// TestTryPreemption_OnQueue_EmptyNodeBeatsOriginatorNode verifies that an empty node that fits the ask beats a node
+// whose only eligible victim is the originator. The regular allocation on the unschedulable node3 can be selected as a
+// queue victim instead, without preempting the originator.
+func TestTryPreemption_OnQueue_EmptyNodeBeatsOriginatorNode(t *testing.T) {
+ appQueueMapping := NewAppQueueMapping()
+ node1 := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ node2 := newNode(nodeID2, map[string]resources.Quantity{"first": 5})
+ node3 := newNode(nodeID3, map[string]resources.Quantity{"first": 5})
+ iterator := getNodeIteratorFn(node1, node2, node3)
+ rootQ, err := createRootQueue(map[string]string{"first": "100"})
+ assert.NilError(t, err)
+ parentQ, err := createManagedQueueGuaranteed(rootQ, "parent", true, map[string]string{"first": "10"}, nil, appQueueMapping)
+ assert.NilError(t, err)
+ childQ1, err := createManagedQueueGuaranteed(parentQ, "child1", false, nil, nil, appQueueMapping)
+ assert.NilError(t, err)
+ childQ2, err := createManagedQueueGuaranteed(parentQ, "child2", false, nil, map[string]string{"first": "5"}, appQueueMapping)
+ assert.NilError(t, err)
+
+ app1 := newApplication(appID1, "default", "root.parent.child1")
+ app1.SetQueue(childQ1)
+ childQ1.AddApplication(app1)
+ appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
+
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ originator := createAllocation("originator", appID1, nodeID1, true, true, 0, false, res)
+ regular := createAllocation("regular", appID1, nodeID3, true, false, 0, false, res)
+ nodes := map[string]*Node{nodeID1: node1, nodeID3: node3}
+ for _, victim := range []*Allocation{originator, regular} {
+ assert.NilError(t, app1.AddAllocation(victim))
+ assert.Check(t, nodes[victim.GetNodeID()].TryAddAllocation(victim), "node %s failed", victim.GetAllocationKey())
+ assert.NilError(t, childQ1.TryIncAllocatedResource(victim.GetAllocatedResource()))
+ }
+ node3.SetSchedulable(false)
+
+ app2, ask, err := creatApp2(childQ2, map[string]resources.Quantity{"first": 5}, "ask", appQueueMapping)
+ assert.NilError(t, err)
+
+ headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 0})
+ preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask, iterator(), false)
+
+ // register predicate handler
+ preemptions := []mock.Preemption{
+ mock.NewPreemption(true, "ask", nodeID1, []string{"originator"}, 0, 0),
+ }
+ plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ result, ok := preemptor.TryPreemption()
+ assert.NilError(t, plugin.GetPredicateError())
+ assert.Assert(t, result != nil, "no result")
+ assert.Assert(t, ok, "no victims found")
+ assert.Equal(t, nodeID2, result.NodeID, "wrong node")
+ assert.Check(t, regular.IsPreempted(), "regular not preempted")
+ assert.Check(t, !originator.IsPreempted(), "originator preempted")
+}
+
+// TestCheckPreemptionPredicates_SatisfactoryCheckUsesVictimList The early exit between batches must score the victim
+// list sent to the shim. Every node sorts a regular allocation first, but the shim is sent a single originator on all
+// nodes except the last, which falls into the second batch, so no node of the first batch may be satisfactory.
+func TestCheckPreemptionPredicates_SatisfactoryCheckUsesVictimList(t *testing.T) {
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1})
+ allocationsByNode := make(map[string][]*Allocation)
+ victimsByNode := make(map[string][]*Allocation)
+ checks := make([]*si.PreemptionPredicatesArgs, 0)
+ preemptions := make([]mock.Preemption, 0)
+ for i := 0; i <= preemptCheckConcurrency; i++ {
+ nodeID := fmt.Sprintf("node-%02d", i)
+ regular := createAllocation(fmt.Sprintf("regular-%02d", i), appID1, nodeID, true, false, 0, false, res)
+ originator := createAllocation(fmt.Sprintf("originator-%02d", i), appID1, nodeID, true, true, 0, false, res)
+ allocationsByNode[nodeID] = []*Allocation{regular, originator}
+ victim := originator
+ if i == preemptCheckConcurrency {
+ victim = regular
+ }
+ victimsByNode[nodeID] = []*Allocation{victim}
+ checks = append(checks, &si.PreemptionPredicatesArgs{
+ AllocationKey: "ask",
+ NodeID: nodeID,
+ PreemptAllocationKeys: []string{victim.GetAllocationKey()},
+ StartIndex: 0,
+ })
+ preemptions = append(preemptions, mock.NewPreemption(true, "ask", nodeID, []string{victim.GetAllocationKey()}, 0, 0))
+ }
+ plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ preemptor := &Preemptor{allocationsByNode: allocationsByNode}
+ result, predicateErrors := preemptor.checkPreemptionPredicates(checks, victimsByNode)
+ assert.NilError(t, plugin.GetPredicateError())
+ assert.Equal(t, 0, len(predicateErrors), "unexpected predicate errors")
+ assert.Assert(t, result != nil, "no result")
+ assert.Assert(t, result.success, "check failed")
+ assert.Equal(t, "node-10", result.nodeID, "wrong node")
+ assert.Equal(t, 1, len(result.victims), "wrong victim count")
+ assert.Equal(t, "regular-10", result.victims[0].GetAllocationKey(), "wrong victim")
+}
+
+// countingPreemptionPlugin accepts every check at its start index and counts the checks; node-00 answers last so the
+// answer order differs from the check order.
+type countingPreemptionPlugin struct {
+ mock.ResourceManagerCallback
+ checks atomic.Int32
+}
+
+func (p *countingPreemptionPlugin) PreemptionPredicates(args *si.PreemptionPredicatesArgs) *si.PreemptionPredicatesResponse {
+ p.checks.Add(1)
+ if args.NodeID == "node-00" {
+ time.Sleep(time.Millisecond)
+ }
+ return &si.PreemptionPredicatesResponse{Success: true, Index: args.StartIndex}
+}
+
+// TestCheckPreemptionPredicates_ResultOrder A better score wins wherever the node sorts, equal scores resolve in check
+// order, a satisfactory batch ends the checks and an unsatisfactory one runs the next. Each node is sent one victim at
+// start index 0, R regular or O originator, and the node IDs sort in the listed order.
+func TestCheckPreemptionPredicates_ResultOrder(t *testing.T) {
+ res := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1})
+ tests := []struct {
+ name string
+ victims string
+ wantNode string
+ wantChecks int
+ }{
+ {"better score beats check order", "OOR", "node-02", 3},
+ {"equal scores resolve in check order", strings.Repeat("R", preemptCheckConcurrency), "node-00", preemptCheckConcurrency},
+ {"satisfactory batch ends the checks", strings.Repeat("R", preemptCheckConcurrency+1), "node-00", preemptCheckConcurrency},
+ {"unsatisfactory batch runs the next batch", strings.Repeat("O", preemptCheckConcurrency) + "R", "node-10", preemptCheckConcurrency + 1},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ victimsByNode := make(map[string][]*Allocation)
+ checks := make([]*si.PreemptionPredicatesArgs, 0, len(tt.victims))
+ for i, kind := range tt.victims {
+ nodeID := fmt.Sprintf("node-%02d", i)
+ victim := createAllocation(fmt.Sprintf("victim-%02d", i), appID1, nodeID, true, kind == 'O', 0, false, res)
+ victimsByNode[nodeID] = []*Allocation{victim}
+ checks = append(checks, &si.PreemptionPredicatesArgs{
+ AllocationKey: "ask",
+ NodeID: nodeID,
+ PreemptAllocationKeys: []string{victim.GetAllocationKey()},
+ StartIndex: 0,
+ })
+ }
+ plugin := &countingPreemptionPlugin{}
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ result, predicateErrors := (&Preemptor{}).checkPreemptionPredicates(checks, victimsByNode)
+ assert.Equal(t, 0, len(predicateErrors), "unexpected predicate errors")
+ assert.Assert(t, result != nil, "no result")
+ assert.Assert(t, result.success, "check failed")
+ assert.Equal(t, tt.wantNode, result.nodeID, "wrong node")
+ assert.Equal(t, tt.wantChecks, int(plugin.checks.Load()), "wrong number of checks")
+ })
+ }
+}
+
// TestTryPreemption_NodeWithCapacityLesserThanAsk Test try preemption on node whose capacity is lesser than ask resource requirements with simple queue hierarchy. Since Node won't accommodate the ask even after preempting all allocations, there is no use in considering the node.
// Guaranteed and Max resource set on both victim queue path and preemptor queue path in 2 levels. victim and preemptor queue are siblings.
// Request (Preemptor) resource type matches with all resource types of the victim. But Guaranteed set only on specific resource type. 2 Victims are available, but 1 should be preempted because further preemption would make usage go below the guaranteed quota
@@ -1123,6 +1348,7 @@ func TestTryPreemption_OnQueue_VictimsOnDifferentNodes(t *testing.T) {
// root.parent.child2. Guaranteed set 5. Request of first:5 is waiting for resources.
// root.parent.child3. Guaranteed not set. 1 Allocation is running on node2. Total usage is first:5.
// High priority ask should not be touched and remaining 2 allocs should be preempted to free up resources
+// Both nodes fit the ask without node preemption and score the same, so node1 is chosen by the check order (lower node ID).
func TestTryPreemption_OnQueue_VictimsAvailable_LowerPriority(t *testing.T) {
appQueueMapping := NewAppQueueMapping()
node1 := newNode(nodeID1, map[string]resources.Quantity{"first": 30})
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org