diff --git a/src/main/git/command-runner/git-admission-waiter-queue.ts b/src/main/git/command-runner/git-admission-waiter-queue.ts new file mode 100644 index 00000000000..1d49496497e --- /dev/null +++ b/src/main/git/command-runner/git-admission-waiter-queue.ts @@ -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> = { + general: { + interactive: createLane(), + status: createLane(), + background: createLane() + }, + network: { + interactive: createLane(), + status: createLane(), + background: createLane() + } + } + private readonly countsByBudget = new Map() + 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 + } +} diff --git a/src/main/git/command-runner/git-subprocess-admission.test.ts b/src/main/git/command-runner/git-subprocess-admission.test.ts index f734a9521b7..252b26df00b 100644 --- a/src/main/git/command-runner/git-subprocess-admission.test.ts +++ b/src/main/git/command-runner/git-subprocess-admission.test.ts @@ -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) diff --git a/src/main/git/command-runner/git-subprocess-admission.ts b/src/main/git/command-runner/git-subprocess-admission.ts index da17a758580..290894af0c2 100644 --- a/src/main/git/command-runner/git-subprocess-admission.ts +++ b/src/main/git/command-runner/git-subprocess-admission.ts @@ -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() - 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 } { - 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) }