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/05 10:31:49 UTC
(yunikorn-core) branch master updated: [YUNIKORN-3459] Decouple ask queue quota accounting from victim sizes in calculateVictimsByNode (#1170)
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 7af297fc [YUNIKORN-3459] Decouple ask queue quota accounting from victim sizes in calculateVictimsByNode (#1170)
7af297fc is described below
commit 7af297fc012e4b3530dfa29159ada733361a737c
Author: hedger9487 <he...@gmail.com>
AuthorDate: Mon Oct 5 16:01:11 2026 +0530
[YUNIKORN-3459] Decouple ask queue quota accounting from victim sizes in calculateVictimsByNode (#1170)
Oversized Victim Rejection: FitInActual(victim.GetAllocatedResource()) checks whether the victim's size fits into the ask queue's remaining guaranteed headroom. If an ask needs 2 cores and the ask queue has 2 cores headroom, preempting an 8-core victim frees 8 cores on the node and fully satisfies the ask while only consuming 2 cores of queue quota. However, because 8 > 2, FitInActual(8 cores) returns false, incorrectly rejecting the 8-core victim and starving the ask.
Predicate Tail Victim Truncation: When an earlier victim is evaluated, calling askQueue.AddAllocation(victim) consumes the ask queue's remaining headroom in the snapshot. Subsequent allocations on the same node (e.g. pods causing PodAntiAffinity or required by the ResourceManager predicate plugin) see an exhausted askQueueRemaining and are discarded before reaching tail, leading to predicate failures.
Closes: #1170
Signed-off-by: mani <ma...@gmail.com>
---
pkg/scheduler/objects/preemption.go | 21 +---
pkg/scheduler/objects/preemption_test.go | 172 +++++++++++++++++++++++++++++--
2 files changed, 167 insertions(+), 26 deletions(-)
diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go
index 56a53775..b4bc41f0 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -248,18 +248,12 @@ func (p *Preemptor) calculateVictimsByNode(nodeAvailable *resources.Resource, po
}
allocationsByQueueSnap := p.duplicateQueueSnapshots()
- // get the current queue snapshot
- askQueue, ok := allocationsByQueueSnap[p.queuePath]
- if !ok {
- log.Log(log.SchedPreemption).Warn("BUG: Queue not found by name", zap.String("queuePath", p.queuePath))
- return -1, nil
- }
// First pass: Check each task to see whether we are able to reduce our shortfall by preempting each
// task in turn, and filter out tasks which will cause their queue to drop below guaranteed capacity.
// If a task could be preempted without violating queue constraints, add it to either the 'head' list or the
// 'tail' list depending on whether the shortfall is reduced. If added to the 'head' list, adjust the node available
- // capacity and the queue guaranteed headroom.
+ // capacity.
head := make([]*Allocation, 0)
tail := make([]*Allocation, 0)
for _, victim := range potentialVictims {
@@ -274,27 +268,14 @@ func (p *Preemptor) calculateVictimsByNode(nodeAvailable *resources.Resource, po
// At times, over-allocation happens because of resource types in usage but not defined as guaranteed.
// So, as an additional check, res types used by ask should be either -ve or zero in victim queue remaining guaranteed resource to confirm
// some relevant useful victim is there.
- // In case of victims densely populated on any specific node, checking/honouring the guaranteed quota on ask or preemptor queue
- // acts as early filtering layer to carry forward only the required victims.
- // For other cases like victims spread over multiple nodes, this doesn't add great value.
if resources.StrictlyGreaterThanOrEquals(preemptableResource, resources.Zero) &&
(remaining == nil || isVictimQueueOverGuaranteed(p.ask.GetAllocatedResource(), remaining)) {
- // Does victimQueue have space equivalent to the resource used by the victim?
- askQueueRemaining := askQueue.GetRemainingGuaranteedResource()
- if askQueueRemaining != nil && askQueueRemaining.FitInActual(victim.GetAllocatedResource()) {
- askQueue.AddAllocation(victim.GetAllocatedResource())
- } else {
- queueSnapshot.AddAllocation(victim.GetAllocatedResource())
- continue
- }
-
// check to see if the shortfall on the node has changed
shortfall := resources.SubEliminateNegative(p.ask.GetAllocatedResource(), nodeCurrentAvailable)
newAvailable := resources.Add(nodeCurrentAvailable, victim.GetAllocatedResource())
newShortfall := resources.SubEliminateNegative(p.ask.GetAllocatedResource(), newAvailable)
if resources.EqualsOrEmpty(shortfall, newShortfall) {
// shortfall did not change, so task should only be considered as a last resort
- askQueue.RemoveAllocation(victim.GetAllocatedResource())
queueSnapshot.AddAllocation(victim.GetAllocatedResource())
tail = append(tail, victim)
} else {
diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go
index 885fed19..70e6e02c 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -670,7 +670,7 @@ func TestTryPreemptionOnNodeWithOGParentAndUGPreemptor(t *testing.T) {
// register predicate handler
preemptions := []mock.Preemption{
- mock.NewPreemption(true, "alloc7", nodeID2, []string{"alloc1"}, 0, 0),
+ mock.NewPreemption(true, "alloc7", nodeID2, []string{"alloc1", "alloc3", "alloc5"}, 0, 0),
}
plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
plugins.RegisterSchedulerPlugin(plugin)
@@ -1289,7 +1289,7 @@ func TestTryPreemption_OnNode_AskResTypesDifferent_GuaranteedSetOnPreemptorSide(
preemptor := NewPreemptor(app2, headRoom, 30*time.Second, ask3, iterator(), false)
// register predicate handler
- preemptions := []mock.Preemption{mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2"}, 0, 0)}
+ preemptions := []mock.Preemption{mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2", "alloc1"}, 0, 0)}
plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
plugins.RegisterSchedulerPlugin(plugin)
defer plugins.UnregisterSchedulerPlugins()
@@ -1712,7 +1712,7 @@ func TestTryPreemption_OnNode_AskResTypesSame_GuaranteedSetOnPreemptorSide(t *te
preemptor := NewPreemptor(app4, headRoom, 30*time.Second, ask4, iterator(), false)
// register predicate handler
- preemptions := []mock.Preemption{mock.NewPreemption(true, "alloc4", nodeID1, []string{"alloc3", "alloc2"}, 1, 1)}
+ preemptions := []mock.Preemption{mock.NewPreemption(true, "alloc4", nodeID1, []string{"alloc3", "alloc2", "alloc1"}, 1, 1)}
plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
plugins.RegisterSchedulerPlugin(plugin)
defer plugins.UnregisterSchedulerPlugins()
@@ -1994,7 +1994,7 @@ func TestTryPreemption_OnNode_UGParent_With_UGPreemptorChild_GNotSetOnVictimChil
preemptor := NewPreemptor(app3, headRoom, 30*time.Second, ask3, iterator(), false)
// register predicate handler
- preemptions := []mock.Preemption{mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2"}, 0, 0)}
+ preemptions := []mock.Preemption{mock.NewPreemption(true, "alloc3", nodeID1, []string{"alloc2", "alloc1"}, 0, 0)}
plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
plugins.RegisterSchedulerPlugin(plugin)
defer plugins.UnregisterSchedulerPlugins()
@@ -2877,8 +2877,168 @@ func TestTryPreemption_PrematureVictimLoopTermination(t *testing.T) {
assert.Assert(t, result != nil)
assert.Equal(t, "preemptor", result.Request.GetAllocationKey())
assert.Equal(t, nodeID1, result.NodeID)
- assert.Check(t, alloc2.IsPreempted())
- assert.Check(t, !alloc1.IsPreempted())
+ assert.Check(t, alloc1.IsPreempted())
+ assert.Check(t, !alloc2.IsPreempted())
+}
+
+// TestTryPreemption_DecoupleAskQueueQuotaFromVictimSize proves that a large candidate victim
+// (e.g. 8 cores) on a node can be preempted to satisfy a small ask (e.g. 2 cores) whose queue has
+// sufficient remaining guaranteed quota (e.g. 2 cores), without being rejected simply because the
+// victim's resource size exceeds the ask queue's remaining guaranteed headroom.
+func TestTryPreemption_DecoupleAskQueueQuotaFromVictimSize(t *testing.T) {
+ appQueueMapping := NewAppQueueMapping()
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 8})
+ 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": "10"}, appQueueMapping)
+ assert.NilError(t, err)
+
+ app1 := newApplication(appID1, "default", "root.parent.child1")
+ app1.SetQueue(childQ1)
+ childQ1.AddApplication(app1)
+ appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
+
+ // alloc1 is 8 cores on the node (the only victim on this node)
+ ask1 := newAllocationAsk("alloc1", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
+ assert.NilError(t, app1.AddAllocationAsk(ask1))
+ alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
+ app1.AddAllocation(alloc1)
+ assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
+ assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
+
+ // Preemptor app in childQ2. Guaranteed is 10, existing usage is 8, remaining guaranteed is 2.
+ app2 := newApplication(appID2, "default", "root.parent.child2")
+ app2.SetQueue(childQ2)
+ childQ2.AddApplication(app2)
+ appQueueMapping.AddAppQueueMapping(app2.ApplicationID, childQ2)
+
+ existingAsk := newAllocationAsk("existing", appID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 8}))
+ assert.NilError(t, app2.AddAllocationAsk(existingAsk))
+ assert.NilError(t, childQ2.TryIncAllocatedResource(existingAsk.GetAllocatedResource()))
+
+ // Preemptor ask needs 2 cores.
+ preemptorAsk := newAllocationAsk("preemptor", appID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
+ assert.NilError(t, app2.AddAllocationAsk(preemptorAsk))
+ childQ2.incPendingResource(preemptorAsk.GetAllocatedResource())
+
+ headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10})
+ preemptor := NewPreemptor(app2, headRoom, 30*time.Second, preemptorAsk, iterator(), false)
+
+ result, ok := preemptor.TryPreemption()
+ assert.Assert(t, ok, "preemption should succeed")
+ assert.Assert(t, result != nil)
+ assert.Equal(t, "preemptor", result.Request.GetAllocationKey())
+ assert.Equal(t, nodeID1, result.NodeID)
+ assert.Check(t, alloc1.IsPreempted(), "alloc1 should be preempted")
+}
+
+// TestTryPreemption_PredicateTailVictimsNotTruncated proves that candidate victims routed to 'tail'
+// (e.g. required for Kubernetes predicate evaluation like PodAntiAffinity, but not directly reducing
+// TestTryPreemption_PredicateTailVictimsNotTruncated proves that candidate victims routed to 'tail'
+// (e.g. required for Kubernetes predicate evaluation like PodAntiAffinity, but not directly reducing
+// the ask's resource shortfall) are not truncated or discarded when an earlier victim in 'head'
+// consumes the ask queue's remaining guaranteed quota in the snapshot.
+// It also proves that when predicates are satisfied at index 0, tail victims are spared (no overkill).
+func TestTryPreemption_PredicateTailVictimsNotTruncated(t *testing.T) {
+ tests := []struct {
+ name string
+ predicatePassIndex int32
+ expectAlloc1Preempt bool
+ expectAlloc2Preempt bool
+ }{
+ {
+ name: "tail victim preempted when required by anti-affinity",
+ predicatePassIndex: 1,
+ expectAlloc1Preempt: true,
+ expectAlloc2Preempt: true,
+ },
+ {
+ name: "tail victim spared when no anti-affinity (no overkill)",
+ predicatePassIndex: 0,
+ expectAlloc1Preempt: true,
+ expectAlloc2Preempt: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ appQueueMapping := NewAppQueueMapping()
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 2})
+ 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": "4"}, 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": "4"}, appQueueMapping)
+ assert.NilError(t, err)
+
+ app1 := newApplication(appID1, "default", "root.parent.child1")
+ app1.SetQueue(childQ1)
+ childQ1.AddApplication(app1)
+ appQueueMapping.AddAppQueueMapping(app1.ApplicationID, childQ1)
+
+ // alloc1 on node: first: 1 (reduces shortfall of "first", goes to head)
+ ask1 := newAllocationAsk("alloc1", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
+ ask1.createTime = time.Now().Add(-5 * time.Second)
+ assert.NilError(t, app1.AddAllocationAsk(ask1))
+ alloc1 := newAllocationWithKey("alloc1", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
+ alloc1.createTime = ask1.createTime
+ app1.AddAllocation(alloc1)
+ assert.Check(t, node.TryAddAllocation(alloc1), "node alloc1 failed")
+ assert.NilError(t, childQ1.TryIncAllocatedResource(ask1.GetAllocatedResource()))
+
+ // alloc2 on node: first: 1 (shortfall already zero, goes to tail, but required for predicate e.g. PodAntiAffinity)
+ ask2 := newAllocationAsk("alloc2", appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
+ ask2.createTime = time.Now().Add(-10 * time.Second)
+ assert.NilError(t, app1.AddAllocationAsk(ask2))
+ alloc2 := newAllocationWithKey("alloc2", appID1, nodeID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
+ alloc2.createTime = ask2.createTime
+ app1.AddAllocation(alloc2)
+ assert.Check(t, node.TryAddAllocation(alloc2), "node alloc2 failed")
+ assert.NilError(t, childQ1.TryIncAllocatedResource(ask2.GetAllocatedResource()))
+
+ // Preemptor app in childQ2.
+ // Guaranteed is first: 4. Existing usage is first: 3. Remaining is first: 1.
+ app2 := newApplication(appID2, "default", "root.parent.child2")
+ app2.SetQueue(childQ2)
+ childQ2.AddApplication(app2)
+ appQueueMapping.AddAppQueueMapping(app2.ApplicationID, childQ2)
+
+ existingAsk := newAllocationAsk("existing", appID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3}))
+ assert.NilError(t, app2.AddAllocationAsk(existingAsk))
+ assert.NilError(t, childQ2.TryIncAllocatedResource(existingAsk.GetAllocatedResource()))
+
+ // Preemptor ask needs first: 1.
+ preemptorAsk := newAllocationAsk("preemptor", appID2, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1}))
+ assert.NilError(t, app2.AddAllocationAsk(preemptorAsk))
+ childQ2.incPendingResource(preemptorAsk.GetAllocatedResource())
+
+ preemptions := []mock.Preemption{
+ mock.NewPreemption(true, "preemptor", nodeID1, []string{"alloc1", "alloc2"}, 0, tt.predicatePassIndex),
+ }
+ plugin := mock.NewPreemptionPredicatePlugin(preemptions, nil, false, false)
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ headRoom := resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10})
+ preemptor := NewPreemptor(app2, headRoom, 30*time.Second, preemptorAsk, iterator(), false)
+
+ result, ok := preemptor.TryPreemption()
+ assert.Assert(t, ok, "preemption should succeed")
+ assert.Assert(t, result != nil)
+ assert.Equal(t, "preemptor", result.Request.GetAllocationKey())
+ assert.Equal(t, nodeID1, result.NodeID)
+ assert.Equal(t, alloc1.IsPreempted(), tt.expectAlloc1Preempt, "alloc1 preempted status mismatch")
+ assert.Equal(t, alloc2.IsPreempted(), tt.expectAlloc2Preempt, "alloc2 preempted status mismatch")
+ })
+ }
}
// TestTryPreemption_PrematureAdditionalVictimsLoopTermination proves that when an early candidate victim
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org