"""Production selection must use static compiler inputs only.""" import argparse import hashlib import json import os from pathlib import Path import shutil import subprocess import tempfile REPO = Path(__file__).resolve().parents[2] def check_selected_program(out): report = json.loads((out / "reports/pipeline_selection.json").read_text()) program = json.loads((out / "pim/config.json").read_text()) digest = hashlib.sha256() for core in range(program["core_cnt"]): digest.update((out / f"pim/core_{core}.pim").read_bytes()) for count in program["array_group_map"][f"core{core}"]: digest.update(f"{count},".encode()) digest.update(b";") assert digest.hexdigest().upper() == report["candidates"][report["selection"]]["identity"], \ "emitted program differs from the analyzed winner" assert report["reused_lowered_candidate"] for name in ("spatial3_scheduled_no_comm.txt", "memory_report.txt"): assert (out / "reports" / name).stat().st_size > 0, name return report def main(): parser = argparse.ArgumentParser() parser.add_argument("--compiler", required=True) args = parser.parse_args() compiler = str(Path(args.compiler).resolve()) model = REPO / "validation/networks/pimcomp_models/googlenet/googlenet-12-pimsim-nn.onnx" with tempfile.TemporaryDirectory(prefix="raptor-static-selection-") as temporary: root = Path(temporary) history = root / "validation/tools/pim/pimcomp/compare/results_comparison.csv" history.parent.mkdir(parents=True) profile = root / "profiles/latency.json" profile.parent.mkdir() trap = root / "bin" trap.mkdir() for name in ("ChipTest", "pimsim", "raptor-periodic", "perf"): path = trap / name path.write_text("#!/bin/sh\necho forbidden execution >> '" + str(root / "executed") + "'\nexit 99\n") path.chmod(0o755) guard = root / "static-audit.so" subprocess.run([os.environ.get("CC", "cc"), "-shared", "-fPIC", str(REPO / "test/PIM/StaticSchedulingAudit.c"), "-o", str(guard)], check=True) env = dict(os.environ, LD_PRELOAD=str(guard), PATH=str(trap) + os.pathsep + os.environ["PATH"], OMP_NUM_THREADS="1") decisions = [] for iteration in range(3): if iteration == 0: history.write_text("network,throughput\ngooglenet,1e99\n") profile.write_text('{"candidate":999,"measured_latency":1}') elif iteration == 1: history.write_bytes(b"invalid historical data\x00") profile.write_text('{"candidate":0,"measured_latency":1e99}') else: history.unlink() shutil.rmtree(profile.parent) out = root / f"run-{iteration}" out.mkdir() command = [compiler, str(model), "-o", str(out / "model"), "--maccel=PIM", "--EmitPimCodegen", "--core-count=138", "--crossbar-count=128", "--crossbar-size=128", "--pipeline=4", "--mlir-disable-threading", "--pim-detect-communication-deadlock", f"--pim-target-config={REPO}/validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json"] result = subprocess.run(command, cwd=root, env=env, capture_output=True, text=True, timeout=180) assert result.returncode == 0, result.stdout + result.stderr report = json.loads((out / "reports/pipeline_selection.json").read_text()) assert report["static_analysis"] and report["temporary_disk_bytes"] == 0 assert report["analyzed_candidates"] > 1, report check_selected_program(out) assert not report["reconstructed_spatial_artifacts"] assert not (out / "dialects").exists() assert not list(out.glob("candidate*")) decisions.append((report["selection"], [(c["identity"], c.get("interval_ns"), c.get("alternative_tie_interval_ns")) for c in report["candidates"]])) # Keep the test's disk peak to one selected program. shutil.rmtree(out) assert decisions[0] == decisions[1] == decisions[2], "historical files or address allocation changed static selection" # Exercise the same winner with and without explicit Spatial artifacts. vgg = REPO / "validation/networks/pimcomp_models/vgg8/vgg8-mnist-reconstructed.onnx" for diagnostics in (False, True): out = root / f"vgg-{diagnostics}" out.mkdir() command = [compiler, str(vgg), "-o", str(out / "model"), "--maccel=PIM", "--EmitPimCodegen", "--pipeline=2", "--core-count=138", "--crossbar-count=128", "--crossbar-size=128", "--mlir-disable-threading", "--pim-detect-communication-deadlock", f"--pim-target-config={REPO}/validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json"] if diagnostics: command += ["--pim-export-spatial-dataflow=all", "--pim-dump-ir", "--pim-periodic-diagnostics"] result = subprocess.run(command, cwd=root, env=env, capture_output=True, text=True, timeout=90) assert result.returncode == 0, result.stdout + result.stderr report = check_selected_program(out) assert report["selection"] != 0 and report["reconstructed_spatial_artifacts"] == diagnostics identity = report["candidates"][report["selection"]]["identity"] schedule = (out / "reports/spatial3_scheduled_no_comm.txt").read_bytes() if diagnostics: assert identity == previous_identity and schedule == previous_schedule for stage in ("spatial3_scheduled_no_comm", "spatial4_scheduled"): for suffix in ("nodes.csv", "edges.csv"): assert (out / f"reports/{stage}.{suffix}").stat().st_size > 0 assert (out / "reports/pipeline_periodic.json").exists() assert (out / "dialects/spatial4_scheduled.mlir").exists() previous_identity, previous_schedule = identity, schedule shutil.rmtree(out) config_path = REPO / "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json" unsupported = json.loads(config_path.read_text()) unsupported["chip_config"]["core_config"]["rob_size"] = 2 network = unsupported["chip_config"]["network_config"] network["net_config_file_path"] = str((config_path.parent / network["net_config_file_path"]).resolve()) hardware = root / "unsupported-target.json" hardware.write_text(json.dumps(unsupported)) out = root / "fallback" out.mkdir() command = [compiler, str(REPO / "validation/networks/pimcomp_models/vgg8/vgg8-mnist-reconstructed.onnx"), "-o", str(out / "model"), "--maccel=PIM", "--EmitPimCodegen", "--pipeline=4", "--core-count=168", "--crossbar-count=96", "--crossbar-size=128", "--mlir-disable-threading", f"--pim-target-config={hardware}"] result = subprocess.run(command, cwd=root, env=env, capture_output=True, text=True, timeout=60) assert result.returncode == 0, result.stdout + result.stderr report = json.loads((out / "reports/pipeline_selection.json").read_text()) assert report["selection"] == 0 and report["analyzed_candidates"] == 0 assert report["generated_candidates"] == 1 and report["lowered_candidates"] == 0 assert "ROB capacity one" in report["reason"] and report["unavailable_estimates"] > 0 assert report["temporary_disk_bytes"] == 0 and not list(root.glob("candidate*")) shutil.rmtree(out) out.mkdir() command[-1] = f"--pim-target-config={config_path}" command += ["--pim-disable-synchronization"] result = subprocess.run(command, cwd=root, env=env, capture_output=True, text=True, timeout=60) assert result.returncode == 0, result.stdout + result.stderr report = json.loads((out / "reports/pipeline_selection.json").read_text()) assert report["selection"] == 0 and report["analyzed_candidates"] == 0 assert report["generated_candidates"] == 1 and report["lowered_candidates"] == 0 assert "synchronization ablation" in report["reason"] assert not (root / "executed").exists(), "production compiler executed a diagnostic or simulator" if __name__ == "__main__": main()