diff --git a/docs/changes/unreleased/1190-cpu-bound.md b/docs/changes/unreleased/1190-cpu-bound.md new file mode 100644 index 0000000000..ddf702b64d --- /dev/null +++ b/docs/changes/unreleased/1190-cpu-bound.md @@ -0,0 +1,52 @@ +--- +kind: added +title: a job subtree is cut when it passes its CPU or process bound +pr: 1190 +surface: [chat, engine] +invalidates: + - "A run's only bounds were its dollar cap and, since #1158, its memory. Nothing bounded a job subtree's CPU or its process count, so a model that spawned busy loops as bash jobs (`yes`, `awk 'BEGIN{for(;;){}}'`) spent no tokens and no bound ever fired. A job subtree is now cut when it passes its bound, and the run is told why." +--- + +A job subtree is the process group #1155 already records at launch with its +leader's identity, plus every process under it. It now has a bound of its own: a +Go-side watch, portable, sampling the subtree's cumulative CPU and its live +process count through the group handle. The bound has two halves that apply in +different places. + +**The process backstop applies always.** A subtree may not hold more than 128 +processes — about four times the measured honest peak of 34 — quota'd or not, +because a fork storm that has not yet accumulated CPU is invisible to a rate and +a cgroup CPU quota does nothing to a process count. + +**The CPU rate rule applies only where nothing else caps the process.** When a +cgroup CPU quota or an affinity mask BINDS this process — the cores it may use +are fewer than the machine's — the quota is already the ceiling on what any +subtree under it can burn, and #1187's park bound is what returns a turn wedged +on a job that never ends. A CPU rule there would only cut honest work early: +inside a four-core cell an ordinary `go test` legitimately uses most of its four +cores, and a share of them would end it. So under a binding quota the CPU rule is +OFF and only the process backstop applies. When NOTHING binds — the process may +use the whole machine — the CPU rule is the only thing between a spinner storm +and the box: a subtree may not sustain more than three fifths of the machine's +cores. On the 20-core box the numbers came from that is 12 cores, clearing the +honest peak (8.5) and cutting the storm (16 loops → 16 cores, 64 → 20). + +Whether a quota binds is read from the process's own cgroup CPU quota (v2 +`cpu.max`, v1 `cpu.cfs_quota_us`/`cpu.cfs_period_us`, walked from the process's +cgroup to the hierarchy root the way the memory bound is read, #1162) and its +scheduler affinity mask; off Linux nothing binds and the machine-cores rule +governs as before. + +**Sustained over half a minute, not a spike.** A build burst can pass twelve +cores for several seconds, so the bound trips only after the subtree has been +over on fifteen consecutive readings at a two-second interval — thirty seconds +sustained. A transient never cuts, and a storm — over on every reading by +construction — is cut half a minute in. A subtree whose usage cannot be read +(off Linux, or a group whose identity no longer matches) is never cut: silence is +not pressure. + +**The run learns it, and it is not a silent kill.** A subtree that passes its +bound is ended and the run is handed the record `the job subtree was not settled +within its bound`, naming the job and what it was holding — #1183's and #1187's +family, said about the subtree. The kill is a requested death, so the registry's +own ending says nothing and this record is the only account of it. diff --git a/internal/processgroup/cores.go b/internal/processgroup/cores.go new file mode 100644 index 0000000000..05d05f808d --- /dev/null +++ b/internal/processgroup/cores.go @@ -0,0 +1,35 @@ +package processgroup + +import "runtime" + +// EffectiveCores is how many processor cores this process may ACTUALLY use: the +// smallest of the machine's logical CPUs, the scheduler affinity mask, and the +// cgroup CPU quota walked from the process's own cgroup upward. A job subtree's +// CPU ceiling is a share of THIS and not of the machine's raw core count, +// because where a job runs the subtree is usually held under a cgroup CPU quota +// or an affinity mask and can never reach more cores than these; a ceiling drawn +// from the machine's cores would sit above anything the quota lets the subtree +// produce and would never trip. +// +// It is a variable so a test — or a caller that already knows the figure — can +// stand a core count in front of the bound without a cgroup or an affinity mask +// to be under; nothing but a test reassigns it. +var EffectiveCores = effectiveCores + +// effectiveCores folds the three readable bounds into the tightest, never below +// one. A bound that cannot be read is a zero that does not participate — the same +// silence rule the memory reader keeps (#1162): an unread bound is not a licence +// to guess a smaller machine. +func effectiveCores() float64 { + cores := float64(runtime.NumCPU()) + if quota := cgroupCPUCores(); quota > 0 && quota < cores { + cores = quota + } + if affinity := affinityCores(); affinity > 0 && float64(affinity) < cores { + cores = float64(affinity) + } + if cores < 1 { + cores = 1 + } + return cores +} diff --git a/internal/processgroup/cores_linux.go b/internal/processgroup/cores_linux.go new file mode 100644 index 0000000000..f50528a527 --- /dev/null +++ b/internal/processgroup/cores_linux.go @@ -0,0 +1,157 @@ +//go:build linux + +package processgroup + +// cores_linux.go reads the two Linux bounds on how many cores this process may +// use — the cgroup CPU quota and the scheduler affinity mask — mirroring the way +// cmd/codeaf reads the cgroup MEMORY bound (#1162): the process's own +// /proc/self/cgroup names its cgroup, and the quota is walked from that cgroup to +// the hierarchy root with the tightest taken, because a parent slice can bound +// tighter than the leaf. + +import ( + "os" + "path/filepath" + "strconv" + "strings" + + "golang.org/x/sys/unix" +) + +// cgroupCPUCores is the CPU quota the cgroup this process runs in places on it, +// in cores (quota / period), or zero when the hierarchy places no quota to give. +func cgroupCPUCores() float64 { + return cgroupCPUCoresAt("/proc/self/cgroup", "/sys/fs/cgroup") +} + +// cgroupCPUCoresAt is [cgroupCPUCores] against a named cgroup file and hierarchy +// root, so the walk can be tested without being in a cgroup. Both generations are +// read: v2 has an empty controller list on its 0:: line and the quota in +// /cpu.max ("quota period", or "max period" for no quota); v1 names cpu +// among its controllers and keeps the quota in cpu//cpu.cfs_quota_us over +// cpu.cfs_period_us (a quota of -1 is no quota). +func cgroupCPUCoresAt(procSelfCgroup, root string) float64 { + raw, err := os.ReadFile(procSelfCgroup) + if err != nil { + return 0 + } + best := 0.0 + for _, line := range strings.Split(string(raw), "\n") { + fields := strings.SplitN(strings.TrimSpace(line), ":", 3) + if len(fields) != 3 { + continue + } + controllers, path := fields[1], fields[2] + var cores float64 + switch { + case controllers == "": + cores = smallestCPUCores(root, path, cpuMaxV2) + case hasController(controllers, "cpu"): + cores = smallestCPUCores(filepath.Join(root, "cpu"), path, cpuMaxV1) + default: + continue + } + if cores > 0 && (best == 0 || cores < best) { + best = cores + } + } + return best +} + +// smallestCPUCores returns the tightest finite core quota that read names at the +// cgroup directory dir or any ancestor up to root, or zero when none bound. +func smallestCPUCores(root, dir string, read func(string) float64) float64 { + root = filepath.Clean(root) + at := filepath.Clean(filepath.Join(root, dir)) + if at != root && !strings.HasPrefix(at, root+string(filepath.Separator)) { + return 0 + } + best := 0.0 + for { + if cores := read(at); cores > 0 { + if best == 0 || cores < best { + best = cores + } + } + if at == root { + break + } + parent := filepath.Dir(at) + if parent == at { + break + } + at = parent + } + return best +} + +// cpuMaxV2 reads a cgroup v2 cpu.max at dir and returns quota / period in cores, +// or zero when the file is absent, unreadable, or says "max". +func cpuMaxV2(dir string) float64 { + raw, err := os.ReadFile(filepath.Join(dir, "cpu.max")) + if err != nil { + return 0 + } + fields := strings.Fields(string(raw)) + if len(fields) != 2 || fields[0] == "max" { + return 0 + } + quota, err := strconv.ParseFloat(fields[0], 64) + if err != nil || quota <= 0 { + return 0 + } + period, err := strconv.ParseFloat(fields[1], 64) + if err != nil || period <= 0 { + return 0 + } + return quota / period +} + +// cpuMaxV1 reads a cgroup v1 cpu.cfs_quota_us over cpu.cfs_period_us at dir and +// returns their ratio in cores, or zero when there is no quota (the file is +// absent or holds -1). +func cpuMaxV1(dir string) float64 { + quota := readInt(filepath.Join(dir, "cpu.cfs_quota_us")) + if quota <= 0 { + return 0 + } + period := readInt(filepath.Join(dir, "cpu.cfs_period_us")) + if period <= 0 { + return 0 + } + return float64(quota) / float64(period) +} + +// readInt reads one integer from a cgroup file, or zero when it cannot. +func readInt(path string) int64 { + raw, err := os.ReadFile(path) + if err != nil { + return 0 + } + n, err := strconv.ParseInt(strings.TrimSpace(string(raw)), 10, 64) + if err != nil { + return 0 + } + return n +} + +// hasController reports whether a cgroup v1 controller list (comma-separated) +// names want. +func hasController(list, want string) bool { + for _, c := range strings.Split(list, ",") { + if c == want { + return true + } + } + return false +} + +// affinityCores is how many CPUs the process's scheduler affinity mask allows, +// or zero when it cannot be read. +func affinityCores() int { + var set unix.CPUSet + if err := unix.SchedGetaffinity(0, &set); err != nil { + return 0 + } + return set.Count() +} diff --git a/internal/processgroup/cores_other.go b/internal/processgroup/cores_other.go new file mode 100644 index 0000000000..91bd5e56db --- /dev/null +++ b/internal/processgroup/cores_other.go @@ -0,0 +1,10 @@ +//go:build !linux + +package processgroup + +// Off Linux there is no /proc/self/cgroup and no sched_getaffinity to read here, +// so neither bound participates and [effectiveCores] falls back to the machine's +// logical CPU count. +func cgroupCPUCores() float64 { return 0 } + +func affinityCores() int { return 0 } diff --git a/internal/processgroup/usage.go b/internal/processgroup/usage.go new file mode 100644 index 0000000000..b3d418024b --- /dev/null +++ b/internal/processgroup/usage.go @@ -0,0 +1,33 @@ +package processgroup + +// A job subtree is a process group, and a process group is addressable — #1155 +// recorded every group at launch with its leader's identity precisely so it +// could be named later. That same handle is what a bound on a subtree's +// resource use needs: the leader's pid is the root of the tree, and the tree is +// exactly what the job started. +// +// WHAT THIS MEASURES, AND IN WHAT UNIT. Two things, because a bound in this +// code is stated in cores and processes and both have to be readable together: +// +// - Processes is how many processes are alive in the subtree at this instant. +// - CPUSeconds is the subtree's CUMULATIVE processor time (user plus system) +// over every process in it. A cumulative figure is what a RATE is taken +// from: the difference of two readings over the wall between them is the +// cores the subtree held, which is the reading this product bounds. One +// reading on its own is not a rate and must not be read as one. +// +// THE READING IS /proc, WHICH IS LINUX'S. A platform without it answers "cannot +// say" rather than a guess, and a bound that cannot say never cuts — the same +// silence rule the admission governor keeps (session's task_pressure.go): a +// person on a machine this cannot measure gets exactly the scheduler they had +// before the bound existed. + +// Usage is what one job subtree is using at one instant. +type Usage struct { + // Processes is how many processes are alive in the subtree. + Processes int + // CPUSeconds is the subtree's cumulative CPU time in seconds, summed over + // every process in it. Subtract two readings and divide by the wall between + // them to get the cores the subtree held. + CPUSeconds float64 +} diff --git a/internal/processgroup/usage_linux.go b/internal/processgroup/usage_linux.go new file mode 100644 index 0000000000..1fc8a37c66 --- /dev/null +++ b/internal/processgroup/usage_linux.go @@ -0,0 +1,111 @@ +//go:build linux + +package processgroup + +import ( + "os" + "strconv" + "strings" +) + +// userHZ is Linux's USER_HZ: the ticks per second the utime and stime fields of +// /proc//stat are counted in. It is 100 on every architecture this product +// ships to, x86-64 and arm64 included, so it is written rather than asked — a +// host read would return the same 100 everywhere a job can run. +const userHZ = 100 + +// Usage reports what the recorded process group's whole subtree is using, or +// says it cannot say. +// +// IT REFUSES WHEN THE IDENTITY NO LONGER MATCHES, for [Group.owned]'s reason: +// the pid may have been reused, and a reading of somebody else's process group +// is worse than no reading — it could cut a subtree that is not ours. A group +// that cannot be attributed is one this bound must leave alone. +func (g Group) Usage() (Usage, bool) { + if ok, _ := g.owned(); !ok { + return Usage{}, false + } + processes, ticks, ok := subtreeTicks(g.pid) + if !ok { + return Usage{}, false + } + return Usage{Processes: processes, CPUSeconds: float64(ticks) / userHZ}, true +} + +// subtreeTicks walks the process tree below root and sums its accumulated CPU +// ticks and its live process count. +// +// THE WALK FOLLOWS THE TREE DOWN, through each process's threads' `children` +// files, rather than reading every process on the machine and sorting out whose +// is whose: what it reads scales with what the job started, not with how busy +// the box is. This is session's task_pressure.go's walk with a CPU figure added; +// the kernel promises `children` exactly only for a stopped tree, so a process +// born or reaped mid-walk can be missed for one reading, which is the right +// precision for a figure re-read on every tick. +// +// `ok` is false only when the root itself could not be read: every other unread +// child is a process that went away between two reads, and one reading that +// missed it is corrected by the next. +func subtreeTicks(root int) (processes int, ticks int64, ok bool) { + seen := map[int]bool{} + for queue := []int{root}; len(queue) > 0; { + pid := queue[0] + queue = queue[1:] + if seen[pid] { + continue + } + seen[pid] = true + dir := "/proc/" + strconv.Itoa(pid) + if raw, err := os.ReadFile(dir + "/stat"); err == nil { + if utime, stime, read := processTicks(string(raw)); read { + ticks += utime + stime + processes++ + ok = true + } + } + threads, err := os.ReadDir(dir + "/task") + if err != nil { + continue + } + for _, thread := range threads { + raw, err := os.ReadFile(dir + "/task/" + thread.Name() + "/children") + if err != nil { + continue + } + for _, field := range strings.Fields(string(raw)) { + if child, err := strconv.Atoi(field); err == nil { + queue = append(queue, child) + } + } + } + } + return processes, ticks, ok +} + +// processTicks reads a process's utime and stime (fields 14 and 15) out of one +// /proc//stat line. +// +// comm (field 2) is parenthesised and may itself contain spaces and +// parentheses, so the fields are counted from after its LAST ')': the state +// (field 3) is then index 0, utime is index 11 and stime is index 12. This is +// the same off-by-the-parens counting identity_linux.go already does for the +// start time. +func processTicks(raw string) (utime, stime int64, ok bool) { + close := strings.LastIndexByte(raw, ')') + if close < 0 { + return 0, 0, false + } + fields := strings.Fields(raw[close+1:]) + if len(fields) <= 12 { + return 0, 0, false + } + user, err := strconv.ParseInt(fields[11], 10, 64) + if err != nil { + return 0, 0, false + } + system, err := strconv.ParseInt(fields[12], 10, 64) + if err != nil { + return 0, 0, false + } + return user, system, true +} diff --git a/internal/processgroup/usage_linux_test.go b/internal/processgroup/usage_linux_test.go new file mode 100644 index 0000000000..e9de44add2 --- /dev/null +++ b/internal/processgroup/usage_linux_test.go @@ -0,0 +1,35 @@ +//go:build linux + +package processgroup + +import ( + "os" + "testing" +) + +// TestUsageReadsALiveProcessGroup is the reader's own contract: a group whose +// leader is still this process is read, and the reading carries a live process +// and a cumulative CPU figure. NOTHING IS SPAWNED AND NOTHING IS SIGNALLED — +// the group is the test process's own and the reading is /proc. +func TestUsageReadsALiveProcessGroup(t *testing.T) { + group := CaptureGroup(os.Getpid()) + usage, ok := group.Usage() + if !ok { + t.Fatal("a live process group could not be read") + } + if usage.Processes < 1 { + t.Fatalf("a live group reported %d processes, want at least itself", usage.Processes) + } + if usage.CPUSeconds < 0 { + t.Fatalf("a group reported %v CPU seconds", usage.CPUSeconds) + } +} + +// TestUsageRefusesAGroupItCannotAttribute is the safe half: a group whose pid +// is not this process's own — identity cannot be checked — says it cannot say, +// so the bound never cuts somebody else's tree. +func TestUsageRefusesAGroupItCannotAttribute(t *testing.T) { + if usage, ok := (Group{}).Usage(); ok { + t.Fatalf("a group with no leader was read anyway: %+v", usage) + } +} diff --git a/internal/processgroup/usage_other.go b/internal/processgroup/usage_other.go new file mode 100644 index 0000000000..4ecf7f23a7 --- /dev/null +++ b/internal/processgroup/usage_other.go @@ -0,0 +1,10 @@ +//go:build !linux + +package processgroup + +// Usage has no answer where there is no /proc to read: the subtree's CPU time +// and process count are Linux's to report here. A platform that cannot say +// answers so, and a bound that cannot say never cuts — it gets the scheduler it +// had before the bound existed rather than a guess at somebody else's process +// group. +func (g Group) Usage() (Usage, bool) { return Usage{}, false } diff --git a/internal/session/jobbound.go b/internal/session/jobbound.go new file mode 100644 index 0000000000..8248be1fc7 --- /dev/null +++ b/internal/session/jobbound.go @@ -0,0 +1,261 @@ +package session + +// A JOB SUBTREE HAS NO CEILING ON WHAT IT BURNS. +// +// #1158 gave a run a soft bound on its MEMORY. It gave it nothing on its CPU +// and nothing on how many processes it may hold, and those are the two a run +// that never spends a token can spend instead. The measured case: a model +// spawned busy loops as bash jobs — `yes`, `awk 'BEGIN{for(;;){}}'` — and one +// attempt drove 64 of them and the next 16, load 30-43 on a 20-core box. +// Nothing inside this program stopped it; only a teardown outside and a +// signature guard did, because every bound the run carried was stated in +// DOLLARS and a busy loop costs none. +// +// So the job subtree — the process group #1155 already records at launch with +// its leader's identity — gets a bound of its own, and a subtree that passes it +// is CUT and the run is TOLD, never stopped silently. +// +// ── WHERE EACH HALF OF THE BOUND APPLIES ── +// +// PROCESSES, ALWAYS. A subtree may not hold more than jobBoundProcesses (128, +// about four times the honest peak of 34). This is the fork storm's ceiling and +// it applies to every subtree, quota'd or not, because a fork that has not yet +// accumulated CPU is invisible to a rate and a cgroup CPU quota does nothing to +// a process count. +// +// CPU, ONLY WHERE NOTHING ELSE CAPS IT. When a cgroup CPU quota or an affinity +// mask BINDS this process — the effective cores it may use are fewer than the +// machine's ([processgroup.EffectiveCores] below runtime.NumCPU) — the quota is +// already the ceiling on what any subtree under it can burn, and #1187's park +// bound is what returns a turn wedged on a job that never ends. Adding a CPU +// rate rule there would only cut honest work early: inside a four-core cell an +// ordinary `go test` legitimately uses most of its four cores, and a share of +// them would end it. So the CPU rule is OFF under a binding quota, and only the +// process backstop applies. +// +// When NOTHING binds — the process may use the whole machine — there is no other +// ceiling, and the CPU rule is the only thing between a spinner storm and the +// box: a subtree may not sustain more than THREE FIFTHS of runtime.NumCPU. On +// the 20-core box the numbers came from that is 12 cores, clearing the honest +// peak (8.5) and cutting the storm (16 loops → 16 cores, 64 → 20). +// +// ── HOW THE CPU RATE IS TAKEN ── +// +// The rate is taken from two readings of the subtree's cumulative CPU over the +// wall between them, and the reading is the subtree's WHOLE tree — the shell +// and every compiler, spinner or child under it — so a command that forks is +// bounded as one job and not as its leader alone. +// +// ── SUSTAINED OVER HALF A MINUTE, NOT A SPIKE ── +// +// A build burst can pass 12 cores for several seconds on a big box, so the bound +// trips only after the subtree has been over on jobBoundStrikes consecutive +// readings — 15 at jobBoundInterval of 2 s, thirty seconds sustained. The +// transient never cuts anything and a real storm, over on every reading by +// construction, is cut half a minute in. +// +// ── PORTABLE FIRST ── +// +// The watch is this program's own and runs everywhere. It reads the subtree +// through the process-group handle #1155 gave, which works wherever a process +// group is a process group; the CPU-and-process reading itself is /proc, so off +// Linux it says it cannot say and the bound never cuts ([processgroup.Usage]). +// The subtree's USAGE needs no cgroup — a job is a process group and not a +// cgroup. Whether a quota BINDS is a different question — how many cores the +// subtree could ever reach — answered from the process's own cgroup CPU quota and +// affinity mask ([processgroup.EffectiveCores]), which off Linux is the machine +// count, so nothing binds and the CPU rule governs as it always did. + +import ( + "math" + "runtime" + "strconv" + "syscall" + "time" + + "github.com/Agent-Field/codeaf/internal/processgroup" +) + +const ( + // jobBoundCoreShare is the share of the machine's cores a job subtree may + // sustain before its bound trips, and it is read only when NOTHING caps the + // process below the machine (see the file header). + jobBoundCoreShare = 3.0 / 5.0 + + // jobBoundProcesses is the process ceiling on one subtree, and it applies + // always. It is about four times the honest peak (34) so it is never what + // separates an honest run from the spinner storm, and it catches the fork + // storm the CPU rate alone would miss. + jobBoundProcesses = 128 + + // jobBoundStrikes is how many consecutive readings over a ceiling the bound + // waits for before it trips: at jobBoundInterval that is thirty seconds, long + // enough that an honest build burst is never cut and a storm still is. + jobBoundStrikes = 15 + + // jobBoundInterval is how often the bound samples a subtree. + jobBoundInterval = 2 * time.Second + + // jobBoundReason is what the run is handed when the bound trips: the family + // the settle turn and the park speak in ([taskAskSettleReason], + // [jobParkBoundReason]), said about the subtree. The work did not settle + // within its bound, so it was ended — said out loud, never as a silent kill. + jobBoundReason = "the job subtree was not settled within its bound" + + // jobBoundRule is the one clause the record carries about itself: the subtree + // is not still running and nothing is coming for it, so a model must not go + // looking for the job or run the command again. + jobBoundRule = "— it was ended for passing its bound: a job subtree may not sustain more than its share of the machine's cores, or grow past its process ceiling, for long. The command was cut, not waited on; do not go looking for it." +) + +// jobSubtreeUsage reads one job subtree's usage. It is a variable so the bound +// can be exercised against injected readings — the storm a test states is a +// sequence of numbers and never real load, which on a shared box is the only +// safe way to prove a bound. +var jobSubtreeUsage = func(group processgroup.Group) (processgroup.Usage, bool) { + return group.Usage() +} + +// jobBoundNow is the clock the bound reads its intervals against. It is a +// variable for [jobSubtreeUsage]'s reason: a test drives the interval without a +// sleep standing in for causality. +var jobBoundNow = time.Now + +// subtreeBound is the running state of the bound on ONE job subtree: the machine +// it runs on, the cores it may actually use, the previous reading a rate is taken +// from, and how many readings in a row have been over. +type subtreeBound struct { + // machineCores is runtime.NumCPU: the whole machine, and the number the CPU + // ceiling is drawn from when nothing caps the process below it. Zero turns the + // bound off — a machine the bound cannot size is one it must not cut on. + machineCores float64 + // effectiveCores is the smallest of the machine, the cgroup CPU quota and the + // affinity mask: how many cores this subtree could ever reach. When it is + // below machineCores a quota or mask BINDS, and the CPU rule is off. + effectiveCores float64 + + last time.Time + lastCPU float64 + have bool + strikes int + // lastCores is the rate the last reading computed, kept so the record can + // name it. + lastCores float64 +} + +// newSubtreeBound builds the bound for the machine this process runs on and the +// cores it can actually use. +func newSubtreeBound() *subtreeBound { + return &subtreeBound{ + machineCores: float64(runtime.NumCPU()), + effectiveCores: processgroup.EffectiveCores(), + } +} + +// quotaBinds reports whether a cgroup CPU quota or an affinity mask holds this +// subtree below the machine's cores. When it does, the quota is already the +// ceiling and the CPU rate rule is off — only the process backstop applies. +func (b *subtreeBound) quotaBinds() bool { + return b.effectiveCores > 0 && b.effectiveCores < b.machineCores +} + +// ceilingCores is the CPU ceiling this subtree may sustain, or +Inf when a quota +// binds and the CPU rule does not apply. +func (b *subtreeBound) ceilingCores() float64 { + if b.quotaBinds() { + return math.Inf(1) + } + return jobBoundCoreShare * b.machineCores +} + +// strike records one reading of the subtree and reports whether the bound has +// tripped. A rate needs two readings, so the first reading can only be judged on +// its process count; that is why the CPU half takes a tick to arm. +func (b *subtreeBound) strike(now time.Time, usage processgroup.Usage, ok bool) bool { + if !ok || b.machineCores <= 0 { + // SILENCE IS NEVER A CUT. A subtree this machine cannot read gets the + // scheduler it had before the bound existed. + b.strikes, b.have = 0, false + return false + } + cores := 0.0 + if b.have && now.After(b.last) { + cores = (usage.CPUSeconds - b.lastCPU) / now.Sub(b.last).Seconds() + } + b.last, b.lastCPU, b.have, b.lastCores = now, usage.CPUSeconds, true, cores + if usage.Processes > jobBoundProcesses || cores > b.ceilingCores() { + b.strikes++ + } else { + b.strikes = 0 + } + return b.strikes >= jobBoundStrikes +} + +// boundStep takes one reading of one job subtree and, when the bound trips, ends +// the subtree and hands the run the record. It reports whether the bound tripped. +// It is the whole body of the watch loop, factored out so a test drives it with +// injected readings and no goroutine and no clock. +func (r *jobRegistry) boundStep(one *job, bound *subtreeBound) bool { + usage, ok := jobSubtreeUsage(one.group) + if !bound.strike(jobBoundNow(), usage, ok) { + return false + } + r.cutSubtreeBound(one, usage, bound.lastCores) + return true +} + +// watchSubtreeBound samples one job subtree on its own beat until the subtree +// ends or the bound cuts it. It runs for the life of the job and stops when the +// job does ([job.done]). +func (r *jobRegistry) watchSubtreeBound(one *job) { + // ONLY A PROCESS SUBTREE HAS ONE. A watch, a task node and a render are jobs + // for everything around them but none of them is a process group with a tree + // under it, so there is nothing here for the bound to measure. + if one.kind != jobKindBash { + return + } + bound := newSubtreeBound() + ticks := time.NewTicker(jobBoundInterval) + defer ticks.Stop() + for { + select { + case <-one.done: + return + case <-ticks.C: + if r.boundStep(one, bound) { + return + } + } + } +} + +// cutSubtreeBound ends a subtree that passed its bound and hands the run the +// record of why. +// +// IT IS A KILL THIS SESSION ASKED FOR, so the job's own ending reports nothing: +// a requested death speaks for itself only through the record written here, and +// a second note from [jobRegistry.settleExit] would be the registry narrating +// what this line just explained. The kin is the park's bound ([parkBoundNote]) — +// the news reaches the model on the same lane, in the same family. +func (r *jobRegistry) cutSubtreeBound(one *job, usage processgroup.Usage, cores float64) { + if !one.requestKill() { + // The job is already ending under somebody else's hand; there is nothing + // left to cut and no reason to say anything twice. + return + } + one.signal(syscall.SIGKILL) + if r.notify == nil { + return + } + r.notify(jobBoundRecord(one, usage, cores)) +} + +// jobBoundRecord is the record a subtree's bound hands the run. It names the +// subtree, says what it was holding, and carries the family's rule about itself. +func jobBoundRecord(one *job, usage processgroup.Usage, cores float64) string { + held := strconv.Itoa(usage.Processes) + " processes" + if cores > 0 { + held += ", " + strconv.FormatFloat(cores, 'f', 1, 64) + " cores" + } + return jobBoundReason + ": job " + strconv.Itoa(one.id) + " was holding " + held + "\n" + jobBoundRule +} diff --git a/internal/session/jobbound_test.go b/internal/session/jobbound_test.go new file mode 100644 index 0000000000..cfa0898075 --- /dev/null +++ b/internal/session/jobbound_test.go @@ -0,0 +1,232 @@ +package session + +// THE STORM THAT SPENT NO TOKENS. A run whose model spawns busy loops as bash +// jobs (`yes`, `awk 'BEGIN{for(;;){}}'`) burns no provider calls, so nothing in +// the run's dollar bound ever fires while the box is pegged. #1158 bounds a +// run's MEMORY and nothing bounds its CPU or its process count. +// +// Two halves, applying in different places (see jobbound.go): the PROCESS +// backstop always, and the CPU rate rule only where NOTHING else caps the +// process — because under a cgroup quota the quota is already the ceiling and a +// CPU rule would cut an honest `go test` inside a cell. +// +// ── NOTHING HERE MAKES LOAD ── +// +// The storm is a sequence of numbers. The subtree's usage reader and the clock +// are the code's own seams ([jobSubtreeUsage], [jobBoundNow]), and the machine +// and effective cores are fields on the bound, so the whole test runs injected +// readings with no `yes`, no `awk`, no cgroup and no subprocess at all — on a +// shared box, a bound proved by real load is the wrong proof and a danger to +// whoever else is working. + +import ( + "strings" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/processgroup" +) + +// subtreeReadings swaps the bound's usage reader and clock for scripted ones. +// Each read hands back the next reading and advances the clock by one step, so +// the rate the bound computes is exactly the scripted one and no wall time +// stands in for causality. +func subtreeReadings(t *testing.T, readings []processgroup.Usage, step time.Duration) { + t.Helper() + originalUsage, originalNow := jobSubtreeUsage, jobBoundNow + t.Cleanup(func() { jobSubtreeUsage, jobBoundNow = originalUsage, originalNow }) + at := time.Unix(0, 0) + index := 0 + jobBoundNow = func() time.Time { return at } + jobSubtreeUsage = func(processgroup.Group) (processgroup.Usage, bool) { + usage := readings[len(readings)-1] + if index < len(readings) { + usage = readings[index] + index++ + } + at = at.Add(step) + return usage, true + } +} + +// storm is a subtree that holds `cores` cores and `processes` processes for +// `samples` readings. Cumulative CPU climbs by cores×step each step, which is +// what a busy loop looks like from the outside and what the bound measures. +func storm(cores float64, processes, samples int, step time.Duration) []processgroup.Usage { + readings := make([]processgroup.Usage, samples) + for i := range readings { + readings[i] = processgroup.Usage{ + Processes: processes, + CPUSeconds: cores * step.Seconds() * float64(i), + } + } + return readings +} + +// driveBound takes reading after reading until the bound trips or the script is +// spent, and answers whether it tripped. +func driveBound(registry *jobRegistry, one *job, bound *subtreeBound, samples int) bool { + for i := 0; i < samples; i++ { + if registry.boundStep(one, bound) { + return true + } + } + return false +} + +// boundWith is the bound as it would be drawn on a machine with `machine` cores +// where this subtree may actually use `effective` of them — the seam that stands +// a machine and a quota in front of the ceiling without a cgroup to be under. +// effective below machine is a quota (or affinity) that BINDS. +func boundWith(effective, machine float64) *subtreeBound { + bound := newSubtreeBound() + bound.machineCores = machine + bound.effectiveCores = effective + return bound +} + +// newTestBound is the bound on the unconstrained box the numbers were measured +// on: 20 cores, nothing capping it, so the CPU ceiling is 12 cores (three fifths +// of 20), above the honest 8.5 and below the storm's 16. +func newTestBound() *subtreeBound { + return boundWith(20, 20) +} + +// long is enough readings to cross the thirty-second window (jobBoundStrikes at +// jobBoundInterval) with room to spare. +const long = jobBoundStrikes + 5 + +// TestAJobSubtreeOverItsBoundIsCutAndTheModelIsTold is the load-bearing half on +// an unconstrained box: the storm is cut, and the run hears about it in the +// bound's own voice rather than a silent kill. +func TestAJobSubtreeOverItsBoundIsCutAndTheModelIsTold(t *testing.T) { + var notes []string + registry := &jobRegistry{notify: func(note string) { notes = append(notes, note) }} + one := &job{id: 7, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + + subtreeReadings(t, storm(16, 16, long, time.Second), time.Second) + + if !driveBound(registry, one, newTestBound(), long) { + t.Fatal("a 16-loop storm never tripped the bound: 16 cores is above the honest peak and must be cut") + } + if !one.killRequested { + t.Fatal("the storm was not cut: no kill was requested for the subtree") + } + if len(notes) != 1 { + t.Fatalf("the run was handed %d records, want exactly one", len(notes)) + } + if !strings.Contains(notes[0], jobBoundReason) { + t.Fatalf("the record is not in the bound's own family: %q", notes[0]) + } + if !strings.Contains(notes[0], "job 7") { + t.Fatalf("the record does not name the subtree it is about: %q", notes[0]) + } +} + +// TestAnHonestVerifyOnAnUnconstrainedBoxIsNotCut is the other half of the +// unquota'd rule: a max-tier verify measured 34 processes and 8.5 cores at once, +// and that must survive untouched. +func TestAnHonestVerifyOnAnUnconstrainedBoxIsNotCut(t *testing.T) { + var notes []string + registry := &jobRegistry{notify: func(note string) { notes = append(notes, note) }} + one := &job{id: 3, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + + subtreeReadings(t, storm(8.5, 34, long, time.Second), time.Second) + + if driveBound(registry, one, newTestBound(), long) { + t.Fatal("an honest parallel verify of 34 processes and 8.5 cores was cut") + } + if one.killRequested { + t.Fatal("a kill was requested for an honest parallel verify") + } + if len(notes) != 0 { + t.Fatalf("an honest run was handed %d records, want none: %q", len(notes), notes) + } +} + +// TestASustainedStormOnAnUnconstrainedBoxIsCut pins the thirty-second window: a +// subtree over the ceiling for the whole window is cut. +func TestASustainedStormOnAnUnconstrainedBoxIsCut(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + subtreeReadings(t, storm(13, 8, long, time.Second), time.Second) + if !driveBound(registry, one, newTestBound(), long) { + t.Fatal("a 13-core storm sustained past the window never tripped on an unconstrained box") + } +} + +// TestAShortBurstOnAnUnconstrainedBoxIsNotCut is why the window is thirty +// seconds: an honest build can pass twelve cores for several seconds, and a +// burst shorter than the window must not be cut. +func TestAShortBurstOnAnUnconstrainedBoxIsNotCut(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + burst := jobBoundStrikes - 3 // over the ceiling, but for fewer than the window's readings + subtreeReadings(t, storm(13, 8, burst, time.Second), time.Second) + if driveBound(registry, one, newTestBound(), burst) { + t.Fatal("a 13-core burst shorter than the thirty-second window was cut") + } +} + +// TestAFullQuotaStormIsNotCutByTheCPURule is the correction's heart: inside a +// four-core cell the quota is already the ceiling, so a subtree using all four +// cores — which is what an honest `go test` there does too — is NOT cut by the +// CPU rule. #1187's park bound is what returns a turn wedged on such a job. +func TestAFullQuotaStormIsNotCutByTheCPURule(t *testing.T) { + var notes []string + registry := &jobRegistry{notify: func(note string) { notes = append(notes, note) }} + one := &job{id: 4, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + + subtreeReadings(t, storm(4.0, 8, long, time.Second), time.Second) + + if driveBound(registry, one, boundWith(4, 20), long) { + t.Fatal("a subtree at its four-core quota was cut by the CPU rule: under a binding quota the CPU rule must be off") + } + if one.killRequested { + t.Fatal("a kill was requested for a subtree inside its quota") + } + if len(notes) != 0 { + t.Fatalf("a quota'd subtree was handed %d records, want none: %q", len(notes), notes) + } +} + +// TestAProcessStormUnderAQuotaIsCutByTheBackstop is the half that still applies +// under a quota: a fork storm, which a CPU quota does nothing to, is caught by +// the process backstop. +func TestAProcessStormUnderAQuotaIsCutByTheBackstop(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + subtreeReadings(t, storm(0.5, 200, long, time.Second), time.Second) + if !driveBound(registry, one, boundWith(4, 20), long) { + t.Fatal("a 200-process storm under a four-core quota was not cut by the process backstop") + } +} + +// TestAForkStormIsCutEvenWithLittleCPU is the process backstop on an +// unconstrained box: a fork that has not yet accumulated CPU, caught by the +// count alone. +func TestAForkStormIsCutEvenWithLittleCPU(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + subtreeReadings(t, storm(0.5, 5000, long, time.Second), time.Second) + if !driveBound(registry, one, newTestBound(), long) { + t.Fatal("a fork storm of 5000 processes never tripped the bound") + } +} + +// TestASubtreeThatCannotBeReadIsNeverCut is the silence rule every reader here +// keeps (task_pressure.go's own): a platform that cannot say what a subtree is +// using never holds, never cuts. +func TestASubtreeThatCannotBeReadIsNeverCut(t *testing.T) { + originalUsage, originalNow := jobSubtreeUsage, jobBoundNow + t.Cleanup(func() { jobSubtreeUsage, jobBoundNow = originalUsage, originalNow }) + at := time.Unix(0, 0) + jobBoundNow = func() time.Time { at = at.Add(time.Second); return at } + jobSubtreeUsage = func(processgroup.Group) (processgroup.Usage, bool) { return processgroup.Usage{}, false } + + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + if driveBound(registry, one, newTestBound(), long) { + t.Fatal("a subtree whose usage could not be read was cut: silence is not pressure") + } +} diff --git a/internal/session/jobs.go b/internal/session/jobs.go index 32b3434460..afe00f61aa 100644 --- a/internal/session/jobs.go +++ b/internal/session/jobs.go @@ -754,6 +754,10 @@ func (r *jobRegistry) start(command string) (*job, error) { } go r.reap(started) + // AND THE SUBTREE GETS ITS OWN BOUND. A bash job is a process group with a + // tree under it, and nothing else bounds what that tree may burn; the watcher + // cuts it when it passes its ceiling and tells the run why (jobbound.go). + go r.watchSubtreeBound(started) return started, nil } @@ -943,6 +947,9 @@ func (r *jobRegistry) adopt(taken *bare.BashCall, how adoption) (*job, error) { // The receive happens INSIDE the goroutine: written as an argument it would // be evaluated here, and the adoption would block until the process exited. go func() { r.settleExit(started, <-taken.Exit()) }() + // The adopted process is a bash subtree like any other and gets the same + // bound ([jobRegistry.watchSubtreeBound]). + go r.watchSubtreeBound(started) return started, nil }