#!/usr/bin/env python3 """Repeatable Partridge solver benchmark runner.""" import argparse import json import os import platform import statistics import subprocess import sys import time from pathlib import Path def cache_values(build_dir): values = {} cache = build_dir / "CMakeCache.txt" if not cache.exists(): return values for line in cache.read_text(encoding="utf-8").splitlines(): if line.startswith(("//", "#")) or "=" not in line: continue key_type, value = line.split("=", 1) key = key_type.split(":", 1)[0] values[key] = value return values def command_output(command, cwd=None): try: return subprocess.run( command, cwd=cwd, check=True, capture_output=True, text=True ).stdout.strip() except (OSError, subprocess.CalledProcessError): return "unknown" def environment(binary): build_dir = binary.resolve().parent cache = cache_values(build_dir) compiler = cache.get("CMAKE_CXX_COMPILER", "unknown") build_type = cache.get("CMAKE_BUILD_TYPE", "unknown") type_flags = cache.get(f"CMAKE_CXX_FLAGS_{build_type.upper()}", "") flags = " ".join( part for part in (cache.get("CMAKE_CXX_FLAGS", ""), type_flags) if part ) source_dir = cache.get("CMAKE_HOME_DIRECTORY") commit = command_output( ["git", "rev-parse", "HEAD"], cwd=source_dir if source_dir else None ) status = command_output( ["git", "status", "--porcelain"], cwd=source_dir if source_dir else None ) compiler_version = ( command_output([compiler, "--version"]).splitlines()[:1] if compiler != "unknown" else ["unknown"] )[0] return { "os": platform.platform(), "compiler": compiler, "compiler_version": compiler_version, "compiler_flags": flags, "linker_flags": cache.get("CMAKE_EXE_LINKER_FLAGS", ""), "build_type": build_type, "commit": commit, "working_tree_dirty": status not in ("", "unknown"), "hardware": { "model": command_output(["sysctl", "-n", "hw.model"]) if sys.platform == "darwin" else platform.node(), "machine": platform.machine() or "unknown", "processor": platform.processor() or "unknown", "logical_cpus": os.cpu_count(), }, "workers": 1, "search_policy": "single-threaded, deterministic, smallest-width valley", "seed": None, } def run_once( binary, order, mode, search_policy, search_route, symmetry, pruning, component_pruning, component_schedule, timeout, ): started = time.monotonic() try: process = subprocess.run( [ str(binary), str(order), mode, search_policy, search_route, symmetry, pruning, component_pruning, component_schedule, ], check=False, capture_output=True, text=True, timeout=timeout, ) except subprocess.TimeoutExpired: return { "status": "timeout", "wall_seconds": time.monotonic() - started, } if process.returncode != 0: return { "status": "error", "returncode": process.returncode, "stderr": process.stderr[-1000:], "wall_seconds": time.monotonic() - started, } try: result = json.loads(process.stdout) except json.JSONDecodeError as error: return { "status": "invalid-output", "detail": str(error), "stdout": process.stdout[-1000:], } result["status"] = "ok" result["wall_seconds"] = time.monotonic() - started return result def distribution(values): if not values: return None median = statistics.median(values) return { "median": median, "min": min(values), "max": max(values), "spread": max(values) - min(values), "median_absolute_deviation": statistics.median( abs(value - median) for value in values ), } def summarize(runs): completed = [run for run in runs if run["status"] == "ok"] return { "attempted": len(runs), "completed": len(completed), "timeouts": sum(run["status"] == "timeout" for run in runs), "errors": sum(run["status"] == "error" for run in runs), "invalid_outputs": sum( run["status"] == "invalid-output" for run in runs ), "solve_seconds": distribution( [run["timing_seconds"]["solve"] for run in completed] ), "construction_seconds": distribution( [run["timing_seconds"]["construction"] for run in completed] ), "validation_seconds": distribution( [run["timing_seconds"]["validation"] for run in completed] ), "render_seconds": distribution( [run["timing_seconds"]["render"] for run in completed] ), "counter_values": completed[0]["counters"] if completed else None, "counters_stable": bool(completed) and all(run["counters"] == completed[0]["counters"] for run in completed), } def summary_failed(summary): return ( summary["completed"] == 0 or summary["errors"] > 0 or summary["invalid_outputs"] > 0 ) def self_test(): sample = distribution([1.0, 3.0, 2.0]) assert sample["median"] == 2.0 assert sample["spread"] == 2.0 assert sample["median_absolute_deviation"] == 1.0 parsed = json.loads( '{"timing_seconds":{"solve":1,"construction":0},"counters":{}}' ) assert parsed["timing_seconds"]["solve"] == 1 runs = [ { "status": "ok", "timing_seconds": { "solve": 1.0, "construction": 0.0, "validation": 0.1, "render": 0.2, }, "counters": {"search_nodes": 3}, }, {"status": "timeout"}, {"status": "error"}, {"status": "invalid-output"}, ] summary = summarize(runs) assert summary["completed"] == 1 assert summary["timeouts"] == 1 assert summary["errors"] == 1 assert summary["invalid_outputs"] == 1 assert summary["construction_seconds"]["median"] == 0.0 assert summary_failed(summary) assert summary_failed(summarize([{"status": "timeout"}])) completed_with_timeout = summarize([runs[0], {"status": "timeout"}]) assert not summary_failed(completed_with_timeout) assert trial_modes(["counters", "plain"], 0) == ["counters", "plain"] assert trial_modes(["counters", "plain"], 1) == ["plain", "counters"] def trial_modes(modes, trial): if len(modes) < 2 or trial % 2 == 0: return modes return list(reversed(modes)) def main(): parser = argparse.ArgumentParser() parser.add_argument("--binary", type=Path) parser.add_argument("--orders", nargs="+", type=int, default=[7, 8, 9]) parser.add_argument("--warmup", type=int, default=1) parser.add_argument("--repetitions", type=int, default=5) parser.add_argument("--timeout", type=float, default=600.0) parser.add_argument( "--candidate-order", dest="search_policy", choices=("ascending", "descending", "best-fit"), default="ascending", ) parser.add_argument("--direct-search", action="store_true") parser.add_argument( "--no-symmetry", action="store_true", help="disable unique-unit-square D4 symmetry breaking", ) parser.add_argument( "--pruning", choices=("all", "valley-capacity", "large-square", "none"), default="all", help="select independently measurable pruning rules", ) parser.add_argument( "--no-pruning", action="store_true", help="disable all pruning (compatibility alias for --pruning none)", ) parser.add_argument( "--component-pruning", choices=("subset-sum", "gcd", "none"), default="none", help="select component-area pruning strength", ) parser.add_argument( "--component-schedule", choices=("boundary", "periodic-8"), default="boundary", help="trigger component checks on boundaries or every eighth depth", ) parser.add_argument( "--measure-overhead", action="store_true", help="also run the compile-time counter-free solver", ) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: self_test() return 0 if args.binary is None: parser.error("--binary is required") if args.warmup < 0 or args.repetitions < 1 or args.timeout <= 0: parser.error("warmup must be non-negative; repetitions and timeout positive") modes = ["counters", "plain"] if not args.measure_overhead: modes = ["counters"] cases = [] failed = False for order in args.orders: mode_results = {mode: {"runs": []} for mode in modes} for trial in range(args.warmup): for mode in trial_modes(modes, trial): run_once( args.binary, order, mode, args.search_policy, "direct" if args.direct_search else "public", "none" if args.no_symmetry else "d4", "none" if args.no_pruning else args.pruning, "none" if args.no_pruning else args.component_pruning, args.component_schedule, args.timeout, ) for trial in range(args.repetitions): for mode in trial_modes(modes, trial): mode_results[mode]["runs"].append( run_once( args.binary, order, mode, args.search_policy, "direct" if args.direct_search else "public", "none" if args.no_symmetry else "d4", "none" if args.no_pruning else args.pruning, "none" if args.no_pruning else args.component_pruning, args.component_schedule, args.timeout, ) ) for result in mode_results.values(): result["summary"] = summarize(result["runs"]) summary = result["summary"] failed = failed or summary_failed(summary) overhead = None counted = mode_results["counters"]["summary"]["solve_seconds"] plain = mode_results.get("plain", {}).get("summary", {}).get("solve_seconds") if counted and plain and plain["median"]: overhead = { "median_seconds": counted["median"] - plain["median"], "median_percent": 100.0 * (counted["median"] / plain["median"] - 1.0), } cases.append({"order": order, "modes": mode_results, "counter_overhead": overhead}) document = { "schema": "partridge-benchmark-v1", "environment": environment(args.binary), "policy": { "orders": args.orders, "warmup_runs": args.warmup, "measured_repetitions": args.repetitions, "per_run_timeout_seconds": args.timeout, "candidate_order": args.search_policy, "search_route": "direct" if args.direct_search else "public", "symmetry_breaking": "none" if args.no_symmetry else "d4", "pruning": "none" if args.no_pruning else args.pruning, "component_pruning": ( "none" if args.no_pruning else args.component_pruning ), "component_schedule": args.component_schedule, "stdout": "captured; rendered grid suppressed by probe", }, "cases": cases, } json.dump(document, sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())