trevato/forge

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

More
Forks1g fMapg aRuns13g rEventsFlows

Where the kernel cannot hold a child forge's cores, the forge pauses and resumes its processes to hold them #11905

scoutcrew opened this

Check · Review · Land · Live — Its check is queued. Next: reviewer, no review yet.

ConversationFiles
Ready to land

scout/forge:scout/11559-where-the-kernel-cannot-hold-a-child-for → main 1 commit fork

Review

Sign in to review this change.
3 files changed +196 −6
src/apps/contain.ts +50 −6
@@ -51,6 +51,7 @@ import {
5151 import { during } from "../perf/jobs.ts";
5252 import { isSize, SIZES, type Size } from "../spawn/sizes.ts";
5353 import { cpuWatch } from "./cpu_rule.ts";
54 import { dutyStep, newDuty, resumeNow, type Duty, type Signaller } from "./cpu_duty.ts";
5455
5556 /** `ceiling`: a declaration was cut to the machine's maximum; `declared`: forge.json's own; `default`: neither. */
5657 export type CeilingFrom = "env" | "pod" | "default" | "size";
@@ -313,6 +314,8 @@ export interface CellSpec {
313314 * it is busy rather than stopped.
314315 */
315316 readonly cpuQuota?: boolean | undefined;
317 /** Where the signals of the pausing fallback go; tests only. */
318 readonly signal?: Signaller | undefined;
316319 /**
317320 * Disk bandwidth the kernel holds the cell to (`io.max`) on the disk
318321 * under `dir` — the forge's data volume — where io is delegated. A child
@@ -633,6 +636,19 @@ export function cellCgroup(
633636 );
634637 }
635638
639 let cpuHolderSaid: string | undefined;
640
641 /** Who holds a quota cell's cores: said once per process, and again only if it changes. */
642 export function noteCpuHolder(kernel: boolean, log: (line: string) => void): string {
643 const who = kernel ? "the kernel" : "the forge, by pausing";
644 if (cpuHolderSaid !== who) {
645 cpuHolderSaid = who;
646 log(`cpu: a child forge's cores are held by ${who}`);
647 console.log(`cpu: a child forge's cores are held by ${who}`);
648 }
649 return who;
650 }
651
636652 /** Where the next pinned cell's cores begin; `affinityArgv` advances it. */
637653 export interface CoreRotor {
638654 next: number;
@@ -730,6 +746,7 @@ export function openCell(
730746 const SMALL = String(1 << 20);
731747
732748 const cgroup = cellCgroup(spec);
749 if (spec.cpuQuota === true) noteCpuHolder(cgroup?.cpuCapped === true, log);
733750 const argv = (command: string): string[] => cellArgv(bwrapArgv(command), spec, cgroup);
734751 const rlimits = cellRlimits(spec.limits);
735752 const bwrapArgv = (command: string): string[] => [
@@ -795,12 +812,21 @@ export function openCell(
795812 cgroup === undefined
796813 ? { bytes: Math.floor(spec.limits.memoryMiB * 1024 * 1024), by: "watch" }
797814 : { bytes: cgroup.maxBytes, by: "kernel" };
798 unwatch = watchTree(pid, watchedLimits(spec.limits, cgroup), cap, writable, spec.label, (why) => {
799 if (breach !== undefined) return;
800 breach = why;
801 log(`stopped: ${why}`);
802 stop(why);
803 });
815 const paused = spec.cpuQuota === true && cgroup?.cpuCapped !== true;
816 unwatch = watchTree(
817 pid,
818 watchedLimits(spec.limits, paused ? { cpuCapped: true } : cgroup),
819 cap,
820 writable,
821 spec.label,
822 (why) => {
823 if (breach !== undefined) return;
824 breach = why;
825 log(`stopped: ${why}`);
826 stop(why);
827 },
828 paused ? { cores: spec.limits.cpus, ...(spec.signal === undefined ? {} : { signal: spec.signal }) } : undefined,
829 );
804830 };
805831 return {
806832 argv,
@@ -957,6 +983,11 @@ interface Watched {
957983 overSince: number | null;
958984 readonly key: number;
959985 readonly label: string | undefined;
986 /** Set where the watch holds the cores by pausing the tree (`apps/cpu_duty.ts`). */
987 readonly duty?: Duty | undefined;
988 readonly signal?: Signaller | undefined;
989 /** The tree's CPU ticks and time at the last duty step. */
990 dutyLast?: { ticks: number; at: number };
960991 /** The memory cap, and who holds it: the kernel (a cgroup) or this watch alone. */
961992 readonly cap: MemoryCap;
962993 /** The tree's CPU ticks so far and resident KiB, at the last tick. */
@@ -1129,6 +1160,7 @@ export function watchTree(
11291160 writable: readonly string[],
11301161 label: string | undefined,
11311162 breach: (why: string) => void,
1163 pause?: { readonly cores: number; readonly signal?: Signaller },
11321164 ): () => void {
11331165 const entry: Watched = {
11341166 root,
@@ -1142,10 +1174,14 @@ export function watchTree(
11421174 key: nextKey++,
11431175 label,
11441176 cap,
1177 ...(pause === undefined ? {} : { duty: newDuty(pause.cores), signal: pause.signal }),
11451178 };
1179 const signal = pause?.signal ?? signalTree;
11461180 watched.add(entry);
11471181 timer ??= setInterval(() => during("cell-watch", tick), WATCH.tickMs);
11481182 return () => {
1183 // A paused tree is always resumed when the watch ends.
1184 if (entry.duty !== undefined) resumeNow(entry.duty, root, signal);
11491185 watched.delete(entry);
11501186 if (watched.size === 0 && timer !== undefined) {
11511187 clearInterval(timer);
@@ -1360,6 +1396,14 @@ function tick(): void {
13601396 ticks += rows.get(pid)?.ticks ?? 0;
13611397 }
13621398 entry.last = { ticks, kib };
1399 if (entry.duty !== undefined) {
1400 const before = entry.dutyLast;
1401 entry.dutyLast = { ticks, at: now };
1402 if (before !== undefined && now > before.at) {
1403 const used = (ticks - before.ticks) / CLK_TCK / ((now - before.at) / 1000);
1404 dutyStep(entry.duty, entry.root, used, entry.signal ?? signalTree);
1405 }
1406 }
13631407 if (kib > limits.memoryMiB * 1024) {
13641408 entry.breach(
13651409 `used ${mib(kib * 1024)} of memory, over its ${limits.memoryMiB} MiB limit (${memorySource(limits.memorySource, limits.ceilingFrom)})`,
src/apps/cpu_duty.ts added +80
@@ -0,0 +1,80 @@
1 /**
2 * Holding a cell's cores without the kernel: pause its process tree for part
3 * of each second. Plain data in (the cores the tree used over the last
4 * second, the limit), plain data out (how long to pause it this second). The
5 * watch in `contain.ts` sends the signals (SIGSTOP, then SIGCONT, never a
6 * kill) where a cell asked for `cpuQuota` and `cpu.max` was not ours to write.
7 */
8
9 export const DUTY_SECOND_MS = 1_000;
10
11 /**
12 * Milliseconds to pause the tree this second. `cores` is what it would use
13 * running the whole second (its demand): a tree that wants 3 cores under a
14 * limit of 1 runs a third of the time, and so is paused for two thirds.
15 * Never the whole second: the tree always runs a little, so it can be read.
16 */
17 export function pauseMs(cores: number, limit: number, second = DUTY_SECOND_MS): number {
18 if (!Number.isFinite(cores) || !Number.isFinite(limit) || limit <= 0 || cores <= limit) return 0;
19 const run = Math.max(MIN_RUN_FRACTION, limit / cores);
20 return Math.round(second * (1 - run));
21 }
22
23 const MIN_RUN_FRACTION = 0.02;
24
25 /**
26 * The tree's demand from what it used: cores measured over a second in which
27 * it was paused for `pausedMs` ran only for the rest, so its demand is the
28 * reading over that running share.
29 */
30 export function demandCores(usedCores: number, pausedMs: number, second = DUTY_SECOND_MS): number {
31 const running = Math.max(MIN_RUN_FRACTION, 1 - pausedMs / second);
32 return Math.max(0, usedCores) / running;
33 }
34
35 /** What the watch sends: to the cell's tree, a signal. */
36 export type Signaller = (root: number, signal: "SIGSTOP" | "SIGCONT") => void;
37
38 /** One cell's pause state: its limit, how long it was paused last second, and the resume that is owed. */
39 export interface Duty {
40 readonly limit: number;
41 paused: number;
42 resume: ReturnType<typeof setTimeout> | undefined;
43 }
44
45 export function newDuty(limit: number): Duty {
46 return { limit, paused: 0, resume: undefined };
47 }
48
49 /**
50 * One second's step: from the cores used over the last second, pause the tree
51 * now and resume it after `pauseMs`. Returns the pause chosen.
52 */
53 export function dutyStep(
54 duty: Duty,
55 root: number,
56 usedCores: number,
57 signal: Signaller,
58 later: (fn: () => void, ms: number) => ReturnType<typeof setTimeout> = (fn, ms) => setTimeout(fn, ms),
59 ): number {
60 resumeNow(duty, root, signal);
61 const ms = pauseMs(demandCores(usedCores, duty.paused), duty.limit);
62 duty.paused = ms;
63 if (ms > 0) {
64 signal(root, "SIGSTOP");
65 duty.resume = later(() => {
66 duty.resume = undefined;
67 signal(root, "SIGCONT");
68 }, ms);
69 (duty.resume as { unref?: () => void }).unref?.();
70 }
71 return ms;
72 }
73
74 /** Resume the tree if it is paused: the close of a cell, and the start of every step. */
75 export function resumeNow(duty: Duty, root: number, signal: Signaller): void {
76 if (duty.resume === undefined) return;
77 clearTimeout(duty.resume);
78 duty.resume = undefined;
79 signal(root, "SIGCONT");
80 }
test/cpu-duty.test.ts added +66
@@ -0,0 +1,66 @@
1 import { expect, test } from "bun:test";
2 import { dutyStep, demandCores, newDuty, pauseMs, resumeNow } from "../src/apps/cpu_duty.ts";
3 import { signalTree, watchTree } from "../src/apps/contain.ts";
4
5 test("under the limit is never paused", () => {
6 expect(pauseMs(0.5, 1)).toBe(0);
7 expect(pauseMs(1, 1)).toBe(0);
8 expect(pauseMs(Number.NaN, 1)).toBe(0);
9 });
10
11 test("a 3-core load under a 1-core limit is held near 1 core", () => {
12 let paused = 0;
13 const used: number[] = [];
14 for (let second = 0; second < 10; second++) {
15 // The tree wants 3 cores while it runs; it runs for what is not paused.
16 const cores = 3 * (1 - paused / 1000);
17 used.push(cores);
18 paused = pauseMs(demandCores(cores, paused), 1);
19 }
20 for (const c of used.slice(2)) expect(Math.abs(c - 1)).toBeLessThan(0.05);
21 });
22
23 test("it always leaves the tree some time to run", () => {
24 expect(pauseMs(10_000, 1)).toBeLessThan(1000);
25 });
26
27 test("a step pauses with SIGSTOP and resumes with SIGCONT, and close resumes a paused tree", () => {
28 const sent: string[] = [];
29 const duty = newDuty(1);
30 const timers: (() => void)[] = [];
31 const later = (fn: () => void) => (timers.push(fn), setTimeout(() => {}, 0));
32 const ms = dutyStep(duty, 7, 3, (_r, s) => sent.push(s), later);
33 expect(ms).toBe(667);
34 expect(sent).toEqual(["SIGSTOP"]);
35 resumeNow(duty, 7, (_r, s) => sent.push(s));
36 expect(sent).toEqual(["SIGSTOP", "SIGCONT"]);
37 resumeNow(duty, 7, (_r, s) => sent.push(s));
38 expect(sent).toEqual(["SIGSTOP", "SIGCONT"]);
39 });
40
41 test("the watch pauses and resumes a busy cell under the fallback and never stops it", async () => {
42 const proc = Bun.spawn(["sh", "-c", `bun -e "const e=Date.now()+8000; while(Date.now()<e){}" & wait`]);
43 const sent: string[] = [];
44 let stopped: string | undefined;
45 const unwatch = watchTree(
46 proc.pid,
47 { memoryMiB: Infinity, cpus: Infinity, diskMiB: Infinity },
48 { bytes: Infinity, by: "watch" },
49 [],
50 undefined,
51 (why) => (stopped = why),
52 { cores: 0.1, signal: (root, s) => (sent.push(s), signalTree(root, s)) },
53 );
54 await Bun.sleep(4500);
55 unwatch();
56 try {
57 expect(sent).toContain("SIGSTOP");
58 expect(sent).toContain("SIGCONT");
59 expect(sent.at(-1)).toBe("SIGCONT");
60 expect(stopped).toBeUndefined();
61 } finally {
62 signalTree(proc.pid, "SIGCONT");
63 signalTree(proc.pid, "SIGKILL");
64 proc.kill("SIGKILL");
65 }
66 }, 15_000);

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs