"""Production selection must use static compiler inputs only.""" import argparse import hashlib import json import os from pathlib import Path import shutil import struct 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()) assert report["strategy"] == "spatial-queue-aware" and report["lowered_candidates"] == 1 for name in ("spatial3_scheduled_no_comm.txt", "memory_report.txt"): assert (out / "reports" / name).stat().st_size > 0, name return report def check_synchronization_ablation(root, compiler, env): vgg = REPO / "validation/networks/pimcomp_models/vgg8/vgg8-mnist-reconstructed.onnx" out = root / "synchronization-ablation" # Ablation keeps normal selection and costs omitted synchronization separately. for arch, cores, crossbars in (("arch-a", 168, 96), ("arch-b", 138, 128)): programs, evidence, schedules = [], [], [] for disabled in (False, True): out.mkdir() command = [compiler, str(vgg), "-o", str(out / "model"), "--maccel=PIM", "--EmitPimCodegen", "--pipeline=4", f"--core-count={cores}", f"--crossbar-count={crossbars}", "--crossbar-size=128", "--mlir-disable-threading", "--pim-detect-communication-deadlock", f"--pim-target-config={REPO}/validation/pimsim_configs/pimcomp/{arch}/throughput_config_1000ms.json"] if disabled: command.append("--pim-disable-synchronization") 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["generated_candidates"] > 1 assert bool(report.get("synchronization_ablation")) == disabled if disabled: assert report["cost_model_program"] == "synchronized-reference" assert "report zero-cost SYNC/WAIT separately" in result.stderr ablation = report["backend_attempts"][0]["synchronization_ablation_cost"] assert ablation["cost_model_program"] == "synchronization-omitted" assert not ablation["used_for_selection"] and ablation["diagnostic_only"] assert all(candidate["eligible"] for candidate in report["candidates"]) evidence.append((report["selection"], [(c["interval_ns"], c["alternative_tie_interval_ns"], c["eligible"], c["stage_core_budgets"], c["task_stage_runs"]) for c in report["candidates"]])) schedules.append((out / "reports/spatial3_scheduled_no_comm.txt").read_bytes()) program = {} sync_count = 0 for path in (out / "pim").rglob("*"): if not path.is_file(): continue data = path.read_bytes() if path.suffix == ".pim": magic, version, count = struct.unpack("<4sII", data[:12]) assert magic == b"PIMB" and version == 1 and len(data) == 12 + 20 * count records = [data[i:i + 20] for i in range(12, len(data), 20)] sync_count += sum(record[0] in (31, 32) for record in records) data = b"".join(record for record in records if record[0] not in (31, 32)) program[str(path.relative_to(out / "pim"))] = hashlib.sha256(data).hexdigest() assert (sync_count == 0) == disabled programs.append(program) shutil.rmtree(out) assert evidence[0] == evidence[1], arch assert schedules[0] == schedules[1], arch assert programs[0] == programs[1], arch 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-no-lrn.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["generated_candidates"] > 1, report assert all("identity" not in c and "request_order" not in c["spatial_cost"] for c in report["candidates"]) check_selected_program(out) assert not (out / "dialects").exists() assert not list(out.glob("candidate*")) decisions.append((report["selection"], [(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 bool((out / "dialects").exists()) == diagnostics identity = {str(p.relative_to(out / "pim")): hashlib.sha256(p.read_bytes()).hexdigest() for p in (out / "pim").rglob("*") if p.is_file()} 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 assert "cannot compare the complete candidate set" in result.stderr report = json.loads((out / "reports/pipeline_selection.json").read_text()) assert report["status"] == "incomplete-costs" and "backend_attempts" not in report assert "ROB capacity one" in report["reason"] shutil.rmtree(out) out.mkdir() result = subprocess.run([*command, "--pim-disable-synchronization"], 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["status"] == "incomplete-costs" and "backend_attempts" not in report assert "ROB capacity one" in report["reason"] shutil.rmtree(out) check_synchronization_ablation(root, compiler, env) assert not (root / "executed").exists(), "production compiler executed a diagnostic or simulator" if __name__ == "__main__": main()