From 0706266a0e09896ac3ba0608b79ca373e888ff96 Mon Sep 17 00:00:00 2001 From: bompus Date: Fri, 6 Nov 2026 13:09:44 -0700 Subject: [PATCH 1/2] Re-vendor the unit observer from house-rules at a4f8dd4 Brings in the fixes made there since e2a593e: per-sample read timing so a stall in the final sample is caught, a missing pswpout counter fails a storage phase, budgets are compared before rounding, and limits can be set by flags. The file is the house-rules copy with sigflint's two header lines naming the new source commit; nothing else is adapted. --- bench/observe-unit.py | 239 +++++++++++++++++++++++++++++++++--------- 1 file changed, 187 insertions(+), 52 deletions(-) diff --git a/bench/observe-unit.py b/bench/observe-unit.py index 82de04f..5eccbcd 100644 --- a/bench/observe-unit.py +++ b/bench/observe-unit.py @@ -1,28 +1,35 @@ #!/usr/bin/env python3 -# Vendored from house-rules, skills/hr-benchmarking/scripts/unit-observer.py, at commit e2a593e +# Vendored from house-rules, skills/hr-benchmarking/scripts/unit-observer.py, at commit a4f8dd4 # (MIT, same author). Change it there, copy it here, and name the new commit in this line. """Continuous observer for one measured unit on a native Linux host. Run it beside a unit (a timed run, a phase) that executes inside one cgroup. It samples once a second until SIGTERM or SIGINT, takes one last sample at that moment, then writes one JSON summary to --output (and every sample to `.samples.jsonl`) and exits 0 when the unit may count, 1 -when it may not. It writes `.ready` after its first sample; start the unit after that. - -Limits are the host benchmarking rules'. Host scope: available memory at least 8 GiB in every -sample (unrounded); memory full PSI at most 10 ms and I/O full PSI at most 100 ms in any observation -interval; coverage with no gap over 2.5 s (monotonic clock) and at least two samples; PSI must be -readable. Competing CPU: the mean over the whole run of host busy cores minus the workload -cgroup's own CPU (usage_usec, descendants included) is at most 4; with --host-only there is no -workload to subtract, so the unit's own CPU counts. Workload scope (the cgroup given with ---workload-cgroup, or `auto` for the observer's parent's cgroup): the same pressure limits on its -own pressure files; in every sample no growth of memory.swap.current or memory.zswap.current over -the first sample, no change of zswpin, zswpout or zswpwb and no change of memory.events high, max, -oom or oom_kill from the first sample, the same directory (device and inode), readable required -counters, and the same set of optional counters (zswap) as in the first sample. A counter that goes -backward is a failure. +when it may not. It writes `.ready` after its first sample; start the unit after that. A +first sample that already fails writes no `.ready`: the observer exits 1 with the reasons in the +summary. + +Limits are these defaults; a protocol that declares others applies its own, through --max-gap-s, +--min-avail-gib, --mem-psi-ms, --io-psi-ms and --max-competing-cores. Host scope: available memory +at least 8 GiB in every sample (unrounded); memory full PSI at most 10 ms and I/O full PSI at most +100 ms per observation interval (each measured interval is scaled to one second by the time +between the reads of that total, and none counts as shorter than half a second); coverage with no +gap over 2.5 s (monotonic clock) and no sample whose reads took longer than that, and at least two +samples; PSI must be readable. Competing CPU: +the mean over the whole run of host busy cores minus the workload cgroup's own CPU (usage_usec, +descendants included) is at most 4; with --host-only there is no workload to subtract, so the +unit's own CPU counts. Workload scope (the cgroup given with --workload-cgroup, or `auto` for the +observer's parent's cgroup): the same pressure limits on its own pressure files; in every sample +no growth of memory.swap.current or memory.zswap.current over the first sample, no change of +zswpin, zswpout or zswpwb and no change of memory.events high, max, oom or oom_kill from the first +sample, the same directory (device and inode), readable required counters, the same set of +optional counters (zswap) as in the first sample, and a memory.current or io.stat that stays +readable once it has been. A CPU or PSI counter that goes backward is a failure. Host-wide swap-out and zswap activity (pswpout, zswpout, zswpwb) is recorded for every interval -and rejects the unit only with --storage-phase, the phases the rules name. +and rejects the unit only with --storage-phase, the phases the rules name; a storage phase also +requires pswpout (and a zswap counter the first sample had) in every sample. Observer cost is recorded (CPU, resident peak, output bytes) and checked against budgets (--max-observer-cores, default 0.05, over the collection interval from the first to the last @@ -48,6 +55,12 @@ MIN_AVAIL_GIB = 8 MEM_PSI_MS = 10 IO_PSI_MS = 100 +# PSI limits are for one nominal observation interval of NOMINAL_INTERVAL_S seconds. Each measured +# interval's stall is scaled to it by the interval's length, and an interval counts as at least +# MIN_PSI_INTERVAL_S, so a stop right after a sample cannot turn a millisecond of stall into a +# rejection. +NOMINAL_INTERVAL_S = 1.0 +MIN_PSI_INTERVAL_S = 0.5 PROC = "/proc" stop = threading.Event() @@ -116,6 +129,11 @@ def own_cgroup(pid): path = line[3:] if path is None: raise OSError("no cgroup v2 path for the observer's parent") + if path.strip("/") == "": + raise OSError( + "the observer's parent is in the root cgroup, so `auto` would watch the whole host; " + "pass the workload cgroup's path" + ) for root in ("/sys/fs/cgroup", "/sys/fs/cgroup/unified"): if os.path.exists(f"{root}/cgroup.controllers"): return root + path @@ -125,7 +143,7 @@ def own_cgroup(pid): def optional_int(path): try: return int(read(path)[0]) - except (FileNotFoundError, ValueError, IndexError): + except (OSError, ValueError, IndexError): return None @@ -139,7 +157,7 @@ def io_bytes(path): if key in total: total[key] += int(value) return total - except (FileNotFoundError, ValueError): + except (OSError, ValueError): return None @@ -147,6 +165,10 @@ def workload_sample(path): """The workload cgroup's counters, or None when a required one is unreadable.""" try: st = os.stat(path) + mem_psi = psi_full(f"{path}/memory.pressure") + mem_psi_t = time.monotonic() # when each total was read, for scaling its delta + io_psi = psi_full(f"{path}/io.pressure") + io_psi_t = time.monotonic() events = {k: v for k, v in keyed(f"{path}/memory.events").items() if k in EVENTS} if set(events) != set(EVENTS): return None @@ -163,25 +185,55 @@ def workload_sample(path): "stat": {k: stat[k] for k in WORKLOAD_STAT if k in stat}, "swap": int(read(f"{path}/memory.swap.current")[0]), "zswap": zswap, - "mem_psi": psi_full(f"{path}/memory.pressure"), - "io_psi": psi_full(f"{path}/io.pressure"), + "mem_psi": mem_psi, + "io_psi": io_psi, + "mem_psi_t": mem_psi_t, + "io_psi_t": io_psi_t, } except (OSError, KeyError, ValueError, IndexError): return None def take(cg): - return { + sample = { "t": time.monotonic(), "wall": time.time(), "cpu": sum(os.times()[:2]), "busy_ticks": host_busy_ticks(), "mem_psi": psi_full(f"{PROC}/pressure/memory"), - "io_psi": psi_full(f"{PROC}/pressure/io"), - "avail_gib": avail_gib(), - "vm": vmstat(), - "workload": workload_sample(cg) if cg else None, } + sample["mem_psi_t"] = time.monotonic() # when each total was read, for scaling its delta + sample["io_psi"] = psi_full(f"{PROC}/pressure/io") + sample["io_psi_t"] = time.monotonic() + sample["avail_gib"] = avail_gib() + sample["vm"] = vmstat() + sample["workload"] = workload_sample(cg) if cg else None + sample["t_end"] = time.monotonic() # a read that blocks past the limit is not a clean sample + return sample + + +def slow_reads(sample): + return sample["t_end"] - sample["t"] > MAX_GAP_S + + +def first_sample_reasons(sample, cg, storage_phase): + """Reasons the first sample already fails, so a unit is not started that cannot count.""" + reasons = [] + if sample["mem_psi"] is None or sample["io_psi"] is None: + reasons.append("host pressure unreadable") + if sample["avail_gib"] < MIN_AVAIL_GIB: + reasons.append(f"available memory below {MIN_AVAIL_GIB:g} GiB") + if slow_reads(sample): + reasons.append(f"a sample's reads took longer than {MAX_GAP_S:g} s") + if storage_phase and "pswpout" not in sample["vm"]: + reasons.append("host swap counter unreadable in a storage phase") + if cg: + workload = sample["workload"] + if workload is None: + reasons.append("workload cgroup counters unreadable or the directory went missing") + elif workload["mem_psi"] is None or workload["io_psi"] is None: + reasons.append("workload pressure unreadable") + return reasons def judge(samples, cg, hz, args): @@ -195,8 +247,10 @@ def judge(samples, cg, hz, args): total = samples[-1]["t"] - samples[0]["t"] out.update(window_s=round(total, 1), max_gap_s=round(max(gaps), 2)) if min(gaps) <= 0 or max(gaps) > MAX_GAP_S: - reasons.append("sampling interval not positive or gap above 2.5 s") + reasons.append(f"sampling interval not positive or gap above {MAX_GAP_S:g} s") return reasons, out + if any(slow_reads(s) for s in samples): + reasons.append(f"a sample's reads took longer than {MAX_GAP_S:g} s") busy = [s["busy_ticks"] for s in samples] if any(b < a for a, b in zip(busy, busy[1:], strict=False)): reasons.append("host CPU counter went backward") @@ -205,37 +259,62 @@ def judge(samples, cg, hz, args): min_avail = min(s["avail_gib"] for s in samples) out["avail_gib_min"] = round(min_avail, 3) if min_avail < MIN_AVAIL_GIB: - reasons.append("available memory below 8 GiB") + reasons.append(f"available memory below {MIN_AVAIL_GIB:g} GiB") reasons += pressure_reasons( - [(s["mem_psi"], s["io_psi"]) for s in samples], gaps, out, "host", "host" + [(s["mem_psi"], s["mem_psi_t"], s["io_psi"], s["io_psi_t"]) for s in samples], + out, + "host", + "host", ) swaps = [{k: b["vm"][k] - a["vm"][k] for k in a["vm"] if k in b["vm"]} for a, b in steps] out["host_swap_deltas_by_interval"] = swaps if args.storage_phase and any(v for d in swaps for v in d.values()): reasons.append("host swap-out or zswap activity in a storage phase") + # pswpout is required; a zswap counter is required once the first sample has it. + needed = {"pswpout", *samples[0]["vm"]} + if args.storage_phase and any(k not in x["vm"] for x in samples for k in needed): + reasons.append("host swap counter unreadable in a storage phase") workload = [s["workload"] for s in samples] w_cores = 0.0 if cg: if any(w is None for w in workload): reasons.append("workload cgroup counters unreadable or the directory went missing") else: - w_reasons, w_cores = workload_reasons(workload, gaps, total, out) + w_reasons, w_cores = workload_reasons(workload, total, out) reasons += w_reasons competing = host_cores - w_cores out["competing_cpu_cores_mean"] = round(competing, 2) if competing > MAX_COMPETING_CORES: - reasons.append("competing CPU above 4 cores") + reasons.append(f"competing CPU above {MAX_COMPETING_CORES:g} cores") return reasons, out -def pressure_reasons(pairs, gaps, out, label, key): - """Memory and I/O full PSI over each observation interval; both must be readable.""" - if any(m is None or i is None for m, i in pairs): +def pressure_reasons(rows, out, label, key): + """Memory and I/O full PSI over each observation interval; both must be readable. + + Each row is (memory total, time it was read, I/O total, time it was read). A delta is scaled by + the time between its own two reads, and reads further apart than MAX_GAP_S leave pressure + unobserved, which rejects. + """ + if any(r[0] is None or r[2] is None for r in rows): return [f"{label} pressure unreadable"] - mem = [b[0] - a[0] for a, b in zip(pairs, pairs[1:], strict=False)] - io = [b[1] - a[1] for a, b in zip(pairs, pairs[1:], strict=False)] + pairs = list(zip(rows, rows[1:], strict=False)) + mem = [b[0] - a[0] for a, b in pairs] + io = [b[2] - a[2] for a, b in pairs] if min(mem + io) < 0: return [f"{label} pressure counter went backward"] + mem_gaps = [b[1] - a[1] for a, b in pairs] + io_gaps = [b[3] - a[3] for a, b in pairs] + if max(mem_gaps + io_gaps) > MAX_GAP_S: + return [f"{label} pressure reads more than {MAX_GAP_S:g} s apart"] + mem = [ + d * NOMINAL_INTERVAL_S / max(g, MIN_PSI_INTERVAL_S) + for d, g in zip(mem, mem_gaps, strict=True) + ] + io = [ + d * NOMINAL_INTERVAL_S / max(g, MIN_PSI_INTERVAL_S) + for d, g in zip(io, io_gaps, strict=True) + ] out[f"{key}_mem_full_psi_ms_max"] = round(max(mem) / 1000, 2) out[f"{key}_io_full_psi_ms_max"] = round(max(io) / 1000, 2) if max(mem) > MEM_PSI_MS * 1000 or max(io) > IO_PSI_MS * 1000: @@ -243,7 +322,18 @@ def pressure_reasons(pairs, gaps, out, label, key): return [] -def workload_reasons(w, gaps, total, out): +def lost(values): + """True when a value that was readable is later not; one never readable is just unknown.""" + seen = False + for value in values: + if value is not None: + seen = True + elif seen: + return True + return False + + +def workload_reasons(w, total, out): reasons = [] first = w[0] cpu = [x["cpu_usec"] for x in w] @@ -267,13 +357,21 @@ def workload_reasons(w, gaps, total, out): reasons.append("workload memory.events high, max, oom or oom_kill changed") out["workload_zswap"] = "absent" if first["zswap"] is None else "present" currents = [x["memory_current"] for x in w] + if lost(currents): + reasons.append("workload memory.current became unreadable") out["workload_memory_current_peak_bytes"] = max(currents) if None not in currents else "unknown" + ios = [x["io_bytes"] for x in w] + if lost(ios): + reasons.append("workload io.stat became unreadable") io_first, io_last = first["io_bytes"], w[-1]["io_bytes"] out["workload_io_bytes_delta"] = ( {k: io_last[k] - io_first[k] for k in io_first} if io_first and io_last else "unknown" ) reasons += pressure_reasons( - [(x["mem_psi"], x["io_psi"]) for x in w], gaps, out, "workload", "workload" + [(x["mem_psi"], x["mem_psi_t"], x["io_psi"], x["io_psi_t"]) for x in w], + out, + "workload", + "workload", ) return reasons, cores @@ -300,17 +398,24 @@ def write_summary(args, samples, reasons, out, status): cpu_s = sum(os.times()[:2]) interval_cpu_s = samples[-1]["cpu"] - samples[0]["cpu"] span = max(samples[-1]["t"] - samples[0]["t"], 1e-9) + cores, rss_mib = interval_cpu_s / span, self_status("VmHWM") / 1024 summary.update( observer_cpu_s=round(cpu_s, 2), observer_cpu_interval_s=round(interval_cpu_s, 3), - observer_cpu_cores_mean=round(interval_cpu_s / span, 3), - observer_rss_peak_mib=round(self_status("VmHWM") / 1024, 1), + observer_cpu_cores_mean=round(cores, 3), + observer_rss_peak_mib=round(rss_mib, 1), samples_bytes=os.path.getsize(args.output + ".samples.jsonl"), ) - if summary["observer_cpu_cores_mean"] > args.max_observer_cores: - reasons.append("observer CPU over budget") - if summary["observer_rss_peak_mib"] > args.max_observer_rss_mib: - reasons.append("observer resident memory over budget") + # The budgets compare the figures before they are rounded for the summary. + if cores > args.max_observer_cores: + reasons.append( + f"observer CPU over budget: {cores:.10g} > {args.max_observer_cores:g} cores" + ) + if rss_mib > args.max_observer_rss_mib: + reasons.append( + "observer resident memory over budget: " + f"{rss_mib:.10g} > {args.max_observer_rss_mib:g} MiB" + ) summary["reasons"] = reasons summary["clean"] = not reasons with open(args.output, "w") as f: @@ -318,13 +423,29 @@ def write_summary(args, samples, reasons, out, status): return summary["clean"] +def positive(text): + """An argparse type for an admission limit: a finite number above zero.""" + try: + value = float(text) + except ValueError: + value = float("nan") + if not value > 0 or value == float("inf"): + raise argparse.ArgumentTypeError(f"must be a positive number, got {text!r}") + return value + + def main(): - global PROC + global PROC, MAX_GAP_S, MAX_COMPETING_CORES, MIN_AVAIL_GIB, MEM_PSI_MS, IO_PSI_MS ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--output", default=os.environ.get("OBSERVER_OUT")) ap.add_argument("--workload-cgroup", default=os.environ.get("WORKLOAD_CGROUP")) ap.add_argument("--host-only", action="store_true") ap.add_argument("--storage-phase", action="store_true") + ap.add_argument("--max-gap-s", type=positive, default=MAX_GAP_S) + ap.add_argument("--min-avail-gib", type=positive, default=MIN_AVAIL_GIB) + ap.add_argument("--mem-psi-ms", type=positive, default=MEM_PSI_MS) + ap.add_argument("--io-psi-ms", type=positive, default=IO_PSI_MS) + ap.add_argument("--max-competing-cores", type=positive, default=MAX_COMPETING_CORES) ap.add_argument("--max-observer-cores", type=float, default=0.05) ap.add_argument("--max-observer-rss-mib", type=float, default=128) ap.add_argument("--proc-root", default="/proc", help="for tests: where /proc/stat etc. live") @@ -332,6 +453,9 @@ def main(): if not args.output: ap.error("--output or OBSERVER_OUT is required") PROC = args.proc_root + MAX_GAP_S, MIN_AVAIL_GIB = args.max_gap_s, args.min_avail_gib + MEM_PSI_MS, IO_PSI_MS = args.mem_psi_ms, args.io_psi_ms + MAX_COMPETING_CORES = args.max_competing_cores for stale in (args.output, args.output + ".ready", args.output + ".samples.jsonl"): if os.path.exists(stale): os.remove(stale) @@ -347,20 +471,31 @@ def main(): if args.workload_cgroup: cg = args.workload_cgroup.rstrip("/") if cg == "auto": - cg = own_cgroup(os.getppid()) + parent = os.getppid() + if parent == 1: + raise OSError( + "`auto` needs a parent inside the workload cgroup, but the parent is pid 1 " + "(a double fork or a service started outside the unit); pass the path" + ) + cg = own_cgroup(parent) if not cg or not os.path.isdir(cg): raise OSError(f"the workload cgroup is not a directory: {cg!r}") out.update(scope="workload-cgroup", workload_cgroup=cg) elif not args.host_only: reasons.append("no workload cgroup given") samples.append(take(cg)) - open(args.output + ".ready", "w").close() - while not stop.wait(1.0): - samples.append(take(cg)) - samples.append(take(cg)) # the boundary at the stop, so the unit's tail is covered - judged, figures = judge(samples, cg, hz, args) - reasons += judged - out.update(figures) + doomed = first_sample_reasons(samples[0], cg, args.storage_phase) + if doomed: + # No .ready: the driver must not start a unit this run cannot count. + reasons += doomed + else: + open(args.output + ".ready", "w").close() + while not stop.wait(1.0): + samples.append(take(cg)) + samples.append(take(cg)) # the boundary at the stop, so the unit's tail is covered + judged, figures = judge(samples, cg, hz, args) + reasons += judged + out.update(figures) except (OSError, ValueError, KeyError, IndexError) as error: status = "failed" reasons.append(f"observer error: {type(error).__name__}: {error}") From 6a7e7acc6dcc9acd38c8252e890303ec96af311c Mon Sep 17 00:00:00 2001 From: bompus Date: Fri, 6 Nov 2026 15:26:25 -0700 Subject: [PATCH 2/2] Note the observer re-vendor beside the 1.0.4 record A dated note says which observer the 1.0.4 campaign ran (the e2a593e copy, still in git history at the freeze commit) and what the new copy does differently, so the protocol's statement about frozen code is not read as covering the file now in the tree. No result or frozen record is edited. --- bench/results/observer-revendor-20261106.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) create mode 100644 bench/results/observer-revendor-20261106.md diff --git a/bench/results/observer-revendor-20261106.md b/bench/results/observer-revendor-20261106.md new file mode 100644 index 0000000..b903e91 --- /dev/null +++ b/bench/results/observer-revendor-20261106.md @@ -0,0 +1,17 @@ +# Unit observer re-vendored, 2026-11-06 + +`bench/observe-unit.py` was replaced with the house-rules unit observer at commit `a4f8dd4`; its header names that commit. Before, it was the copy at `e2a593e`. Nothing else was adapted. No measurement was run, and no recorded result changes. + +## What the release 1.0.4 record says about the observer + +The 1.0.4 campaign ran the copy at `e2a593e`: the file's git blob at the freeze commit `2685da5` is the same as the blob just before this change (`82de04f2`). Read the statement in `protocol-release-1.0.4.md` that the frozen `bench/` code is unchanged since `2685da5` as covering the observer that ran; the file in the tree now differs from it. To check that record against its observer, use `git show 2685da5:bench/observe-unit.py`. + +## What the new observer does differently + +- It scales each pressure interval to one second before comparing it with the limit (memory full 10 ms, I/O full 100 ms), dividing by the time between the two reads with a floor of 0.5 s, and rejects reads more than a set gap apart. The protocol's "pressure within the same limits per interval" describes the limits, which are unchanged; the old copy compared the unscaled difference between reads. +- It times the reads of each sample, so a stall in the final sample is caught. +- A first sample that already fails writes no `.ready` file, so `run-admitted.sh` reports no first sample and does not start the unit. +- A storage phase requires the `pswpout` counter. +- Budgets are compared before rounding. Limits can be set with flags; the defaults are the ones the protocols state, and the observer's own cost limits are still 0.05 cores and 128 MiB. + +`run-admitted.sh` uses only `OBSERVER_OUT`, `--workload-cgroup auto`, the `.ready` file and the top-level `clean` key, and those are unchanged.