john/McRogueFace
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Wiki Activity

feat(bench): Gauntlet OOM safety guards + The Crucible wall-clock benchmark; closes #353 closes #354

Browse source 
The Gauntlet's auto-ramp had a step ceiling but no memory ceiling. GRID TITAN
ramps grid side S with cost ~ S*S, but renders into a fixed viewport so frame
time can stay under budget while allocation runs away -- observed hard-locking
the desktop mid grid-allocation (all RAM consumed, forced logout).

#353 -- memory safety (defense in depth, weakest failure mode last):
  * tests/benchmarks/gauntlet/safety.py: rss_mb/vmsize_mb probes +
    install_address_space_cap() (RLIMIT_AS backstop; a runaway Grid() then
    aborts THIS process via std::bad_alloc instead of taking the machine down).
  * Trial.predict_bytes(load) + Trial.max_load: a trial refuses a load whose
    predicted footprint exceeds a 512 MB budget BEFORE allocating it. GRID
    TITAN implements both (S*S*~28 B; max_load=4300 ~= 494 MB).
  * RampController: pre-allocation predict/cap check + post-set_load RSS
    watchdog; records stop_reason (budget/hard_cap/max_load/mem_predict/
    mem_rss/max_steps). run_gauntlet.py installs the address-space cap at start.
  * tests/unit/gauntlet_safety_test.py proves the ramp bails on the memory
    guards before over-allocating (synthetic runaway trial records the largest
    load it was asked to build).

#354 -- The Crucible (tests/benchmarks/crucible.py): a headless, deterministic
wall-clock microbenchmark of fixed "comically extreme but tractable" configs
(grid alloc, cell fill, layer writes, turn-manager swarm, FOV, A*, entity
churn). Safe (<512 MB, ~seconds), display-free, cross-version (missing APIs
report "unsupported"). Emits JSON; MCRF_CRUCIBLE_BASELINE diffs two builds.

This is the safe replacement for the windowed-Gauntlet A/B that crashed the
desktop. First result, current master vs the 0.2.8 release artifact (both
headless): grid_alloc -58.9%, grid_fill -25.4%, layer_fill -56.1%,
entity_churn -43.0%, fov_storm -10.3%, step_swarm -6.2%, path_queries -3.2%;
geomean 0.673 (32.7% faster overall), peak RSS ~103 vs ~119 MB. Tracks the
#332 SoA rewrite + #329 entity indexing.

Suite 314/314.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
John McCardle 2026-07-11 10:28:13 -04:00
commit f68a770346
7 changed files with 707 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

412
tests/benchmarks/crucible.py Normal file
View file

@ -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()

12
tests/benchmarks/gauntlet/run_gauntlet.py
View file

@ -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)

69
tests/benchmarks/gauntlet/safety.py Normal file
View file

@ -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)

50
tests/benchmarks/gauntlet/scoring.py
View file

@ -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)

14
tests/benchmarks/gauntlet/trials/__init__.py
View file

@ -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

12
tests/benchmarks/gauntlet/trials/grid_titan.py
View file

@ -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