#!/usr/bin/env python3 """Run reproducible multi-position CPU/CUDA benchmarks.""" from __future__ import annotations import argparse import json import math import random import re import statistics import subprocess import sys from pathlib import Path HERE = Path(__file__).resolve().parent WORLD_MIN = -1_875_000 WORLD_MAX_TOP_LEFT_5 = 1_874_996 def positions(count: int) -> list[tuple[int, int]]: known = (51_069, 419_124) result = [known] hard_target = max(1, count // 2) offsets = sorted( ((dx, dz) for dz in range(-8, 9) for dx in range(-8, 9) if dx or dz), key=lambda item: (abs(item[0]) + abs(item[1]), item[1], item[0]), ) for dx, dz in offsets: if len(result) >= hard_target: break result.append((known[0] + dx, known[1] + dz)) fixed = [ (0, 0), (WORLD_MIN, WORLD_MIN), (WORLD_MIN, WORLD_MAX_TOP_LEFT_5), (WORLD_MAX_TOP_LEFT_5, WORLD_MIN), (WORLD_MAX_TOP_LEFT_5, WORLD_MAX_TOP_LEFT_5), (-1, -1), (1, 1), (-1_000_000, 1_000_000), (1_000_000, -1_000_000), ] for position in fixed: if len(result) >= count: break if position not in result: result.append(position) rng = random.Random(20_261_004) while len(result) < count: result.append( ( rng.randint(WORLD_MIN, WORLD_MAX_TOP_LEFT_5), rng.randint(WORLD_MIN, WORLD_MAX_TOP_LEFT_5), ) ) return result[:count] def run(command: list[str]) -> str: completed = subprocess.run(command, text=True, capture_output=True, check=False) if completed.returncode != 0: sys.stderr.write(completed.stdout) sys.stderr.write(completed.stderr) raise RuntimeError(f"command failed ({completed.returncode}): {' '.join(command)}") return completed.stdout def fields(output: str) -> dict[str, float]: result: dict[str, float] = {} for key, value in re.findall(r"([a-zA-Z0-9_]+)=([0-9.eE+-]+)", output.splitlines()[0]): result[key] = float(value) return result def percentile(values: list[float], fraction: float) -> float: ordered = sorted(values) index = math.ceil(fraction * len(ordered)) - 1 return ordered[max(0, min(index, len(ordered) - 1))] def stats(values: list[float]) -> dict[str, float]: return { "min": min(values), "median": statistics.median(values), "mean": statistics.fmean(values), "p95": percentile(values, 0.95), "max": max(values), } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--samples", type=int, default=64) parser.add_argument("--cpu-threads", type=int, default=16) parser.add_argument("--cpu-old-count", type=int, default=1 << 28) parser.add_argument("--cuda-old-count", type=int, default=1 << 34) parser.add_argument("--cuda-threads", type=int, default=256) parser.add_argument("--skip-cpu", action="store_true") parser.add_argument("--skip-cuda", action="store_true") parser.add_argument("--output", type=Path) args = parser.parse_args() if args.samples <= 0: parser.error("samples must be positive") rows: list[dict[str, object]] = [] sample_positions = positions(args.samples) for index, (x, z) in enumerate(sample_positions): row: dict[str, object] = {"index": index, "x": x, "z": z} common = ["--x", str(x), "--z", str(z), "-n", "5"] if not args.skip_cpu: new_cpu = fields(run([ str(HERE / "slime_pattern_cpp"), "search", *common, "-p", str(args.cpu_threads), "--max-results", "0", ])) old_cpu = fields(run([ str(HERE / "slime_bruteforce_cpp_bench"), *common, "--count", str(args.cpu_old_count), "-p", str(args.cpu_threads), ])) row["new_cpu_seconds"] = new_cpu["seconds"] row["old_cpu_projected_seconds"] = old_cpu["projected_full48_seconds"] row["cpu_speedup"] = old_cpu["projected_full48_seconds"] / new_cpu["seconds"] row["lower18_roots"] = int(new_cpu["lower18_roots"]) row["lower35_checked"] = int(new_cpu["lower35_checked"]) row["exact_matches_cpu"] = int(new_cpu["exact_matches"]) row["old_cpu_measured_seeds"] = args.cpu_old_count if not args.skip_cuda: new_cuda = fields(run([ str(HERE / "slime_pattern_cuda"), *common, "--threads", str(args.cuda_threads), "--max-results", "0", ])) old_cuda = fields(run([ str(HERE / "slime_bruteforce_cuda_bench"), *common, "--count", str(args.cuda_old_count), ])) new_cuda_seconds = new_cuda["milliseconds"] / 1000.0 row["new_cuda_seconds"] = new_cuda_seconds row["old_cuda_projected_seconds"] = old_cuda["projected_full48_seconds"] row["cuda_speedup"] = ( old_cuda["projected_full48_seconds"] / new_cuda_seconds if new_cuda_seconds > 0.0 else None ) row["exact_matches_cuda"] = int(new_cuda["exact_matches"]) row["old_cuda_measured_seeds"] = args.cuda_old_count rows.append(row) print(json.dumps(row, separators=(",", ":")), flush=True) summary: dict[str, object] = { "samples": args.samples, "positions": sample_positions, "rows": rows, } numeric_keys = [ "lower18_roots", "lower35_checked", "new_cpu_seconds", "old_cpu_projected_seconds", "cpu_speedup", "new_cuda_seconds", "old_cuda_projected_seconds", "cuda_speedup", ] summary["statistics"] = { key: stats([float(row[key]) for row in rows if row.get(key) is not None]) for key in numeric_keys if key in rows[0] and any(row.get(key) is not None for row in rows) } summary["aggregate"] = { "old_cpu_measured_seeds": sum(int(row.get("old_cpu_measured_seeds", 0)) for row in rows), "old_cuda_measured_seeds": sum(int(row.get("old_cuda_measured_seeds", 0)) for row in rows), "lower35_checked": sum(int(row.get("lower35_checked", 0)) for row in rows), "exact_matches_cpu": sum(int(row.get("exact_matches_cpu", 0)) for row in rows), "exact_matches_cuda": sum(int(row.get("exact_matches_cuda", 0)) for row in rows), } print("SUMMARY " + json.dumps(summary, separators=(",", ":")), flush=True) if args.output is not None: args.output.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())