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 3d84b78f [YUNIKORN-3442] Fix preemption victim selection vector
truncation for predicate constraints (#1156)
3d84b78f is described below
commit 3d84b78f0448f3aa6c11077f4ac1b7649e84ecdd
Author: hedger9487 <[email protected]>
AuthorDate: Fri Sep 18 13:03:31 2026 +0530
[YUNIKORN-3442] Fix preemption victim selection vector truncation for
predicate constraints (#1156)
Separates nodeVictims from extraVictims in TryPreemption().
Unconditionally preserves all nodeVictims returned by tryNodes().
Restricts the secondary victim pruning loop strictly to extraVictims
(queue-level victims).
Closes: #1156
Signed-off-by: mani <[email protected]>
---
pkg/scheduler/objects/preemption.go | 54 ++++++++++++++-------------
pkg/scheduler/objects/preemption_test.go | 64 ++++++++++++++++++++++++++++++++
2 files changed, 92 insertions(+), 26 deletions(-)
diff --git a/pkg/scheduler/objects/preemption.go
b/pkg/scheduler/objects/preemption.go
index e44c5b93..f08af1b4 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -613,20 +613,19 @@ func (p *Preemptor) TryPreemption() (*AllocationResult,
bool) {
p.application.executeReservationReleasedCallback(released)
// try to find a node to schedule on and victims to preempt
- nodeID, victims, ok := p.tryNodes()
+ nodeID, nodeVictims, ok := p.tryNodes()
if !ok {
// no preemption possible
return nil, false
}
// look for additional victims in case we have not yet made enough
capacity in the queue
- extraVictims, ok := p.calculateAdditionalVictims(victims)
+ extraVictims, ok := p.calculateAdditionalVictims(nodeVictims)
if !ok {
// not enough resources were preempted
return nil, false
}
- victims = append(victims, extraVictims...)
- if len(victims) == 0 {
+ if len(nodeVictims)+len(extraVictims) == 0 {
return nil, false
}
@@ -638,13 +637,21 @@ func (p *Preemptor) TryPreemption() (*AllocationResult,
bool) {
fitIn := p.nodeAvailableMap[nodeID].FitIn(p.ask.GetAllocatedResource())
- // Since there could be more victims than the actual need, ensure only
required victims are filtered finally
+ // Victims selected for the chosen node by tryNodes() are required for
node capacity
+ // and predicate constraints (e.g. PodAntiAffinity) evaluated by the
ResourceManager plugin.
+ // They must not be truncated based on raw ask resource demand.
+ var finalVictims []*Allocation
+ for _, victim := range nodeVictims {
+ finalVictims = append(finalVictims, victim)
+ victimsTotalResource.AddTo(victim.GetAllocatedResource())
+ }
+
+ // Since there could be more extra victims than the actual need, ensure
only required victims are filtered finally
// to do: There is room for improvements especially when there are more
victims. victims could be chosen based
// on different criteria. for example, victims could be picked up
either from specific node (bin packing) or
// from multiple nodes (fair) given the choices.
- var finalVictims []*Allocation
hasVictimsOnOtherNodes := false
- for _, victim := range victims {
+ for _, victim := range extraVictims {
// Victims from any node is acceptable as long as chosen node
has enough space to accommodate the ask
// Otherwise, preempting victims from 'n' different nodes
doesn't help to achieve the goal.
if victim.GetNodeID() != nodeID {
@@ -664,27 +671,22 @@ func (p *Preemptor) TryPreemption() (*AllocationResult,
bool) {
}
}
- hasShortfall := victimsTotalResource.IsEmpty()
- if !hasShortfall {
- for k, victimVal := range victimsTotalResource.Resources {
- if needVal, ok :=
p.ask.GetAllocatedResource().Resources[k]; ok {
- var avail resources.Quantity
- if !fitIn && !hasVictimsOnOtherNodes {
- avail =
p.nodeAvailableMap[nodeID].Resources[k]
- }
- if avail+victimVal < needVal {
- hasShortfall = true
- break
- }
- }
- }
- }
-
- if hasShortfall {
- // there is shortfall, so preemption doesn't help
+ if victimsTotalResource.IsEmpty() {
p.ask.LogAllocationFailure(common.PreemptionShortfall, true)
return nil, false
}
+ for k, victimVal := range victimsTotalResource.Resources {
+ if needVal, ok := p.ask.GetAllocatedResource().Resources[k]; ok
{
+ var avail resources.Quantity
+ if !fitIn && !hasVictimsOnOtherNodes {
+ avail = p.nodeAvailableMap[nodeID].Resources[k]
+ }
+ if avail+victimVal < needVal {
+
p.ask.LogAllocationFailure(common.PreemptionShortfall, true)
+ return nil, false
+ }
+ }
+ }
// Has any victim released?
// (Placeholder ?) Allocation chosen as victim earlier in the beginning
of preemption cycle but released in the meantime also should be prevented from
@@ -744,7 +746,7 @@ func (p *Preemptor) TryPreemption() (*AllocationResult,
bool) {
log.Log(log.SchedPreemption).Info("Reserving node for ask after
preemption",
zap.String("allocationKey", p.ask.GetAllocationKey()),
zap.String("nodeID", nodeID),
- zap.Int("collected victim count", len(victims)),
+ zap.Int("collected victim count",
len(nodeVictims)+len(extraVictims)),
zap.Int("preempted victim count", len(finalVictims)))
return newReservedAllocationResult(nodeID, p.ask), true
}
diff --git a/pkg/scheduler/objects/preemption_test.go
b/pkg/scheduler/objects/preemption_test.go
index 82d818ca..9734f0aa 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -2510,3 +2510,67 @@ func TestTryPreemption_NodeAvailableDeficit(t
*testing.T) {
assert.Equal(t, nodeID1, result.NodeID)
assert.Check(t, alloc1.IsPreempted(), "alloc1 should be preempted")
}
+
+// TestTryPreemption_PredicateVictimsNotTruncated verifies that when predicate
evaluation (e.g. anti-affinity)
+// requires multiple victims on a node, Core does not drop victims that exceed
the raw resource requirement of the ask.
+func TestTryPreemption_PredicateVictimsNotTruncated(t *testing.T) {
+ appQueueMapping := NewAppQueueMapping()
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 10})
+ iterator := getNodeIteratorFn(node)
+ rootQ, err := createRootQueue(map[string]string{"first": "20"})
+ assert.NilError(t, err)
+ parentQ, err := createManagedQueueGuaranteed(rootQ, "parent", true,
map[string]string{"first": "20"}, map[string]string{"first": "10"},
appQueueMapping)
+ assert.NilError(t, err)
+ childQ1, err := createManagedQueueGuaranteed(parentQ, "child1", false,
nil, nil, appQueueMapping)
+ assert.NilError(t, err)
+ childQ2, err := createManagedQueueGuaranteed(parentQ, "child2", false,
map[string]string{"first": "20"}, map[string]string{"first": "15"},
appQueueMapping)
+ assert.NilError(t, err)
+
+ app1 := newApplication(appID1, "default", "root.parent.child1")
+ app1.SetQueue(childQ1)
+ childQ1.AddApplication(app1)
+ appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
+
+ // alloc1: created earlier, usage first: 5
+ ask1 := newAllocationAsk("alloc1", appID1,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+ ask1.createTime = time.Now().Add(-1 * time.Minute)
+ assert.NilError(t, app1.AddAllocationAsk(ask1))
+ alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+ alloc1.createTime = ask1.createTime
+ app1.AddAllocation(alloc1)
+ assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
+ assert.NilError(t,
childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
+
+ // alloc2: created newer, usage first: 5
+ ask2 := newAllocationAsk("alloc2", appID1,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+ ask2.createTime = time.Now()
+ assert.NilError(t, app1.AddAllocationAsk(ask2))
+ alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1,
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+ alloc2.createTime = ask2.createTime
+ app1.AddAllocation(alloc2)
+ assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
+ assert.NilError(t,
childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
+
+ // Preemptor ask in childQ2 needs first: 5
+ app2, ask3, err := creatApp2(childQ2,
map[string]resources.Quantity{"first": 5}, "alloc3", appQueueMapping)
+ assert.NilError(t, err)
+
+ headRoom :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10})
+ preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask3,
iterator(), false)
+
+ // Shim indicates that BOTH alloc2 (index 0) and alloc1 (index 1) must
be preempted
+ preemptions := []mock.Preemption{
+ mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2",
"alloc1"}, 0, 1),
+ }
+ plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false,
false)
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ result, ok := preemptor.TryPreemption()
+ assert.Assert(t, result != nil, "no result")
+ assert.Assert(t, ok, "no victims found")
+ assert.Equal(t, "alloc3", result.Request.GetAllocationKey(), "wrong
alloc")
+ assert.Equal(t, nodeID1, result.NodeID, "wrong node")
+ assert.Check(t, alloc2.IsPreempted(), "alloc2 not preempted")
+ assert.Check(t, alloc1.IsPreempted(), "alloc1 not preempted")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
