This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch branch-1.9
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/branch-1.9 by this push:
new 257fdef7 [YUNIKORN-3318] Deadlock in preemption init un-reserving apps
(#1103)
257fdef7 is described below
commit 257fdef7e24b052f433a6ad074bc5a0f8381bd79
Author: Wilfred Spiegelenburg <[email protected]>
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 <[email protected]>
(cherry picked from commit 3dbbb5d73823841dcd7af754984bb0ab53585da7)
---
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: [email protected]
For additional commands, e-mail: [email protected]
