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 2a271ec1 [YUNIKORN-3352] Replace the O(n) ask-priority rescan with a
pending histogram (#1122)
2a271ec1 is described below
commit 2a271ec187eba9a79a24be5496c5cf2a46c6ea56
Author: Tigerquoll <[email protected]>
AuthorDate: Thu Aug 20 19:34:42 2026 +1000
[YUNIKORN-3352] Replace the O(n) ask-priority rescan with a pending
histogram (#1122)
updateAskMaxPriority recomputed an application's highest pending ask
priority by
rescanning every one of its asks, and it ran on every allocate, on every
deallocate and on every ask removal whose priority was at or above the
current
maximum. Over a scheduling cycle that is quadratic in the number of asks an
application holds.
Replace the rescan with a small histogram of pending asks per priority
value,
maintained incrementally: incPendingPriority when an ask becomes pending,
decPendingPriority when it leaves the pending set. askMaxPriority is then
O(1) in
the common case, and only a bucket emptying at the current maximum costs a
walk -
over the number of distinct priorities, not the number of asks. The queue is
notified only when the value actually changes, where the rescan
re-published it
unconditionally.
No scheduling behaviour changes: the golden decision-trace tests added in
YUNIKORN-3338 reproduce byte-for-byte, goldens not regenerated.
Generated by Author with assistance from Claude Code.
Signed-off-by: Tigerquoll <[email protected]>
Closes: #1122
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/scheduler/objects/application.go | 118 +++--
pkg/scheduler/objects/application_property_test.go | 517 +++++++++++++++++++++
pkg/scheduler/objects/application_test.go | 194 ++++++++
3 files changed, 798 insertions(+), 31 deletions(-)
diff --git a/pkg/scheduler/objects/application.go
b/pkg/scheduler/objects/application.go
index 6d630423..71decadc 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -115,7 +115,8 @@ type Application struct {
rejectedMessage string // If the application
is rejected, save the rejected message
stateLog []*StateLogEntry // state log for this
application
placeholderData map[string]*PlaceholderData // track placeholder
and gang related info
- askMaxPriority int32 // highest priority
value of outstanding asks
+ askMaxPriority int32 // highest priority
value of outstanding (pending) asks
+ pendingPriorities map[int32]int // count of pending
(non-allocated) asks per priority value
hasPlaceholderAlloc bool // Whether there is at
least one allocated placeholder
runnableInQueue bool // whether the
application is runnable/schedulable in the queue. Default is true.
runnableByUserLimit bool // whether the
application is runnable/schedulable based on user/group quota. Default is true.
@@ -155,6 +156,7 @@ func NewApplication(siApp *si.AddApplicationRequest, ugi
security.UserGroup, eve
rejectedMessage: "",
stateLog: make([]*StateLogEntry, 0),
askMaxPriority: configs.MinPriority,
+ pendingPriorities: make(map[int32]int),
sortedRequests: sortedRequests{},
sendStateChangeEvents: true,
runnableByUserLimit: true,
@@ -597,6 +599,7 @@ func (sa *Application) removeAsksInternal(allocKey string,
detail si.EventRecord
}
sa.requests = make(map[string]*Allocation)
sa.sortedRequests = sortedRequests{}
+ sa.pendingPriorities = make(map[int32]int)
sa.askMaxPriority = configs.MinPriority
sa.queue.UpdateApplicationPriority(sa.ApplicationID,
sa.askMaxPriority)
} else {
@@ -612,13 +615,12 @@ func (sa *Application) removeAsksInternal(allocKey
string, detail si.EventRecord
deltaPendingResource =
ask.GetAllocatedResource()
sa.pending = resources.Sub(sa.pending,
deltaPendingResource)
sa.pending.Prune()
+ // the removed ask was pending: drop it from
the priority histogram
+ sa.removeFromPriorities(ask.GetPriority())
}
delete(sa.requests, allocKey)
sa.sortedRequests.remove(ask)
sa.appEvents.SendRemoveAskEvent(sa.ApplicationID,
ask.allocationKey, ask.GetAllocatedResource(), detail)
- if priority := ask.GetPriority(); priority >=
sa.askMaxPriority {
- sa.updateAskMaxPriority()
- }
}
}
// clean up the queue pending resources
@@ -664,6 +666,9 @@ func (sa *Application) AddAllocationAsk(ask *Allocation)
error {
var oldAskResource *resources.Resource = nil
if oldAsk := sa.requests[ask.GetAllocationKey()]; oldAsk != nil &&
!oldAsk.IsAllocated() {
oldAskResource = oldAsk.GetAllocatedResource().Clone()
+ // the old ask was pending and is being replaced: drop it from
the priority histogram so the
+ // new ask's addAllocationAskInternal (via addToPriorities)
nets correctly.
+ sa.removeFromPriorities(oldAsk.GetPriority())
}
// Check if we need to change state based on the ask added, there are
two cases:
@@ -778,15 +783,68 @@ func (sa *Application) RecoverAllocationAsk(alloc
*Allocation) {
}
}
+// addToPriorities records that an ask at priority p just became pending
(added, or
+// deallocated back to pending). Call with sa.Lock() held.
+func (sa *Application) addToPriorities(p int32) {
+ sa.pendingPriorities[p]++
+ if p > sa.askMaxPriority {
+ sa.setAskMaxPriority(p)
+ }
+}
+
+// removeFromPriorities records that a pending ask at priority p just left the
pending set
+// (allocated, or removed). Call with sa.Lock() held.
+func (sa *Application) removeFromPriorities(p int32) {
+ // n is the count the band is left with. A band is removed as soon as
it empties, so a band that
+ // is present always holds at least one ask - n < 0 therefore means
there was no band at all,
+ // not a band that happened to be sitting at zero.
+ switch n := sa.pendingPriorities[p] - 1; {
+ case n < 0:
+ // nothing to decrement: an inc/dec pair was missed somewhere.
Report it rather than
+ // absorbing it, because askMaxPriority is only correct while
those pairs match.
+ log.Log(log.SchedApplication).DPanic("no pending priority band
to decrement",
+ zap.String("appID", sa.ApplicationID),
+ zap.Int32("priority", p))
+ case n > 0:
+ // other pending asks remain at this priority, so the band
survives and the maximum -
+ // whichever band it points at - cannot have moved.
+ sa.pendingPriorities[p] = n
+ default:
+ // that was the last pending ask at this priority, so the band
goes away
+ delete(sa.pendingPriorities, p)
+ if p != sa.askMaxPriority {
+ // the band that got deleted was not the max priority,
so we don't change the existing
+ // cached value
+ return
+ }
+ // the top band emptied, so the maximum has to be recomputed.
This walks the bands, not the
+ // asks: O(distinct priorities in use), not O(number of asks).
An empty map yields
+ // MinPriority, which is what the full rescan this replaced
also produced with nothing
+ // pending.
+ newMax := configs.MinPriority
+ for band := range sa.pendingPriorities {
+ newMax = max(newMax, band)
+ }
+ sa.setAskMaxPriority(newMax)
+ }
+}
+
+// setAskMaxPriority updates askMaxPriority and only propagates to the queue
when the value
+// actually changes. Call with sa.Lock() held.
+func (sa *Application) setAskMaxPriority(v int32) {
+ if v == sa.askMaxPriority {
+ return
+ }
+ sa.askMaxPriority = v
+ sa.queue.UpdateApplicationPriority(sa.ApplicationID, v)
+}
+
func (sa *Application) addAllocationAskInternal(ask *Allocation) {
sa.requests[ask.GetAllocationKey()] = ask
- // update app priority
- allocated := ask.IsAllocated()
- priority := ask.GetPriority()
- if !allocated && priority > sa.askMaxPriority {
- sa.askMaxPriority = priority
- sa.queue.UpdateApplicationPriority(sa.ApplicationID,
sa.askMaxPriority)
+ // update app priority: a recovered (already allocated) ask must never
enter the pending histogram
+ if !ask.IsAllocated() {
+ sa.addToPriorities(ask.GetPriority())
}
if ask.IsPlaceholder() {
@@ -862,10 +920,8 @@ func (sa *Application) allocateAsk(ask *Allocation)
(*resources.Resource, error)
return nil, fmt.Errorf("unable to allocate previously allocated
ask %s on app %s", ask.GetAllocationKey(), sa.ApplicationID)
}
- if ask.GetPriority() >= sa.askMaxPriority {
- // recalculate downward
- sa.updateAskMaxPriority()
- }
+ // the ask just left the pending set
+ sa.removeFromPriorities(ask.GetPriority())
delta := ask.GetAllocatedResource()
sa.pending = resources.Sub(sa.pending, delta)
@@ -881,11 +937,19 @@ func (sa *Application) deallocateAsk(ask *Allocation)
(*resources.Resource, erro
return nil, fmt.Errorf("unable to deallocate pending ask %s on
app %s", ask.GetAllocationKey(), sa.ApplicationID)
}
- askPriority := ask.GetPriority()
- if askPriority > sa.askMaxPriority {
- // increase app priority
- sa.askMaxPriority = askPriority
- sa.queue.UpdateApplicationPriority(sa.ApplicationID,
askPriority)
+ // The ask returns to the pending set, but only if it still IS this
application's ask: an ask that
+ // has already been dropped from sa.requests must not be counted again.
That is reachable today:
+ // removeAsksInternal("") wipes sa.requests while leaving
sa.allocations intact until the shim
+ // confirms the releases, and a release arriving in that window reaches
RollbackAllocation, which
+ // finds the entry in sa.allocations and deallocates it. The identity
comparison (not just a
+ // presence check) also covers a stale ask object that has since been
replaced by a new ask under
+ // the same key.
+ // This matches the converged behaviour of the full rescan this change
replaced:
+ // updateAskMaxPriority derived the max by scanning sa.requests, so an
ask absent from sa.requests
+ // never influenced it. Without the guard that pre-existing accounting
drift would turn into a
+ // permanent leak in the incremental histogram instead.
+ if sa.requests[ask.GetAllocationKey()] == ask {
+ sa.addToPriorities(ask.GetPriority())
}
delta := ask.GetAllocatedResource()
@@ -2075,18 +2139,6 @@ func (sa *Application)
removeAllocationInternal(allocationKey string, releaseTyp
return alloc
}
-func (sa *Application) updateAskMaxPriority() {
- value := configs.MinPriority
- for _, v := range sa.requests {
- if v.IsAllocated() {
- continue
- }
- value = max(value, v.GetPriority())
- }
- sa.askMaxPriority = value
- sa.queue.UpdateApplicationPriority(sa.ApplicationID, value)
-}
-
func (sa *Application) hasZeroAllocations() bool {
return resources.IsZero(sa.pending) &&
resources.IsZero(sa.allocatedResource)
}
@@ -2282,6 +2334,10 @@ func (sa *Application) GetAskMaxPriority() int32 {
func (sa *Application) cleanupAsks() {
sa.requests = make(map[string]*Allocation)
sa.sortedRequests = nil
+ // a Failed app can still hold pending asks: reset the histogram or the
consistency check would
+ // fire on terminal apps.
+ sa.pendingPriorities = make(map[int32]int)
+ sa.askMaxPriority = configs.MinPriority
}
func (sa *Application) cleanupTrackedResource() {
diff --git a/pkg/scheduler/objects/application_property_test.go
b/pkg/scheduler/objects/application_property_test.go
new file mode 100644
index 00000000..36d23f30
--- /dev/null
+++ b/pkg/scheduler/objects/application_property_test.go
@@ -0,0 +1,517 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package objects
+
+import (
+ "fmt"
+ "math/rand"
+ "sort"
+ "strconv"
+ "testing"
+ "time"
+
+ "gotest.tools/v3/assert"
+
+ "github.com/apache/yunikorn-core/pkg/common/configs"
+ "github.com/apache/yunikorn-core/pkg/common/resources"
+ siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
+ "github.com/apache/yunikorn-scheduler-interface/lib/go/si"
+)
+
+// TestApplicationPropertyFuzzHistogram drives an Application through a long
randomized sequence
+// of the real ask-management entry points (AddAllocationAsk, AllocateAsk,
DeallocateAsk,
+// RemoveAllocationAsk, RecoverAllocationAsk, AddAllocation,
RollbackAllocation and FSM-to-Failed
+// cleanup) and after every single step verifies that the incrementally
maintained
+// askMaxPriority/pendingPriorities bookkeeping matches a reference model
rebuilt from a simple set
+// of maps tracked alongside the application.
+//
+// The point is coverage of state combinations rather than of entry points:
TestMaxAskPriority pins
+// the handful of transitions that are easy to reason about by hand, while
this test reaches the
+// interleavings that are not - a replaced ask whose priority differs from the
one it displaces, a
+// rollback of an allocation whose ask has already been dropped, an allocate
that empties the top
+// priority bucket while lower buckets still hold pending asks. Those are
precisely the cases where
+// incremental bookkeeping and a full rescan can disagree.
+func TestApplicationPropertyFuzzHistogram(t *testing.T) {
+ // A "ghost" rollback (case 7 reverting an allocation whose ask
sa.requests no longer holds) needs
+ // a specific remove-then-release interleaving, so it is rare enough
that individual seeds
+ // legitimately see none - measured, 6 of the 20 seeds below produce
zero. Assert that coverage
+ // over the whole seed set rather than per seed: that still fails
loudly if a change stops the
+ // fuzzer reaching the state at all, without making the test flaky on
the seeds that never do.
+ totalGhostRollbacks := 0
+ for seed := int64(0); seed < 20; seed++ {
+ t.Run(fmt.Sprintf("seed-%d", seed), func(t *testing.T) {
+ totalGhostRollbacks += runPropertyFuzz(t, seed)
+ })
+ }
+ assert.Assert(t, totalGhostRollbacks > 0, "no seed ever rolled back an
ask missing from sa.requests")
+}
+
+// runPropertyFuzz executes one seeded run and returns the number of ghost
rollbacks it performed, so
+// the caller can assert that coverage across the whole seed set.
+func runPropertyFuzz(t *testing.T, seed int64) int { //nolint:funlen
+ t.Helper()
+
+ // Pin the Completing->Completed state timer to an effectively-infinite
duration for the
+ // duration of this run. enter_Completing (application_state.go) arms a
real time.AfterFunc
+ // timer (completingTimeout, default 30s) that - completely
independently of this goroutine's
+ // step loop - fires HandleApplicationEvent(CompleteApplication) and,
on reaching Completed,
+ // app.cleanupAsks() wipes sa.requests/sortedRequests/pendingPriorities
out from under this
+ // test. Under normal speed this fuzz loop finishes in a couple of
seconds so the default 30s
+ // would not fire, but on a loaded or CI machine it could - and if it
does, the wipe is
+ // invisible to the reference model,
+ // producing exactly the kind of run-dependent, seed-varying divergence
this test exists to
+ // catch. Neutralizing the timer removes that residual,
timing-dependent nondeterminism source
+ // entirely rather than relying on the test finishing "fast enough".
+ SetCompletingTimeout(time.Hour)
+ defer SetCompletingTimeout(30 * time.Second) // restore the documented
production default
+
+ // AddAllocation (case 6) charges the confirmed allocation to the
process-global user manager
+ // (application.go addAllocationInternal -> incUserResourceUsage) and
only RollbackAllocation
+ // gives it back, so any allocation still confirmed when this run ends
leaves usage attributed to
+ // the shared getTestUserGroup() user. Sibling tests in this package
assert exact user/group
+ // totals (assertUserGroupResource in utilities_test.go), so hand the
trackers back clean rather
+ // than leaking this run's fuzz totals into whatever test happens to
run next.
+ defer setupUGM()
+
+ rng := rand.New(rand.NewSource(seed)) //nolint:gosec // deterministic
PRNG is the point: reproducible fuzzing
+ appID := fmt.Sprintf("fuzz-app-%d", seed)
+ app := newApplication(appID, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 1})
+ // nextCreationTime hands out a strictly increasing, deterministic
"creationTime" SI allocation
+ // tag value to every ask this run constructs (see newFuzzAsk). Without
this,
+ // NewAllocationFromSI (allocation.go) falls back to time.Now()
whenever the tag is absent,
+ // which makes every Allocation's createTime - and therefore its
position among LessThan ties in
+ // sortedRequests - depend on wall-clock time instead of on the
(deterministic) sequence of
+ // operations this fuzzer performs. Pinning it removes that dependency
so a fixed seed produces
+ // bit-for-bit identical Allocation state on every run.
+ nextCreationTime := int64(0)
+
+ // keyPriority records the priority assigned to every ask key ever
created (in AddAllocationAsk
+ // or RecoverAllocationAsk); priority is immutable per ask for the
lifetime of the key so this
+ // map is only ever added to (or wiped wholesale on clear-all/cleanup),
never mutated in place.
+ keyPriority := make(map[string]int32)
+ // pendingKeys is the reference model of what the pending histogram
should contain: keys that are
+ // currently pending (added or deallocated back to pending, and NOT yet
allocated, removed, or
+ // recovered-as-allocated).
+ pendingKeys := make(map[string]bool)
+ // allocatedKeys is the reference model of keys that are currently
allocated (via AllocateAsk or
+ // RecoverAllocationAsk) and not yet deallocated or removed.
+ allocatedKeys := make(map[string]bool)
+ // confirmedKeys mirrors sa.allocations: keys handed to AddAllocation
and not yet removed from
+ // sa.allocations again (only a successful RollbackAllocation does that
here - application.go
+ // ~line 886). Note that it is deliberately NOT wiped by the
clear-all/Failed cleanup branches in
+ // case 5: removeAsksInternal and cleanupAsks only ever touch
sa.requests, so a stale entry does
+ // survive in sa.allocations there and the model must say so too.
+ confirmedKeys := make(map[string]bool)
+ nextKey := 0
+
+ // Coverage high-water marks: proof this fuzz run actually reaches
non-trivial states rather
+ // than passing trivially by never exercising the product. Checked at
the end of this function.
+ maxPending, maxAllocated, maxDistinctPendingPriorities := 0, 0, 0
+ // successfulRollbacks counts RollbackAllocation calls that actually
returned without error, i.e.
+ // that really ran deallocateAsk. A rejected call (wrong app state,
unknown key) mutates nothing,
+ // so counting attempts would let this operation decay into a no-op
unnoticed.
+ successfulRollbacks := 0
+ // ghostRollbacks counts the subset of those that rolled back an ask
sa.requests no longer holds
+ // (case 7); replacedAsks counts the AddAllocationAsk calls that took
the replace-existing-ask
+ // branch (case 8). Both are narrow branches that a small change to the
candidate filters could
+ // stop reaching entirely, so both are asserted - replacedAsks per run
at the end of this
+ // function, ghostRollbacks over the whole seed set by the caller.
+ ghostRollbacks := 0
+ replacedAsks := 0
+
+ const steps = 5000
+ for i := 0; i < steps; i++ {
+ switch rng.Intn(9) {
+ case 0: // AddAllocationAsk
+ key := fmt.Sprintf("ask-%d-%d", seed, nextKey)
+ nextKey++
+ priority := int32(rng.Intn(11) - 5) //nolint:gosec //
bounded to -5..5, no overflow
+ placeholder := rng.Intn(2) == 0
+ nextCreationTime++
+ var ask *Allocation
+ if placeholder {
+ ask = newFuzzAsk(key, appID, "tg-"+key, res,
true, priority, "", nextCreationTime)
+ } else {
+ ask = newFuzzAsk(key, appID, "", res, false,
priority, "", nextCreationTime)
+ }
+ addErr := app.AddAllocationAsk(ask)
+ if addErr == nil {
+ // unique keys every time, so this operation
always takes the "brand new ask" path
+ // through AddAllocationAsk; the
replace-existing-ask branch is driven by case 8.
+ keyPriority[key] = priority
+ pendingKeys[key] = true
+ }
+
+ case 1: // AllocateAsk
+ if len(pendingKeys) > 0 {
+ key := pickRandomKey(rng, pendingKeys)
+ if _, allocErr := app.AllocateAsk(key);
allocErr == nil {
+ delete(pendingKeys, key)
+ allocatedKeys[key] = true
+ }
+ }
+ // if pendingKeys is empty this step is a no-op; the
invariant assertion below still
+ // runs for uniform per-step coverage even though
nothing changed.
+
+ case 2: // DeallocateAsk
+ if len(allocatedKeys) > 0 {
+ key := pickRandomKey(rng, allocatedKeys)
+ if _, deallocErr := app.DeallocateAsk(key);
deallocErr == nil {
+ delete(allocatedKeys, key)
+ // the ask returns to pending at the
SAME priority it was created with; priority
+ // is immutable, so keyPriority already
has the right value.
+ pendingKeys[key] = true
+ }
+ }
+
+ case 3: // RemoveAllocationAsk(key) - remove a single existing
key
+ if key, ok := pickRandomExistingKey(rng, keyPriority);
ok {
+ app.RemoveAllocationAsk(key)
+ // removeAsksInternal deletes the key from
sa.requests unconditionally regardless of
+ // whether it was pending or allocated
(application.go ~line 592-611: the pending
+ // histogram/queue accounting is only adjusted
`if !ask.IsAllocated()`, but the
+ // sa.requests delete always happens) - so the
reference model must drop the key from
+ // every tracking map regardless of its prior
pending/allocated state.
+ delete(pendingKeys, key)
+ delete(allocatedKeys, key)
+ delete(keyPriority, key)
+ }
+
+ case 4: // RecoverAllocationAsk - build an already-allocated
ask and recover it
+ key := fmt.Sprintf("recovered-%d-%d", seed, nextKey)
+ nextKey++
+ priority := int32(rng.Intn(11) - 5) //nolint:gosec //
bounded to -5..5, no overflow
+ nextCreationTime++
+ alloc := newFuzzAsk(key, appID, "", res, false,
priority, "recovered-node", nextCreationTime)
+ assert.Assert(t, alloc != nil, "NewAllocationFromSI
unexpectedly returned nil")
+ app.RecoverAllocationAsk(alloc)
+ // a recovered ask is already allocated (NodeID set =>
allocated=true in
+ // NewAllocationFromSI) so addAllocationAskInternal
must NOT add it to the pending
+ // histogram: record it only as allocated, never as
pending. This is the exact invariant
+ // this fuzz operation is probing.
+ keyPriority[key] = priority
+ allocatedKeys[key] = true
+
+ case 5: // occasionally clear-all or fail-and-cleanup,
otherwise another RemoveAllocationAsk-ish no-op
+ switch rng.Intn(50) {
+ case 0: // ~1-in-50 (of the ~1-in-8 chance for case 5):
RemoveAllocationAsk("") clear-all
+ app.RemoveAllocationAsk("")
+ pendingKeys = make(map[string]bool)
+ allocatedKeys = make(map[string]bool)
+ keyPriority = make(map[string]int32)
+ case 1: // ~1-in-50: attempt to drive to Failed (only
if not already terminal), which
+ // triggers cleanupAsks(). FailApplication is
only a valid FSM transition from
+ // New/Accepted/Running (-> Failing) and then
from Failing (-> Failed)
+ // (application_state.go eventDesc()). If the
app is currently sitting in some other
+ // non-terminal state - most notably
Completing, which removeAsksInternal drives it
+ // into once sa.pending AND
sa.allocatedResource are both zero (application.go ~line
+ // 635) - both calls below are silently
rejected by the FSM, sa.requests is left
+ // completely untouched, and cleanupAsks()
never runs. Since case 6 confirms
+ // allocations, sa.allocatedResource is no
longer always zero: a confirmed allocation
+ // now holds the app out of Completing on its
own, so which of the two branches below
+ // is taken depends on the confirm/rollback
history as well as on the pending asks.
+ // Only non-placeholder allocations are
confirmed, so getPlaceholderAllocations() is
+ // always empty and the
hasPlaceHolderAllocations part of that condition never
+ // changes the outcome. The previous version of
this test wiped the reference model
+ // unconditionally here, which diverged from
the product whenever that rejection
+ // happened: the model would claim empty while
the product still held real
+ // pending/allocated asks. Gating the wipe on
the actual post-attempt state (the
+ // operation's real result) keeps the model
faithful to what the product actually did.
+ // cleanupAsks() only clears
sa.requests/sortedRequests/pendingPriorities, so
+ // confirmedKeys (sa.allocations) is
intentionally left alone by both wipes below.
+ // The extra successfulRollbacks condition
exists because Failed is an absorbing
+ // state - application_state.go has no
transition out of it other than Expire - and
+ // RollbackAllocation refuses to do anything
outside Accepted/Running, so a run that
+ // fails at step 30 spends its remaining 4970
steps unable to exercise case 7 at all
+ // (measured: seeds 7 and 12 failed at step
31/55 and then had every one of their
+ // ~500 rollback attempts rejected on state).
Holding the failure back until the
+ // rollback path has run once costs no Failed
coverage worth having: this branch
+ // still comes up roughly every 400 steps
afterwards.
+ if successfulRollbacks > 0 && !app.IsFailed() {
+ _ =
app.HandleApplicationEvent(FailApplication) //nolint:errcheck //
New/Accepted/Running -> Failing
+ _ =
app.HandleApplicationEvent(FailApplication) //nolint:errcheck // Failing ->
Failed, runs cleanupAsks()
+ if app.IsFailed() {
+ pendingKeys =
make(map[string]bool)
+ allocatedKeys =
make(map[string]bool)
+ keyPriority =
make(map[string]int32)
+ }
+ }
+ default:
+ // no-op filler so cases 5's sub-branches don't
dominate step count; still asserted.
+ }
+
+ case 6: // AddAllocation - confirm an allocated ask so it lands
in sa.allocations
+ // RollbackAllocation (case 7) looks its target up in
sa.allocations, not in sa.requests
+ // (application.go ~line 868), so without a
confirmation step there is nothing for it to
+ // roll back and the whole operation would be dead
code. Production confirms by handing
+ // AddAllocation the very same *Allocation that already
sits in sa.requests - either the
+ // ask that AllocateAsk just flipped to allocated
(partition.go ~line 1319-1336) or the
+ // one RecoverAllocationAsk just added (~line
1263-1264) - so the fuzzer does the same and
+ // passes app.GetAllocationAsk(key) rather than
building a second Allocation for the same
+ // key. That matters here: RollbackAllocation
deallocates the object it found in
+ // sa.allocations, and deallocateAsk only counts an ask
as pending again when it still IS
+ // the object sa.requests holds for that key - so a
second Allocation built for the same
+ // key would fail that identity check for a reason the
product itself cannot produce,
+ // making the assertions test fiction.
+ if key, ok := pickRandomFilteredKey(rng, allocatedKeys,
func(k string) bool {
+ if confirmedKeys[k] {
+ // already in sa.allocations:
confirming the same key twice would add its
+ // resource to sa.allocatedResource and
to the user tracker a second time, which
+ // no production path does.
+ return false
+ }
+ // Placeholders take the other branch of
addAllocationInternal (application.go ~line
+ // 1926) which arms the placeholder execution
timer through initPlaceholderTimer:
+ // execTimeout is defaultPlaceholderTimeout for
these apps (NewApplication ~line 174),
+ // not zero, so the timer really is armed and
timeoutPlaceholderProcessing can mutate
+ // the state this test asserts on from another
goroutine. That is exactly the
+ // async-interference class the
SetCompletingTimeout pin at the top of this function
+ // removes, and confirming real allocations is
all RollbackAllocation needs, so
+ // placeholders are never confirmed.
+ ask := app.GetAllocationAsk(k)
+ return ask != nil && !ask.IsPlaceholder()
+ }); ok {
+ ask := app.GetAllocationAsk(key)
+ assert.Assert(t, ask != nil, "seed=%d step=%d:
allocated key %s missing from sa.requests", seed, i, key)
+ app.AddAllocation(ask)
+ // AddAllocation touches
sa.allocations/allocatedResource and the app state only: it
+ // leaves sa.requests, the pending histogram
and sortedRequests alone, so the pending
+ // and allocated reference sets are unchanged
by design.
+ confirmedKeys[key] = true
+ }
+
+ case 7: // RollbackAllocation - revert a confirmed allocation
back to a pending ask
+ // This is the new C1 caller under test:
RollbackAllocation runs deallocateAsk
+ // (application.go ~line 875), which is the function
that does addToPriorities, so a
+ // still-tracked ask has to reappear in the pending
histogram at its original priority.
+ // Candidates deliberately also include confirmed keys
whose ask is no longer in
+ // sa.requests at all. That "ghost" state IS reachable
in production: sa.allocations is
+ // only cleaned up once the shim confirms the release,
so every path that drops asks while
+ // leaving sa.allocations behind opens a window for a
SCHEDULING_FAILED_ON_RM release to
+ // still route to RollbackAllocation -
RemoveAllocationAsk on an allocated key, or
+ // removeAsksInternal("") from
timeoutPlaceholderProcessing case 2, which is reached from
+ // Running for a soft gang whose ResumeApplication
transition is rejected (that event is
+ // only valid from New/Accepted, application_state.go
~line 125) so the app stays Running
+ // and remains rollback-eligible. deallocateAsk must
not count such an ask as pending
+ // again - nothing tracks it any more, so nothing would
ever take it back out of the
+ // histogram - so the reference model below only marks
the key pending again when it is
+ // still tracked.
+ // The only combination excluded is a confirmed key
that is still tracked but already
+ // pending (deallocated without being removed):
ask.deallocate() rejects that outright, so
+ // picking it would only ever burn a step.
+ if key, ok := pickRandomFilteredKey(rng, confirmedKeys,
func(k string) bool {
+ if allocatedKeys[k] {
+ return true
+ }
+ // ghost: sa.requests no longer holds the ask,
but sa.allocations still does
+ _, tracked := keyPriority[k]
+ return !tracked
+ }); ok {
+ if _, rollbackErr :=
app.RollbackAllocation(key); rollbackErr == nil {
+ // only a successful call ran
deallocateAsk: RollbackAllocation refuses outright
+ // unless the app is Accepted or
Running (application.go ~line 864) and bails out
+ // before touching anything if the ask
is no longer allocated, so a returned error
+ // means the product state is unchanged
and the model must be too.
+ delete(confirmedKeys, key)
+ delete(allocatedKeys, key)
+ successfulRollbacks++
+ if _, tracked := keyPriority[key];
tracked {
+ // same as DeallocateAsk (case
2): the ask goes back to pending at its
+ // immutable creation priority,
which keyPriority already holds.
+ pendingKeys[key] = true
+ } else {
+ // ghost rollback: the ask is
gone from sa.requests, so deallocateAsk must
+ // leave the pending histogram
completely untouched.
+ ghostRollbacks++
+ }
+ }
+ }
+
+ case 8: // AddAllocationAsk re-using an existing key - the
replace-existing-ask branch
+ // AddAllocationAsk (application.go ~line 668) handles
a key that is already in
+ // sa.requests and still pending separately: it has to
unwind the old ask from the
+ // pending histogram before addAllocationAskInternal
counts the replacement, or the key
+ // is counted twice and its old priority never drops
out again. Case 0 only ever mints
+ // fresh keys, so without this operation that branch is
never executed here at all.
+ if len(pendingKeys) > 0 {
+ key := pickRandomKey(rng, pendingKeys)
+ existing := app.GetAllocationAsk(key)
+ assert.Assert(t, existing != nil, "seed=%d
step=%d: pending key %s missing from sa.requests", seed, i, key)
+ priority := int32(rng.Intn(11) - 5)
//nolint:gosec // bounded to -5..5, no overflow
+ nextCreationTime++
+ // keep the placeholder/task-group shape of the
ask being replaced: production
+ // re-sends the same pod's ask, it never turns
a placeholder into a regular ask.
+ replacement := newFuzzAsk(key, appID,
existing.GetTaskGroup(), res, existing.IsPlaceholder(), priority, "",
nextCreationTime)
+ if replaceErr :=
app.AddAllocationAsk(replacement); replaceErr == nil {
+ // the replacement carries its OWN
priority: the key stays pending but the model
+ // must account for it at the new
priority from here on.
+ keyPriority[key] = priority
+ replacedAsks++
+ }
+ }
+ }
+
+ distinctPriorities := assertFuzzInvariants(t, app, keyPriority,
pendingKeys, seed, i)
+
+ if len(pendingKeys) > maxPending {
+ maxPending = len(pendingKeys)
+ }
+ if len(allocatedKeys) > maxAllocated {
+ maxAllocated = len(allocatedKeys)
+ }
+ if distinctPriorities > maxDistinctPendingPriorities {
+ maxDistinctPendingPriorities = distinctPriorities
+ }
+ }
+
+ // Guard against this fuzz run passing trivially (e.g. because a change
elsewhere caused every
+ // operation to be rejected/no-op'd): confirm it actually drove the
application through
+ // non-trivial pending/allocated/multi-priority states, so a real
regression in the product's
+ // bookkeeping (e.g. removeFromPriorities) has states to be caught in.
+ t.Logf("seed=%d coverage: maxPending=%d maxAllocated=%d
maxDistinctPendingPriorities=%d successfulRollbacks=%d ghostRollbacks=%d
replacedAsks=%d", seed, maxPending, maxAllocated, maxDistinctPendingPriorities,
successfulRollbacks, ghostRollbacks, replacedAsks)
+ assert.Assert(t, maxPending > 0, "seed=%d: fuzz run never observed any
pending asks", seed)
+ assert.Assert(t, maxAllocated > 0, "seed=%d: fuzz run never observed
any allocated asks", seed)
+ assert.Assert(t, maxDistinctPendingPriorities > 1, "seed=%d: fuzz run
never observed a multi-priority pending histogram", seed)
+ // RollbackAllocation is rejected outright unless the app is Accepted
or Running, so a change that
+ // leaves the app parked in some other state (or that stops case 6
producing confirmed
+ // allocations) would silently turn case 7 into a no-op and stop
covering deallocateAsk's new
+ // caller entirely, while every assertion above kept passing.
+ assert.Assert(t, successfulRollbacks > 0, "seed=%d: fuzz run never
completed a RollbackAllocation", seed)
+ // The replace-existing-ask branch (case 8) must not double count the
key in the histogram. It is
+ // only reached while pendingKeys is non-empty, so assert it really
happened rather than trusting
+ // that condition to keep holding. The ghost-rollback count is returned
instead of asserted here:
+ // it is too rare to demand per seed, see
TestApplicationPropertyFuzzHistogram.
+ assert.Assert(t, replacedAsks > 0, "seed=%d: fuzz run never took the
replace-existing-ask branch", seed)
+
+ return ghostRollbacks
+}
+
+// newFuzzAsk builds an Allocation directly from an si.Allocation (bypassing
the shared
+// newAllocationAsk* test helpers in utilities_test.go, which leave the
CreationTime tag unset and
+// so fall back to time.Now() in NewAllocationFromSI). creationTime is a
caller-supplied, strictly
+// increasing logical clock value (see nextCreationTime in runPropertyFuzz)
encoded into the
+// si.AllocationTags[siCommon.CreationTime] tag, which NewAllocationFromSI
parses back out as the
+// Allocation's createTime - making construction order (and therefore
sortedRequests tie-breaking)
+// fully deterministic for a fixed seed instead of depending on wall-clock
time.
+func newFuzzAsk(key, appID, taskGroup string, res *resources.Resource,
placeholder bool, priority int32, nodeID string, creationTime int64)
*Allocation {
+ alloc := &si.Allocation{
+ AllocationKey: key,
+ ApplicationID: appID,
+ PartitionName: "default",
+ ResourcePerAlloc: res.ToProto(),
+ TaskGroupName: taskGroup,
+ Placeholder: placeholder,
+ Priority: priority,
+ NodeID: nodeID,
+ AllocationTags: map[string]string{siCommon.CreationTime:
strconv.FormatInt(creationTime, 10)},
+ }
+ return NewAllocationFromSI(alloc)
+}
+
+// pickRandomKey returns a random key from a non-empty set of keys
(map[string]bool).
+func pickRandomKey(rng *rand.Rand, keys map[string]bool) string {
+ list := make([]string, 0, len(keys))
+ for k := range keys {
+ list = append(list, k)
+ }
+ // sort for determinism: ranging a map yields a randomized order, so
without this the same
+ // rng seed would not reproduce the same operation sequence, defeating
seed-based replay.
+ sort.Strings(list)
+ return list[rng.Intn(len(list))]
+}
+
+// pickRandomFilteredKey returns a random key from keys for which keep()
returns true, or ok=false if
+// none qualifies. It exists because the AddAllocation/RollbackAllocation
operations are only defined
+// on a subset of a tracking set (an allocated key that is not confirmed yet,
a confirmed key that is
+// still allocated); picking blindly and then dropping the step would make
those operations fire far
+// less often than their share of the step budget suggests.
+// The candidate list is sorted before indexing for the same determinism
reason as pickRandomKey.
+func pickRandomFilteredKey(rng *rand.Rand, keys map[string]bool, keep
func(string) bool) (string, bool) {
+ list := make([]string, 0, len(keys))
+ for k := range keys {
+ if keep(k) {
+ list = append(list, k)
+ }
+ }
+ if len(list) == 0 {
+ return "", false
+ }
+ sort.Strings(list)
+ return list[rng.Intn(len(list))], true
+}
+
+// pickRandomExistingKey returns a random key from keyPriority (any key ever
created that hasn't
+// been fully removed yet), or ok=false if there are none.
+func pickRandomExistingKey(rng *rand.Rand, keyPriority map[string]int32)
(string, bool) {
+ if len(keyPriority) == 0 {
+ return "", false
+ }
+ list := make([]string, 0, len(keyPriority))
+ for k := range keyPriority {
+ list = append(list, k)
+ }
+ // sort for determinism: ranging a map yields a randomized order (Go
randomizes map iteration
+ // order per-process), so without this the same rng seed would pick a
different key here on
+ // different runs, cascading into a different overall operation
sequence (which ask gets
+ // removed determines the FSM state transitions that follow it) and
defeating seed-based replay
+ // - this was the residual nondeterminism source alongside
pickRandomKey above.
+ sort.Strings(list)
+ return list[rng.Intn(len(list))], true
+}
+
+// assertFuzzInvariants rebuilds the expected pending-ask histogram and max
from the reference model
+// (keyPriority + pendingKeys) and compares it against the application's
incrementally maintained
+// state. On any mismatch it fails with the seed and step number embedded in
the message so the
+// failure is reproducible via `go test -run .../seed-<seed>`.
+// It returns the number of distinct pending priorities observed this step, so
the caller can track
+// coverage high-water marks without a second pass over pendingKeys.
+func assertFuzzInvariants(t *testing.T, app *Application, keyPriority
map[string]int32, pendingKeys map[string]bool, seed int64, step int)
(distinctPendingPriorities int) {
+ t.Helper()
+
+ wantHistogram := make(map[int32]int)
+ wantMax := configs.MinPriority
+ for k := range pendingKeys {
+ p := keyPriority[k]
+ wantHistogram[p]++
+ if p > wantMax {
+ wantMax = p
+ }
+ }
+
+ app.RLock()
+ gotMax := app.askMaxPriority
+ gotHistogram := make(map[int32]int, len(app.pendingPriorities))
+ for p, c := range app.pendingPriorities {
+ gotHistogram[p] = c
+ }
+ app.RUnlock()
+
+ assert.Equal(t, gotMax, wantMax, "seed=%d step=%d: askMaxPriority
mismatch", seed, step)
+ assert.Equal(t, len(gotHistogram), len(wantHistogram), "seed=%d
step=%d: pendingPriorities histogram size mismatch, got=%v want=%v", seed,
step, gotHistogram, wantHistogram)
+ for p, wantCount := range wantHistogram {
+ assert.Equal(t, gotHistogram[p], wantCount, "seed=%d step=%d:
pendingPriorities[%d] mismatch, got histogram=%v want histogram=%v", seed,
step, p, gotHistogram, wantHistogram)
+ }
+
+ return len(wantHistogram)
+}
diff --git a/pkg/scheduler/objects/application_test.go
b/pkg/scheduler/objects/application_test.go
index 70db068f..27732e04 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -2646,6 +2646,30 @@ func
TestTryAllocatePreemptNodeWithReservationsNotPossibleToCancel(t *testing.T)
assert.Assert(t, allocs[1].IsPreempted(), "alloc2 should have been
preempted")
}
+// assertMaxPriorityConsistent recomputes askMaxPriority (and the
pendingPriorities histogram it is
+// derived from) via a full scan over sa.requests and compares against the
incrementally maintained
+// values. This guards the incremental accounting that replaced the old O(N^2)
+// updateAskMaxPriority rescan.
+func assertMaxPriorityConsistent(t *testing.T, app *Application) {
+ t.Helper()
+ app.RLock()
+ defer app.RUnlock()
+ wantMax := configs.MinPriority
+ wantHistogram := make(map[int32]int)
+ for _, req := range app.requests {
+ if req.IsAllocated() {
+ continue
+ }
+ wantMax = max(wantMax, req.GetPriority())
+ wantHistogram[req.GetPriority()]++
+ }
+ assert.Equal(t, app.askMaxPriority, wantMax, "askMaxPriority
inconsistent with full scan over requests")
+ assert.Equal(t, len(app.pendingPriorities), len(wantHistogram),
"pendingPriorities histogram size mismatch")
+ for p, count := range wantHistogram {
+ assert.Equal(t, app.pendingPriorities[p], count,
"pendingPriorities count mismatch for priority %d", p)
+ }
+}
+
func TestMaxAskPriority(t *testing.T) {
app := newApplication(appID1, "default", "root.unknown")
if app == nil || app.ApplicationID != appID1 {
@@ -2660,32 +2684,38 @@ func TestMaxAskPriority(t *testing.T) {
// initial state
assert.Equal(t, app.GetAskMaxPriority(), configs.MinPriority, "wrong
default priority")
+ assertMaxPriorityConsistent(t, app)
// p=10 added
ask1 := newAllocationAskPriority("prio-10", appID1, res, 10)
err = app.AddAllocationAsk(ask1)
assert.NilError(t, err, "ask should have been updated on app")
assert.Equal(t, app.GetAskMaxPriority(), int32(10), "wrong priority
after adding p=10")
+ assertMaxPriorityConsistent(t, app)
// p=5 added
ask2 := newAllocationAskPriority("prio-5", appID1, res, 5)
err = app.AddAllocationAsk(ask2)
assert.NilError(t, err, "ask should have been added to app")
assert.Equal(t, app.GetAskMaxPriority(), int32(10), "wrong priority
after adding p=5")
+ assertMaxPriorityConsistent(t, app)
// p=15 added
ask3 := newAllocationAskPriority("prio-15", appID1, res, 15)
err = app.AddAllocationAsk(ask3)
assert.NilError(t, err, "ask should have been added to app")
assert.Equal(t, app.GetAskMaxPriority(), int32(15), "wrong priority
after adding p=15")
+ assertMaxPriorityConsistent(t, app)
// p=10 removed
app.RemoveAllocationAsk(ask1.GetAllocationKey())
assert.Equal(t, app.GetAskMaxPriority(), int32(15), "wrong priority
after removing p=10")
+ assertMaxPriorityConsistent(t, app)
// p=15 removed
app.RemoveAllocationAsk(ask3.GetAllocationKey())
assert.Equal(t, app.GetAskMaxPriority(), int32(5), "wrong priority
after removing p=15")
+ assertMaxPriorityConsistent(t, app)
// re-add removed asks
err = app.AddAllocationAsk(ask1)
@@ -2694,26 +2724,109 @@ func TestMaxAskPriority(t *testing.T) {
assert.NilError(t, err, "ask should have been added to app")
assert.Equal(t, app.GetAskMaxPriority(), int32(15), "wrong priority
after re-adding asks")
+ assertMaxPriorityConsistent(t, app)
// update to allocated for p=15
_, err = app.AllocateAsk(ask3.GetAllocationKey())
assert.NilError(t, err, "ask should have been updated")
assert.Equal(t, app.GetAskMaxPriority(), int32(10), "wrong priority
after updating p=15 to allocated")
+ assertMaxPriorityConsistent(t, app)
// update to allocated for p=5
_, err = app.AllocateAsk(ask2.GetAllocationKey())
assert.NilError(t, err, "ask should have been updated")
assert.Equal(t, app.GetAskMaxPriority(), int32(10), "wrong priority
after updating p=5 to allocated")
+ assertMaxPriorityConsistent(t, app)
// update to unallocated for p=5
_, err = app.DeallocateAsk(ask2.GetAllocationKey())
assert.NilError(t, err, "ask should have been updated")
assert.Equal(t, app.GetAskMaxPriority(), int32(10), "wrong priority
after updating p=5 to unallocated")
+ assertMaxPriorityConsistent(t, app)
// update to unallocated for p=15
_, err = app.DeallocateAsk(ask3.GetAllocationKey())
assert.NilError(t, err, "ask should have been updated")
assert.Equal(t, app.GetAskMaxPriority(), int32(15), "wrong priority
after updating p=15 to unallocated")
+ assertMaxPriorityConsistent(t, app)
+}
+
+// TestAddAllocationAskReplaceExistingPendingAsk covers the
replace-existing-ask branch of
+// AddAllocationAsk: re-submitting an ask under a key that is already tracked
and still pending.
+// The displaced ask has to leave the pending histogram before
addAllocationAskInternal counts the
+// replacement, or a single allocation key is counted twice and the priority
it was originally
+// submitted at never drops out again. The full rescan this replaced could not
get that wrong: it
+// derived the maximum from sa.requests, which only ever holds one ask per key.
+func TestAddAllocationAskReplaceExistingPendingAsk(t *testing.T) {
+ app := newApplication(appID1, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ err = app.AddAllocationAsk(newAllocationAskPriority(aKey, appID1, res,
5))
+ assert.NilError(t, err, "ask should have been added to app")
+ assertMaxPriorityConsistent(t, app)
+
+ // same key, still pending, different priority: the replace branch
+ replacement := newAllocationAskPriority(aKey, appID1, res, 3)
+ err = app.AddAllocationAsk(replacement)
+ assert.NilError(t, err, "ask should have been updated on app")
+
+ app.RLock()
+ assert.Equal(t, len(app.pendingPriorities), 1, "pending histogram must
only hold the replacement's priority")
+ assert.Equal(t, app.pendingPriorities[3], 1, "wrong pending count for
the replacement priority")
+ app.RUnlock()
+ assert.Equal(t, app.GetAskMaxPriority(), int32(3), "wrong priority
after replacing p=5 with p=3")
+ assertMaxPriorityConsistent(t, app)
+
+ // allocating the only ask must empty the histogram: a double counted
key would leave the
+ // replaced ask's priority behind.
+ _, err = app.AllocateAsk(aKey)
+ assert.NilError(t, err, "ask should have been allocated")
+ app.RLock()
+ assert.Equal(t, len(app.pendingPriorities), 0, "allocating the only ask
must empty the pending histogram")
+ app.RUnlock()
+ assert.Equal(t, app.GetAskMaxPriority(), configs.MinPriority, "wrong
priority after allocating the only ask")
+ assertMaxPriorityConsistent(t, app)
+}
+
+// TestRollbackAllocationAskNotTracked covers deallocateAsk running for an ask
that sa.requests no
+// longer holds. removeAsksInternal("") wipes sa.requests and the pending
histogram but deliberately
+// leaves sa.allocations alone until the shim confirms the releases, so a
SCHEDULING_FAILED_ON_RM
+// release arriving in that window reaches RollbackAllocation, which looks the
entry up in
+// sa.allocations and deallocates it. deallocateAsk must not count that ask as
pending again: the
+// application does not track it any more, so nothing would ever take it back
out of the histogram.
+func TestRollbackAllocationAskNotTracked(t *testing.T) {
+ setupUGM()
+ defer setupUGM()
+ app := newApplication(appID1, "default", "root.default")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ ask := newAllocationAskPriority(aKey, appID1, res, 5)
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "ask should have been added to app")
+ _, err = app.AllocateAsk(aKey)
+ assert.NilError(t, err, "ask should have been allocated")
+ // confirm the allocation so it lands in sa.allocations, which is what
RollbackAllocation looks in
+ app.AddAllocation(ask)
+
+ // wipe the asks: sa.allocations (and so the rollback target) survives
this
+ app.RemoveAllocationAsk("")
+ assert.Assert(t, app.GetAllocationAsk(aKey) == nil, "ask should no
longer be tracked in requests")
+ assert.Assert(t, app.IsAccepted() || app.IsRunning(), "app must still
be rollback-eligible, is %s", app.CurrentState())
+
+ _, err = app.RollbackAllocation(aKey)
+ assert.NilError(t, err, "rollback of the confirmed allocation should
have succeeded")
+
+ app.RLock()
+ assert.Equal(t, len(app.pendingPriorities), 0, "rollback of an
untracked ask must not change the pending histogram")
+ app.RUnlock()
+ assert.Equal(t, app.GetAskMaxPriority(), configs.MinPriority, "rollback
of an untracked ask must not change askMaxPriority")
+ assertMaxPriorityConsistent(t, app)
}
func TestAskEvents(t *testing.T) {
@@ -3964,6 +4077,87 @@ func TestTryPlaceHolderAllocateDifferentNodes(t
*testing.T) {
assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
}
+// revertTriggerPredicatePlugin is a test-local predicate plugin whose
Predicates call has the side
+// effect of marking a specific placeholder as preempted, then returning nil
(i.e. the predicate
+// check itself still "passes"). This is used to reach the narrow window in
tryPlaceholderAllocate
+// between the loop's IsPreempted() guard and the ph.SetReleased(true) call,
so that SetReleased
+// fails and the revert path is exercised (see
TestTryPlaceHolderAllocateRevertsOnPreemptedPlaceholder).
+type revertTriggerPredicatePlugin struct {
+ mockCommon.ResourceManagerCallback
+ ph *Allocation
+}
+
+func (p *revertTriggerPredicatePlugin) Predicates(_ *si.PredicatesArgs) error {
+ _ = p.ph.MarkPreempted() //nolint:errcheck
+ return nil
+}
+
+// TestTryPlaceHolderAllocateRevertsOnPreemptedPlaceholder drives
tryPlaceholderAllocate down the
+// path where ph.SetReleased(true) fails because the placeholder became
preempted in between the
+// loop's own IsPreempted() guard and the SetReleased(true) call. This
exercises the FIRST revert
+// branch inside the phAllocs loop (injected via node.preReserveConditions):
sa.deallocateAsk(request),
+// ClearRelease() on both sides, and continue.
+//
+// Because ph is registered only at the Application level
(app.AddAllocation/addPlaceholderData) and
+// never actually consumes capacity on the *Node object itself (matching the
pattern of every other
+// TestTryPlaceHolderAllocate* test in this file, none of which call
node.AddAllocation(ph) either),
+// the node's tracked available resources are unaffected by ph and still show
room for reqFit. As a
+// result, once the first branch reverts and falls through to the fallback
ForEachNode retry
+// (phFit/reqFit, using node.preAllocateCheck/preAllocateConditions), that
retry also finds room,
+// proceeds to TryAddAllocation + allocateAsk + SetReleased(true) on the SAME
still-preempted ph, and
+// that SetReleased(true) call fails again for the same reason - so the
SECOND/fallback revert branch
+// (node.RemoveAllocation, deallocateAsk, ClearRelease on both sides) is
exercised too, verified by the
+// log line "allocation is already preempted, so not proceeding further and
reverting to old state"
+// appearing twice when run with -v. The function then returns nil with all
state reverted, which is
+// exactly what the assertions below check.
+func TestTryPlaceHolderAllocateRevertsOnPreemptedPlaceholder(t *testing.T) {
+ // node capacity equals the placeholder's (and the real ask's) resource
size; this keeps the
+ // scenario minimal (single node, single ph, single ask) while both
revert branches above still
+ // get exercised, since node-level capacity tracking is independent of
the ph's Application-level
+ // bookkeeping in this test setup.
+ node := newNode(nodeID1, map[string]resources.Quantity{"first": 5})
+ nodeMap := map[string]*Node{nodeID1: node}
+ iterator := getNodeIteratorFn(node)
+ getNode := func(nodeID string) *Node {
+ return nodeMap[nodeID]
+ }
+
+ app := newApplication(appID0, "default", "root.default")
+
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ res :=
resources.NewResourceFromMap(map[string]resources.Quantity{"first": 5})
+ ph := newPlaceholderAlloc(appID0, nodeID1, res, tg1)
+ app.AddAllocation(ph)
+ app.addPlaceholderData(ph)
+ assertPlaceholderData(t, app, tg1, 1, 0, 0, res)
+
+ // same size as the placeholder so we go down the main swap path, not
the "placeholder is
+ // larger, cancel it" branch.
+ ask := newAllocationAsk(aKey, appID0, res)
+ ask.taskGroupName = tg1
+ err = app.AddAllocationAsk(ask)
+ assert.NilError(t, err, "ask should have been added to app")
+
+ // register a plugin whose Predicates call marks ph preempted right
before SetReleased(true)
+ // is invoked, forcing that call to fail and the revert path to run.
+ plugin := &revertTriggerPredicatePlugin{ph: ph}
+ plugins.RegisterSchedulerPlugin(plugin)
+ defer plugins.UnregisterSchedulerPlugins()
+
+ result := app.tryPlaceholderAllocate(iterator, getNode)
+ assert.Assert(t, result == nil, "result should be nil: both the first
swap attempt and the fallback retry failed SetReleased(true) on the
still-preempted placeholder and reverted")
+ assert.Assert(t, ph.IsPreempted(), "placeholder should remain marked
preempted (side effect stuck)")
+ assert.Assert(t, !ph.IsReleased(), "placeholder should not be released:
the failing SetReleased(true) call never set it")
+ assert.Assert(t, !ask.IsAllocated(), "ask should have been reverted
back to pending by deallocateAsk")
+ assert.Assert(t, ask.GetRelease() == nil, "ask's release link should
have been cleared by the revert")
+ assert.Assert(t, ph.GetRelease() == nil, "placeholder's release link
should have been cleared by the revert")
+
+ assertMaxPriorityConsistent(t, app)
+}
+
func TestTryNodesNoReserve(t *testing.T) {
app := newApplication(appID0, "default", "root.default")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
