feat(queuev2): integrate InMemQueue with cache.BudgetManager#8205
feat(queuev2): integrate InMemQueue with cache.BudgetManager#8205arzonus wants to merge 6 commits into
Conversation
… tracking Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>
Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>
Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>
… manager Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>
Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>
…ScheduledQueue Signed-off-by: Seva Kaloshin <seva.kaloshin@gmail.com>
arzonus
changed the title
Code Review
|
| Auto-apply | Compact |
|
|
Was this helpful? React with 👍 / 👎 | Gitar
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
| func (q *inMemQueueImpl) RTrimBySize() (persistence.HistoryTaskKey, bool) { | ||
| if len(q.tasks) == 0 { | ||
| return persistence.MinimumHistoryTaskKey, false | ||
| } | ||
|
|
||
| if maxSize <= 0 { | ||
| q.tasks = nil | ||
| return persistence.MinimumHistoryTaskKey, true | ||
| trimmed := false | ||
| // Use q.Len() (not UsedCount) because reserve may fail for tasks inserted while at capacity. | ||
| for len(q.tasks) > 0 && int64(len(q.tasks)) > q.budgetMgr.CapacityCount() { | ||
| q.tasks[len(q.tasks)-1] = nil // release GC ref | ||
| q.tasks = q.tasks[:len(q.tasks)-1] | ||
| _ = q.budgetMgr.ReleaseCountWithCallback(q.cacheID, func() (int64, error) { return 1, nil }) | ||
| trimmed = true | ||
| } | ||
|
|
||
| trimmed := len(q.tasks) > maxSize | ||
| if trimmed { | ||
| // Nil out the tail elements so the backing array releases references to | ||
| // the removed Task interface values. Without this, interface values beyond | ||
| // maxSize remain rooted in the array until it is reallocated. | ||
| for i := maxSize; i < len(q.tasks); i++ { | ||
| q.tasks[i] = nil | ||
| } | ||
| q.tasks = q.tasks[:maxSize] | ||
| if len(q.tasks) == 0 { | ||
| return persistence.MinimumHistoryTaskKey, trimmed | ||
| } |
There was a problem hiding this comment.
⚠️ Bug: RTrimBySize over-releases budget count for untracked tail tasks
In RTrimBySize (queue_mem.go:174-179), each trimmed task triggers ReleaseCountWithCallback(... return 1 ...), decrementing the budget manager's usedCount by one per removed task. However, the loop condition is len(q.tasks) > CapacityCount(), which can trim tasks that were never reserved.
Concretely (the exact scenario the new test 'RTrimBySize trims untracked task inserted when at capacity' sets up): capacity=3, 3 tasks reserved → usedCount=3, then a 4th putTask whose ReserveCountWithCallback fails and is rolled back → queue len=4, usedCount still 3. RTrimBySize sees len 4 > 3, trims the 4th (untracked) task and releases 1 → usedCount becomes 2 while the queue still holds 3 tracked tasks. The budget now undercounts by 1.
This drift is permanent and accumulates each time a task is inserted at capacity, eventually skewing UsedCount()/metrics low (and on a shared manager, can be driven toward 0/clamped) even though the cache is full. Note the new test asserts only q.Len() here and deliberately does NOT assert mgr.UsedCount(), masking the inconsistency.
Suggested fix: only release count for tasks that were actually reserved. Compute the release amount as the overlap between trimmed tasks and the reserved count, e.g. clamp released = min(tasksTrimmed, UsedCount()) for this cacheID, or reserve every appended task unconditionally (track real length) so used count always equals len. Either way, after RTrimBySize the invariant UsedCount == len(q.tasks) should hold.
Was this helpful? React with 👍 / 👎
| trimmed := false | ||
| // Use q.Len() (not UsedCount) because reserve may fail for tasks inserted while at capacity. | ||
| for len(q.tasks) > 0 && int64(len(q.tasks)) > q.budgetMgr.CapacityCount() { | ||
| q.tasks[len(q.tasks)-1] = nil // release GC ref | ||
| q.tasks = q.tasks[:len(q.tasks)-1] | ||
| _ = q.budgetMgr.ReleaseCountWithCallback(q.cacheID, func() (int64, error) { return 1, nil }) | ||
| trimmed = true | ||
| } |
There was a problem hiding this comment.
💡 Performance: RTrimBySize calls dynamic CapacityCount() and release callback per task
The trim loop in RTrimBySize (queue_mem.go:174-178) evaluates q.budgetMgr.CapacityCount() on every iteration (each invokes the maxCount dynamic-config IntPropertyFn) and issues a separate ReleaseCountWithCallback per removed task. When trimming a large tail (e.g. after a runtime capacity decrease of TimerProcessorCacheMaxSize), this is O(n) config lookups and O(n) budget-manager calls, despite commit 8036460 claiming a 'batch release'.
Suggested fix: read capacity := q.budgetMgr.CapacityCount() once before the loop, count the number of tasks removed, then issue a single ReleaseCountWithCallback(q.cacheID, func() (int64, error) { return removed, nil }) after the loop.
Hoist CapacityCount() out of the loop and batch the release into one call.:
capacity := q.budgetMgr.CapacityCount()
var removed int64
for len(q.tasks) > 0 && int64(len(q.tasks)) > capacity {
q.tasks[len(q.tasks)-1] = nil // release GC ref
q.tasks = q.tasks[:len(q.tasks)-1]
removed++
}
if removed > 0 {
_ = q.budgetMgr.ReleaseCountWithCallback(q.cacheID, func() (int64, error) { return removed, nil })
}
trimmed := removed > 0
- Apply fix
Check the box to apply the fix or reply for a change | Was this helpful? React with 👍 / 👎
| func newInMemQueueWithBudget(mgr cache.Manager, cacheID string) *inMemQueueImpl { | ||
| return &inMemQueueImpl{budgetMgr: mgr, cacheID: cacheID} | ||
| } |
There was a problem hiding this comment.
💡 Bug: inMemQueueImpl methods nil-deref if constructed without budget manager
All mutating methods now unconditionally call q.budgetMgr.* (e.g. queue_mem.go:76,89,162,177,190). newInMemQueueWithBudget is the only constructor and the timer factory always passes a non-nil manager, so today this is safe. However, unlike cachedScheduledQueue which explicitly treats its budgetMgr as optional ('optional; stopped in Stop() if non-nil'), inMemQueueImpl has no nil guard. A future caller passing nil (mirroring the newCachedScheduledQueue(..., nil) pattern used throughout the tests) would panic with a nil pointer dereference.
Suggested fix: either document/enforce that mgr must be non-nil in newInMemQueueWithBudget (e.g. panic with a clear message on nil), or guard each q.budgetMgr call with a nil check so the queue degrades to untracked behavior.
Was this helpful? React with 👍 / 👎
1 participant
What changed?
Integrated
inMemQueueImpl(the in-memory task cache backingCachedQueueReader) with the existingcache.BudgetManager. Each shard'sInMemQueuenow tracks its task count via a dedicated per-shardcache.BudgetManagerinstance. ChangedRTrimBySizeto drop its staticmaxSize intparameter — eviction is now driven by the budget manager's capacity, backed by the dynamicTimerProcessorCacheMaxSizeconfig property.Why?
Previously,
CachedQueueReaderenforced its size limit by passing a staticMaxSizesnapshot toRTrimBySizeon every insert. Withcache.BudgetManager:BudgetManagerUsedCount,BudgetManagerCapacityCount, etc.) via the existing budget manager metric framework.TimerProcessorCacheMaxSizeis consumed as a liveIntPropertyFn, so runtime config changes take effect immediately without restart.RTrimBySizetrims based onq.Len() > CapacityCount()(actual queue length vs. budget capacity) rather thanUsedCount(), which correctly handles tasks inserted while at capacity (whose optimistic reserve was rolled back but whose insertion was not skipped, per design).How did you test it?
Unit tests:
go test -race ./service/history/queuev2/...New
TestInMemQueueBudgettable-driven test covers:PutTasksreserves count for inserted tasks; dedup does not double-reserve.RTrimBySizetrims to capacity and releases count; no-op when within capacity.RTrimBySizecorrectly trims untracked tasks inserted when at capacity.LTrimbatch-releases count for removed tasks.Clearreleases all count.RTrimBySizeon one corrects the combined total.Full pre-PR pipeline:
Potential risks
TimerProcessorCacheMaxSizesemantics shift: previously enforced per-shard at construction time; now re-evaluated dynamically on everyRTrimBySizecall. Runtime decreases will cause the next eviction cycle to trim down to the new limit.cachedScheduledQueuelifecycle (per shard, perTimerProcessorEnableCachedScheduledQueuefeature flag). No budget manager is created when the flag is off.newCachedScheduledQueuesignature gains abudgetMgr cache.Managerparameter — callers in tests must passnilwhen no budget is needed.Release notes
N/A — internal implementation change to timer queue cache size enforcement. No API, IDL, or schema changes.
Documentation Changes
N/A