You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by wi...@apache.org on 2026/07/13 05:40:02 UTC
(yunikorn-core) branch master updated: [YUNIKORN-3318] Deadlock in preemption init un-reserving apps (#1103)
This is an automated email from the ASF dual-hosted git repository.
wilfred-s 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 3dbbb5d7 [YUNIKORN-3318] Deadlock in preemption init un-reserving apps (#1103)
3dbbb5d7 is described below
commit 3dbbb5d73823841dcd7af754984bb0ab53585da7
Author: Wilfred Spiegelenburg <wi...@apache.org>
AuthorDate: Mon Jul 13 15:39:38 2026 +1000
[YUNIKORN-3318] Deadlock in preemption init un-reserving apps (#1103)
If an application A holds a reservation R on a node N the preemption
initialisation will deadlock.
During the initWorkingState() call reservations are released from a node
if they are older than the timeout, not a required node allocation, and
the priority is lower than the allocation we are trying to preempt for.
Fix time calculation to use the reservation creation time for accuracy.
Track the number of reservations released.
Correctly determine if a node is still reserved after releases.
fix unit tests
Closes: #1103
Signed-off-by: Wilfred Spiegelenburg <wi...@apache.org>
---
pkg/scheduler/objects/application.go | 24 +++++-------
pkg/scheduler/objects/application_test.go | 61 ++++++++++++++++---------------
pkg/scheduler/objects/preemption.go | 54 ++++++++++++++++-----------
pkg/scheduler/objects/preemption_test.go | 47 ++++++++++++++++++++++++
4 files changed, 122 insertions(+), 64 deletions(-)
diff --git a/pkg/scheduler/objects/application.go b/pkg/scheduler/objects/application.go
index fd2636d8..694924a4 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -47,17 +47,18 @@ import (
)
var (
- reservationDelay = 2 * time.Second
- reservationWaitTimeout = 60 * time.Minute
- // Make it configurable
+ // Make these settings configurable?
+ reservationDelay = 2 * time.Second
+ reservationWaitTimeout = 60 * time.Minute
completingTimeout = 30 * time.Second
terminatedTimeout = 3 * 24 * time.Hour
defaultPlaceholderTimeout = 15 * time.Minute
+ // global app loggers (rate limited)
+ initAppLogOnce sync.Once
+ initReqNodeLogOnce sync.Once
+ rateLimitedAppLog *log.RateLimitedLogger
+ rateLimitedReqNodeLog *log.RateLimitedLogger
)
-var initAppLogOnce sync.Once
-var rateLimitedAppLog *log.RateLimitedLogger
-var initReqNodeLogOnce sync.Once
-var rateLimitedReqNodeLog *log.RateLimitedLogger
const (
Soft string = "Soft"
@@ -1425,13 +1426,8 @@ func (sa *Application) tryReservedAllocate(headRoom *resources.Resource, nodeIte
}
if !sa.checkHeadRooms(ask, userHeadroom, headRoom) {
- // Cancel the reservation after wait time expires
- createTime := reserve.createTime
-
- askAge := time.Since(createTime.Add(reservationWaitTimeout))
-
- // Has wait time reached?
- if askAge > reservationWaitTimeout {
+ // Cancel the reservation after wait time expires, but not for required node asks
+ if ask.GetRequiredNode() == "" && time.Since(reserve.createTime) > reservationWaitTimeout {
num := sa.unReserveInternal(reserve)
sa.queue.UnReserve(sa.ApplicationID, num)
log.Log(log.SchedApplication).Info("Cancelled reservation as wait time expired",
diff --git a/pkg/scheduler/objects/application_test.go b/pkg/scheduler/objects/application_test.go
index c3a3947a..7233788d 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -2536,16 +2536,17 @@ func TestTryAllocatePreemptNodeWithReservations(t *testing.T) {
// pass the time and try again
ask4.createTime = ask4.createTime.Add(-30 * time.Second)
- reservationWaitTimeout = -60 * time.Second
+ defWaitTimeout := reservationWaitTimeout
+ reservationWaitTimeout = 60 * time.Second
+ defer func() {
+ reservationWaitTimeout = defWaitTimeout
+ }()
result3 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
assert.Assert(t, result3 != nil, "result3 expected")
assert.Equal(t, Reserved, result3.ResultType, "expected reservation")
alloc3 := result3.Request
assert.Assert(t, alloc3 != nil, "alloc3 expected")
- assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
-
- // reset wait timeout
- reservationWaitTimeout = 60 * time.Minute
+ assert.Assert(t, allocs[1].IsPreempted(), "alloc2 should have been preempted")
}
func TestTryAllocatePreemptNodeWithReservationsWithHighPriority(t *testing.T) {
@@ -2558,7 +2559,11 @@ func TestTryAllocatePreemptNodeWithReservationsWithHighPriority(t *testing.T) {
// pass the time and try again
ask4.createTime = ask4.createTime.Add(-30 * time.Second)
- reservationWaitTimeout = -60 * time.Second
+ defWaitTimeout := reservationWaitTimeout
+ reservationWaitTimeout = 60 * time.Second
+ defer func() {
+ reservationWaitTimeout = defWaitTimeout
+ }()
result3 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
assert.Assert(t, result3 == nil, "result3 expected")
@@ -2570,10 +2575,7 @@ func TestTryAllocatePreemptNodeWithReservationsWithHighPriority(t *testing.T) {
assert.Equal(t, Reserved, result4.ResultType, "expected reservation")
alloc3 := result4.Request
assert.Assert(t, alloc3 != nil, "alloc3 expected")
- assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
-
- // reset wait timeout
- reservationWaitTimeout = 60 * time.Minute
+ assert.Assert(t, allocs[1].IsPreempted(), "alloc2 should have been preempted")
}
// TestTryAllocatePreemptNodeWithReservationsNotPossibleToCancel Ensures reservations cannot be cancelled because of the following constraints:
@@ -2596,18 +2598,21 @@ func TestTryAllocatePreemptNodeWithReservationsNotPossibleToCancel(t *testing.T)
preemptionAttemptsRemaining := 10
// on first attempt, should see a reservation on node2 since we're after the reservation timeout
- var alloc11 *Allocation
- ask5.createTime = ask5.createTime.Add(-10 * time.Second)
result1 := app1.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
- assert.Assert(t, result1 != nil, "result1 expected")
- alloc11 = result1.Request
- assert.Equal(t, "alloc5", alloc11.allocationKey, "wrong node assignment")
- assert.Assert(t, result1.Request != nil, "alloc1 expected")
+ assert.Assert(t, result1 != nil, "result expected")
assert.Equal(t, "node2", result1.NodeID, "wrong node assignment")
assert.Equal(t, Reserved, result1.ResultType, "expected reservation")
- allocs = append(allocs, alloc11)
+ alloc1 := result1.Request
+ assert.Assert(t, alloc1 != nil, "expected allocation result")
+ allocs = append(allocs, alloc1)
+ assert.Equal(t, "alloc5", alloc1.allocationKey, "wrong allocation assignment")
+ // fake the partition processing and reserve the node and app
err = getNode("node2").Reserve(app1, ask5)
- assert.NilError(t, err)
+ assert.NilError(t, err, "node reservation should not have failed")
+ err = app1.reserveInternal(getNode("node2"), ask5)
+ assert.NilError(t, err, "app reservation should not have failed")
+
+ assert.Equal(t, len(getNode("node1").GetReservations()), 1, "reservation expected on node1")
// Set higher priority than the reserved ask priority but no preemption because reserved ask waiting time not exceeded
ask4.priority = 1
@@ -2618,28 +2623,27 @@ func TestTryAllocatePreemptNodeWithReservationsNotPossibleToCancel(t *testing.T)
// Both Node 1 & 2 has reservations, one allocation has required node set and another had marked for "triggered preemption" flag
// Still, preemption doesn't yield any positive outcome
ask4.createTime = ask4.createTime.Add(-30 * time.Second)
- reservationWaitTimeout = -60 * time.Second
+ defWaitTimeout := reservationWaitTimeout
+ reservationWaitTimeout = 0
+ defer func() {
+ reservationWaitTimeout = defWaitTimeout
+ }()
ask4.preemptCheckTime = ask4.preemptCheckTime.Add(-30 * time.Second)
result4 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
- assert.Assert(t, result4 == nil, "result3 expected")
+ assert.Assert(t, result4 == nil, "result4 expected")
// Ensure reserved ask waiting time exceeds
// Ensure reserved allocation doesn't have required node set and not marked for "triggered preemption" flag
- // Still, preemption doesn't yield any positive outcome
ask5.requiredNode = ""
ask3.preemptionTriggered = false
ask4.createTime = ask4.createTime.Add(-30 * time.Second)
- reservationWaitTimeout = -60 * time.Second
ask4.preemptCheckTime = ask4.preemptCheckTime.Add(-30 * time.Second)
result5 := app3.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 18}), true, 30*time.Second, &preemptionAttemptsRemaining, iterator, iterator, getNode)
- assert.Assert(t, result5 != nil, "result3 expected")
+ assert.Assert(t, result5 != nil, "result5 expected")
assert.Equal(t, Reserved, result5.ResultType, "expected reservation")
alloc3 := result5.Request
assert.Assert(t, alloc3 != nil, "alloc3 expected")
- assert.Assert(t, allocs[0].IsPreempted(), "alloc1 should have been preempted")
-
- // reset wait timeout
- reservationWaitTimeout = 60 * time.Minute
+ assert.Assert(t, allocs[1].IsPreempted(), "alloc2 should have been preempted")
}
func TestMaxAskPriority(t *testing.T) {
@@ -3328,13 +3332,12 @@ func TestTryAllocateWithReservedHeadRoomChecking(t *testing.T) {
assert.Equal(t, len(app.reservations), 1)
// pass the time and try again
- reservationWaitTimeout = -60 * time.Second
+ app.reservations[ask.allocationKey].createTime = time.Now().Add(-90 * time.Minute)
result = app.tryReservedAllocate(headRoom, iter)
assert.Assert(t, result == nil, "result is expected to be nil due to insufficient headroom")
assert.Equal(t, len(app.reservations), 0)
// reset wait timeout
- reservationWaitTimeout = 60 * time.Minute
}
func TestUpdateRunnableStatus(t *testing.T) {
diff --git a/pkg/scheduler/objects/preemption.go b/pkg/scheduler/objects/preemption.go
index 4849c8d0..dc791184 100644
--- a/pkg/scheduler/objects/preemption.go
+++ b/pkg/scheduler/objects/preemption.go
@@ -133,10 +133,11 @@ func (p *Preemptor) initQueueSnapshots() {
}
// initWorkingState builds helper data structures required to compute a solution
-func (p *Preemptor) initWorkingState() {
+// returns the number of reservations released while preparing the data structures
+func (p *Preemptor) initWorkingState() int {
// return if we have already run
if p.nodeAvailableMap != nil {
- return
+ return 0
}
// ensure queue snapshots are populated
@@ -158,38 +159,48 @@ func (p *Preemptor) initWorkingState() {
queueByAlloc[allocation.GetAllocationKey()] = victims
}
}
-
+ // total reservation cancelled to make sure we update the global counters
+ totalReservationCancel := 0
// walk node iterator and track available resources per node
p.iterator.ForEachNode(func(node *Node) bool {
- hasOtherReservations := false
+ isReserved := false
if node.IsReserved() && !node.isReservedForAllocation(p.ask.GetAllocationKey()) {
- hasOtherReservations = true
+ leftCount := 0
for _, res := range node.GetReservations() {
+ leftCount++
// Is Allocation daemon set?
// Has this allocation already triggered preemption?
if res.alloc.requiredNode != "" || res.alloc.HasTriggeredPreemption() {
continue
}
- createTime := res.alloc.GetCreateTime()
- // Take reservation delay also into account
- askAge := time.Since(createTime.Add(reservationWaitTimeout).Add(reservationDelay))
-
// Cancel reservation based on its priority and waiting time in reservation queue
- if res.alloc.GetPriority() < p.ask.priority && askAge > reservationWaitTimeout {
- num := res.app.UnReserve(res.node, res.alloc)
- res.app.GetQueue().UnReserve(res.app.ApplicationID, num)
- log.Log(log.SchedApplication).Info("Cancelled reservation to consider node for preemption",
- zap.String("triggered by appID", p.application.ApplicationID),
- zap.String("triggered by allocationKey", p.ask.allocationKey),
- zap.String("affected application ID", res.appID),
- zap.String("affected allocationKey", res.allocKey),
- zap.String("node", res.nodeID),
- zap.Int("reservations count", num))
- hasOtherReservations = false
+ if res.alloc.GetPriority() < p.ask.priority && time.Since(res.createTime) > reservationWaitTimeout {
+ log.Log(log.SchedPreemption).Info("Cancelling reservation to consider node for preemption",
+ zap.String("triggeringAppID", p.application.ApplicationID),
+ zap.String("triggeringAllocationKey", p.ask.allocationKey),
+ zap.String("reservingAppID", res.appID),
+ zap.String("reservingAllocationKey", res.allocKey),
+ zap.String("node", node.NodeID))
+ num := 0
+ if p.application.ApplicationID == res.appID {
+ num = res.app.unReserveInternal(res)
+ res.app.queue.UnReserve(res.app.ApplicationID, num)
+ } else {
+ num = res.app.UnReserve(res.node, res.alloc)
+ res.app.GetQueue().UnReserve(res.app.ApplicationID, num)
+ }
+ totalReservationCancel += num
+ leftCount -= num
}
}
+ log.Log(log.SchedPreemption).Debug("Reservations left on node are cleanup",
+ zap.String("triggeringAppID", p.application.ApplicationID),
+ zap.String("triggeringAllocationKey", p.ask.allocationKey),
+ zap.String("node", node.NodeID),
+ zap.Int("leftCount", leftCount))
+ isReserved = leftCount > 0
}
- if !node.IsSchedulable() || hasOtherReservations || !node.FitInNode(p.ask.GetAllocatedResource()) {
+ if !node.IsSchedulable() || isReserved || !node.FitInNode(p.ask.GetAllocatedResource()) {
// node is not available, remove any potential victims from consideration
delete(allocationsByNode, node.NodeID)
} else {
@@ -205,6 +216,7 @@ func (p *Preemptor) initWorkingState() {
p.allocationsByNode = allocationsByNode
p.queueByAlloc = queueByAlloc
p.nodeAvailableMap = nodeAvailableMap
+ return totalReservationCancel
}
// checkPreemptionQueueGuarantees verifies that it's possible to free enough resources to fit the given ask
diff --git a/pkg/scheduler/objects/preemption_test.go b/pkg/scheduler/objects/preemption_test.go
index 599e0f66..627eaa34 100644
--- a/pkg/scheduler/objects/preemption_test.go
+++ b/pkg/scheduler/objects/preemption_test.go
@@ -2246,3 +2246,50 @@ func TestTryPreemption_AskQueue_Under_DiffParent_With_OG_And_UG_ResTypes(t *test
})
}
}
+
+func Test_PreemptForAppOnReservedNode(t *testing.T) {
+ rootQ, err := createRootQueue(map[string]string{"first": "6"})
+ assert.NilError(t, err, "root queue create failed")
+ var childQ *Queue
+ childQ, err = createManagedQueueGuaranteed(rootQ, "child", false, map[string]string{"first": "10"}, map[string]string{"first": "10"}, nil)
+ assert.NilError(t, err, "child queue create failed")
+ node1 := newNode("node1", map[string]resources.Quantity{"first": 6})
+ node1.SetOccupiedResource(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 2}))
+ iterator := getNodeIteratorFn(node1)
+ getNode := func(id string) *Node {
+ if id == node1.NodeID {
+ return node1
+ }
+ return nil
+ }
+
+ app := newApplication(appID1, "default", childQ.QueuePath)
+ app.SetQueue(childQ)
+ childQ.AddApplication(app)
+
+ // (1) no-priority ask, RESERVED on node1 ask does not fit in free node resources
+ askLow := newAllocationAsk(aKey, appID1, resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5}))
+ // must be more than wait timeout (default 60min) + delay (default 2sec) ago
+ askLow.createTime = time.Now().Add(-90 * time.Minute)
+ assert.NilError(t, app.AddAllocationAsk(askLow), "ask addition to app should not have failed")
+ assert.NilError(t, app.Reserve(node1, askLow), "reservation on node should not have failed")
+ nodeRes := node1.GetReservations()
+ assert.Equal(t, len(nodeRes), 1, "expected 1 reservation")
+ nodeRes[0].createTime = nodeRes[0].createTime.Add(-90 * time.Minute)
+
+ // priority ask on the SAME app that will enter preemption as node is full (usage + reservation)
+ askHigh := newAllocationAskPriority(aKey2, "repro-app", resources.NewResourceFromMap(map[string]resources.Quantity{"first": 3}), 100)
+ askHigh.allowPreemptOther = true
+ askHigh.createTime = time.Now().Add(-10 * time.Second)
+ assert.NilError(t, app.AddAllocationAsk(askHigh), "add high ask")
+
+ remaining := 2
+ // waiting ask fits on the node, fits in the queue created 10 sec ago (larger than preemption delay)
+ // headroom == max for the queue
+ // preemption turned on with a 1-second delay and at least 1 attempt remaining
+ // unreserve must not block on the node and reserve the node for this preemption
+ // NOTE: deadlock detection in locking fails this test if regressed
+ result := app.tryAllocate(resources.NewResourceFromMap(map[string]resources.Quantity{"first": 10}), true, 1*time.Second, &remaining, iterator, iterator, getNode)
+ assert.Assert(t, result != nil, "expected and allocation result back")
+ assert.Equal(t, result.ResultType, Reserved, "expected result type to be Reserved")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org