mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
perf(git): bound eligible route selection
This commit is contained in:
@@ -10,16 +10,78 @@ type WaiterLane = {
|
||||
items: AdmissionWaiter[]
|
||||
head: number
|
||||
count: number
|
||||
baseEligible: boolean
|
||||
headroomEligible: boolean
|
||||
version: number
|
||||
}
|
||||
|
||||
const TIERS = ['interactive', 'status', 'background'] as const
|
||||
const createLane = (): WaiterLane => ({ items: [], head: 0, count: 0 })
|
||||
const createLane = (): WaiterLane => ({
|
||||
items: [],
|
||||
head: 0,
|
||||
count: 0,
|
||||
baseEligible: false,
|
||||
headroomEligible: false,
|
||||
version: 0
|
||||
})
|
||||
type TierLanes = Record<GitAdmissionTier, Map<string | null, WaiterLane>>
|
||||
const createTierLanes = (): TierLanes => ({
|
||||
interactive: new Map(),
|
||||
status: new Map(),
|
||||
background: new Map()
|
||||
})
|
||||
type Candidate = { lane: WaiterLane; waiter: AdmissionWaiter; version: number }
|
||||
|
||||
class CandidateHeap {
|
||||
private readonly items: Candidate[] = []
|
||||
|
||||
push(candidate: Candidate): void {
|
||||
this.items.push(candidate)
|
||||
let index = this.items.length - 1
|
||||
while (index > 0) {
|
||||
const parent = Math.floor((index - 1) / 2)
|
||||
if (this.items[parent].waiter.id <= candidate.waiter.id) {
|
||||
break
|
||||
}
|
||||
this.items[index] = this.items[parent]
|
||||
index = parent
|
||||
}
|
||||
this.items[index] = candidate
|
||||
}
|
||||
|
||||
peek(valid: (candidate: Candidate) => boolean): Candidate | null {
|
||||
// Eligibility changes invalidate by version so route updates stay O(log N).
|
||||
while (this.items.length > 0 && !valid(this.items[0])) {
|
||||
this.pop()
|
||||
}
|
||||
return this.items[0] ?? null
|
||||
}
|
||||
|
||||
private pop(): void {
|
||||
const last = this.items.pop()
|
||||
if (!last || this.items.length === 0) {
|
||||
return
|
||||
}
|
||||
let index = 0
|
||||
while (true) {
|
||||
const left = index * 2 + 1
|
||||
if (left >= this.items.length) {
|
||||
break
|
||||
}
|
||||
const right = left + 1
|
||||
const child =
|
||||
right < this.items.length && this.items[right].waiter.id < this.items[left].waiter.id
|
||||
? right
|
||||
: left
|
||||
if (this.items[child].waiter.id >= last.waiter.id) {
|
||||
break
|
||||
}
|
||||
this.items[index] = this.items[child]
|
||||
index = child
|
||||
}
|
||||
this.items[index] = last
|
||||
}
|
||||
}
|
||||
|
||||
export class GitAdmissionWaiterQueue {
|
||||
private readonly lanes: Record<AdmissionClass, TierLanes> = {
|
||||
@@ -27,6 +89,23 @@ export class GitAdmissionWaiterQueue {
|
||||
network: createTierLanes()
|
||||
}
|
||||
private readonly countsByBudget = new Map<string, number>()
|
||||
private readonly baseCandidates: Record<AdmissionClass, Record<GitAdmissionTier, CandidateHeap>> =
|
||||
{
|
||||
general: {
|
||||
interactive: new CandidateHeap(),
|
||||
status: new CandidateHeap(),
|
||||
background: new CandidateHeap()
|
||||
},
|
||||
network: {
|
||||
interactive: new CandidateHeap(),
|
||||
status: new CandidateHeap(),
|
||||
background: new CandidateHeap()
|
||||
}
|
||||
}
|
||||
private readonly headroomCandidates: Record<AdmissionClass, CandidateHeap> = {
|
||||
general: new CandidateHeap(),
|
||||
network: new CandidateHeap()
|
||||
}
|
||||
private totalCount = 0
|
||||
|
||||
get count(): number {
|
||||
@@ -46,6 +125,9 @@ export class GitAdmissionWaiterQueue {
|
||||
for (const key of waiter.budgetKeys) {
|
||||
this.countsByBudget.set(key, (this.countsByBudget.get(key) ?? 0) + 1)
|
||||
}
|
||||
if (lane.count === 1) {
|
||||
this.publishLaneHead(waiter.admissionClass, waiter.tier, lane)
|
||||
}
|
||||
}
|
||||
|
||||
dequeue(waiter: AdmissionWaiter): void {
|
||||
@@ -64,9 +146,35 @@ export class GitAdmissionWaiterQueue {
|
||||
this.countsByBudget.set(key, current - 1)
|
||||
}
|
||||
}
|
||||
const previousHead = lane.items[lane.head]
|
||||
this.compact(lane)
|
||||
if (lane.count === 0) {
|
||||
lane.version += 1
|
||||
lanes.delete(waiter.route)
|
||||
} else if (lane.items[lane.head] !== previousHead) {
|
||||
lane.version += 1
|
||||
this.publishLaneHead(waiter.admissionClass, waiter.tier, lane)
|
||||
}
|
||||
}
|
||||
|
||||
updateRouteEligibility(
|
||||
admissionClass: AdmissionClass,
|
||||
route: string | null,
|
||||
baseEligible: boolean,
|
||||
headroomEligible: boolean
|
||||
): void {
|
||||
for (const tier of TIERS) {
|
||||
const lane = this.lanes[admissionClass][tier].get(route)
|
||||
if (
|
||||
!lane ||
|
||||
(lane.baseEligible === baseEligible && lane.headroomEligible === headroomEligible)
|
||||
) {
|
||||
continue
|
||||
}
|
||||
lane.baseEligible = baseEligible
|
||||
lane.headroomEligible = headroomEligible
|
||||
lane.version += 1
|
||||
this.publishLaneHead(admissionClass, tier, lane)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -89,39 +197,66 @@ export class GitAdmissionWaiterQueue {
|
||||
nextFitting(
|
||||
admissionClass: AdmissionClass,
|
||||
effectiveTier: (waiter: AdmissionWaiter) => number,
|
||||
slotKindFor: (waiter: AdmissionWaiter) => AdmissionSlotKind | null,
|
||||
allowBase: boolean,
|
||||
allowHeadroom: boolean,
|
||||
abort: (waiter: AdmissionWaiter) => void
|
||||
): SelectedWaiter | null {
|
||||
let selected: (SelectedWaiter & { tier: number }) | null = null
|
||||
for (const rawTier of TIERS) {
|
||||
for (const lane of this.lanes[admissionClass][rawTier].values()) {
|
||||
for (let index = lane.head; index < lane.items.length; index += 1) {
|
||||
const waiter = lane.items[index]
|
||||
if (waiter.state !== 'queued') {
|
||||
continue
|
||||
while (true) {
|
||||
const candidates: SelectedWaiter[] = []
|
||||
if (allowBase) {
|
||||
// Aging preserves order within a raw tier, so only its oldest eligible head can win.
|
||||
for (const tier of TIERS) {
|
||||
const candidate = this.peekValid(this.baseCandidates[admissionClass][tier], 'base')
|
||||
if (candidate) {
|
||||
candidates.push({ waiter: candidate.waiter, slotKind: 'base' })
|
||||
}
|
||||
if (waiter.signal?.aborted) {
|
||||
abort(waiter)
|
||||
continue
|
||||
}
|
||||
const slotKind = slotKindFor(waiter)
|
||||
if (slotKind) {
|
||||
const tier = effectiveTier(waiter)
|
||||
if (
|
||||
!selected ||
|
||||
tier < selected.tier ||
|
||||
(tier === selected.tier && waiter.id < selected.waiter.id)
|
||||
) {
|
||||
selected = { waiter, slotKind, tier }
|
||||
}
|
||||
}
|
||||
// Every waiter in this route lane has the same budget shape, so a
|
||||
// blocked head proves the rest cannot fit either.
|
||||
break
|
||||
}
|
||||
}
|
||||
if (allowHeadroom) {
|
||||
const candidate = this.peekValid(this.headroomCandidates[admissionClass], 'headroom')
|
||||
if (candidate) {
|
||||
candidates.push({ waiter: candidate.waiter, slotKind: 'headroom' })
|
||||
}
|
||||
}
|
||||
const selected = candidates.sort(
|
||||
(left, right) =>
|
||||
effectiveTier(left.waiter) - effectiveTier(right.waiter) ||
|
||||
left.waiter.id - right.waiter.id ||
|
||||
(left.slotKind === 'base' ? -1 : 1)
|
||||
)[0]
|
||||
if (!selected || !selected.waiter.signal?.aborted) {
|
||||
return selected ?? null
|
||||
}
|
||||
abort(selected.waiter)
|
||||
}
|
||||
return selected
|
||||
}
|
||||
|
||||
private publishLaneHead(
|
||||
admissionClass: AdmissionClass,
|
||||
tier: GitAdmissionTier,
|
||||
lane: WaiterLane
|
||||
): void {
|
||||
const waiter = lane.items[lane.head]
|
||||
if (!waiter || waiter.state !== 'queued') {
|
||||
return
|
||||
}
|
||||
const candidate = { lane, waiter, version: lane.version }
|
||||
if (lane.baseEligible) {
|
||||
this.baseCandidates[admissionClass][tier].push(candidate)
|
||||
}
|
||||
if (tier === 'interactive' && lane.headroomEligible) {
|
||||
this.headroomCandidates[admissionClass].push(candidate)
|
||||
}
|
||||
}
|
||||
|
||||
private peekValid(heap: CandidateHeap, slotKind: AdmissionSlotKind): Candidate | null {
|
||||
return heap.peek(
|
||||
({ lane, waiter, version }) =>
|
||||
version === lane.version &&
|
||||
waiter === lane.items[lane.head] &&
|
||||
waiter.state === 'queued' &&
|
||||
(slotKind === 'base' ? lane.baseEligible : lane.headroomEligible)
|
||||
)
|
||||
}
|
||||
|
||||
private compact(lane: WaiterLane): void {
|
||||
|
||||
@@ -411,6 +411,52 @@ describe('GitAdmissionScheduler', () => {
|
||||
expect(scheduler.snapshot()).toMatchObject({ queued: 0, queuedWaiters: [] })
|
||||
})
|
||||
|
||||
it('selects one eligible route without scanning thousands of saturated routes', async () => {
|
||||
const routeCount = 2_000
|
||||
const scheduler = new GitAdmissionScheduler({
|
||||
generalCap: routeCount,
|
||||
generalHeadroom: 0,
|
||||
routeCap: 1,
|
||||
routeHeadroom: 0
|
||||
})
|
||||
const running = await Promise.all(
|
||||
Array.from({ length: routeCount }, (_, index) =>
|
||||
scheduler.acquire({ ...local('background'), wslDistro: `distro-${index}` })
|
||||
)
|
||||
)
|
||||
let aborted = false
|
||||
let abortedReads = 0
|
||||
const abortListeners = new Set<() => void>()
|
||||
const signal = {
|
||||
get aborted() {
|
||||
abortedReads += 1
|
||||
return aborted
|
||||
},
|
||||
addEventListener: (_event: string, listener: () => void) => abortListeners.add(listener),
|
||||
removeEventListener: (_event: string, listener: () => void) => abortListeners.delete(listener)
|
||||
} as unknown as AbortSignal
|
||||
const queued = Array.from({ length: routeCount }, (_, index) =>
|
||||
scheduler
|
||||
.acquire({ ...local('background'), wslDistro: `distro-${index}`, signal })
|
||||
.then((grant) => grant.release())
|
||||
)
|
||||
abortedReads = 0
|
||||
|
||||
running.at(-1)?.release()
|
||||
await queued.at(-1)
|
||||
|
||||
expect(abortedReads).toBeLessThanOrEqual(2)
|
||||
expect(scheduler.snapshot().queued).toBe(routeCount - 1)
|
||||
|
||||
aborted = true
|
||||
for (const listener of abortListeners) {
|
||||
listener()
|
||||
}
|
||||
await Promise.allSettled(queued.slice(0, -1))
|
||||
running.slice(0, -1).forEach((grant) => grant.release())
|
||||
expect(scheduler.snapshot()).toMatchObject({ queued: 0, queuedWaiters: [] })
|
||||
})
|
||||
|
||||
it('captures killswitch state in each release closure', async () => {
|
||||
const scheduler = schedulerWithOneSlot()
|
||||
_resetGitAdmissionForTests(scheduler)
|
||||
|
||||
@@ -74,6 +74,7 @@ export class GitAdmissionScheduler {
|
||||
onAbort: () => this.abort(waiter)
|
||||
}
|
||||
this.waiters.enqueue(waiter)
|
||||
this.refreshRouteEligibility(admissionClass, route)
|
||||
request.signal?.addEventListener('abort', waiter.onAbort, { once: true })
|
||||
if (request.signal?.aborted) {
|
||||
this.abort(waiter)
|
||||
@@ -172,10 +173,12 @@ export class GitAdmissionScheduler {
|
||||
private drain(admissionClass: AdmissionClass): void {
|
||||
while (true) {
|
||||
const now = this.config.now()
|
||||
const globalBudget = this.ensureBudget(admissionClass)
|
||||
const selected = this.waiters.nextFitting(
|
||||
admissionClass,
|
||||
(waiter) => this.effectiveTier(waiter, now),
|
||||
(waiter) => this.slotKindFor(waiter),
|
||||
globalBudget.baseUsed < globalBudget.baseCapacity,
|
||||
globalBudget.headroomUsed < globalBudget.headroomCapacity,
|
||||
(waiter) => this.abort(waiter)
|
||||
)
|
||||
if (!selected) {
|
||||
@@ -196,6 +199,7 @@ export class GitAdmissionScheduler {
|
||||
budget.headroomUsed += 1
|
||||
}
|
||||
}
|
||||
this.refreshRouteEligibility(waiter.admissionClass, waiter.route)
|
||||
this.waiters.dequeue(waiter)
|
||||
const queueWaitMs = Math.max(0, now - waiter.enqueuedAt)
|
||||
this.publishEvent(waiter, slotKind, 'grant', queueWaitMs)
|
||||
@@ -231,6 +235,7 @@ export class GitAdmissionScheduler {
|
||||
budget.headroomUsed -= 1
|
||||
}
|
||||
}
|
||||
this.refreshRouteEligibility(waiter.admissionClass, waiter.route)
|
||||
this.publishEvent(waiter, slotKind, 'release', queueWaitMs)
|
||||
this.pruneRouteBudgets(waiter.budgetKeys)
|
||||
this.drain(waiter.admissionClass)
|
||||
@@ -288,6 +293,16 @@ export class GitAdmissionScheduler {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private refreshRouteEligibility(admissionClass: AdmissionClass, route: string | null): void {
|
||||
const budget = route ? this.ensureBudget(`route:${admissionClass}:${route}`) : null
|
||||
this.waiters.updateRouteEligibility(
|
||||
admissionClass,
|
||||
route,
|
||||
!budget || budget.baseUsed < budget.baseCapacity,
|
||||
!budget || budget.headroomUsed < budget.headroomCapacity
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
let scheduler = new GitAdmissionScheduler()
|
||||
|
||||
Reference in New Issue
Block a user