mirror of
https://github.com/stablyai/orca.git
synced 2026-10-25 08:04:05 +00:00
perf(git): keep admission queue drains linear
This commit is contained in:
@@ -0,0 +1,132 @@
|
||||
import type { AdmissionClass, AdmissionSlotKind, AdmissionWaiter } from './git-admission-state'
|
||||
import type { GitAdmissionTier } from './git-exec-options'
|
||||
|
||||
type SelectedWaiter = {
|
||||
waiter: AdmissionWaiter
|
||||
slotKind: AdmissionSlotKind
|
||||
}
|
||||
|
||||
type WaiterLane = {
|
||||
items: AdmissionWaiter[]
|
||||
head: number
|
||||
count: number
|
||||
}
|
||||
|
||||
const TIERS = ['interactive', 'status', 'background'] as const
|
||||
const createLane = (): WaiterLane => ({ items: [], head: 0, count: 0 })
|
||||
|
||||
export class GitAdmissionWaiterQueue {
|
||||
private readonly lanes: Record<AdmissionClass, Record<GitAdmissionTier, WaiterLane>> = {
|
||||
general: {
|
||||
interactive: createLane(),
|
||||
status: createLane(),
|
||||
background: createLane()
|
||||
},
|
||||
network: {
|
||||
interactive: createLane(),
|
||||
status: createLane(),
|
||||
background: createLane()
|
||||
}
|
||||
}
|
||||
private readonly countsByBudget = new Map<string, number>()
|
||||
private totalCount = 0
|
||||
|
||||
get count(): number {
|
||||
return this.totalCount
|
||||
}
|
||||
|
||||
enqueue(waiter: AdmissionWaiter): void {
|
||||
const lane = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
lane.items.push(waiter)
|
||||
lane.count += 1
|
||||
this.totalCount += 1
|
||||
for (const key of waiter.budgetKeys) {
|
||||
this.countsByBudget.set(key, (this.countsByBudget.get(key) ?? 0) + 1)
|
||||
}
|
||||
}
|
||||
|
||||
dequeue(waiter: AdmissionWaiter): void {
|
||||
const lane = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
lane.count -= 1
|
||||
this.totalCount -= 1
|
||||
for (const key of waiter.budgetKeys) {
|
||||
const current = this.countsByBudget.get(key)
|
||||
if (current === 1) {
|
||||
this.countsByBudget.delete(key)
|
||||
} else if (current !== undefined) {
|
||||
this.countsByBudget.set(key, current - 1)
|
||||
}
|
||||
}
|
||||
this.compact(lane)
|
||||
}
|
||||
|
||||
hasBudget(key: string): boolean {
|
||||
return this.countsByBudget.has(key)
|
||||
}
|
||||
|
||||
snapshot(): AdmissionWaiter[] {
|
||||
return (['general', 'network'] as const)
|
||||
.flatMap((admissionClass) =>
|
||||
TIERS.flatMap((tier) => {
|
||||
const lane = this.lanes[admissionClass][tier]
|
||||
return lane.items.slice(lane.head).filter((waiter) => waiter.state === 'queued')
|
||||
})
|
||||
)
|
||||
.sort((left, right) => left.id - right.id)
|
||||
}
|
||||
|
||||
nextFitting(
|
||||
admissionClass: AdmissionClass,
|
||||
effectiveTier: (waiter: AdmissionWaiter) => number,
|
||||
slotKindFor: (waiter: AdmissionWaiter) => AdmissionSlotKind | null,
|
||||
abort: (waiter: AdmissionWaiter) => void
|
||||
): SelectedWaiter | null {
|
||||
let selected: (SelectedWaiter & { tier: number }) | null = null
|
||||
for (const rawTier of TIERS) {
|
||||
const lane = this.lanes[admissionClass][rawTier]
|
||||
for (let index = lane.head; index < lane.items.length; index += 1) {
|
||||
const waiter = lane.items[index]
|
||||
if (waiter.state !== 'queued') {
|
||||
continue
|
||||
}
|
||||
if (waiter.signal?.aborted) {
|
||||
abort(waiter)
|
||||
continue
|
||||
}
|
||||
const slotKind = slotKindFor(waiter)
|
||||
if (!slotKind) {
|
||||
continue
|
||||
}
|
||||
const tier = effectiveTier(waiter)
|
||||
if (
|
||||
!selected ||
|
||||
tier < selected.tier ||
|
||||
(tier === selected.tier && waiter.id < selected.waiter.id)
|
||||
) {
|
||||
selected = { waiter, slotKind, tier }
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
return selected
|
||||
}
|
||||
|
||||
private compact(lane: WaiterLane): void {
|
||||
let head = lane.head
|
||||
while (head < lane.items.length && lane.items[head].state !== 'queued') {
|
||||
head += 1
|
||||
}
|
||||
lane.head = head
|
||||
if (lane.count === 0) {
|
||||
lane.items.length = 0
|
||||
lane.head = 0
|
||||
return
|
||||
}
|
||||
const tombstones = lane.items.length - head - lane.count
|
||||
if (head < 256 && (lane.items.length < 256 || tombstones <= lane.count)) {
|
||||
return
|
||||
}
|
||||
lane.items = lane.items.slice(head).filter((candidate) => candidate.state === 'queued')
|
||||
lane.head = 0
|
||||
}
|
||||
}
|
||||
@@ -349,6 +349,26 @@ describe('GitAdmissionScheduler', () => {
|
||||
expect(order).toEqual(Array.from({ length: 20 }, (_, index) => index))
|
||||
})
|
||||
|
||||
it('drains a large saturated FIFO burst without retaining settled waiters', async () => {
|
||||
const scheduler = new GitAdmissionScheduler({ generalCap: 1, generalHeadroom: 0 })
|
||||
const running = await scheduler.acquire(local('status'))
|
||||
const count = 4_000
|
||||
const order: number[] = []
|
||||
const queued = Array.from({ length: count }, (_, index) =>
|
||||
scheduler.acquire(local('background')).then((grant) => {
|
||||
order.push(index)
|
||||
grant.release()
|
||||
})
|
||||
)
|
||||
|
||||
expect(scheduler.snapshot().queued).toBe(count)
|
||||
running.release()
|
||||
await Promise.all(queued)
|
||||
|
||||
expect(order).toEqual(Array.from({ length: count }, (_, index) => index))
|
||||
expect(scheduler.snapshot()).toMatchObject({ queued: 0, queuedWaiters: [] })
|
||||
})
|
||||
|
||||
it('captures killswitch state in each release closure', async () => {
|
||||
const scheduler = schedulerWithOneSlot()
|
||||
_resetGitAdmissionForTests(scheduler)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { uncRouteKey } from '../../providers/working-directory-validation'
|
||||
import { classifyGitCommand } from '../wsl-direct-git-read-commands'
|
||||
import { createAbortError } from './abort-error'
|
||||
import { GitAdmissionWaiterQueue } from './git-admission-waiter-queue'
|
||||
import {
|
||||
ADMISSION_TIER_VALUE,
|
||||
AdmissionEventPublisher,
|
||||
@@ -42,7 +43,7 @@ function routeKey(request: GitAdmissionRequest): string | null {
|
||||
export class GitAdmissionScheduler {
|
||||
private readonly config: AdmissionSchedulerConfig
|
||||
private readonly budgets = new Map<string, AdmissionBudget>()
|
||||
private readonly waiters: AdmissionWaiter[] = []
|
||||
private readonly waiters = new GitAdmissionWaiterQueue()
|
||||
private nextWaiterId = 0
|
||||
private readonly eventPublisher: AdmissionEventPublisher
|
||||
|
||||
@@ -72,13 +73,17 @@ export class GitAdmissionScheduler {
|
||||
reject,
|
||||
onAbort: () => this.abort(waiter)
|
||||
}
|
||||
this.waiters.enqueue(waiter)
|
||||
request.signal?.addEventListener('abort', waiter.onAbort, { once: true })
|
||||
if (request.signal?.aborted) {
|
||||
this.abort(waiter)
|
||||
return
|
||||
}
|
||||
this.waiters.push(waiter)
|
||||
this.drain()
|
||||
// Adding a blocked waiter cannot make an older waiter runnable. Avoid a
|
||||
// queue scan for every arrival while the fixed-size budget is saturated.
|
||||
if (this.slotKindFor(waiter)) {
|
||||
this.drain(admissionClass)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -87,9 +92,9 @@ export class GitAdmissionScheduler {
|
||||
queuedWaiters: { id: number; args: readonly string[]; tier: AdmissionWaiter['tier'] }[]
|
||||
budgets: Record<string, { baseUsed: number; headroomUsed: number }>
|
||||
} {
|
||||
const queuedWaiters = this.waiters.filter((waiter) => waiter.state === 'queued')
|
||||
const queuedWaiters = this.waiters.snapshot()
|
||||
return {
|
||||
queued: queuedWaiters.length,
|
||||
queued: this.waiters.count,
|
||||
queuedWaiters: queuedWaiters.map(({ id, args, tier }) => ({ id, args, tier })),
|
||||
budgets: Object.fromEntries(
|
||||
[...this.budgets].map(([key, budget]) => [
|
||||
@@ -156,35 +161,28 @@ export class GitAdmissionScheduler {
|
||||
})
|
||||
}
|
||||
|
||||
private drain(): void {
|
||||
let granted = true
|
||||
while (granted) {
|
||||
granted = false
|
||||
private slotKindFor(waiter: AdmissionWaiter): AdmissionSlotKind | null {
|
||||
return this.fits(waiter, 'base')
|
||||
? 'base'
|
||||
: waiter.tier === 'interactive' && this.fits(waiter, 'headroom')
|
||||
? 'headroom'
|
||||
: null
|
||||
}
|
||||
|
||||
private drain(admissionClass: AdmissionClass): void {
|
||||
while (true) {
|
||||
const now = this.config.now()
|
||||
const ordered = this.waiters
|
||||
.filter((waiter) => waiter.state === 'queued')
|
||||
.sort(
|
||||
(left, right) =>
|
||||
this.effectiveTier(left, now) - this.effectiveTier(right, now) || left.id - right.id
|
||||
)
|
||||
for (const waiter of ordered) {
|
||||
if (waiter.signal?.aborted) {
|
||||
this.abort(waiter)
|
||||
continue
|
||||
}
|
||||
const slotKind = this.fits(waiter, 'base')
|
||||
? 'base'
|
||||
: waiter.tier === 'interactive' && this.fits(waiter, 'headroom')
|
||||
? 'headroom'
|
||||
: null
|
||||
if (!slotKind) {
|
||||
continue
|
||||
}
|
||||
this.grant(waiter, slotKind, now)
|
||||
granted = true
|
||||
const selected = this.waiters.nextFitting(
|
||||
admissionClass,
|
||||
(waiter) => this.effectiveTier(waiter, now),
|
||||
(waiter) => this.slotKindFor(waiter),
|
||||
(waiter) => this.abort(waiter)
|
||||
)
|
||||
if (!selected) {
|
||||
return
|
||||
}
|
||||
this.grant(selected.waiter, selected.slotKind, now)
|
||||
}
|
||||
this.pruneBudgets()
|
||||
}
|
||||
|
||||
private grant(waiter: AdmissionWaiter, slotKind: AdmissionSlotKind, now: number): void {
|
||||
@@ -198,6 +196,7 @@ export class GitAdmissionScheduler {
|
||||
budget.headroomUsed += 1
|
||||
}
|
||||
}
|
||||
this.waiters.dequeue(waiter)
|
||||
const queueWaitMs = Math.max(0, now - waiter.enqueuedAt)
|
||||
this.publishEvent(waiter, slotKind, 'grant', queueWaitMs)
|
||||
queueMicrotask(() => {
|
||||
@@ -210,7 +209,6 @@ export class GitAdmissionScheduler {
|
||||
queueWaitMs,
|
||||
release: this.releaseOnce(waiter, slotKind, queueWaitMs)
|
||||
})
|
||||
this.removeWaiter(waiter)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -234,7 +232,8 @@ export class GitAdmissionScheduler {
|
||||
}
|
||||
}
|
||||
this.publishEvent(waiter, slotKind, 'release', queueWaitMs)
|
||||
this.drain()
|
||||
this.pruneRouteBudgets(waiter.budgetKeys)
|
||||
this.drain(waiter.admissionClass)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -249,18 +248,14 @@ export class GitAdmissionScheduler {
|
||||
Math.max(0, this.config.now() - waiter.enqueuedAt)
|
||||
)()
|
||||
}
|
||||
const wasQueued = waiter.state === 'queued'
|
||||
waiter.state = 'settled'
|
||||
waiter.signal?.removeEventListener('abort', waiter.onAbort)
|
||||
this.removeWaiter(waiter)
|
||||
waiter.reject(createAbortError())
|
||||
this.drain()
|
||||
}
|
||||
|
||||
private removeWaiter(waiter: AdmissionWaiter): void {
|
||||
const index = this.waiters.indexOf(waiter)
|
||||
if (index !== -1) {
|
||||
this.waiters.splice(index, 1)
|
||||
if (wasQueued) {
|
||||
this.waiters.dequeue(waiter)
|
||||
this.pruneRouteBudgets(waiter.budgetKeys)
|
||||
}
|
||||
waiter.reject(createAbortError())
|
||||
}
|
||||
|
||||
private publishEvent(
|
||||
@@ -274,21 +269,20 @@ export class GitAdmissionScheduler {
|
||||
waiter,
|
||||
slotKind,
|
||||
queueWaitMs,
|
||||
queued: this.waiters.filter((candidate) => candidate.state === 'queued').length,
|
||||
queued: this.waiters.count,
|
||||
budgets: this.budgets
|
||||
})
|
||||
}
|
||||
|
||||
private pruneBudgets(): void {
|
||||
const queuedKeys = new Set(
|
||||
this.waiters.flatMap((waiter) => (waiter.state === 'queued' ? waiter.budgetKeys : []))
|
||||
)
|
||||
for (const [key, budget] of this.budgets) {
|
||||
private pruneRouteBudgets(keys: readonly string[]): void {
|
||||
for (const key of keys) {
|
||||
const budget = this.budgets.get(key)
|
||||
if (
|
||||
budget &&
|
||||
key.startsWith('route:') &&
|
||||
budget.baseUsed === 0 &&
|
||||
budget.headroomUsed === 0 &&
|
||||
!queuedKeys.has(key)
|
||||
!this.waiters.hasBudget(key)
|
||||
) {
|
||||
this.budgets.delete(key)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user