diff --git a/tests/benchmarks/crucible.py b/tests/benchmarks/crucible.py new file mode 100644 index 0000000..b1b7fe0 --- /dev/null +++ b/tests/benchmarks/crucible.py @@ -0,0 +1,412 @@ +"""THE CRUCIBLE -- McRogueFace headless wall-clock microbenchmark. + +The Gauntlet answers "how much load until the frame budget breaks?" by ramping +until failure -- it needs a display, varies run-to-run with desktop noise, and +(pre-2026-07-11) could ramp a viewport-bounded grid straight into OOM. The +Crucible answers a different, complementary question: + + "How many milliseconds of CPU does the engine spend on a FIXED, comically + extreme but tractable amount of work?" + +Every benchmark runs a fixed configuration to completion and times it with +time.perf_counter. No ramp, no display, no rendering -- so it is: + + * SAFE -- every config is sized to fit in well under 512 MB and finish + in a couple of seconds; it cannot OOM the machine. + * HEADLESS -- pure engine CPU paths (allocation, cell writes, TCOD sync, + turn manager, pathfinding, FOV, layer writes). No frame_time. + * DETERMINISTIC -- fixed seeds; the same build produces the same work every + run, so A/B deltas between builds are real signal, not noise. + +This is the right tool for comparing two builds (e.g. current vs the 0.2.8 +release artifact) on the CPU-bound paths the #329/#332/#348 perf work touched. + +Run (this build): + ./mcrogueface --headless --exec ../tests/benchmarks/crucible.py + +Write JSON + compare against another build's JSON: + MCRF_CRUCIBLE_OUT=/tmp/cur.json ./mcrogueface --headless --exec .../crucible.py + MCRF_CRUCIBLE_BASELINE=/tmp/cur.json ./other/mcrogueface --headless --exec .../crucible.py + +Env vars: + MCRF_CRUCIBLE_OUT path to write this run's JSON (optional) + MCRF_CRUCIBLE_BASELINE path to a prior run's JSON to diff against (optional) + MCRF_CRUCIBLE_REPS override repetition count multiplier (default 1) + MCRF_CRUCIBLE_ONLY comma-separated benchmark names to run (default all) +""" +import os +import sys +import gc +import json +import time +import random + +import mcrfpy + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "gauntlet")) +try: + import safety + _rss = safety.rss_mb + _cap = safety.install_address_space_cap +except Exception: # pragma: no cover - safety is in-tree, but stay robust + def _rss(): + try: + with open("/proc/self/statm") as f: + return int(f.read().split()[1]) * 4096 / (1024.0 * 1024.0) + except Exception: + return 0.0 + + def _cap(cap_mb=2500): + return None + + +# -------------------------------------------------------------------------- +# Timing harness +# -------------------------------------------------------------------------- +class Bench: + """A named fixed-configuration benchmark. `run` does one full unit of work + and returns an integer 'work count' (for a work/ms figure). It is called + `reps` times; we report the FASTEST rep (least contended) in ms plus the + resident-memory high-water mark observed across reps.""" + + def __init__(self, name, desc, reps, fn): + self.name = name + self.desc = desc + self.reps = reps + self.fn = fn + + def measure(self, seed): + best_ms = None + total_work = 0 + rss_peak = _rss() + for i in range(self.reps): + gc.collect() + rng = random.Random(seed + i) + try: + t0 = time.perf_counter() + work = self.fn(rng) + dt_ms = (time.perf_counter() - t0) * 1000.0 + except (AttributeError, TypeError) as ex: + # A missing API on an older/newer build -- report honestly as + # unsupported rather than crashing the whole comparison run. + return { + "name": self.name, "desc": self.desc, "reps": self.reps, + "best_ms": 0.0, "work": 0, "us_per_work": 0.0, + "rss_peak_mb": round(_rss(), 1), + "status": "unsupported", "error": "%s: %s" % (type(ex).__name__, ex), + } + total_work = work + if best_ms is None or dt_ms < best_ms: + best_ms = dt_ms + r = _rss() + if r > rss_peak: + rss_peak = r + gc.collect() + return { + "name": self.name, + "desc": self.desc, + "reps": self.reps, + "best_ms": round(best_ms, 3), + "work": total_work, + "us_per_work": round(best_ms * 1000.0 / total_work, 4) if total_work else 0.0, + "rss_peak_mb": round(rss_peak, 1), + "status": "ok", + } + + +# -------------------------------------------------------------------------- +# Fixed workloads -- "comically extreme, but tractable" +# -------------------------------------------------------------------------- +def _open_grid(side, layers=False): + g = mcrfpy.Grid(grid_size=(side, side), texture=mcrfpy.default_texture) + if layers: + base = mcrfpy.TileLayer(z_index=-2, name="base", texture=mcrfpy.default_texture) + g.add_layer(base) + base.fill(0) + color = mcrfpy.ColorLayer(z_index=-1, name="overlay") + g.add_layer(color) + color.fill(mcrfpy.Color(20, 26, 38, 120)) + return g + + +def bench_grid_alloc(rng): + """Construct + destroy a big grid with two layers. Stresses #332 SoA plane + allocation + layer allocation + TCOD map alloc/free.""" + side = 600 + n = 12 + for _ in range(n): + g = _open_grid(side, layers=True) + del g + gc.collect() + return n * side * side # cells allocated + + +def bench_grid_fill(rng): + """Set walkable/transparent across every cell of a large grid. Each write + goes through the GridPoint wrapper -> setWalkable/Transparent + TCOD sync.""" + side = 400 + g = _open_grid(side) + for y in range(side): + for x in range(side): + p = g.at(x, y) + p.walkable = True + p.transparent = True + return side * side + + +def bench_layer_fill(rng): + """Bulk ColorLayer/TileLayer writes: full fills + many fill_rect windows.""" + side = 500 + g = _open_grid(side, layers=True) + color = g.layer("overlay") + tile = g.layer("base") + reps = 60 + for i in range(reps): + c = mcrfpy.Color(i % 256, (i * 3) % 256, (i * 7) % 256, 180) + ox = rng.randint(0, side - 64) + oy = rng.randint(0, side - 64) + color.fill_rect((ox, oy), (64, 64), c) + tile.fill_rect((ox, oy), (64, 64), (i % 10)) + return reps * 64 * 64 + + +def bench_layer_edit_buffer(rng): + """#335 edit() zero-copy buffer path. With numpy: whole-plane vectorized + writes. Without numpy (light build): a coarse memoryview stripe write so the + benchmark still exercises the buffer + __exit__ invalidation.""" + side = 400 + g = _open_grid(side, layers=True) + color = g.layer("overlay") + reps = 40 + try: + import numpy as np + for i in range(reps): + with color.edit() as view: + a = np.asarray(view) + a[:, :, 0] = (i * 5) % 256 + a[:, :, 3] = 200 + return reps * side * side + except ImportError: + for i in range(reps): + with color.edit() as view: + # touch one row per rep (cheap, still crosses the buffer boundary) + row = i % side + for x in range(side): + view[row, x, 0] = (x + i) % 256 + return reps * side + + +def bench_step_swarm(rng): + """Turn manager: many entities random-walking (NOISE4) for several turns. + Stresses executeBehavior + isCellWalkable + spatial hash + (post-fix) the + single view invalidation per step().""" + side = 60 + g = _open_grid(side) + for y in range(side): + for x in range(side): + p = g.at(x, y) + p.walkable = True + p.transparent = True + n_ent = 1000 + for _ in range(n_ent): + e = mcrfpy.Entity(grid_pos=(rng.randint(0, side - 1), rng.randint(0, side - 1)), + texture=mcrfpy.default_texture, sprite_index=84) + e.move_speed = 0 + g.entities.append(e) + e.set_behavior(mcrfpy.Behavior.NOISE4) + turns = 30 + g.step(n=turns) + return n_ent * turns + + +def bench_fov_storm(rng): + """Field of view recomputed from many origins on a large open grid.""" + side = 200 + g = _open_grid(side) + for y in range(side): + for x in range(side): + p = g.at(x, y) + p.walkable = True + p.transparent = True + # scatter some blocking walls so FOV actually casts + for _ in range(side * side // 20): + x = rng.randint(0, side - 1) + y = rng.randint(0, side - 1) + g.at(x, y).transparent = False + n = 1500 + for _ in range(n): + g.compute_fov((rng.randint(0, side - 1), rng.randint(0, side - 1)), radius=20) + return n + + +def bench_path_queries(rng): + """A* pathfinding between many random walkable pairs on an open grid.""" + side = 120 + g = _open_grid(side) + for y in range(side): + for x in range(side): + p = g.at(x, y) + p.walkable = True + p.transparent = True + n = 600 + found = 0 + for _ in range(n): + a = (rng.randint(0, side - 1), rng.randint(0, side - 1)) + b = (rng.randint(0, side - 1), rng.randint(0, side - 1)) + path = g.find_path(a, b) + if path is not None: + found += 1 + return n + + +def bench_entity_churn(rng): + """Add then remove many entities repeatedly -- spatial hash + cache churn.""" + side = 80 + g = _open_grid(side) + for y in range(side): + for x in range(side): + g.at(x, y).walkable = True + reps = 8 + n_ent = 800 + for _ in range(reps): + ents = [] + for _ in range(n_ent): + e = mcrfpy.Entity(grid_pos=(rng.randint(0, side - 1), rng.randint(0, side - 1)), + texture=mcrfpy.default_texture, sprite_index=84) + g.entities.append(e) + ents.append(e) + col = g.entities + while len(col): + col.remove(col[len(col) - 1]) + return reps * n_ent * 2 + + +BENCHES = [ + Bench("grid_alloc", "construct+destroy 600x600 grid w/2 layers x12", 3, bench_grid_alloc), + Bench("grid_fill", "set walkable+transparent on 400x400 cells", 3, bench_grid_fill), + Bench("layer_fill", "500x500 layers: fills + 60 fill_rect windows", 3, bench_layer_fill), + Bench("layer_edit_buffer", "#335 edit() buffer writes on 400x400 ColorLayer", 3, bench_layer_edit_buffer), + Bench("step_swarm", "1000 NOISE4 entities x 30 turns on 60x60", 3, bench_step_swarm), + Bench("fov_storm", "1500 compute_fov(r=20) on 200x200", 3, bench_fov_storm), + Bench("path_queries", "600 A* queries on open 120x120", 3, bench_path_queries), + Bench("entity_churn", "add+remove 800 entities x8 on 80x80", 3, bench_entity_churn), +] + + +# -------------------------------------------------------------------------- +# Reporting +# -------------------------------------------------------------------------- +def _git_short(): + # The engine forbids subprocess spawns; read HEAD directly. + try: + gitdir = os.path.normpath(os.path.join(HERE, "..", "..", ".git")) + head = open(os.path.join(gitdir, "HEAD")).read().strip() + if head.startswith("ref:"): + ref = head.split(" ", 1)[1] + return open(os.path.join(gitdir, ref)).read().strip()[:10] + return head[:10] + except Exception: + return "unknown" + + +def _print_table(results): + print("=" * 78) + print("THE CRUCIBLE -- wall-clock (fastest of %d reps), headless" % results[0]["reps"]) + print("-" * 78) + print("%-18s %10s %14s %10s %s" % ("bench", "best_ms", "us/work", "rss_mb", "desc")) + print("-" * 78) + for r in results: + if r.get("status") == "unsupported": + print("%-18s %10s %14s %10.1f %s" + % (r["name"], "UNSUPP", "-", r["rss_peak_mb"], r["desc"])) + else: + print("%-18s %10.2f %14.4f %10.1f %s" + % (r["name"], r["best_ms"], r["us_per_work"], r["rss_peak_mb"], r["desc"])) + print("=" * 78) + + +def _print_compare(results, baseline): + base_by = {b["name"]: b for b in baseline.get("benches", [])} + print("A/B vs baseline (%s @ %s) -- lower ms is better; -N%% = faster now" + % (baseline.get("version", "?"), baseline.get("commit", "?"))) + print("-" * 78) + print("%-18s %12s %12s %10s" % ("bench", "base_ms", "now_ms", "delta")) + print("-" * 78) + ratios = [] + for r in results: + b = base_by.get(r["name"]) + if r.get("status") == "unsupported" or (b and b.get("status") == "unsupported"): + print("%-18s %12s %12s %10s" % (r["name"], "--", "--", "unsupp")) + continue + if not b: + print("%-18s %12s %12.2f %10s" % (r["name"], "--", r["best_ms"], "new")) + continue + base_ms = b["best_ms"] + now_ms = r["best_ms"] + if base_ms > 0: + pct = (now_ms - base_ms) / base_ms * 100.0 + ratios.append(now_ms / base_ms) + tag = "%+.1f%%" % pct + else: + tag = "n/a" + print("%-18s %12.2f %12.2f %10s" % (r["name"], base_ms, now_ms, tag)) + print("-" * 78) + if ratios: + geo = 1.0 + for x in ratios: + geo *= x + geo = geo ** (1.0 / len(ratios)) + print("geomean now/base: %.3f (%.1f%% %s overall)" + % (geo, abs(geo - 1.0) * 100.0, "faster" if geo < 1.0 else "slower")) + print("=" * 78) + + +def main(): + cap = _cap() + if cap is not None: + print("[safety] address-space cap: %.0f MB" % cap) + + only = os.environ.get("MCRF_CRUCIBLE_ONLY", "").strip() + only_set = set(s.strip() for s in only.split(",") if s.strip()) if only else None + + results = [] + for b in BENCHES: + if only_set and b.name not in only_set: + continue + sys.stdout.write(" running %-18s ... " % b.name) + sys.stdout.flush() + res = b.measure(seed=1337) + results.append(res) + sys.stdout.write("%.2f ms (rss %.0f MB)\n" % (res["best_ms"], res["rss_peak_mb"])) + sys.stdout.flush() + + print("") + _print_table(results) + + record = { + "schema": 1, + "kind": "crucible", + "version": getattr(mcrfpy, "__version__", "?"), + "commit": _git_short(), + "benches": results, + } + + out = os.environ.get("MCRF_CRUCIBLE_OUT", "").strip() + if out: + with open(out, "w") as f: + json.dump(record, f, indent=2) + print("wrote %s" % out) + + base_path = os.environ.get("MCRF_CRUCIBLE_BASELINE", "").strip() + if base_path and os.path.exists(base_path): + with open(base_path) as f: + baseline = json.load(f) + print("") + _print_compare(results, baseline) + + sys.exit(0) + + +if __name__ == "__main__": + main() diff --git a/tests/benchmarks/gauntlet/run_gauntlet.py b/tests/benchmarks/gauntlet/run_gauntlet.py index 1735b46..87b8f82 100644 --- a/tests/benchmarks/gauntlet/run_gauntlet.py +++ b/tests/benchmarks/gauntlet/run_gauntlet.py @@ -17,6 +17,7 @@ import mcrfpy from gauntlet_main import Gauntlet, disable_vsync import baseline_io +import safety def _print_summary(record): @@ -37,6 +38,17 @@ def _print_summary(record): def main(): + # Machine-saver: hard-cap this process's address space so a runaway trial + # allocation aborts THIS process cleanly instead of exhausting system RAM + # and hard-locking the desktop (observed 2026-07-11). The per-trial + # predict_bytes / RSS guards should stop the ramp long before this fires; + # this is the backstop for when they don't. + cap = safety.install_address_space_cap() + if cap is not None: + print("[safety] address-space cap: %.0f MB (RSS ceiling %.0f MB)" + % (cap, safety.DEFAULT_RSS_CEILING_MB)) + sys.stdout.flush() + disable_vsync() app = Gauntlet(autorun=True) diff --git a/tests/benchmarks/gauntlet/safety.py b/tests/benchmarks/gauntlet/safety.py new file mode 100644 index 0000000..aa31baa --- /dev/null +++ b/tests/benchmarks/gauntlet/safety.py @@ -0,0 +1,69 @@ +"""Memory safety guards for The Gauntlet. + +The Gauntlet auto-ramps load geometrically until the frame budget breaks. Some +trials (GRID TITAN especially) scale COST but not necessarily FRAME TIME with +load: a huge SxS grid rendered into a fixed viewport can stay under the frame +budget while its cell/layer/TCOD-map allocation grows as S*S. With no memory +ceiling the ramp then walks straight into OOM -- observed hard-locking the +desktop (kernel OOM-killer / swap-thrash) mid-allocation. + +Defense in depth, weakest failure mode last: + +1. predict_bytes() -- a trial that can estimate its footprint refuses a load + whose predicted allocation exceeds the budget BEFORE allocating it. Graceful: + the ramp records the last passing load and stops. +2. RSS watchdog -- after any set_load, if resident memory exceeds the + ceiling the ramp stops gracefully. Catches trials that cannot predict. +3. RLIMIT_AS -- a hard address-space cap. If 1 and 2 are both outrun by a + single oversized allocation, the process dies cleanly (std::bad_alloc -> + abort) instead of taking the machine down. This is the machine-saver, not a + graceful path -- it exists so a bug in 1/2 can never again lock the desktop. + +Budgets are deliberately conservative. On a 24 GB box the default 1500 MB RSS +ceiling / 2500 MB address-space cap keep the whole run to a small fraction of +RAM; tune via env vars MCRF_GAUNTLET_RSS_MB / MCRF_GAUNTLET_AS_MB if needed. +""" +import os +import resource + +_PAGE = resource.getpagesize() + +DEFAULT_RSS_CEILING_MB = int(os.environ.get("MCRF_GAUNTLET_RSS_MB", "1500")) +DEFAULT_AS_CAP_MB = int(os.environ.get("MCRF_GAUNTLET_AS_MB", "2500")) + + +def rss_mb(): + """Resident set size of this process, in MB (Linux /proc/self/statm).""" + with open("/proc/self/statm") as f: + resident_pages = int(f.read().split()[1]) + return resident_pages * _PAGE / (1024.0 * 1024.0) + + +def vmsize_mb(): + """Virtual address-space size of this process, in MB.""" + with open("/proc/self/statm") as f: + vm_pages = int(f.read().split()[0]) + return vm_pages * _PAGE / (1024.0 * 1024.0) + + +def install_address_space_cap(cap_mb=DEFAULT_AS_CAP_MB): + """Hard-limit total address space so a runaway allocation aborts the process + cleanly instead of exhausting system RAM. Returns the effective cap in MB, or + None if the platform/hard-limit would not allow setting it. + + The cap is max(cap_mb, current_vmsize + 512) so we never set a limit below + what is already mapped (which would abort immediately on the next malloc).""" + try: + soft, hard = resource.getrlimit(resource.RLIMIT_AS) + except (ValueError, OSError): + return None + floor_mb = vmsize_mb() + 512.0 + target_mb = max(float(cap_mb), floor_mb) + target_bytes = int(target_mb * 1024 * 1024) + if hard != resource.RLIM_INFINITY and target_bytes > hard: + target_bytes = hard + try: + resource.setrlimit(resource.RLIMIT_AS, (target_bytes, hard)) + except (ValueError, OSError): + return None + return target_bytes / (1024.0 * 1024.0) diff --git a/tests/benchmarks/gauntlet/scoring.py b/tests/benchmarks/gauntlet/scoring.py index 14ecd22..8c80fa1 100644 --- a/tests/benchmarks/gauntlet/scoring.py +++ b/tests/benchmarks/gauntlet/scoring.py @@ -25,6 +25,17 @@ SETTLE_MS = 1000 HOLD_MS = 2000 MAX_STEPS = 40 # safety ceiling so a trial that never breaks budget still terminates +try: + from safety import rss_mb, DEFAULT_RSS_CEILING_MB +except ImportError: # allow importing scoring.py without the package on sys.path + def rss_mb(): + return 0.0 + DEFAULT_RSS_CEILING_MB = 1500 + +# Grid-data budget for a single trial's predicted footprint (bytes). A load +# whose predict_bytes() exceeds this is refused before allocation. +MEM_BUDGET_MB = 512 + def percentile(values, q): """Nearest-rank percentile of an unsorted list. q in [0, 1].""" @@ -66,7 +77,8 @@ class RampController: def __init__(self, trial, metrics_provider, budget_ms=BUDGET_MS, hard_cap_ms=HARD_CAP_MS, settle_ms=SETTLE_MS, hold_ms=HOLD_MS, max_steps=MAX_STEPS, - hard_cap_strikes=HARD_CAP_STRIKES, on_finish=None): + hard_cap_strikes=HARD_CAP_STRIKES, on_finish=None, + mem_budget_mb=MEM_BUDGET_MB, rss_ceiling_mb=DEFAULT_RSS_CEILING_MB): self.hard_cap_strikes = hard_cap_strikes self.strikes = 0 self.trial = trial @@ -77,6 +89,8 @@ class RampController: self.hold_ms = hold_ms self.max_steps = max_steps self.on_finish = on_finish + self.mem_budget_mb = mem_budget_mb + self.rss_ceiling_mb = rss_ceiling_mb self.k = 0 self.load = trial.base_load @@ -86,6 +100,7 @@ class RampController: self.last_pass = None # dict of last passing window self.done = False self.result = None + self.stop_reason = None # budget | hard_cap | max_load | mem_predict | mem_rss | max_steps # -- public API ------------------------------------------------------- def start(self): @@ -152,20 +167,41 @@ class RampController: "metrics_at_peak": dict(metrics), } if self.k + 1 > self.max_steps: - self._finish() + self._finish("max_steps") return - self.k += 1 - self.load = load_at_step(self.trial.base_load, self.trial.growth, self.k) + + next_k = self.k + 1 + next_load = load_at_step(self.trial.base_load, self.trial.growth, next_k) + + # -- memory guards: never allocate past the budget/cap ------------ + cap = getattr(self.trial, "max_load", None) + if cap is not None and next_load > cap: + self._finish("max_load") + return + predicted = self.trial.predict_bytes(next_load) + if predicted is not None and predicted > self.mem_budget_mb * 1024 * 1024: + self._finish("mem_predict") + return + + self.k = next_k + self.load = next_load self.trial.set_load(self.load) + + # -- RSS watchdog: bail if the allocation we just did overshot ----- + if self.rss_ceiling_mb and rss_mb() > self.rss_ceiling_mb: + self._finish("mem_rss") + return + self.phase = "settle" self.phase_start = None self.samples = [] else: - self._finish() + self._finish("hard_cap" if forced_fail else "budget") - def _finish(self): + def _finish(self, reason="budget"): self.phase = "done" self.done = True + self.stop_reason = reason if self.last_pass is not None: lp = self.last_pass self.result = { @@ -175,6 +211,7 @@ class RampController: "p95_ms": lp["p95_ms"], "samples": lp["samples"], "metrics_at_peak": lp["metrics_at_peak"], + "stop_reason": reason, } else: # First load already failed -- record a zero score honestly. @@ -185,6 +222,7 @@ class RampController: "p95_ms": round(percentile(self.samples, 0.95), 3), "samples": len(self.samples), "metrics_at_peak": self.metrics_provider(), + "stop_reason": reason, } if self.on_finish: self.on_finish(self) diff --git a/tests/benchmarks/gauntlet/trials/__init__.py b/tests/benchmarks/gauntlet/trials/__init__.py index f413395..5802946 100644 --- a/tests/benchmarks/gauntlet/trials/__init__.py +++ b/tests/benchmarks/gauntlet/trials/__init__.py @@ -19,6 +19,20 @@ class Trial: base_load = 50 growth = 1.6 + # -- safety (see safety.py) ------------------------------------------- + # Absolute hard cap on load; the ramp will not set a load above this even + # if the frame budget still holds. None = no explicit cap (the RSS watchdog + # and address-space backstop still apply). Trials whose cost grows faster + # than their frame time (e.g. grid side -> S*S cells) MUST set this. + max_load = None + + def predict_bytes(self, load): + """Estimate the resident footprint (bytes) this trial would allocate at + `load`. Return None if unpredictable. The ramp refuses a load whose + prediction exceeds the memory budget BEFORE allocating it, so a + geometric jump cannot OOM between two frame-time samples.""" + return None + # -- lifecycle -------------------------------------------------------- def __init__(self): self.scene = None diff --git a/tests/benchmarks/gauntlet/trials/grid_titan.py b/tests/benchmarks/gauntlet/trials/grid_titan.py index 02b58fe..5f9b992 100644 --- a/tests/benchmarks/gauntlet/trials/grid_titan.py +++ b/tests/benchmarks/gauntlet/trials/grid_titan.py @@ -24,6 +24,18 @@ class GridTitan(Trial): base_load = 20 growth = 1.4 + # Cost scales as S*S (cells) but the render is viewport-bounded, so frame + # time can stay under budget while allocation runs away. Cap the grid side + # and predict the footprint so the ramp bails BEFORE a fatal allocation. + # ~28 bytes/cell: 2 uint8 planes + int32 TileLayer + RGBA ColorLayer + TCOD + # map cell + slack. max_load derived from a 512 MB grid-data budget. + CELL_BYTES_EST = 28 + max_load = 4300 # ~4300^2 * 28 B ~= 494 MB + + def predict_bytes(self, load): + side = max(4, int(load)) + return side * side * self.CELL_BYTES_EST + def setup(self, scene, ui): super().setup(scene, ui) self.grid = None diff --git a/tests/unit/gauntlet_safety_test.py b/tests/unit/gauntlet_safety_test.py new file mode 100644 index 0000000..9373a4d --- /dev/null +++ b/tests/unit/gauntlet_safety_test.py @@ -0,0 +1,144 @@ +"""Headless test for The Gauntlet memory safety guards. + +Reproduces (without allocating anything) the failure that hard-locked the +desktop on 2026-07-11: a trial whose frame time never breaks the budget while +its allocation grows geometrically. Asserts the RampController now stops via a +memory guard instead of ramping unbounded, and never sets a load past the cap. + +Also exercises the address-space cap + RSS probes in safety.py. + +Run: + ./mcrogueface --headless --exec ../tests/unit/gauntlet_safety_test.py +""" +import os +import sys + +import mcrfpy # noqa: F401 (engine interpreter; not used directly) + +HERE = os.path.dirname(os.path.abspath(__file__)) +GAUNTLET = os.path.normpath(os.path.join(HERE, "..", "benchmarks", "gauntlet")) +sys.path.insert(0, GAUNTLET) + +import safety # noqa: E402 +from scoring import RampController, load_at_step # noqa: E402 +from trials import Trial # noqa: E402 +from trials.grid_titan import GridTitan # noqa: E402 + +failures = [] + + +def check(cond, msg): + if not cond: + failures.append(msg) + print(" FAIL: " + msg) + else: + print(" ok: " + msg) + + +class RunawayTrial(Trial): + """A trial that would allocate S*S bytes and never breaks the frame budget. + set_load() records the largest load it was ever asked to build so the test + can prove the guard refused the fatal one BEFORE allocation.""" + key = "runaway" + name = "RUNAWAY" + unit = "side" + base_load = 100 + growth = 1.5 + max_load = 5000 + CELL_BYTES = 32 + + def __init__(self): + super().__init__() + self.peak_asked = 0 + + def predict_bytes(self, load): + return int(load) * int(load) * self.CELL_BYTES + + def set_load(self, level_value): + # NEVER actually allocate -- just record the request. + self.peak_asked = max(self.peak_asked, int(level_value)) + self.load = int(level_value) + + +def drive(ramp, always_ms=4.0, max_iters=100000): + """Feed synthetic 'always under budget' samples until the ramp finishes.""" + ramp.start() + now = 0.0 + it = 0 + while not ramp.done and it < max_iters: + now += 16.0 + ramp.sample(now, {"frame_time": always_ms}) + it += 1 + return it < max_iters + + +def test_safety_probes(): + print("[safety.py probes]") + check(safety.rss_mb() > 0.0, "rss_mb() positive (%.1f MB)" % safety.rss_mb()) + check(safety.vmsize_mb() > 0.0, "vmsize_mb() positive (%.1f MB)" % safety.vmsize_mb()) + + +def test_predict_cap_bails_before_alloc(): + print("[predict_bytes guard]") + trial = RunawayTrial() + # 64 MB budget: predict_bytes(load) = load^2 * 32 exceeds it at load ~1448. + ramp = RampController(trial, lambda: {"frame_time": 4.0}, + mem_budget_mb=64, rss_ceiling_mb=0) + finished = drive(ramp) + check(finished, "ramp terminated (did not loop unbounded)") + check(ramp.stop_reason in ("mem_predict", "max_load"), + "stopped via memory guard, not budget: %r" % ramp.stop_reason) + # The refused load must never have been handed to set_load. + budget_bytes = 64 * 1024 * 1024 + check(trial.predict_bytes(trial.peak_asked) <= budget_bytes, + "largest allocated load (%d -> %d B) stayed within budget (%d B)" + % (trial.peak_asked, trial.predict_bytes(trial.peak_asked), budget_bytes)) + check(ramp.result["max_load"] > 0, "recorded a real max_load (%d)" + % ramp.result["max_load"]) + + +def test_absolute_max_load_cap(): + print("[max_load cap]") + trial = RunawayTrial() + # Huge memory budget so only the absolute max_load cap can stop it. + ramp = RampController(trial, lambda: {"frame_time": 4.0}, + mem_budget_mb=10 ** 9, rss_ceiling_mb=0) + finished = drive(ramp) + check(finished, "ramp terminated under max_load cap") + check(ramp.stop_reason == "max_load", "stopped via max_load: %r" % ramp.stop_reason) + check(trial.peak_asked <= trial.max_load, + "never set a load above max_load (%d <= %d)" + % (trial.peak_asked, trial.max_load)) + + +def test_grid_titan_declares_guards(): + print("[GRID TITAN guards]") + gt = GridTitan() + check(gt.max_load is not None and gt.max_load > 0, + "GRID TITAN declares max_load (%r)" % gt.max_load) + # Predicted footprint at the cap must be a sane, tractable number (< 1 GB). + pb = gt.predict_bytes(gt.max_load) + check(pb is not None and pb < 1024 * 1024 * 1024, + "GRID TITAN cap footprint tractable (%.0f MB at S=%d)" + % (pb / 1024.0 / 1024.0, gt.max_load)) + + +def main(): + test_safety_probes() + test_predict_cap_bails_before_alloc() + test_absolute_max_load_cap() + test_grid_titan_declares_guards() + + print("=" * 50) + if failures: + print("GAUNTLET SAFETY TEST FAILED (%d issues)" % len(failures)) + for f in failures: + print(" - " + f) + print("FAIL") + sys.exit(1) + print("PASS") + sys.exit(0) + + +if __name__ == "__main__": + main()