diff --git a/README.md b/README.md index ef20369f..09311a60 100644 --- a/README.md +++ b/README.md @@ -1114,6 +1114,14 @@ host has room even counting those, commands start at once: a burst of gated test runs on an idle host starts together. Once only the raw load and memory have room, commands start one at a time, 5 seconds after the previous admission. +Waiting commands start first come, first served. Each records its place in a +queue when it starts waiting, and counts every command queued ahead of it as +one more recent admission: it starts before them only while the host has room +for all of them, and only the first in line waits out the 5-second spacing. A +command's wait is therefore bounded by the work queued ahead of it, never by +commands that started waiting later. A place is freed once its command starts, +its gate is cancelled, or its gate process exits. + The load limit applies only while a gated command is still running. Otherwise the load comes from work the gate cannot wait out, so commands start one at a time as if the raw load had room. A freshly booted hosted macOS runner is one @@ -1128,11 +1136,13 @@ an account missing from it, as in arbitrary-UID containers, uses its UID can write it. Two kinds of command go ahead of the rest. `quest gate --quick ` waits -at most a minute, then starts the command whatever the host's load and memory; +at most a minute, then starts the command whatever the host's load, memory, and +place in the queue; pass it only for cheap commands, such as a formatter or `quest check`, that cost less to run than to hold back. A check that lands PRs goes first: with `QUEST_GATE_PRIORITY=1` in its environment, every gate inside the check waits -for memory but never for load, whether or not other gated commands run. Both +for memory but never for load or the queue, whether or not other gated commands +run. Both record their admissions, so the other callers count them against the host's headroom. Set the variable where the check starts, such as a merge queue build's workflow or the land daemon's sandbox, rather than relying on a slot in @@ -1142,7 +1152,10 @@ the host's load. `quest agent` drops the variable, as it drops an admission. Each admission records the gate's process ID and, except on Windows, when the gate started, and counts as running until that process exits; a reused process ID counts as running only on Windows, where it makes a command wait for the -load again. The gate holds no lock while the command runs, and it owns the +load again. A place in the queue records the same, except that on Windows the +waiting gate refreshes it, and it frees a minute after the last refresh, so a +reused process ID cannot hold the queue. The gate holds no lock while the +command runs, and it owns the command's descendants. Once the command exits, any descendant still running is killed, so a build or test run leaves nothing behind. @@ -1204,8 +1217,8 @@ first one as Just's copy, not as a second signal: a SIGTERM to Just's process group reaches the gate both directly and through Just. A repeated SIGTERM still kills, though when sent to Just or the gate alone, or when the two copies arrive together, it takes a third; a repeated Ctrl-C, which Just does not -forward, kills at the second. There is no fairness guarantee, and no timeout except -`--quick`'s: any other command waits as long as the host stays busy. +forward, kills at the second. There is no timeout except `--quick`'s: any other +command waits its turn for as long as the host stays busy. ## Route work by size diff --git a/quest/m1/gate-fifo-admission.md b/quest/m1/gate-fifo-admission.md deleted file mode 100644 index 701d92f0..00000000 --- a/quest/m1/gate-fifo-admission.md +++ /dev/null @@ -1,44 +0,0 @@ -# [M] Gate admission is first come, first served - -## Goal - -`quest gate` admits waiting commands in arrival order, so no caller is passed -by commands that started waiting after it, and its wait is bounded by the work -queued ahead of it. The host's load and memory still decide when the head of -the queue starts. `quick`, `priority`, and `host` keep their meanings. - -Today admission has no queue: every waiter polls the 1-minute load average and -available memory, and whichever sees room first starts. With about 12 agents -gating on one desktop, a diff-scoped `just check` that ran in 84 s waited 77 -minutes to be admitted -([moq-dev/moq.pro#3751](https://github.com/moq-dev/moq.pro/pull/3751)). - -## Plan - -Decided 2026-11-09 (maintainer, `/quest-plan` of moq-dev/moq.pro#3751's -follow-up): - -- Plan it: ✅ in m1, beside [gate-cpu-weight](/quest/m1/gate-cpu-weight.md) - (rejected: m0; drop). -- Mechanism: ✅ FIFO admission: waiters start in arrival order, with no cost - estimate (rejected: shortest job first from measured costs, which needs a - cost model and can starve full gates; a slot reserved for small commands; - leaving the choice to the implementer). - -The admission ledger (`src/cli/gate.ts`) already records each admission under -an atomic versioned publish; the queue can live in the same ledger, recording -each waiter when it starts waiting. A waiter whose gate process died leaves -the queue the way a dead admission leaves the ledger today. `quick` callers -still start after `QUICK_MS` whatever their place, and a `host` reservation -still holds everyone back. A `priority` caller, which today ignores load so it -goes first while the host is loaded, keeps going ahead of the queue rather than -taking a place in it. - -`admit` drops a ledger entry once it is older than `RECENT_MS` and its process -no longer runs; a queued waiter's entry must stay while its gate lives, however -long it waits, and must not count as an admission against anyone's headroom. -`admit`'s doc comment, which today promises no fairness, states the new order. - -## Related - -- [Gate scopes yield the CPU to the interactive session](/quest/m1/gate-cpu-weight.md) - changes where admitted commands run; this changes when they are admitted diff --git a/src/cli/gate.ts b/src/cli/gate.ts index e8018663..8838b465 100644 --- a/src/cli/gate.ts +++ b/src/cli/gate.ts @@ -15,16 +15,19 @@ import { delegated, nest, questScope, startTime, systemdScopes } from "./supervi /** * One admitted command: a unique id, the wall-clock time it was admitted, and the process ID of * the gate that runs it, which holds a slot until that command ends, with when that gate started - * (see `processStart`), so a reused process ID holds nothing. + * (see `processStart`), so a reused process ID holds nothing. `queued` marks a caller still + * waiting for its turn instead, as of `at`: it holds a place in the queue, the ledger's order of + * queued entries, and no slot. */ export interface Admission { id: string; at: number; pid?: number; + queued?: boolean; start?: string; } -/** One version of the host's admission ledger. Versions only grow; 0 is the empty ledger. */ +/** One version of the host's admission ledger and queue. Versions only grow; 0 is the empty ledger. */ export interface Ledger { version: number; admissions: Admission[]; @@ -57,7 +60,7 @@ export interface Limits { memory: number; } -export type Waiting = "load" | "memory"; +export type Waiting = "load" | "memory" | "queue"; /** * Admissions that only fit the host's raw load and memory, not its headroom, @@ -248,6 +251,7 @@ function parse(text: string): Admission[] { typeof entry?.id === "string" && Number.isFinite(entry?.at) && (entry.pid === undefined || (Number.isSafeInteger(entry.pid) && entry.pid > 0)) && + (entry.queued === undefined || entry.queued === true) && (entry.start === undefined || typeof entry.start === "string"), ); return valid ? (value as Admission[]) : []; @@ -401,10 +405,19 @@ export function fileHost(directory: string, now = Date.now): Host { * one, the load is work the gate cannot wait out, as on a freshly booted macOS * runner, so the raw load counts as having room. * - * Each caller decides against the newest ledger and publishes its admission - * atomically, so concurrent callers never exceed the rule and one always wins - * while it has room. No fairness promise, and no admission timeout except a - * quick caller's (see `Precedence`). + * Callers are admitted first come, first served: each records its place in the + * queue when it starts waiting, and counts every caller queued ahead of it as + * one more recent admission, so it starts before them only while the host has + * room for them all. Only the head of the queue waits out the spacing. A place + * lasts until its caller is admitted, cancelled, or its gate ends; one whose + * gate's start time is unknown, as on Windows, could pass for a reused process + * ID, so its caller refreshes it, and it frees a minute after the last refresh. + * A priority caller goes ahead of the queue, and a quick one starts at its + * deadline whatever its place (see `Precedence`). + * + * Each caller decides against the newest ledger and publishes atomically, so + * concurrent callers never exceed the rule and one always wins while it has + * room. No admission timeout except a quick caller's. */ export async function admit( limits: Limits, @@ -414,18 +427,29 @@ export async function admit( precedence: Precedence = {}, ) { const id = randomUUID(); + // A place in the queue has an id of its own, so publishing the admission adds a new entry. + const place = randomUUID(); // Every admission names its gate's start, so a reused process ID holds no slot. const self = { pid: process.pid, start: processStart(process.pid) }; const deadline = precedence.quick ? host.now() + QUICK_MS : Infinity; - for (;;) { - signal.throwIfAborted(); + + /** One decision against the newest ledger: true once admitted. */ + const turn = async (): Promise => { const ledger = host.ledger(); // Sampled after the read, so only a backward clock step dates an admission in the future. const now = host.now(); - // A running admission stays in the ledger past the minute, so later callers still see its slot. - const running = new Set(ledger.admissions.filter((admission) => host.running(admission))); - const kept = ledger.admissions.filter((admission) => admission.at > now - RECENT_MS || running.has(admission)); - const recent = kept.filter((admission) => admission.at > now - RECENT_MS); + const live = new Set(ledger.admissions.filter((admission) => host.running(admission))); + const kept = ledger.admissions.filter((admission) => + admission.queued + ? // A place whose gate cannot be told from a reused process ID lasts only while refreshed. + admission.id === place || + (live.has(admission) && (admission.start !== undefined || admission.at > now - RECENT_MS)) + : // A running admission stays in the ledger past the minute, so later callers still see its slot. + admission.at > now - RECENT_MS || live.has(admission), + ); + const admissions = kept.filter((admission) => !admission.queued); + const running = new Set(admissions.filter((admission) => live.has(admission))); + const recent = admissions.filter((admission) => admission.at > now - RECENT_MS); if (kept.some((admission) => admission.at > now)) { // Date those now, so they still count and space admissions but expire a minute after // the step. A forward step can only expire admissions up to a minute early. @@ -433,13 +457,28 @@ export async function admit( ledger, kept.map((admission) => ({ ...admission, at: Math.min(admission.at, now) })), ); - continue; + return false; } - if (now >= deadline) { - // Recorded like any admission, so it still counts against the others' headroom. - if (host.publish(ledger, [...kept, { id, at: now, ...self }])) return; - continue; + const position = kept.findIndex((admission) => admission.id === place); + const mine = kept[position]; + if ( + !precedence.priority && + (mine === undefined || (mine.start === undefined && mine.at <= now - RECENT_MS / 2)) + ) { + // Take a place at the back, or refresh one that would otherwise expire where it stands. + const entry = { id: place, at: now, ...self, queued: true }; + host.publish( + ledger, + mine ? kept.map((admission) => (admission === mine ? entry : admission)) : [...kept, entry], + ); + return false; } + const ahead = kept.slice(0, Math.max(position, 0)).filter((admission) => admission.queued).length; + // An admission takes the place of this caller's entry in the queue. + const admitted = (at: number) => + host.publish(ledger, [...kept.filter((admission) => admission.id !== place), { id, at, ...self }]); + // Recorded like any admission, so it still counts against the others' headroom. + if (now >= deadline) return admitted(now); // Priority ignores load, whether or not a gated command runs; memory still applies. const load = precedence.priority ? 0 : running.size > 0 ? host.load() : Math.min(host.load(), limits.load); const memory = host.memory(); @@ -447,19 +486,43 @@ export async function admit( if (load > limits.load) waiting?.("load"); else if (memory < limits.memory) waiting?.("memory"); else { - const headroom = - load + recent.length < limits.load && memory - recent.length * limits.memory >= limits.memory; - const last = Math.max(-Infinity, ...recent.map((admission) => admission.at)); - wait = headroom ? 0 : last + SETTLE_MS - now; - if (wait <= 0) { - // A caller descheduled since sampling would record a stale time, which lets the - // next one skip its spacing; decide again once the stall reaches that interval. - const at = host.now(); - if (at - now < SETTLE_MS && host.publish(ledger, [...kept, { id, at, ...self }])) return; - continue; + // Every caller queued ahead takes a slot before this one, so it starts first only while + // the host has room for them all. + const taken = recent.length + ahead; + const headroom = load + taken < limits.load && memory - taken * limits.memory >= limits.memory; + if (!headroom && ahead > 0) waiting?.("queue"); + else { + const last = Math.max(-Infinity, ...recent.map((admission) => admission.at)); + wait = headroom ? 0 : last + SETTLE_MS - now; + if (wait <= 0) { + // A caller descheduled since sampling would record a stale time, which lets the + // next one skip its spacing; decide again once the stall reaches that interval. + const at = host.now(); + return at - now < SETTLE_MS && admitted(at); + } } } await host.sleep(Math.min(wait, deadline - now), signal); + return false; + }; + + try { + for (;;) { + signal.throwIfAborted(); + if (await turn()) return; + } + } finally { + // A cancelled caller leaves the queue at once, rather than once its gate ends. + for (let ledger = host.ledger(); ledger.admissions.some((admission) => admission.id === place);) { + if ( + host.publish( + ledger, + ledger.admissions.filter((admission) => admission.id !== place), + ) + ) + break; + ledger = host.ledger(); + } } } @@ -635,6 +698,7 @@ export function owner(env = process.env, cgroup = "/proc/self/cgroup"): Owner { const notices: Record = { load: "quest gate: waiting for the load average to fall", memory: "quest gate: waiting for free memory", + queue: "quest gate: waiting for commands queued earlier to start", }; /** Run from the CLI. */ diff --git a/test/gate.test.ts b/test/gate.test.ts index 63e3f81c..016c6db7 100644 --- a/test/gate.test.ts +++ b/test/gate.test.ts @@ -261,6 +261,99 @@ test("a quick caller starts after at most QUICK_MS, and its admission is recorde expect(slow.ledger().admissions).toEqual([{ id: expect.any(String), at: QUICK_MS, ...me }]); }); +test("waiting callers start in arrival order, a later one first only while the host has room for both", async () => { + const limits = { load: 4, memory: GiB }; + // A build still runs, and an earlier caller waits for its turn. + const build = { id: "build", at: -10 * RECENT_MS, ...me }; + const earlier = { id: "earlier", at: 0, ...me, queued: true }; + const directory = tempDir(); + const state = { now: 0, load: 3, memory: Infinity }; + const events: string[] = []; + const host = testHost(directory, state, events); + expect(host.publish(host.ledger(), [build, earlier])).toBe(true); + const sleep = host.sleep; + host.sleep = async (ms, signal) => { + await sleep(ms, signal); + // The earlier caller takes its turn. + const ledger = host.ledger(); + const rest = ledger.admissions.filter(({ id }) => id !== "earlier"); + if (rest.length < ledger.admissions.length) + expect(host.publish(ledger, [...rest, { id: "started", at: state.now, ...me }])).toBe(true); + }; + // Alone, this caller would start at once; behind the earlier one, it waits, then spaces itself. + await admit(limits, host, signal, (reason) => events.push(reason)); + expect(events).toEqual(["queue", `sleep ${SETTLE_MS}`, `sleep ${SETTLE_MS}`]); + expect(host.ledger().admissions).toEqual([ + build, + { id: "started", at: SETTLE_MS, ...me }, + { id: expect.any(String), at: 2 * SETTLE_MS, ...me }, + ]); + // With room for both, it starts at once and leaves the earlier caller's place. + const roomy = testHost(tempDir(), { now: 0, load: 2, memory: Infinity }, events); + expect(roomy.publish(roomy.ledger(), [build, earlier])).toBe(true); + events.length = 0; + await admit(limits, roomy, signal, (reason) => events.push(reason)); + expect(events).toEqual([]); + expect(roomy.ledger().admissions).toEqual([build, earlier, { id: expect.any(String), at: 0, ...me }]); +}); + +test("a place frees once its gate ends or its caller cancels, or a minute after its last refresh without a start", async () => { + const limits = { load: 4, memory: GiB }; + const build = { id: "build", at: -10 * RECENT_MS, ...me }; + const exited = spawnSync(process.execPath, ["-e", ""]).pid; + const directory = tempDir(); + const state = { now: 0, load: 3, memory: Infinity }; + const host = testHost(directory, state); + expect( + host.publish(host.ledger(), [ + build, + { id: "ended", at: 0, pid: exited, queued: true }, + { id: "reused", at: 0, pid: process.pid, start: "an earlier process", queued: true }, + { id: "unrefreshed", at: -RECENT_MS, pid: process.pid, queued: true }, + ]), + ).toBe(true); + const events: string[] = []; + await admit(limits, testHost(directory, state, events), signal, (reason) => events.push(reason)); + expect(events).toEqual([]); + expect(host.ledger().admissions).toEqual([build, { id: expect.any(String), at: 0, ...me }]); + // A place refreshed within the minute still holds, and a cancelled caller behind it leaves the queue. + const fresh = { id: "fresh", at: -RECENT_MS / 2, pid: process.pid, queued: true }; + const queued = testHost(tempDir(), state); + expect(queued.publish(queued.ledger(), [build, fresh])).toBe(true); + const controller = new AbortController(); + const waited: string[] = []; + queued.sleep = async () => { + expect(queued.ledger().admissions).toEqual([build, fresh, expect.objectContaining({ queued: true, ...me })]); + controller.abort(new Error("cancelled")); + }; + await expect(admit(limits, queued, controller.signal, (reason) => waited.push(reason))).rejects.toThrow( + "cancelled", + ); + expect(waited).toEqual(["queue"]); + expect(queued.ledger().admissions).toEqual([build, fresh]); +}); + +test("a priority caller goes ahead of the queue, and a quick one starts at its deadline whatever its place", async () => { + const limits = { load: 4, memory: GiB }; + const build = { id: "build", at: -10 * RECENT_MS, ...me }; + const earlier = { id: "earlier", at: 0, ...me, queued: true }; + const directory = tempDir(); + const state = { now: 0, load: 3, memory: Infinity }; + const host = testHost(directory, state); + expect(host.publish(host.ledger(), [build, earlier])).toBe(true); + const events: string[] = []; + await admit(limits, testHost(directory, state, events), signal, undefined, { priority: true }); + expect(events).toEqual([]); + await admit(limits, testHost(directory, state, events), signal, undefined, { quick: true }); + expect(events).toEqual(Array.from({ length: QUICK_MS / SETTLE_MS }, () => `sleep ${SETTLE_MS}`)); + expect(host.ledger().admissions).toEqual([ + build, + earlier, + { id: expect.any(String), at: 0, ...me }, + { id: expect.any(String), at: QUICK_MS, ...me }, + ]); +}); + test("a lone quick or priority caller starts at once on load from outside the gate", async () => { // No gated command runs, so the load is work the gate cannot wait out, whatever the precedence. const state = { now: 0, load: 12, memory: Infinity }; @@ -348,7 +441,7 @@ test("a caller that dies mid-admission blocks no one past a minute", async () => await admit({ load: 1, memory: 1 }, testHost(directory, state, events), signal); expect(events).toEqual([]); // The stale draft is gone; one still being written stays. - expect(readdirSync(directory).sort()).toEqual(["2", "live.tmp"]); + expect(readdirSync(directory).sort()).toEqual(["3", "live.tmp"]); // A version cut short, as by power loss, reads as empty rather than wedging every caller. writeFileSync(join(directory, "100"), ""); expect(host.ledger()).toEqual({ version: 100, admissions: [] });