trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns17g rEventsFlows

Checks at once and shards per check default to what the pod's CPUs hold, so a big machine needs no override #2551

Merged torquecrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

torque/forge:torque/2549-checks-at-once-and-shards-per-check-defa → main 3 commits fork 2 threads · 2 unresolved

  • Merged. torque/2549-checks-at-once-and-shards-per-check-defa was merged into main as 1761ba7.

Review

Sign in to review this change.
7 files changed +271 −21
src/checks/service.ts +10 −5
@@ -61,6 +61,7 @@ import { appendRunLog } from "../runs/log.ts";
6161 import { runningLimitSentence, runningRuns } from "../settings/quotas.ts";
6262 import { exportTree, readFileAt } from "../apps/checkout.ts";
6363 import { parseCheck } from "../apps/manifest.ts";
64 import { workerCpus } from "../../scripts/shards.ts";
6465 import { accountActive } from "../core/accounts.ts";
6566 import { childEnv, runStep } from "../apps/process.ts";
6667 import { noteReached } from "../apps/egress_report.ts";
@@ -68,6 +69,7 @@ import {
6869 capacityCount,
6970 noteBooted,
7071 capacityEnv,
72 cpuLimit,
7173 egressSetting,
7274 } from "../core/capacity.ts";
7375 import {
@@ -241,9 +243,10 @@ export function provideChecksForTest(app: App, handle: ChecksHandle): void {
241243 export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
242244 // Read on every pick (`core/capacity.ts`): an admin's change is honoured
243245 // by the next check, not the next boot. `concurrency` is the boot default.
244 noteBooted(app.db, "checks.concurrency", String(options.concurrency ?? 1));
246 const booted = options.concurrency ?? 1;
247 noteBooted(app.db, "checks.concurrency", String(booted));
245248 const capacity = (): number =>
246 capacityCount(app.db, "checks.concurrency", options.concurrency ?? 1);
249 capacityCount(app.db, "checks.concurrency", booted);
247250 const gate = options.gate ?? new Gate(capacity());
248251 const placement = options.placement ?? placementFromEnv();
249252 /** Full gates on land in flight: not slots in the queue, but stopped with it. */
@@ -788,9 +791,11 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
788791 }
789792 : undefined;
790793 try {
794 const cellCpus = resolveLimits("step", manifest.limits).cpus;
791795 const env = childEnv({
792 // The shard count an admin set (`core/capacity.ts`), never the pod's.
793 ...capacityEnv(app.db),
796 // The shard count an admin set (`core/capacity.ts`), else the
797 // default sized to what this check's cell holds, never the pod's.
798 ...capacityEnv(app.db, workerCpus(cpuLimit(), cellCpus)),
794799 CI: "1",
795800 FORGE_URL: options.forgeUrl,
796801 FORGE_CHECK: run.id,
@@ -798,7 +803,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
798803 FORGE_CHECK_IMPACT: IMPACT_FILE,
799804 // The cell's CPU limit, which `availableParallelism()` may not
800805 // report inside it; the shard count follows this (#2257).
801 FORGE_CELL_CPUS: String(resolveLimits("step", manifest.limits).cpus),
806 FORGE_CELL_CPUS: String(cellCpus),
802807 ...(base === undefined
803808 ? {}
804809 : {
src/core/capacity.ts +81 −11
@@ -13,7 +13,7 @@
1313
1414 import type { Database } from "bun:sqlite";
1515 import { readFileSync } from "node:fs";
16 import { totalmem } from "node:os";
16 import { availableParallelism, totalmem } from "node:os";
1717
1818 export type CapacityKey =
1919 | "crew.concurrency"
@@ -34,6 +34,8 @@ interface Definition {
3434 | { readonly type: "choice"; readonly values: readonly string[] };
3535 /** What it is when neither the store nor the environment says. */
3636 readonly fallback: string;
37 /** The default from the pod's CPUs, in place of `fallback` (#2549). */
38 readonly sized?: (cpus: number) => number;
3739 readonly help: string;
3840 }
3941
@@ -50,14 +52,16 @@ export const CAPACITY: Record<CapacityKey, Definition> = {
5052 env: "FORGE_CHECKS_CONCURRENCY",
5153 kind: { type: "count", min: 1, max: 16 },
5254 fallback: "1",
53 help: "How many repository checks run at once. Lowering it lets running ones finish; the next one waits.",
55 sized: (cpus) => checksForCpus(cpus),
56 help: "How many repository checks run at once. Lowering it lets running ones finish; the next one waits. Empty: the pod's CPUs over three per shard (shards × 3), from 1 to 16 — 80 CPUs give 6.",
5457 },
5558 "checks.shards": {
5659 label: "Test shards per check",
5760 env: "FORGE_CHECK_SHARDS",
5861 kind: { type: "count", min: 1, max: 32 },
59 fallback: "",
60 help: "How many parallel shards a check's test run uses (handed to checks and crew shells as FORGE_CHECK_SHARDS). Empty: as many as the machine has cores.",
62 fallback: "1",
63 sized: (cpus) => shardsForCpus(cpus),
64 help: "How many parallel shards a check's test run uses (handed to checks and crew shells as FORGE_CHECK_SHARDS). Empty: the pod's CPUs, at most 4 — and, in a check, no more than its cell holds.",
6165 },
6266 "children.max": {
6367 label: "Forges this one may send",
@@ -106,14 +110,72 @@ export function validCapacity(key: CapacityKey, value: string): boolean {
106110 }
107111
108112 /** The boot default: the environment, when it holds a valid value. */
109 export function capacityDefault(key: CapacityKey): string {
110 const env = process.env[CAPACITY[key].env]?.trim() ?? "";
111 if (env !== "" && validCapacity(key, env)) return env;
113 export function capacityDefault(key: CapacityKey, cpus?: number): string {
114 const env = explicitDefault(key);
115 if (env !== undefined) return env;
116 const sized = CAPACITY[key].sized;
117 if (sized !== undefined) return String(sized(cpus ?? cpuLimit()));
112118 if (key === "children.max") return String(childrenForMemory(memoryLimitBytes()));
113119 if (key === "crew.concurrency") return String(crewForMemory(memoryLimitBytes()));
114120 return CAPACITY[key].fallback;
115121 }
116122
123 /** The environment's value for `key`, when it holds a valid one. */
124 function explicitDefault(key: CapacityKey): string | undefined {
125 const env = process.env[CAPACITY[key].env]?.trim() ?? "";
126 return env !== "" && validCapacity(key, env) ? env : undefined;
127 }
128
129 /** Most shards a check takes by default: past four, a test run waits on git, not CPU. */
130 const SHARDS_DEFAULT_MAX = 4;
131 /** CPUs a shard is given by default: its bun test, the git children it starts, and the forge it boots. */
132 const CPUS_PER_SHARD = 3;
133
134 /** Shards per check by default: `min(4, cpus)`, at least one. 80 CPUs give 4; 2 give 2. */
135 export function shardsForCpus(cpus: number): number {
136 const n = Math.floor(cpus);
137 return Number.isFinite(n) ? Math.max(1, Math.min(SHARDS_DEFAULT_MAX, n)) : 1;
138 }
139
140 /**
141 * Checks at once by default: `floor(cpus / (shards × 3))`, held to 1–16, so
142 * every check's shards have three CPUs each. 80 CPUs: 4 shards, 80 / 12 = 6
143 * checks; 16 CPUs: 1; 4 CPUs: 4 shards, 4 / 12 → 1 (#2549).
144 */
145 export function checksForCpus(cpus: number, shards = shardsForCpus(cpus)): number {
146 const n = Math.floor(cpus / (shards * CPUS_PER_SHARD));
147 return Number.isFinite(n) ? Math.max(1, Math.min(16, n)) : 1;
148 }
149
150 let cpusForTest: number | undefined;
151
152 /**
153 * A test's CPU count in place of the pod's, so a default sized from it is
154 * the same on a laptop and an 80-core runner; `undefined` reads the pod again.
155 */
156 export function provideCpusForTest(cpus: number | undefined): void {
157 cpusForTest = cpus;
158 }
159
160 /**
161 * The pod's CPU limit: cgroup v2 `cpu.max` ("quota period", in µs), rounded
162 * down and never below one, else `os.availableParallelism()` — "max" (no
163 * quota), an unreadable file or anything not two numbers falls back. Never
164 * more than the machine has.
165 */
166 export function cpuLimit(file = "/sys/fs/cgroup/cpu.max"): number {
167 if (cpusForTest !== undefined) return cpusForTest;
168 const machine = availableParallelism();
169 try {
170 const [quota, period] = readFileSync(file, "utf8").trim().split(/\s+/).map(Number);
171 if (quota !== undefined && period !== undefined && quota > 0 && period > 0)
172 return Math.max(1, Math.min(machine, Math.floor(quota / period)));
173 } catch {
174 // No cgroup v2 CPU controller here: the machine's count.
175 }
176 return machine;
177 }
178
117179 const GIB = 1024 ** 3;
118180 const MIB = 1024 ** 2;
119181
@@ -215,10 +277,18 @@ export function egressSetting(db: Database): "enforce" | "report" {
215277 return capacityValue(db, "cells.egress") === "report" ? "report" : "enforce";
junipercrew

capacityEnv now always sets FORGE_CHECK_SHARDS, and the test covers this. A crew shell with no cpus argument gets the pod's count, not its cell's. That is fine today, but say so in the doc comment.
216278 }
217279
218 /** What a check's commands and a crew shell are told: the shard count, when one is set. */
219 export function capacityEnv(db: Database): Record<string, string> {
220 const shards = capacityValue(db, "checks.shards");
221 return shards === "" ? {} : { FORGE_CHECK_SHARDS: shards };
280 /**
281 * What a check's commands and a crew shell are told: the shard count in
282 * force. An admin's value or the environment's is passed as it is; the
283 * default is sized from `cpus` — a check passes what its cell holds, so a
284 * small cell is never told the pod's count.
285 */
286 export function capacityEnv(db: Database, cpus?: number): Record<string, string> {
287 const shards =
288 storedCapacity(db, "checks.shards") ??
289 explicitDefault("checks.shards") ??
290 String(shardsForCpus(cpus ?? cpuLimit()));
291 return { FORGE_CHECK_SHARDS: shards };
222292 }
223293
224294 export interface CapacityChange {
src/main.ts +2 −1
@@ -333,7 +333,8 @@ const checks =
333333 ? undefined
334334 : startChecks(app, {
335335 forgeUrl,
336 concurrency: positive(process.env.FORGE_CHECKS_CONCURRENCY, 1),
336 // Sized to the pod's CPUs unless the environment says (#2549).
337 concurrency: positive(process.env.FORGE_CHECKS_CONCURRENCY, Number(capacityDefault("checks.concurrency"))),
junipercrew

This line is well past the width of its neighbours. Compute const checksDefault = Number(capacityDefault("checks.concurrency")) above and pass that.
337338 redHoldMs: RED_HOLD_MS,
338339 // A roll lets running checks finish this long before cutting them off.
339340 drainMs: positive(process.env.FORGE_CHECKS_DRAIN_S, 300) * 1000,
test/capacity-cpus.test.ts added +143
@@ -0,0 +1,143 @@
1 /**
2 * Checks at once and shards per check default to what the pod's CPUs hold
3 * (#2549): shards `min(4, cpus)`, checks `floor(cpus / (shards × 3))` held
4 * to 1–16, so the vxrail's 80 cores need no override and the sentinel no
5 * longer calls 4 shards and 6 checks away from their defaults.
6 */
7
8 import { afterEach, describe, expect, test } from "bun:test";
9 import { writeFileSync } from "node:fs";
10 import { mkdtemp, rm } from "node:fs/promises";
11 import { availableParallelism, tmpdir } from "node:os";
12 import { join } from "node:path";
13 import { createApp } from "../src/app.ts";
14 import {
15 CAPACITY,
16 capacityDefault,
17 capacityEnv,
18 capacityValue,
19 checksForCpus,
20 cpuLimit,
21 provideCpusForTest,
22 setCapacity,
23 shardsForCpus,
24 } from "../src/core/capacity.ts";
25 import { settingsAway } from "../src/crew/sentinel.ts";
26 import { createUser } from "../src/core/users.ts";
27
28 const ENV = ["FORGE_CHECK_SHARDS", "FORGE_CHECKS_CONCURRENCY"] as const;
29 const saved = ENV.map((k) => process.env[k]);
30 afterEach(() => {
31 ENV.forEach((k, i) => {
32 if (saved[i] === undefined) delete process.env[k];
33 else process.env[k] = saved[i];
34 });
35 provideCpusForTest(4);
36 });
37
38 describe("the defaults follow the CPUs", () => {
39 // By hand: shards = min(4, cpus); checks = floor(cpus / (shards * 3)), 1..16.
40 test.each([
41 // cpus, shards, checks
42 [4, 4, 1], // 4 / 12 = 0.33 → held to 1
43 [16, 4, 1], // 16 / 12 = 1.33 → 1
44 [80, 4, 6], // 80 / 12 = 6.67 → 6
45 [2, 2, 1], // 2 / 6 = 0.33 → 1
46 [1, 1, 1],
47 [1000, 4, 16], // 1000 / 12 = 83 → held to 16
48 ])("%d CPUs: %d shards, %d checks at once", (cpus, shards, checks) => {
49 expect(shardsForCpus(cpus)).toBe(shards);
50 expect(checksForCpus(cpus)).toBe(checks);
51 delete process.env.FORGE_CHECK_SHARDS;
52 delete process.env.FORGE_CHECKS_CONCURRENCY;
53 expect(capacityDefault("checks.shards", cpus)).toBe(String(shards));
54 expect(capacityDefault("checks.concurrency", cpus)).toBe(String(checks));
55 provideCpusForTest(cpus);
56 expect(capacityDefault("checks.shards")).toBe(String(shards));
57 expect(capacityDefault("checks.concurrency")).toBe(String(checks));
58 });
59
60 test("the environment still wins over the CPUs", () => {
61 process.env.FORGE_CHECK_SHARDS = "8";
62 process.env.FORGE_CHECKS_CONCURRENCY = "3";
63 expect(capacityDefault("checks.shards", 80)).toBe("8");
64 expect(capacityDefault("checks.concurrency", 80)).toBe("3");
65 });
66
67 test("both help lines say how the default is reached", () => {
68 expect(CAPACITY["checks.shards"].help).toContain("at most 4");
69 expect(CAPACITY["checks.concurrency"].help).toContain("shards × 3");
70 });
71 });
72
73 describe("a stored value overrides the default; the sentinel compares with it", () => {
74 test("80 CPUs: 4 shards and 6 checks are the defaults, 8 shards is away", async () => {
75 delete process.env.FORGE_CHECK_SHARDS;
76 delete process.env.FORGE_CHECKS_CONCURRENCY;
77 provideCpusForTest(80);
78 const dataDir = await mkdtemp(join(tmpdir(), "forge-capacity-cpus-"));
79 const app = createApp({ dataDir, port: 0 });
80 try {
81 const a = await createUser(app.db, {
82 origin: "seed",
83 trusted: true,
84 username: "alice",
85 email: "[email protected]",
86 password: "a-good-passphrase",
87 adminIfFirst: true,
88 });
89 if (!a.ok) throw new Error(a.error);
90 expect(capacityValue(app.db, "checks.shards")).toBe("4");
91 expect(capacityEnv(app.db)).toEqual({ FORGE_CHECK_SHARDS: "4" });
92 // A check's cell of 2 CPUs is given 2, not the pod's 4.
93 expect(capacityEnv(app.db, 2)).toEqual({ FORGE_CHECK_SHARDS: "2" });
94
95 // The owner's choice for the vxrail, set more than a day ago.
96 setCapacity(app.db, "checks.shards", "4", a.value.id);
97 setCapacity(app.db, "checks.concurrency", "6", a.value.id);
98 setCapacity(app.db, "crew.concurrency", "5", a.value.id);
99 const old = new Date(Date.now() - 3 * 24 * 3600_000).toISOString();
100 app.db.query("UPDATE forge_setting SET updated_at = ?1").run(old);
101 expect(settingsAway(app.db, Date.now()).map((f) => f.key)).toEqual(["setting:crew.concurrency"]);
102
103 // A stored value overrides the default, even a cell's smaller one.
104 setCapacity(app.db, "checks.shards", "8", a.value.id);
105 app.db.query("UPDATE forge_setting SET updated_at = ?1").run(old);
106 expect(capacityValue(app.db, "checks.shards")).toBe("8");
107 expect(capacityEnv(app.db, 2)).toEqual({ FORGE_CHECK_SHARDS: "8" });
108 const away = settingsAway(app.db, Date.now()).find((f) => f.key === "setting:checks.shards");
109 expect(away?.evidence[0]).toBe('checks.shards = "8", default "4"');
110 } finally {
111 app.close();
112 await rm(dataDir, { recursive: true, force: true });
113 }
114 });
115 });
116
117 describe("cpuLimit reads the pod's cgroup", () => {
118 const withFile = async (text: string | undefined, run: (file: string) => void) => {
119 provideCpusForTest(undefined);
120 const dir = await mkdtemp(join(tmpdir(), "forge-cpu-max-"));
121 const file = join(dir, "cpu.max");
122 if (text !== undefined) writeFileSync(file, text);
123 try {
124 run(file);
125 } finally {
126 await rm(dir, { recursive: true, force: true });
127 }
128 };
129 const machine = availableParallelism();
130
131 test("quota over period, rounded down, never more than the machine", async () => {
132 await withFile("150000 100000\n", (f) => expect(cpuLimit(f)).toBe(Math.min(machine, 1)));
133 await withFile("50000 100000\n", (f) => expect(cpuLimit(f)).toBe(1));
134 await withFile(`${machine * 100000} 100000\n`, (f) => expect(cpuLimit(f)).toBe(machine));
135 await withFile(`${machine * 200000} 100000\n`, (f) => expect(cpuLimit(f)).toBe(machine));
136 });
137
138 test("no quota, garbage or an unreadable file falls back to the machine", async () => {
139 await withFile("max 100000\n", (f) => expect(cpuLimit(f)).toBe(machine));
140 await withFile("nonsense", (f) => expect(cpuLimit(f)).toBe(machine));
141 await withFile(undefined, (f) => expect(cpuLimit(f)).toBe(machine));
142 });
143 });
test/capacity.test.ts +13 −4
@@ -22,8 +22,8 @@ import { AppsRuntime } from "../src/apps/runtime.ts";
2222 import { startChecks, type ChecksHandle } from "../src/checks/service.ts";
2323 import {
2424 capacityChanges,
25 capacityDefault,
2625 capacityCount,
26 capacityDefault,
2727 noteBooted,
2828 capacityEnv,
2929 capacityValue,
@@ -292,14 +292,23 @@ describe("capacity: only a forge admin reads or changes it", () => {
292292 body: new URLSearchParams({ key: "checks.shards", reset: "1" }),
293293 });
294294 expect(reset.status).toBe(303);
295 expect(capacityEnv(app.db)).toEqual({});
295 // Back at the default, which is sized from the CPUs it is given (#2549).
296 const fromEnv = process.env.FORGE_CHECK_SHARDS;
297 delete process.env.FORGE_CHECK_SHARDS;
298 try {
299 expect(capacityEnv(app.db, 80)).toEqual({ FORGE_CHECK_SHARDS: "4" });
300 expect(capacityEnv(app.db, 2)).toEqual({ FORGE_CHECK_SHARDS: "2" });
301 } finally {
302 if (fromEnv !== undefined) process.env.FORGE_CHECK_SHARDS = fromEnv;
303 }
304 const shardsDefault = capacityDefault("checks.shards");
296305
297306 const changes = capacityChanges(app.db);
298307 expect(changes.map((c) => [c.key, c.from, c.to, c.actorName])).toEqual([
299 ["checks.shards", "3", "", "alice"],
308 ["checks.shards", "3", shardsDefault, "alice"],
300309 ["cells.egress", "enforce", "report", "alice"],
301310 ["crew.concurrency", capacityDefault("crew.concurrency"), "3", "alice"],
302 ["checks.shards", "", "3", "alice"],
311 ["checks.shards", shardsDefault, "3", "alice"],
303312 ]);
304313 const after = await (
305314 await fetch(`${servers.url}/settings/capacity`, { headers: { cookie } })
test/helpers/preload.ts +4
@@ -3,6 +3,10 @@
33 * the forge. Only what the whole suite must agree on belongs here.
44 */
55
6 import { provideCpusForTest } from "../../src/core/capacity.ts";
67 import { installCheapHash } from "./cheap-hash.ts";
78
89 installCheapHash();
10 // Defaults sized from CPUs (#2549) read as on a 4-CPU pod — 4 shards, one
11 // check at once — whatever machine runs the suite.
12 provideCpusForTest(4);
test/snapshot-restore-parallel.test.ts +18
@@ -11,6 +11,7 @@ import { mkdtempSync, readFileSync } from "node:fs";
1111 import { rm } from "node:fs/promises";
1212 import { tmpdir } from "node:os";
1313 import { join } from "node:path";
14 import type { runGit } from "../src/git/env.ts";
1415 import { restoreSnapshot } from "../src/snapshot/restore.ts";
1516 import { IN_FLIGHT } from "../src/snapshot/snapshot.ts";
1617 import { dirStore, type Store } from "../src/snapshot/store.ts";
@@ -140,4 +141,21 @@ describe("a restore reads in parallel", () => {
140141 "could not restore alice/first",
141142 );
142143 });
144
145 test("when clones fail at once, the one named is the manifest's earliest, not the race's winner", async () => {
146 const order = ["alice/first", "alice/second", "alice/third"];
147 const files = order.map((r) => ({ path: `repos/${r}.bundle`, bytes: 13 }));
148 const store = dirStore(join(root, "store-5"));
149 for (const f of files) await store.put(prefix + f.path, "not a bundle\n");
150 await store.put(`${prefix}manifest.json`, manifestOf(files));
151 // Every clone fails; the first entry's fails last.
152 const git: typeof runGit = async (args) => {
153 const first = args.some((a) => a.endsWith("first.bundle"));
154 await Bun.sleep(first ? 60 : 5);
155 return { ok: false, error: { code: 128, stderr: "not a bundle", args } };
156 };
157 await expect(restoreSnapshot(store, NAME, join(root, "data-5"), { git })).rejects.toThrow(
158 "could not restore alice/first",
159 );
160 });
143161 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs