Something went wrong. Try again.
The agentic engineering control plane for the posthuman future
Something went wrong. Try again.
13 kB · 314 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315#!/usr/bin/env python3"""Bench xenomorphic on the classic "cat huge file" stress.
Generates a deterministic dense output stream (long lines, ASCII digits),spawns xenomorphic with one ``-e`` that pipes it into the shell, and printsPERF_DEBUG-derived stats.
Examples:
python3 scripts/bigfile.py # default 200000 lines python3 scripts/bigfile.py --lines 1000000 # bigger python3 scripts/bigfile.py --lines 100000 --warmup 3 --repeat 5 python3 scripts/bigfile.py --help"""from __future__ import annotations
import argparseimport csvimport osimport statisticsimport subprocessimport sysimport textwrapfrom pathlib import Path
# Allow running from project root or from scripts/ directly.HERE = Path(__file__).resolve().parentPROJECT_ROOT = HERE.parentsys.path.insert(0, str(PROJECT_ROOT))
from xenobench import ( # noqa: E402 DEFAULT_MARKER, PhaseStats, find_binary, run_benchmark,)
def generate_workload(path: Path, lines: int, cols: int) -> None: """Write ``lines`` rows of `index | <filler>` deterministically.
Each row is "%07d <filler>" with ``cols-8`` filler chars. The filler is ascii whitespace; the workload streams densely. Uses awk so the workload generation itself doesn't pollute Python's CPU/heap picture.
Earlier versions read the line count from ``ARGV[2]``, which in a shell pipeline ``seq N | awk '...' > file`` is the output path the shell opens for redirection -- not the line count. With nothing coerced numerically to 0, awk emitted nothing. Pass the count via ``-v`` and the render width via ``-v W``. """ filler_cols = max(1, cols - len("NNNNNNN ")) # Use -v for N (lines) AND W (filler width); the printf "%Ws" trick # needs W as a number, not a sprintf format string at runtime. awk_body = ( "BEGIN{p=sprintf(\"%" + str(filler_cols) + "s\",\"x\");" "for(i=0;i<N;i++) printf \"%07d %s\\n\", i, p}" ) awk_body_sh = awk_body.replace("'", "'\\''") cmd = ( f"awk -v N={lines} -v W={filler_cols} '{awk_body_sh}' > {path}" ) completed = subprocess.run( ["bash", "-c", cmd], capture_output=True, text=True, check=False) if completed.returncode != 0: raise RuntimeError( f"workload generation failed: {completed.stderr or completed.stdout}")
def workload_size(path: Path) -> int: return path.stat().st_size
def fmt_phase(p: PhaseStats) -> str: if not p.samples: return "no samples" return (f"n={p.count:5d} total={p.total_ms:7.1f}ms " f"min={p.min_ms:5.2f}ms p50={p.percentile(0.50):5.2f}ms " f"p95={p.percentile(0.95):5.2f}ms max={p.max_ms:5.2f}ms")
def run_once(commands: list[str], marker: Path, binary: Path, timeout_s: float, log_dir: Path, cwd: Path, settle_s: float): return run_benchmark( commands, binary=binary, cwd=cwd, marker=marker, timeout_s=timeout_s, settle_after_marker_s=settle_s, log_dir=log_dir, extra_env={"XDG_CONFIG_HOME": "/tmp/xeno-bench-noop-doesntexist"}, )
def main() -> int: parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent(__doc__ or ""), ) parser.add_argument("--binary", type=Path, default=None, help="Path to the xenomorphic binary. " "Defaults to $XENOBENCH_BINARY or auto-resolve.") parser.add_argument("--lines", type=int, default=200_000, help="Lines to generate (default: 200000).") parser.add_argument("--cols", type=int, default=120, help="Approx characters per line (default: 120).") parser.add_argument("--workload-reuse", type=Path, default=None, help="Reuse a pre-made workload instead of generating one.") parser.add_argument("--warmup", type=int, default=1, help="Discarded warmup runs before measurement (default: 1).") parser.add_argument("--repeat", type=int, default=5, help="Measured runs after warmup (default: 5).") parser.add_argument("--timeout", type=float, default=600.0, help="Per-run timeout in seconds (default: 600).") parser.add_argument("--marker", type=Path, default=DEFAULT_MARKER, help="Path to the touch sentinel file (default: /tmp/xeno-bench-ready).") parser.add_argument("--keep-workload", action="store_true", help="Reuse the same generated workload across runs.") parser.add_argument("--workload-dir", type=Path, default=Path("/tmp"), help="Where to write generated workloads (default: /tmp).") parser.add_argument("--log-dir", type=Path, default=Path("/tmp"), help="Where to write PERF_DEBUG logs (default: /tmp).") parser.add_argument("--csv", type=Path, default=None, help="Write one CSV row per measured run to this path.") parser.add_argument("--summary-csv", type=Path, default=None, help="Write one CSV row summarizing medians across " "all measured runs.") parser.add_argument("--cwd", type=Path, default=Path("/tmp"), help="Spawn the PTY in this directory (passes -C to " "the binary). Default: /tmp (skips any project-" "dir profile / direnv init).") parser.add_argument("--settle", type=float, default=1.5, help="Seconds to wait AFTER the marker file appears " "before SIGTERM. Lets the renderer drain the " "trailing bytes PTY->renderthread. Default: 1.5.") args = parser.parse_args()
if not args.cwd.is_dir(): print(f"bigfile: --cwd={args.cwd} is not an existing directory", file=sys.stderr) return 2 print(f"bigfile: PTY cwd={args.cwd} (override: --cwd <dir>)", file=sys.stderr)
try: binary = args.binary or find_binary() except FileNotFoundError as exc: print(f"bigfile: {exc}", file=sys.stderr) return 2
workload: Path cleanup_workload = False if args.workload_reuse is not None: workload = args.workload_reuse else: args.workload_dir.mkdir(parents=True, exist_ok=True) workload = args.workload_dir / f"xeno-bench-{args.lines}lines.txt" if not workload.exists(): generate_workload(workload, args.lines, args.cols) cleanup_workload = args.keep_workload size_bytes = workload_size(workload) print(f"bigfile: workload={workload} ({size_bytes/1e6:.2f} MiB)", file=sys.stderr)
cmd = ( f"wc -l < {workload}; " f"sync; " f"cat {workload}; " f"sync; " f"touch {args.marker}" )
runs = []
if args.csv: args.csv.parent.mkdir(parents=True, exist_ok=True)
# Warmup -> discard for w in range(args.warmup): print(f"bigfile: warmup run {w+1}/{args.warmup}", file=sys.stderr) out = run_once([cmd], args.marker, binary, args.timeout, args.log_dir, args.cwd, args.settle) for warn in out.warnings: print(f"bigfile: WARN ({warn})", file=sys.stderr)
# Measured runs. for r in range(args.repeat): print(f"bigfile: measured run {r+1}/{args.repeat}", file=sys.stderr) out = run_once([cmd], args.marker, binary, args.timeout, args.log_dir, args.cwd, args.settle) for warn in out.warnings: print(f"bigfile: WARN ({warn})", file=sys.stderr)
print(f" log: {out.log_path}", file=sys.stderr) print(f" elapsed:{out.elapsed_s:7.3f}s " f"bytes:{out.total_bytes} " f"frames:{out.frame_count} " f"fps:{out.fps:5.1f} " f"last_vtDirty:{out.last_vt_dirty}", file=sys.stderr) for name, stats in out.phase_stats.items(): print(f" {name:>26}: {fmt_phase(stats)}", file=sys.stderr) print(f" {'(gap frames)':>26}: {fmt_phase(out.gap_stats)}", file=sys.stderr) print(f" {'(updateToSync)':>26}: " f"{fmt_phase(out.latency_update_to_sync)}", file=sys.stderr) if out.timed_out: print(f" TIMED OUT", file=sys.stderr) runs.append(out)
if args.csv: mode = "a" if args.csv.exists() else "w" with args.csv.open(mode, newline="") as fh: writer = csv.writer(fh) if mode == "w": writer.writerow([ "lines", "bytes", "elapsed_s", "fps", "vt_MBps", "frames", "last_vt_dirty", "vt_write_ms_total", "vt_write_p95_ms", "updateRenderState_ms_total", "updateRenderState_p95_ms", "buildFrame+setFrame_ms_total", "buildFrame+setFrame_p95_ms", "renderer.render_ms_total", "renderer.render_p95_ms", "gap_ms_total", "gap_p95_ms", "log_path", ]) phases = out.phase_stats
def prop(name: str, key: str) -> float: return getattr(phases.get(name, PhaseStats(name=name)), key)
def p95(name: str) -> float: return phases.get(name, PhaseStats(name=name)).percentile(0.95)
writer.writerow([ args.lines, out.total_bytes, f"{out.elapsed_s:.4f}", f"{out.fps:.2f}", f"{out.vt_throughput_MBps:.2f}", out.frame_count, out.last_vt_dirty, prop("vt_write", "total_ms"), p95("vt_write"), prop("updateRenderState", "total_ms"), p95("updateRenderState"), prop("buildFrame+setFrame", "total_ms"), p95("buildFrame+setFrame"), prop("renderer.render", "total_ms"), p95("renderer.render"), out.gap_stats.total_ms, out.gap_stats.percentile(0.95), str(out.log_path), ])
# Summary across measured runs. def med(key): vals = [getattr(r, key) for r in runs if not r.timed_out] return statistics.median(vals) if vals else float("nan")
def med_phase(name): vals = [] for r in runs: if r.timed_out: continue s = r.phase_stats.get(name) if s and s.samples: vals.append(s.total_ms) return statistics.median(vals) if vals else float("nan")
summary = { "lines": args.lines, "workload_bytes": size_bytes, "runs": len([r for r in runs if not r.timed_out]), "elapsed_s_median": med("elapsed_s"), "fps_median": med("fps"), "vt_MBps_median": med("vt_throughput_MBps"), "vt_write_ms_total_median": med_phase("vt_write"), "updateRenderState_ms_total_median": med_phase("updateRenderState"), "buildFrame+setFrame_ms_total_median": med_phase("buildFrame+setFrame"), "renderer_render_ms_total_median": med_phase("renderer.render"), "gap_ms_total_median": med_phase("gap"), } print("", file=sys.stderr) print(f"bigfile SUMMARY (median of {summary['runs']} runs):", file=sys.stderr) for k, v in summary.items(): if isinstance(v, float): print(f" {k}: {v:.4f}", file=sys.stderr) else: print(f" {k}: {v}", file=sys.stderr)
if args.summary_csv: args.summary_csv.parent.mkdir(parents=True, exist_ok=True) mode = "a" if args.summary_csv.exists() else "w" with args.summary_csv.open(mode, newline="") as fh: writer = csv.writer(fh) if mode == "w": writer.writerow(list(summary.keys())) writer.writerow([str(v) for v in summary.values()])
if cleanup_workload: try: workload.unlink() except FileNotFoundError: pass
return 0 if all(r.exit_code in (None, 0) for r in runs) else 1
if __name__ == "__main__": sys.exit(main())