diff --git a/validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json b/validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json new file mode 100644 index 0000000..34e6284 --- /dev/null +++ b/validation/tools/pim/pimcomp/compare/pimcomp_architecture_sync_evidence.json @@ -0,0 +1,24 @@ +{ + "schema": 1, + "status": "documentary_evidence_plus_repository_checks", + "architectures": { + "arch-a": { + "pimcomp_identity": "ISAAC-like static/deterministic timing model", + "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN", + "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json", + "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration." + }, + "arch-b": { + "pimcomp_identity": "PUMA-like architecture", + "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN", + "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json", + "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST." + }, + "arch-c": { + "pimcomp_identity": "ISSCC 2023 ReRAM architecture row", + "primary_classification": "MAPPING_NOT_ESTABLISHED", + "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json", + "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved." + } + } +} diff --git a/validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py new file mode 100644 index 0000000..3169e82 --- /dev/null +++ b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_architecture_sync_contract.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Small, offline architecture-evidence helpers used by the sync experiment. + +The adversarial experiment owns the dynamic investigation. This module keeps +the stable repository/configuration facts it needs in one place and exposes a +deliberately small API so the experiment can also be imported as a library. +""" + +from __future__ import annotations + +import hashlib +import json +import subprocess +from pathlib import Path +from typing import Any + + +REPO = Path(__file__).resolve().parents[5] +VALIDATION = REPO / "validation" +PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN" +PIMSIM_ROOT = REPO / "backend-simulators/pim/pimsim-nn" +RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator" +EVIDENCE_PATH = Path(__file__).with_name("pimcomp_architecture_sync_evidence.json") + + +def _default_evidence() -> dict[str, Any]: + return { + "schema": 1, + "status": "documentary_evidence_plus_repository_checks", + "architectures": { + "arch-a": { + "pimcomp_identity": "ISAAC-like static/deterministic timing model", + "primary_classification": "STATIC_TIMING_CONTRACT_MAPPING_UNPROVEN", + "config": "validation/pimsim_configs/pimcomp/arch-a/throughput_config_1000ms.json", + "hardware_reference": "ISAAC (HPCA 2016), documentary mapping requires review of the cited paper/configuration.", + }, + "arch-b": { + "pimcomp_identity": "PUMA-like architecture", + "primary_classification": "HARDWARE_SYNC_EXISTS_BUT_NOT_MODELED_BY_PIMSIM_NN", + "config": "validation/pimsim_configs/pimcomp/arch-b/throughput_config_1000ms.json", + "hardware_reference": "PUMA, documentary valid/count synchronization is not encoded in ordinary PIMCOMP LD/ST.", + }, + "arch-c": { + "pimcomp_identity": "ISSCC 2023 ReRAM architecture row", + "primary_classification": "MAPPING_NOT_ESTABLISHED", + "config": "validation/pimsim_configs/pimcomp/arch-c/throughput_config_1000ms.json", + "hardware_reference": "ISSCC 2023 ReRAM reference; cross-system mapping is unresolved.", + }, + }, + } + + +def load_evidence() -> dict[str, Any]: + if EVIDENCE_PATH.is_file(): + return json.loads(EVIDENCE_PATH.read_text(encoding="utf-8")) + return _default_evidence() + + +def source_contract() -> dict[str, Any]: + """Return the contract claims used for report labeling. + + These labels are intentionally conservative: they are not a substitute + for a paper citation and never turn an unordered relation into a safe one. + """ + + return { + "ld_st": "ordinary timed global-memory accesses in the Rust model", + "send_recv": "explicit point-to-point synchronization modeled by the simulator", + "wait_sync": "instruction-level synchronization when emitted", + "valid_count": "not encoded by PIMCOMP exported LD/ST instructions", + "static_timing": "not proven as a PIMCOMP-to-ISAAC contract", + } + + +def classify_contract( + architecture: str, _contract: dict[str, Any], manifest: dict[str, Any] +) -> str: + return manifest["architectures"][architecture]["primary_classification"] + + +def architecture_source_evidence(architecture: str, manifest: dict[str, Any]) -> dict[str, Any]: + return dict(manifest["architectures"][architecture]) + + +def git_identity(path: Path) -> dict[str, Any]: + result: dict[str, Any] = {"path": str(path), "commit": None, "worktree_status": []} + if not path.exists(): + result["error"] = "missing" + return result + try: + result["commit"] = subprocess.run( + ["git", "-C", str(path), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + status = subprocess.run( + ["git", "-C", str(path), "status", "--short"], + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + result["worktree_status"] = status + except (OSError, subprocess.CalledProcessError) as exc: + result["error"] = f"{type(exc).__name__}: {exc}" + return result + + +def tree_hash(root: Path, pattern: str) -> dict[str, str]: + values: dict[str, str] = {} + for path in sorted(root.glob(pattern)): + if not path.is_file(): + continue + digest = hashlib.sha256(path.read_bytes()).hexdigest() + values[str(path.relative_to(root))] = digest + return values + + +def instruction_graph(artifact: Path) -> tuple[dict[tuple[int, int], list[tuple[int, int]]], dict[str, Any]]: + """Build same-core program-order edges from the exported JSON streams. + + Cross-core synchronization is added only when an artifact explicitly + carries matching SEND/RECV metadata. Ordinary global LD/ST creates no + edge by construction; that is the relation this experiment is testing. + """ + + graph: dict[tuple[int, int], list[tuple[int, int]]] = {} + streams: dict[int, list[dict[str, Any]]] = {} + for path in sorted(artifact.glob("core_*.json"), key=lambda item: int(item.stem.split("_")[1])): + try: + instructions = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + core = int(path.stem.split("_")[1]) + 1 + streams[core] = instructions if isinstance(instructions, list) else [] + for index in range(len(streams[core]) - 1): + graph.setdefault((core, index), []).append((core, index + 1)) + + sends: dict[tuple[int, int], tuple[int, int]] = {} + recvs: dict[tuple[int, int], tuple[int, int]] = {} + for core, instructions in streams.items(): + for index, instruction in enumerate(instructions): + op = str(instruction.get("op", instruction.get("operation", ""))).lower() + peer = instruction.get("core") + if peer is None: + continue + key = (core, int(peer) + 1) + if op == "send": + sends.setdefault(key, (core, index)) + elif op == "recv": + recvs.setdefault(key, (core, index)) + for key, send in sends.items(): + recv = recvs.get((key[1], key[0])) + if recv is not None: + graph.setdefault(send, []).append(recv) + return graph, {"cores": sorted(streams), "explicit_send_recv_edges": len(sends)} + + +def make_identical_inputs(model: Path, batch_size: int, out: Path) -> list[Path]: + import numpy as np + + import sys + + compare_dir = Path(__file__).resolve().parent + if str(compare_dir) not in sys.path: + sys.path.insert(0, str(compare_dir)) + import compare_raptor_pimcomp_model as compare # noqa: PLC0415 + + inputs, _ = compare.onnx_io(model) + arrays = [] + for _index, _name, element_type, shape in inputs: + dtype = compare._ONNX_TO_NP[element_type] + arrays.append(np.full(shape, 1.0, dtype=dtype)) + flattened = np.concatenate( + [compare.flatten_pimcomp_input(array) for array in arrays] + ) if arrays else np.empty(0, dtype=np.float32) + samples = [[flattened.copy()] for _ in range(batch_size)] + return [ + Path(path) + for path in compare.write_input_batch_binaries( + samples, out / "inputs/pimcomp_isolated" + ) + ] + + +if __name__ == "__main__": + print(json.dumps(load_evidence(), indent=2, sort_keys=True)) diff --git a/validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py new file mode 100644 index 0000000..2342d7a --- /dev/null +++ b/validation/tools/pim/pimcomp/compare/test_PIMCOMP_global_memory_sync.py @@ -0,0 +1,479 @@ +#!/usr/bin/env python3 +"""Reusable build, execution, and provenance helpers for PIMCOMP audits.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from bisect import bisect_left +from collections import defaultdict +from pathlib import Path +from typing import Any + +import numpy as np + + +REPO = Path(__file__).resolve().parents[5] +VALIDATION = REPO / "validation" +CONFIG_ROOT = VALIDATION / "pimsim_configs/pimcomp" +PIMCOMP_ROOT = REPO / "third_party/PIMCOMP-NN" +PIMSIM_NN_ROOT = REPO / "backend-simulators/pim/pimsim-nn" +RUST_ROOT = REPO / "backend-simulators/pim/pim-simulator" +RUST_BINARY = RUST_ROOT / "target/release/pim-simulator" +COMPARE_SCRIPT = Path(__file__).with_name("compare_raptor_pimcomp_model.py") +PYTHON = REPO / ".venv/bin/python" + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import compare_raptor_pimcomp_model as compare # noqa: E402 + + +class ExperimentError(RuntimeError): + pass + + +def architecture_configs(architecture: str) -> tuple[Path, Path]: + root = CONFIG_ROOT / architecture + return root / "throughput_config_1000ms.json", root / "latency_config.json" + + +def check_prerequisites(throughput: Path, latency: Path) -> None: + required = { + "PIMCOMP backend": PIMCOMP_ROOT / "build/PIMCOMP-NN", + "PIMCOMP frontend": PIMCOMP_ROOT / "frontend/frontend.py", + "Raptor compiler": REPO / "build_release/Release/bin/onnx-mlir", + "Rust simulator source": RUST_ROOT, + "pimsim-nn build": PIMSIM_NN_ROOT / "build", + "throughput config": throughput, + "latency config": latency, + } + missing = [f"{label}: {path}" for label, path in required.items() if not path.exists()] + if missing: + raise ExperimentError("missing prerequisites:\n" + "\n".join(missing)) + + +def _run(cmd: list[str], cwd: Path, log: Path, timeout: float = 0.0) -> subprocess.CompletedProcess[str]: + log.parent.mkdir(parents=True, exist_ok=True) + try: + result = subprocess.run( + [str(value) for value in cmd], + cwd=cwd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + timeout=None if timeout <= 0 else timeout, + ) + except subprocess.TimeoutExpired as exc: + log.write_text((exc.stdout or "") + "\nTIMEOUT\n", encoding="utf-8") + raise ExperimentError(f"command timed out: {' '.join(map(str, cmd))}") from exc + log.write_text(result.stdout, encoding="utf-8") + if result.returncode: + raise ExperimentError( + f"command failed ({result.returncode}): {' '.join(map(str, cmd))}\n" + f"see {log}\n{result.stdout[-3000:]}" + ) + return result + + +def build_simulator(args: Any, out: Path) -> None: + _run( + [ + "cargo", "build", "--release", "--no-default-features", + "--package", "pim-simulator", "--bin", "pim-simulator", + ], + RUST_ROOT, + out / "cargo_build.log", + float(getattr(args, "timeout", 0.0)), + ) + if not RUST_BINARY.is_file(): + raise ExperimentError(f"Rust simulator binary was not produced: {RUST_BINARY}") + + +def make_model(path: Path) -> None: + import onnx + from onnx import TensorProto, helper, numpy_helper + + shape = [1, 64, 8, 8] + weights = [] + for name in ("w0", "w1"): + weight = np.zeros((64, 64, 3, 3), dtype=np.float32) + for channel in range(64): + weight[channel, channel, 1, 1] = 1.0 + weights.append(numpy_helper.from_array(weight, name=name)) + input_value = helper.make_tensor_value_info("input", TensorProto.FLOAT, shape) + output_value = helper.make_tensor_value_info("output", TensorProto.FLOAT, shape) + nodes = [ + helper.make_node( + "Conv", ["input", "w0"], ["hidden"], name="conv0", + kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1], + dilations=[1, 1], group=1, + ), + helper.make_node( + "Conv", ["hidden", "w1"], ["output"], name="conv1", + kernel_shape=[3, 3], strides=[1, 1], pads=[1, 1, 1, 1], + dilations=[1, 1], group=1, + ), + ] + graph = helper.make_graph(nodes, "pimcomp_sync_two_conv", [input_value], [output_value], weights) + model = helper.make_model(graph, opset_imports=[helper.make_operatorsetid("", 13)]) + model.ir_version = min(model.ir_version, 8) + onnx.checker.check_model(model) + path.parent.mkdir(parents=True, exist_ok=True) + onnx.save(model, path) + + +def make_inputs( + model: Path, batch_size: int, seed: int, out: Path +) -> tuple[list[list[np.ndarray]], list[Path], list[Path], list[np.ndarray]]: + del seed # deterministic values are intentional; the seed remains in the report. + inputs_desc, _ = compare.onnx_io(model) + if len(inputs_desc) != 1: + raise ExperimentError("the synchronization model must have exactly one input") + _index, _name, element_type, shape = inputs_desc[0] + dtype = compare._ONNX_TO_NP[element_type] + arrays: list[np.ndarray] = [ + np.full(shape, float(10 ** index), dtype=dtype) for index in range(max(8, batch_size)) + ] + input_batch = [[array] for array in arrays] + raptor_paths = [ + Path(path) + for path in compare.write_input_batch_binaries(input_batch, out / "inputs/raptor") + ] + pimcomp_batch = [[compare.flatten_pimcomp_input(array)] for array in arrays] + pimcomp_paths = [ + Path(path) + for path in compare.write_input_batch_binaries(pimcomp_batch, out / "inputs/pimcomp") + ] + return input_batch, raptor_paths, pimcomp_paths, arrays + + +def make_references( + model: Path, input_batch: list[list[np.ndarray]], architecture_out: Path, _args: Any +) -> list[Path]: + import onnxruntime as ort + + input_desc, output_desc = compare.onnx_io(model) + session = ort.InferenceSession(str(model), providers=["CPUExecutionProvider"]) + references: list[Path] = [] + for index, sample in enumerate(input_batch): + values = {input_desc[item][1]: sample[item] for item in range(len(input_desc))} + outputs = session.run(None, values) + directory = architecture_out / "reference" / f"iteration_{index:06d}" + directory.mkdir(parents=True, exist_ok=True) + for output, descriptor in zip(outputs, output_desc): + output_index, name, _dtype, _shape = descriptor + filename = f"output{output_index}_{compare.sanitize_output_name(name)}.csv" + np.savetxt(directory / filename, np.asarray(output).reshape(-1), delimiter=",") + references.append(directory) + return references + + +def compile_artifact(args: Any, model: Path, architecture_out: Path, throughput_config: Path) -> dict[str, Any]: + comparison = architecture_out / "comparison" + command = [ + str(PYTHON), str(COMPARE_SCRIPT), + "--model", str(model), + "--out-dir", str(comparison), + "--common-dir", str(architecture_out / "common"), + "--pimcomp-config", str(throughput_config), + "--pimsim-mode", "throughput", + "--pimsim-time-ms", "1000", + "--batch-size", str(args.batch_size), + "--pimcomp-pipeline", "batch", + "--pimcomp-replication", "balance", + "--raptor-extra-arg=--pipeline=4", + "--seed", str(args.seed), + "--timeout-seconds", str(args.timeout), + ] + if args.no_fast: + command.append("--no-fast") + log = architecture_out / "comparison_compile.log" + try: + _run(command, REPO, log, args.timeout) + except ExperimentError as exc: + report = comparison / "pimcomp/comparison_report.json" + if not report.is_file(): + raise + report_data = json.loads(report.read_text(encoding="utf-8")) + raptor_error = "; ".join( + str(item.get("error", "")) + for item in report_data.get("failures", []) + if "RAPTOR" in str(item.get("stage", "")).upper() + ) or str(exc) + else: + report = comparison / "pimcomp/comparison_report.json" + report_data = json.loads(report.read_text(encoding="utf-8")) + raptor_error = "; ".join( + str(item.get("error", "")) + for item in report_data.get("failures", []) + if "RAPTOR" in str(item.get("stage", "")).upper() + ) or None + + paths = report_data.get("paths", {}) + pimcomp = Path(paths["pimcomp_exported_pim"]) if paths.get("pimcomp_exported_pim") else comparison / "pimcomp/exported" + pimsim = Path(paths["pimcomp_pimsim_nn"]) if paths.get("pimcomp_pimsim_nn") else comparison / "pimcomp/pimsim_nn" + raptor = Path(paths["raptor_pim"]) if paths.get("raptor_pim") else comparison / "raptor/pim.missing" + if not pimcomp.is_dir(): + raise ExperimentError(f"PIMCOMP Rust artifact missing; see {report}") + return { + "artifact": pimcomp, + "pimsim_artifact": pimsim if pimsim.is_dir() else None, + "raptor_artifact": raptor if raptor.is_dir() else Path(), + "raptor_error": raptor_error, + "comparison_report": report, + } + + +def _instruction_op(instruction: dict[str, Any]) -> str: + return str(instruction.get("op", instruction.get("operation", ""))).lower() + + +def _address(instruction: dict[str, Any], registers: dict[int, int], register: str) -> int | None: + try: + base = int(registers[int(instruction[register])]) + except (KeyError, TypeError, ValueError): + return None + offset = instruction.get("offset") or {} + select = int(offset.get("offset_select", 0)) + value = int(offset.get("offset_value", 0)) + # LD's global operand is r1 (selector bit 2); ST's global operand is rd + # (selector bit 1). The local simulator uses the same asymmetric ISA. + selector_bit = 1 if register == "rd" else 2 + return base + value if select & selector_bit else base + + +def _static_instruction( + core_file_index: int, + artifact_format: str, + instruction_index: int, + instruction: dict[str, Any], + address: int, + size: int, + artifact: Path, +) -> dict[str, Any]: + core = core_file_index + 1 + core_file = f"core_{core_file_index}.json" + binary_file = f"core_{core_file_index}.pim" + return { + "core": core, + # JSON PIMCOMP streams enter the Rust executor after an initial + # synthetic slot; Raptor's emitted binary/JSON streams do not. + "pc": instruction_index if artifact_format == "binary+json" else instruction_index - 1, + "artifact_pc": instruction_index, + "address": address, + "size": size, + "instruction_file": core_file, + "execution_file": binary_file if (artifact / binary_file).is_file() else core_file, + "artifact_format": artifact_format, + } + + +def analyze_artifact(artifact: Path) -> dict[str, Any]: + core_paths = sorted(artifact.glob("core_*.json"), key=lambda path: int(path.stem.split("_")[1])) + if not core_paths: + raise ExperimentError(f"artifact has no core_*.json files: {artifact}") + artifact_format = "binary+json" if any(artifact.glob("core_*.pim")) else "json" + stores: list[dict[str, Any]] = [] + loads: list[dict[str, Any]] = [] + counts: dict[str, int] = defaultdict(int) + instruction_files: dict[str, str] = {} + participating = 0 + for path in core_paths: + core_file_index = int(path.stem.split("_")[1]) + instructions = json.loads(path.read_text(encoding="utf-8")) + if instructions: + participating += 1 + instruction_files[f"core_{core_file_index}"] = str(path) + registers: dict[int, int] = {} + for index, instruction in enumerate(instructions): + op = _instruction_op(instruction) + counts[op] += 1 + if op in {"sldi", "lldi"} and "rd" in instruction and "imm" in instruction: + registers[int(instruction["rd"])] = int(instruction["imm"]) + continue + if op not in {"ld", "st"}: + continue + address = _address(instruction, registers, "rd" if op == "st" else "rs1") + if address is None: + continue + size = int(instruction.get("size", instruction.get("len", 0))) + if size <= 0: + continue + item = _static_instruction( + core_file_index, artifact_format, index, instruction, address, size, artifact + ) + (stores if op == "st" else loads).append(item) + + loads_by_address = sorted(loads, key=lambda item: int(item["address"])) + starts = [int(item["address"]) for item in loads_by_address] + dependencies: list[dict[str, Any]] = [] + for store in stores: + begin = int(store["address"]) + end = begin + int(store["size"]) + first = bisect_left(starts, end) + for load in loads_by_address[:first]: + if load["core"] == store["core"]: + continue + load_begin = int(load["address"]) + load_end = load_begin + int(load["size"]) + if load_end <= begin: + continue + dependencies.append({ + "overlap": { + "address_begin": max(begin, load_begin), + "address_end": min(end, load_end), + }, + "writer": dict(store), + "reader": dict(load), + "explicit_sync_ordering_evidence": False, + }) + dependencies.sort(key=lambda item: ( + item["overlap"]["address_begin"], item["writer"]["core"], + item["writer"]["pc"], item["reader"]["core"], item["reader"]["pc"], + )) + return { + "artifact_format": artifact_format, + "instruction_files": instruction_files, + "stores": stores, + "loads": loads, + "cross_core_dependencies": dependencies, + "cross_core_dependency_count": len(dependencies), + "participating_core_count": participating, + "instruction_counts": dict(sorted(counts.items())), + "representative_dependency": dependencies[0] if dependencies else None, + } + + +def trace_events(path: Path) -> list[dict[str, Any]]: + if not path.is_file(): + return [] + events = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if line.strip(): + events.append(json.loads(line)) + return events + + +def _prov(event: dict[str, Any]) -> tuple[int, ...]: + return tuple(int(value) for value in event.get("provenance", [])) + + +def _overlap(a: dict[str, Any], b: dict[str, Any]) -> bool: + return max(int(a["address"]), int(b["address"])) < min( + int(a["address"]) + int(a["size"]), int(b["address"]) + int(b["size"]) + ) + + +def dynamic_analysis(events: list[dict[str, Any]], static: dict[str, Any], config: Path) -> dict[str, Any]: + stores = [event for event in events if event.get("event") == "global_store"] + loads = [event for event in events if event.get("event") == "global_load"] + input_stores = [event for event in events if event.get("event") == "external_input_store"] + event_keys = { + (int(event.get("core", -1)), int(event.get("pc", -1)), int(event.get("address", -1)), int(event.get("size", -1))) + for event in stores + loads + } + executed = sum( + 1 for dependency in static["cross_core_dependencies"] + if ( + int(dependency["writer"]["core"]), int(dependency["writer"]["pc"]), + int(dependency["writer"]["address"]), int(dependency["writer"]["size"]), + ) in event_keys + and ( + int(dependency["reader"]["core"]), int(dependency["reader"]["pc"]), + int(dependency["reader"]["address"]), int(dependency["reader"]["size"]), + ) in event_keys + ) + cross_links = [] + for load in loads: + for writer in load.get("last_writers", []): + if int(writer.get("core", -1)) == 0 or not writer.get("provenance"): + continue + cross_links.append((load, writer)) + host_races = [] + for load in loads: + expected = (int(load.get("core_iteration", -1)),) + if any(_overlap(load, source) and _prov(load) and _prov(load) != expected for source in input_stores): + host_races.append(load) + reused_versions = { + int(version) + for store in stores + for version in store.get("overwritten_versions", []) + } + mixed = [event for event in events if event.get("event") == "cross_sample_data_mix"] + return { + "executed_cross_core_dependencies": executed, + "sample_dependent_cross_core_links": len(cross_links), + "host_input_lifetime_races": len(host_races), + "send_recv_generation_races": 0, + "mixed_sample_operations": len(mixed), + "global_memory_versions_reused": len(reused_versions), + "global_memory_generation_races": 0, + "host_input_race_contaminates_test": bool(host_races), + "config": str(config), + } + + +def run_rust( + artifact: Path, + inputs: list[Path], + references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], + out: Path, + args: Any, + *, + schedule_policy: str = "greedy", + schedule_seed: int = 0, + schedule_target: str | None = None, + schedule_deferral_budget: int | None = None, + target_stall: str | None = None, + channel_last: bool = False, +) -> dict[str, Any]: + if not inputs: + raise ExperimentError("Rust run requires at least one input") + out.mkdir(parents=True, exist_ok=True) + output = out / "output.bin" + batch_outputs = out / "iterations" + shutil.rmtree(batch_outputs, ignore_errors=True) + dump = compare.build_dump_ranges(artifact / "config.json", outputs_desc) + mode = "latency" if len(inputs) == 1 else "throughput" + command = [ + str(RUST_BINARY), "--folder", str(artifact), "--output", str(output), + "--dump", dump, "--mode", mode, "--batch-size", str(len(inputs)), + "--input-dir", str(inputs[0].parent), "--batch-output-dir", str(batch_outputs), + "--provenance-trace", str(out / "provenance.jsonl"), + "--diagnostic-schedule-policy", schedule_policy, + "--diagnostic-schedule-seed", str(schedule_seed), + ] + if schedule_target: + command += ["--diagnostic-schedule-target", schedule_target] + if schedule_deferral_budget is not None: + command += ["--diagnostic-schedule-deferral-budget", str(schedule_deferral_budget)] + if target_stall: + command += ["--diagnostic-target-stall", target_stall] + log = out / "simulator.log" + try: + _run(command, RUST_ROOT, log, float(args.timeout)) + except ExperimentError as exc: + return { + "passed": False, "max_diffs": {}, "error": str(exc), "completed": False, + "command": [str(value) for value in command], + } + failed: list[int] = [] + max_diffs: dict[str, float] = {} + for index, reference in enumerate(references[: len(inputs)]): + result = compare.compare_simulator_outputs( + batch_outputs / f"output_{index:06d}.bin", outputs_desc, reference, + threshold=args.threshold, rtol=args.rtol, channel_last=channel_last, + ) + if not result.passed: + failed.append(index) + for name, value in result.max_diffs.items(): + max_diffs[name] = max(max_diffs.get(name, 0.0), value) + return { + "passed": not failed, "max_diffs": max_diffs, + "failed_iterations": failed, "completed": True, + "command": [str(value) for value in command], + "trace": str(out / "provenance.jsonl"), + } diff --git a/validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py b/validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py new file mode 100644 index 0000000..2a60b30 --- /dev/null +++ b/validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py @@ -0,0 +1,1401 @@ +#!/usr/bin/env python3 +"""Search PIMCOMP and Raptor global-memory reuse with legal diagnostic scheduling. + +One documented invocation is: + + .venv/bin/python validation/tools/pim/pimcomp/correctness/test_PIMCOMP_adversarial_memory_sync.py \ + --out-dir /tmp/pimcomp-adversarial-sync --batch-size 4 --seed 0 --self-check + +The experiment uses identical external input bytes for throughput runs so the +known host-input lifetime issue cannot decide the intermediate-memory result. +The Rust scheduler remains greedy unless a diagnostic policy is explicitly +selected. +""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import sys +from bisect import bisect_right +from pathlib import Path +from typing import Any + + +REPO = Path(__file__).resolve().parents[5] +SCRIPT = Path(__file__).resolve() +HELPER_DIR = REPO / "validation/tools/pim/pimcomp/compare" +GLOBAL_SCRIPT = HELPER_DIR / "test_PIMCOMP_global_memory_sync.py" +AUDIT_SCRIPT = HELPER_DIR / "test_PIMCOMP_architecture_sync_contract.py" +PYTHON = REPO / ".venv/bin/python" + +sys.path.insert(0, str(HELPER_DIR)) +import test_PIMCOMP_architecture_sync_contract as audit # noqa: E402 +import test_PIMCOMP_global_memory_sync as global_sync # noqa: E402 + + +INVALID = "INVALID_SYNCHRONIZATION_REPRODUCER" +GENERATION_RACE = "PIMCOMP_GLOBAL_MEMORY_GENERATION_RACE_CONFIRMED" +READ_BEFORE_PRODUCE = "PIMCOMP_GLOBAL_MEMORY_READ_BEFORE_PRODUCE_CONFIRMED" +UNORDERED_COUNTEREXAMPLE = "PIMCOMP_GLOBAL_MEMORY_REUSE_UNORDERED_AND_COUNTEREXAMPLE_FOUND" +UNORDERED_NO_COUNTEREXAMPLE = "PIMCOMP_GLOBAL_MEMORY_REUSE_UNORDERED_NO_COUNTEREXAMPLE_FOUND" +PROVEN_ORDERED = "PIMCOMP_GLOBAL_MEMORY_REUSE_PROVEN_ORDERED" +RAPTOR_UNAVAILABLE = "RAPTOR_ARTIFACT_UNAVAILABLE_FOR_CONFIGURATION" + + +class ExperimentError(RuntimeError): + pass + + +def write_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, indent=2, sort_keys=True, default=str) + "\n", encoding="utf-8") + + +def relative(path: Path, root: Path) -> str: + try: + return str(path.resolve().relative_to(root.resolve())) + except ValueError: + return str(path) + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def event_key(item: dict[str, Any]) -> tuple[int, int, int, int]: + return (int(item["core"]), int(item["pc"]), int(item["address"]), int(item["size"])) + + +def matching_events( + events: list[dict[str, Any]], event_name: str, item: dict[str, Any] +) -> list[dict[str, Any]]: + return sorted( + [event for event in events if event.get("event") == event_name and event_key(event) == event_key(item)], + key=lambda event: int(event.get("cycle", 0)), + ) + + +def indexed_events(events: list[dict[str, Any]], event_name: str) -> dict[tuple[int, int, int, int], list[dict[str, Any]]]: + index: dict[tuple[int, int, int, int], list[dict[str, Any]]] = {} + for event in events: + if event.get("event") == event_name: + index.setdefault(event_key(event), []).append(event) + for values in index.values(): + values.sort(key=lambda event: int(event.get("cycle", 0))) + return index + + +def core_hb_edges( + graph: dict[tuple[int, int], list[tuple[int, int]]] +) -> dict[int, list[tuple[int, int, int]]]: + """Keep only cross-core edges; same-core order is represented by PC order.""" + edges: dict[int, list[tuple[int, int, int]]] = {} + for (core, pc), successors in graph.items(): + for target_core, target_pc in successors: + if target_core != core: + edges.setdefault(core, []).append((pc, target_core, target_pc)) + for values in edges.values(): + values.sort() + return edges + + +def core_hb_reachable( + edges: dict[int, list[tuple[int, int, int]]], + source: tuple[int, int], + target: tuple[int, int], +) -> bool: + if source[0] == target[0]: + return source[1] <= target[1] + earliest = {source[0]: source[1]} + pending = [source[0]] + while pending: + core = pending.pop() + entry_pc = earliest[core] + for send_pc, target_core, receive_pc in edges.get(core, []): + if send_pc < entry_pc: + continue + if receive_pc < earliest.get(target_core, 1 << 60): + earliest[target_core] = receive_pc + pending.append(target_core) + return earliest.get(target[0], 1 << 60) <= target[1] + + +def instruction_evidence( + item: dict[str, Any] | None, + static: dict[str, Any], + artifact: Path, + report_root: Path, + role: str, +) -> dict[str, Any] | None: + if item is None: + return None + event_name = "global_store" if role in {"writer", "overwrite", "static_writer"} else "global_load" + static_items = static["stores"] if event_name == "global_store" else static["loads"] + match = next( + ( + entry for entry in static_items + if event_key(entry) == event_key(item) + ), + None, + ) + artifact_rel = relative(artifact, report_root) + core_file = int(item["core"]) - 1 + evidence = { + "role": role, + "event": event_name, + "core": int(item["core"]), + "core_file_index": core_file, + "core_iteration": item.get("core_iteration"), + "cycle": item.get("cycle"), + "pc": int(item["pc"]), + "address": int(item["address"]), + "size": int(item["size"]), + "provenance": list(item.get("provenance", [])), + "version": item.get("version"), + "versions": list(item.get("versions", [])), + "overwritten_versions": list(item.get("overwritten_versions", [])), + "last_writers": list(item.get("last_writers", [])), + "artifact": artifact_rel, + "instruction_file": f"{artifact_rel}/{match['instruction_file']}" if match else None, + "executed_instruction_file": f"{artifact_rel}/{match['execution_file']}" if match else None, + "instruction_index": match.get("artifact_pc") if match else int(item["pc"]), + "artifact_pc": match.get("artifact_pc") if match else None, + "artifact_format": match.get("artifact_format") if match else static.get("artifact_format"), + } + return evidence + + +def static_dependency_evidence( + dependency: dict[str, Any], static: dict[str, Any], artifact: Path, report_root: Path, + hb_ordered: bool | None = None, +) -> dict[str, Any]: + if hb_ordered is None: + graph, _ = audit.instruction_graph(artifact) + hb_ordered = core_hb_reachable( + core_hb_edges(graph), + (int(dependency["reader"]["core"]), int(dependency["reader"]["artifact_pc"])), + (int(dependency["writer"]["core"]), int(dependency["writer"]["artifact_pc"])), + ) + return { + "range": dependency["overlap"], + "hb_ordered": bool(dependency.get("hb_ordered", hb_ordered)), + "explicit_sync_ordering_evidence": dependency.get("explicit_sync_ordering_evidence", False), + "writer": instruction_evidence(dependency["writer"], static, artifact, report_root, "static_writer"), + "reader": instruction_evidence(dependency["reader"], static, artifact, report_root, "static_reader"), + } + + +def enrich_order( + order: dict[str, Any], static: dict[str, Any], artifact: Path, report_root: Path +) -> dict[str, Any]: + order["smoking_guns"] = { + "old_generation_store": instruction_evidence( + order.get("old_store"), static, artifact, report_root, "writer" + ), + "overwrite_store": instruction_evidence( + order.get("next_store"), static, artifact, report_root, "overwrite" + ), + "consumer_load": instruction_evidence( + order.get("load"), static, artifact, report_root, "reader" + ), + } + return order + + +def provenance(item: dict[str, Any]) -> tuple[int, ...]: + return tuple(int(value) for value in item.get("provenance", [])) + + +def overlaps(left: dict[str, Any], right: dict[str, Any]) -> bool: + return max(int(left["address"]), int(right["address"])) < min( + int(left["address"]) + int(left["size"]), + int(right["address"]) + int(right["size"]), + ) + + +def writer_seen_by_load(store: dict[str, Any], load: dict[str, Any]) -> bool: + old_provenance = provenance(store) + return any( + int(writer.get("core", -1)) == int(store["core"]) + and int(writer.get("pc", -1)) == int(store["pc"]) + and provenance(writer) == old_provenance + and ( + int(writer.get("version", -1)) in {int(version) for version in load.get("versions", [])} + or int(writer.get("cycle", -1)) == int(store.get("cycle", -2)) + ) + for writer in load.get("last_writers", []) + ) + + +def provenance_context_index( + events: list[dict[str, Any]], +) -> dict[tuple[int, int], tuple[list[int], list[tuple[int, ...]]]]: + result: dict[tuple[int, int], tuple[list[int], list[tuple[int, ...]]]] = {} + for event in events: + if event.get("event") != "local_compute" or not provenance(event): + continue + key = (int(event.get("core", -1)), int(event.get("core_iteration", -1))) + cycles, values = result.setdefault(key, ([], [])) + cycles.append(int(event.get("cycle", -1))) + values.append(provenance(event)) + return result + + +def consumer_context_provenance( + events: list[dict[str, Any]], load: dict[str, Any], + context_index: dict[tuple[int, int], tuple[list[int], list[tuple[int, ...]]]] | None = None, +) -> tuple[int, ...] | None: + if context_index is not None: + key = (int(load["core"]), int(load.get("core_iteration", -1))) + indexed = context_index.get(key) + if indexed: + cycles, values = indexed + position = bisect_right(cycles, int(load["cycle"])) - 1 + return values[position] if position >= 0 else None + context = [ + event + for event in events + if event.get("event") == "local_compute" + and int(event.get("core", -1)) == int(load["core"]) + and int(event.get("core_iteration", -1)) == int(load.get("core_iteration", -2)) + and int(event.get("cycle", -1)) < int(load["cycle"]) + and provenance(event) + ] + return provenance(context[-1]) if context else None + + +def dependency_candidates( + events: list[dict[str, Any]], static: dict[str, Any], artifact: Path +) -> tuple[list[dict[str, Any]], dict[str, int]]: + stores = [event for event in events if event.get("event") == "global_store"] + loads = [event for event in events if event.get("event") == "global_load"] + candidates: list[dict[str, Any]] = [] + graph, _ = audit.instruction_graph(artifact) + hb_edges = core_hb_edges(graph) + store_index = indexed_events(events, "global_store") + load_index = indexed_events(events, "global_load") + context_index = provenance_context_index(events) + ordered = 0 + unordered = 0 + for dependency in static["cross_core_dependencies"]: + writer = dependency["writer"] + reader = dependency["reader"] + source = (int(reader["core"]), int(reader["artifact_pc"])) + target = (int(writer["core"]), int(writer["artifact_pc"])) + hb_ordered = core_hb_reachable(hb_edges, source, target) + if hb_ordered: + ordered += 1 + else: + unordered += 1 + writer_events = store_index.get(event_key(writer), []) + reader_events = load_index.get(event_key(reader), []) + for old_store in writer_events: + old_provenance = provenance(old_store) + if not old_provenance: + continue + for load in reader_events: + if int(load["cycle"]) <= int(old_store["cycle"]): + continue + consumer_provenance = consumer_context_provenance(events, load, context_index) + if consumer_provenance != old_provenance: + continue + next_stores = [ + event + for event in writer_events + if int(event["cycle"]) > int(load["cycle"]) + and int(event.get("core_iteration", 0)) > int(old_store.get("core_iteration", 0)) + and provenance(event) + and provenance(event) != old_provenance + ] + if not next_stores: + continue + next_store = next_stores[0] + if provenance(load) == old_provenance: + if not writer_seen_by_load(old_store, load): + continue + elif provenance(load) != provenance(next_store): + continue + candidates.append( + { + "dependency": dependency, + "hb_ordered": hb_ordered, + "old_store": old_store, + "load": load, + "next_store": next_store, + "consumer_context_provenance": list(consumer_provenance), + "normal_slack": int(next_store["cycle"]) - int(load["cycle"]), + "target_spec": target_spec(dependency, old_store, load), + } + ) + break + candidates.sort(key=lambda item: (item["hb_ordered"], item["normal_slack"], event_key(item["load"]))) + unique: list[dict[str, Any]] = [] + seen: set[tuple[Any, ...]] = set() + for candidate in candidates: + old = candidate["old_store"] + load = candidate["load"] + key = ( + old["core"], old["pc"], load["core"], load["pc"], + load["address"], load["size"], + ) + if key not in seen: + seen.add(key) + unique.append(candidate) + return unique, {"hb_ordered": ordered, "hb_unordered": unordered} + + +def representative_dependency_candidate( + events: list[dict[str, Any]], static: dict[str, Any], artifact: Path +) -> dict[str, Any] | None: + graph, _ = audit.instruction_graph(artifact) + hb_edges = core_hb_edges(graph) + context_index = provenance_context_index(events) + store_index = indexed_events(events, "global_store") + load_index = indexed_events(events, "global_load") + for dependency in static["cross_core_dependencies"]: + writer = dependency["writer"] + reader = dependency["reader"] + hb_ordered = core_hb_reachable( + hb_edges, + (int(reader["core"]), int(reader["artifact_pc"])), + (int(writer["core"]), int(writer["artifact_pc"])), + ) + writer_events = store_index.get(event_key(writer), []) + for load in load_index.get(event_key(reader), []): + context = consumer_context_provenance(events, load, context_index) + old_store = next( + ( + store for store in reversed(writer_events) + if int(store["cycle"]) < int(load["cycle"]) + and context + and provenance(store) == context + and writer_seen_by_load(store, load) + ), + None, + ) + if old_store is None: + continue + return { + "dependency": dependency, + "hb_ordered": hb_ordered, + "old_store": old_store, + "load": load, + "next_store": None, + "consumer_context_provenance": list(context), + "normal_slack": None, + "target_spec": target_spec(dependency, old_store, load), + "reuse_target": False, + } + return None + + +def read_before_produce_events( + events: list[dict[str, Any]], static: dict[str, Any], artifact: Path | None = None, + report_root: Path | None = None, +) -> list[dict[str, Any]]: + result = [] + seen: set[tuple[Any, ...]] = set() + store_index = indexed_events(events, "global_store") + load_index = indexed_events(events, "global_load") + context_index = provenance_context_index(events) + for dependency in static["cross_core_dependencies"]: + writer = dependency["writer"] + reader = dependency["reader"] + writer_events = store_index.get(event_key(writer), []) + for load in load_index.get(event_key(reader), []): + if load.get("versions"): + continue + context = consumer_context_provenance(events, load, context_index) + if not context: + continue + prior_expected = any( + int(store["cycle"]) < int(load["cycle"]) + and provenance(store) == context + for store in writer_events + ) + later_writer = next( + ( + store for store in writer_events + if int(store["cycle"]) > int(load["cycle"]) + and provenance(store) == context + ), + None, + ) + if prior_expected or later_writer is None: + continue + key = (load.get("core"), load.get("pc"), load.get("cycle"), load.get("address")) + if key in seen: + continue + seen.add(key) + item = { + "load": load, + "dependency": dependency, + "later_writer": later_writer, + "expected_provenance": list(context), + } + if artifact is not None and report_root is not None: + item["smoking_gun"] = { + "reader": instruction_evidence(load, static, artifact, report_root, "reader"), + "required_writer": instruction_evidence( + later_writer, static, artifact, report_root, "writer" + ), + "relationship": "consumer loaded before the expected generation was stored", + } + result.append(item) + return result + + +def target_spec( + dependency: dict[str, Any], old_store: dict[str, Any], load: dict[str, Any] +) -> str: + overlap = dependency["overlap"] + writer_min = int(old_store.get("core_iteration", 0)) + 1 + return ":".join( + str(value) + for value in ( + old_store["core"], + old_store["pc"], + load["core"], + load["pc"], + overlap["address_begin"], + overlap["address_end"], + load.get("core_iteration", 0), + writer_min, + ) + ) + + +def target_order( + events: list[dict[str, Any]], candidate: dict[str, Any] +) -> dict[str, Any]: + old = candidate["old_store"] + baseline_load = candidate["load"] + writer_events = matching_events(events, "global_store", old) + reader_events = matching_events(events, "global_load", baseline_load) + old_prov = provenance(old) + loads = [ + event for event in reader_events + if int(event.get("core_iteration", -1)) == int(baseline_load.get("core_iteration", -2)) + ] + load = loads[0] if loads else None + expected = list(consumer_context_provenance(events, load)) if load and consumer_context_provenance(events, load) else [] + old_stores = [ + event for event in writer_events + if load + and int(event["cycle"]) < int(load["cycle"]) + and provenance(event) == tuple(expected) + ] + old_store = old_stores[-1] if old_stores else None + next_store = None + if old_store and load: + next_store = next( + ( + event for event in writer_events + if int(event["cycle"]) > int(old_store["cycle"]) + and int(event["core_iteration"]) > int(old_store["core_iteration"]) + and provenance(event) != old_prov + ), + None, + ) + observed = list(load.get("provenance", [])) if load else [] + inversion = bool(next_store and load and int(next_store["cycle"]) < int(load["cycle"])) + newer_observed = bool( + next_store + and load + and provenance(next_store) + and provenance(next_store) == tuple(observed) + and int(next_store["cycle"]) < int(load["cycle"]) + ) + future_expected = bool( + load + and expected + and any( + int(event["cycle"]) > int(load["cycle"]) + and provenance(event) == tuple(expected) + for event in writer_events + ) + ) + read_before_produce = bool(load and expected and not old_store and future_expected) + return { + "old_store": old_store, + "load": load, + "next_store": next_store, + "expected_provenance": expected, + "observed_provenance": observed, + "order_inverted": inversion, + "wrong_generation_observed": newer_observed and observed != expected, + "read_before_produce": read_before_produce, + } + + +def delay_values(slack: int) -> list[int]: + if slack <= 0: + return [1] + values = { + max(1, slack // 4), + max(1, slack // 2), + max(1, (3 * slack) // 4), + max(1, slack - 1), + slack, + slack + 1, + 2 * slack, + } + return sorted(values) + + +def sanitize_isolated_diagnostics(raw: dict[str, Any], identical: bool) -> dict[str, Any]: + result = copy.deepcopy(raw) + result["raw_host_input_provenance_mismatches"] = raw["host_input_lifetime_races"] + result["host_input_lifetime_races"] = 0 if identical else raw["host_input_lifetime_races"] + result["host_input_race_contaminates_test"] = not identical + return result + + +def run_fixed_search( + candidate: dict[str, Any], arch_out: Path, isolated_inputs: list[Path], references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], args: argparse.Namespace, + baseline_events: list[dict[str, Any]], target_index: int, static: dict[str, Any], + report_root: Path, channel_last: bool, +) -> dict[str, Any]: + if candidate["normal_slack"] is None: + return { + "normal_slack": None, + "tested_delays": [], + "all_delays": [], + "runs": [], + "minimum_stall_for_inversion": None, + "minimum_stall_for_provenance_failure": None, + "baseline_order": enrich_order( + target_order(baseline_events, candidate), static, + candidate["dependency"]["artifact"], report_root, + ), + "skipped": "no next-generation store observed in baseline", + } + values = delay_values(int(candidate["normal_slack"])) + runs_by_delay: dict[int, dict[str, Any]] = {} + + def run_delay(delay: int) -> dict[str, Any]: + run_out = arch_out / "fixed_stall" / f"target_{target_index:02d}" / f"delay_{delay}" + result = global_sync.run_rust( + candidate["dependency"]["artifact"], isolated_inputs, references, outputs_desc, + run_out, args, schedule_policy="greedy", schedule_seed=args.seed, + schedule_target=candidate["target_spec"], + target_stall=( + f"{candidate['load']['core']}:{candidate['load']['pc']}:{candidate['load'].get('core_iteration', 0)}:{delay}" + ), + channel_last=channel_last, + ) + trace_path = run_out / "provenance.jsonl" + events = global_sync.trace_events(trace_path) if trace_path.is_file() else [] + order = target_order(events, candidate) if events else {"order_inverted": False, "wrong_generation_observed": False, "read_before_produce": False} + order = enrich_order(order, static, candidate["dependency"]["artifact"], report_root) + item = { + "delay": delay, + "result": result, + "trace": relative(trace_path, arch_out), + "order": order, + } + runs_by_delay[delay] = item + return item + + baseline_order = enrich_order( + target_order(baseline_events, candidate), static, + candidate["dependency"]["artifact"], report_root, + ) + for delay in values: + run_delay(delay) + inverted_delays = [delay for delay, item in runs_by_delay.items() if item["order"].get("order_inverted")] + first_inversion = min(inverted_delays) if inverted_delays else None + if first_inversion is not None: + lower = max( + [delay for delay, item in runs_by_delay.items() if delay < first_inversion and not item["order"].get("order_inverted")] + or [0] + ) + upper = first_inversion + while upper - lower > 1: + middle = (lower + upper) // 2 + item = runs_by_delay.get(middle) or run_delay(middle) + if item["order"].get("order_inverted"): + upper = middle + else: + lower = middle + first_inversion = upper + runs = [runs_by_delay[delay] for delay in sorted(runs_by_delay)] + return { + "normal_slack": candidate["normal_slack"], + "tested_delays": values, + "all_delays": sorted(runs_by_delay), + "runs": runs, + "minimum_stall_for_inversion": first_inversion, + "minimum_stall_for_provenance_failure": next( + (item["delay"] for item in runs if item["order"].get("wrong_generation_observed")), None + ), + "baseline_order": baseline_order, + } + + +def run_adversarial( + candidate: dict[str, Any], arch_out: Path, isolated_inputs: list[Path], references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], args: argparse.Namespace, + target_index: int, static: dict[str, Any], report_root: Path, channel_last: bool, +) -> dict[str, Any]: + run_out = arch_out / "adversarial" / f"target_{target_index:02d}" + result = global_sync.run_rust( + candidate["dependency"]["artifact"], isolated_inputs, references, outputs_desc, + run_out, args, schedule_policy="adversarial", schedule_seed=args.seed, + schedule_target=candidate["target_spec"], schedule_deferral_budget=args.deferral_budget, + channel_last=channel_last, + ) + trace_path = run_out / "provenance.jsonl" + events = global_sync.trace_events(trace_path) if trace_path.is_file() else [] + order = target_order(events, candidate) if events else { + "order_inverted": False, "wrong_generation_observed": False, "read_before_produce": False, + } + order = enrich_order(order, static, candidate["dependency"]["artifact"], report_root) + scheduler_events = [ + event for event in events if event.get("event") in {"scheduler_defer", "scheduler_prefer", "scheduler_force"} + ] + return { + "result": result, + "trace": relative(trace_path, arch_out), + "order": order, + "scheduler_events": len(scheduler_events), + "deferrals": sum(event.get("event") == "scheduler_defer" for event in scheduler_events), + "force_reasons": sorted({event.get("reason") for event in scheduler_events if event.get("event") == "scheduler_force"}), + "completed": bool(events and result.get("error") is None), + } + + +def run_randomized( + candidate: dict[str, Any], arch_out: Path, isolated_inputs: list[Path], references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], args: argparse.Namespace, + static: dict[str, Any], report_root: Path, channel_last: bool, +) -> list[dict[str, Any]]: + results = [] + for seed in range(args.randomized_seeds): + run_out = arch_out / "randomized" / f"seed_{seed:04d}" + result = global_sync.run_rust( + candidate["dependency"]["artifact"], isolated_inputs, references, outputs_desc, + run_out, args, schedule_policy="randomized", schedule_seed=seed, + schedule_target=candidate["target_spec"], + channel_last=channel_last, + ) + trace_path = run_out / "provenance.jsonl" + events = global_sync.trace_events(trace_path) if trace_path.is_file() else [] + order = target_order(events, candidate) if events else { + "wrong_generation_observed": False, + "order_inverted": False, + "read_before_produce": False, + } + order = enrich_order(order, static, candidate["dependency"]["artifact"], report_root) + diagnostics = ( + sanitize_isolated_diagnostics( + global_sync.dynamic_analysis( + events, static, candidate["dependency"]["artifact"] / "config.json" + ), + True, + ) + if events + else {} + ) + results.append({ + "seed": seed, + "result": result, + "trace": relative(trace_path, arch_out), + "order": order, + "diagnostics": diagnostics, + "read_before_produce": read_before_produce_events( + events, static, candidate["dependency"]["artifact"], report_root + ) if events else [], + }) + return results + + +def architecture_classification( + validation: dict[str, Any], candidates: list[dict[str, Any]], + fixed: list[dict[str, Any]], adversarial: list[dict[str, Any]], + randomized: list[dict[str, Any]], hb_counts: dict[str, int], +) -> str: + if not validation["valid"]: + return INVALID + read_before = ( + any(item["order"].get("read_before_produce") for item in adversarial) + or any( + item["order"].get("read_before_produce") + for search in fixed for item in search["runs"] + ) + or any(item["read_before_produce"] for item in randomized) + ) + generation_race = ( + any(item["order"].get("wrong_generation_observed") for item in adversarial) + or any( + item["order"].get("wrong_generation_observed") + for search in fixed for item in search["runs"] + ) + or any(item["order"].get("wrong_generation_observed") for item in randomized) + ) + if generation_race: + return GENERATION_RACE + if read_before: + return READ_BEFORE_PRODUCE + unordered = [candidate for candidate in candidates if not candidate["hb_ordered"]] + if ( + any(item["order"].get("order_inverted") for item in adversarial) + or any(item["order"].get("order_inverted") for search in fixed for item in search["runs"]) + or any(item["order"].get("order_inverted") for item in randomized) + ): + return UNORDERED_COUNTEREXAMPLE + if unordered or hb_counts["hb_unordered"]: + return UNORDERED_NO_COUNTEREXAMPLE + return PROVEN_ORDERED + + +def aggregate_classifications(classifications: list[str]) -> str: + if INVALID in classifications: + return INVALID + if RAPTOR_UNAVAILABLE in classifications: + return RAPTOR_UNAVAILABLE + for classification in ( + GENERATION_RACE, + READ_BEFORE_PRODUCE, + UNORDERED_COUNTEREXAMPLE, + UNORDERED_NO_COUNTEREXAMPLE, + ): + if classification in classifications: + return classification + return PROVEN_ORDERED + + +def architecture_interpretation(architecture: str, classification: str) -> dict[str, str]: + if architecture == "arch-a": + return { + "pimcomp_model": "PIMCOMP_ARCH_A_MODEL_PERMITS_DELAY", + "original_hardware": "ADVERSARIAL_DELAY_NOT_PROVEN_LEGAL_FOR_ORIGINAL_ARCH_A", + "note": "ISAAC static timing/documentary mapping remains unproven for generic PIMCOMP global memory.", + } + if architecture == "arch-b": + return { + "pimcomp_model": "PIMCOMP_ARCH_B_ABSTRACTION_NOT_GENERATION_SAFE" if classification in {GENERATION_RACE, READ_BEFORE_PRODUCE, UNORDERED_COUNTEREXAMPLE} else "PIMCOMP_ARCH_B_ABSTRACTION_NO_COUNTEREXAMPLE", + "original_hardware": "PUMA_VALID_COUNT_NOT_AUTOMATICALLY_INHERITED", + "note": "A PUMA valid/count buffer is documentary hardware evidence; PIMCOMP ordinary LD/ST does not encode it.", + } + return { + "pimcomp_model": "PIMCOMP_ARCH_C_ABSTRACT_MODEL_NOT_GENERATION_SAFE" if classification in {GENERATION_RACE, READ_BEFORE_PRODUCE, UNORDERED_COUNTEREXAMPLE} else "PIMCOMP_ARCH_C_ABSTRACT_MODEL_NO_COUNTEREXAMPLE", + "original_hardware": "MAPPING_NOT_ESTABLISHED", + "note": "The Arch-C row scales the cited 4MB ReRAM processor to a larger simulated mesh; ordering equivalence is unresolved.", + } + + +def unavailable_raptor_artifact( + architecture: str, + manifest: dict[str, Any], + throughput_config: Path, + latency_config: Path, + report_root: Path, + comparison_report: Path, + error: str | None, +) -> dict[str, Any]: + return { + "source": "raptor", + "identity": manifest["architectures"][architecture]["pimcomp_identity"], + "classification": RAPTOR_UNAVAILABLE, + "architecture_interpretation": { + "pimcomp_model": "RAPTOR_ARTIFACT_NOT_AVAILABLE", + "original_hardware": "NOT_TESTED", + "note": "Raptor did not emit an executable artifact for this configuration; no synchronization claim is made.", + }, + "config": { + "throughput": relative(throughput_config, REPO), + "latency": relative(latency_config, REPO), + "throughput_sha256": sha256(throughput_config), + "latency_sha256": sha256(latency_config), + }, + "artifact": { + "source": "raptor", + "rust_export": None, + "pimsim_export": None, + "format": None, + "hashes": {}, + "instruction_files": {}, + }, + "validation": { + "valid": False, + "artifact_available": False, + "batch1": {"passed": False, "error": error}, + "isolated_throughput": {"passed": False, "error": error}, + "active_core_count": 0, + "cross_core_dependency_count": 0, + "executed_cross_core_dependencies": 0, + "memory_ranges_reused": 0, + "sample_dependent_cross_core_links": 0, + "host_input_race_contaminates_test": False, + }, + "happens_before": {"hb_ordered": 0, "hb_unordered": 0}, + "static_instruction_counts": {}, + "diagnostics": { + "host_input_lifetime_races": 0, + "raw_host_input_provenance_mismatches": 0, + "send_recv_generation_races": 0, + "mixed_sample_operations": 0, + "global_memory_versions_reused": 0, + "generation_races": 0, + "read_before_produce": 0, + }, + "smoking_guns": { + "availability": { + "comparison_report": relative(comparison_report, report_root), + "error": error, + }, + "static_dependency": None, + "first_generation_race": None, + "first_read_before_produce": None, + }, + "targets": [], + "source_identity": audit.architecture_source_evidence(architecture, manifest), + } + + +def run_artifact_experiment( + architecture: str, + source: str, + artifact: Path, + pimsim_artifact: Path | None, + source_out: Path, + inputs: list[Path], + isolated_inputs: list[Path], + references: list[Path], + outputs_desc: list[tuple[int, str, int, list[int]]], + args: argparse.Namespace, + manifest: dict[str, Any], + throughput_config: Path, + latency_config: Path, + report_root: Path, +) -> dict[str, Any]: + """Run the same evidence-producing experiment on one executable artifact.""" + source_out.mkdir(parents=True, exist_ok=True) + channel_last = source == "pimcomp" + static = global_sync.analyze_artifact(artifact) + static["artifact"] = artifact + static["artifact_hashes"] = { + "config": sha256(artifact / "config.json"), + "core_instruction_files": audit.tree_hash(artifact, "core_*.json"), + "core_binary_files": audit.tree_hash(artifact, "core_*.pim"), + } + static_output = copy.deepcopy(static) + static_output.pop("artifact", None) + write_json(source_out / "analysis/global_memory_dependencies.json", static_output) + + batch1 = global_sync.run_rust( + artifact, inputs[:1], references[:1], outputs_desc, + source_out / "rust/batch1", args, channel_last=channel_last, + ) + distinct = global_sync.run_rust( + artifact, inputs[:args.batch_size], references[:args.batch_size], outputs_desc, + source_out / "rust/distinct_throughput", args, channel_last=channel_last, + ) + isolated_references = [references[0]] * args.batch_size + isolated = global_sync.run_rust( + artifact, isolated_inputs, isolated_references, outputs_desc, + source_out / "rust/input_isolated", args, channel_last=channel_last, + ) + isolated_trace = source_out / "rust/input_isolated/provenance.jsonl" + isolated_events = global_sync.trace_events(isolated_trace) + raw_dynamic = global_sync.dynamic_analysis(isolated_events, static, artifact / "config.json") + dynamic = sanitize_isolated_diagnostics(raw_dynamic, True) + static_candidates, hb_counts = dependency_candidates(isolated_events, static, artifact) + candidates = [candidate for candidate in static_candidates if not candidate["hb_ordered"]][: args.max_targets] + if not candidates: + fallback = representative_dependency_candidate(isolated_events, static, artifact) + if fallback is not None and not fallback["hb_ordered"]: + candidates = [fallback] + for candidate in candidates: + candidate["dependency"]["artifact"] = artifact + candidate["static_smoking_gun"] = static_dependency_evidence( + candidate["dependency"], static, artifact, report_root, candidate["hb_ordered"] + ) + + validation = { + "batch1": batch1, + "distinct_throughput": distinct, + "isolated_throughput": isolated, + "input_layout": "flattened PIMCOMP" if source == "pimcomp" else "NCHW Raptor", + "active_core_count": static["participating_core_count"], + "cross_core_dependency_count": static["cross_core_dependency_count"], + "executed_cross_core_dependencies": dynamic["executed_cross_core_dependencies"], + "memory_ranges_reused": dynamic["global_memory_versions_reused"], + "sample_dependent_cross_core_links": dynamic["sample_dependent_cross_core_links"], + "host_input_race_contaminates_test": dynamic["host_input_race_contaminates_test"], + } + validation["valid"] = bool( + batch1.get("passed") + and isolated.get("passed") + and validation["active_core_count"] >= 2 + and validation["cross_core_dependency_count"] > 0 + and validation["executed_cross_core_dependencies"] > 0 + and validation["memory_ranges_reused"] > 0 + and validation["sample_dependent_cross_core_links"] > 0 + and not validation["host_input_race_contaminates_test"] + ) + + fixed_results = [] + adversarial_results = [] + randomized_results: list[dict[str, Any]] = [] + for index, candidate in enumerate(candidates): + fixed_results.append( + run_fixed_search( + candidate, source_out, isolated_inputs, isolated_references, + outputs_desc, args, isolated_events, index, static, report_root, channel_last, + ) + ) + adversarial_results.append( + run_adversarial( + candidate, source_out, isolated_inputs, isolated_references, + outputs_desc, args, index, static, report_root, channel_last, + ) + ) + if index == 0 and args.randomized_seeds: + randomized_results = run_randomized( + candidate, source_out, isolated_inputs, isolated_references, + outputs_desc, args, static, report_root, channel_last, + ) + classification = architecture_classification( + validation, candidates, fixed_results, adversarial_results, + randomized_results, hb_counts, + ) + first_proof = next( + ( + target["order"]["smoking_guns"] + for target in adversarial_results + if target["order"].get("wrong_generation_observed") + ), + None, + ) + first_read_before = next( + ( + event["smoking_gun"] + for result in randomized_results + for event in result["read_before_produce"] + if event.get("smoking_gun") + ), + None, + ) + artifact_info = { + "source": source, + "rust_export": relative(artifact, report_root), + "pimsim_export": relative(pimsim_artifact, report_root) if pimsim_artifact else None, + "format": static["artifact_format"], + "hashes": static["artifact_hashes"], + "instruction_files": static["instruction_files"], + } + return { + "source": source, + "identity": manifest["architectures"][architecture]["pimcomp_identity"], + "classification": classification, + "architecture_interpretation": architecture_interpretation(architecture, classification), + "config": { + "throughput": relative(throughput_config, REPO), + "latency": relative(latency_config, REPO), + "throughput_sha256": sha256(throughput_config), + "latency_sha256": sha256(latency_config), + }, + "artifact": artifact_info, + "validation": validation, + "happens_before": hb_counts, + "static_instruction_counts": static["instruction_counts"], + "diagnostics": { + "host_input_lifetime_races": dynamic["host_input_lifetime_races"], + "raw_host_input_provenance_mismatches": dynamic["raw_host_input_provenance_mismatches"], + "send_recv_generation_races": dynamic["send_recv_generation_races"], + "mixed_sample_operations": dynamic["mixed_sample_operations"], + "global_memory_versions_reused": dynamic["global_memory_versions_reused"], + "fixed_generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for search in fixed_results for item in search["runs"] + ), + "adversarial_generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for item in adversarial_results + ), + "randomized_generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for item in randomized_results + ), + "randomized_read_before_produce": sum( + len(item["read_before_produce"]) for item in randomized_results + ), + "generation_races": sum( + item["order"].get("wrong_generation_observed", False) + for item in adversarial_results + ) + sum( + item["order"].get("wrong_generation_observed", False) + for search in fixed_results for item in search["runs"] + ) + sum( + item["order"].get("wrong_generation_observed", False) + for item in randomized_results + ), + "read_before_produce": sum( + item["order"].get("read_before_produce", False) + for item in adversarial_results + ) + sum( + item["order"].get("read_before_produce", False) + for search in fixed_results for item in search["runs"] + ) + sum( + len(item["read_before_produce"]) for item in randomized_results + ), + }, + "smoking_guns": { + "static_dependency": ( + static_dependency_evidence( + static["representative_dependency"], static, artifact, report_root + ) if static["representative_dependency"] else None + ), + "first_generation_race": first_proof, + "first_read_before_produce": first_read_before, + }, + "targets": [ + { + "candidate": { + key: value for key, value in candidate.items() if key != "dependency" + } | { + "dependency": { + key: value for key, value in candidate["dependency"].items() if key != "artifact" + } + }, + "fixed": fixed_results[index], + "adversarial": adversarial_results[index], + "randomized": randomized_results if index == 0 else [], + } + for index, candidate in enumerate(candidates) + ], + "source_identity": audit.architecture_source_evidence(architecture, manifest), + } + + +def markdown(report: dict[str, Any]) -> str: + def evidence_text(evidence: dict[str, Any] | None) -> str: + if evidence is None: + return "not observed" + provenance_text = evidence.get("provenance", []) + versions = evidence.get("versions", []) + version_text = evidence.get("version") + if versions: + version_text = versions if len(versions) <= 8 else versions[:3] + ["...", *versions[-3:]] + return ( + f"core {evidence['core']} (core_{evidence['core_file_index']}.json / " + f"{evidence.get('executed_instruction_file')}) " + f"PC {evidence['pc']} iter {evidence.get('core_iteration')} " + f"cycle {evidence.get('cycle')} range " + f"[{evidence['address']},{evidence['address'] + evidence['size']}) " + f"provenance {provenance_text} version {version_text}\n" + f" instruction: {evidence.get('instruction_file')}\n" + f" executed: {evidence.get('executed_instruction_file')}" + ) + + lines = [ + "PIMCOMP/Raptor Adversarial Global-Memory Synchronization Test", + "===============================================================", + "", + "Scheduler:", + " default: greedy / ASAP", + " diagnostics: bounded target delay, randomized, adversarial", + f" seed: {report['seed']}", + f" batch: {report['batch_size']}", + "", + ] + for architecture, architecture_item in report["architectures"].items(): + lines += [ + architecture, + "-" * len(architecture), + f"Contract classification: {architecture_item['contract_classification']}", + "", + ] + for source, item in architecture_item["artifacts"].items(): + validation = item["validation"] + lines += [ + f"{source.upper()} ARTIFACT", + f" artifact: {item['artifact']['rust_export']}", + f" format: {item['artifact']['format']}", + f" classification: {item['classification']}", + f" single inference: {'PASS' if validation['batch1']['passed'] else 'FAIL'}", + f" input-isolated greedy throughput: {'PASS' if validation['isolated_throughput']['passed'] else 'FAIL'}", + f" cores: {validation['active_core_count']}; cross-core ST→LD dependencies: {validation['cross_core_dependency_count']}", + f" executed dependencies: {validation['executed_cross_core_dependencies']}; reused ranges: {validation['memory_ranges_reused']}", + f" HB ordered/unordered: {item['happens_before']['hb_ordered']}/{item['happens_before']['hb_unordered']}", + f" host-input contamination: {validation['host_input_race_contaminates_test']}", + f" adversarial targets: {len(item['targets'])}; proven generation races: {item['diagnostics']['generation_races']}", + "", + ] + static = item["smoking_guns"].get("static_dependency") + if static: + lines += [ + " STATIC DEPENDENCY:", + f" range [{static['range']['address_begin']},{static['range']['address_end']})", + f" writer: {evidence_text(static['writer'])}", + f" reader: {evidence_text(static['reader'])}", + f" HB LD_N→ST_N+1: {'YES' if static['hb_ordered'] else 'NO'}", + "", + ] + for index, target in enumerate(item["targets"]): + candidate = target["candidate"] + order = target["adversarial"]["order"] + next_store = candidate.get("next_store") + next_store_text = next_store["cycle"] if next_store else "not observed" + lines += [ + f" TARGET {index}: range [{candidate['dependency']['overlap']['address_begin']},{candidate['dependency']['overlap']['address_end']})", + f" writer core {candidate['old_store']['core']} PC {candidate['old_store']['pc']} iter {candidate['old_store']['core_iteration']}", + f" reader core {candidate['load']['core']} PC {candidate['load']['pc']} iter {candidate['load']['core_iteration']}", + f" normal LD cycle {candidate['load']['cycle']}; next ST cycle {next_store_text}; slack {candidate['normal_slack']}", + f" HB LD_N→ST_N+1: {'YES' if candidate['hb_ordered'] else 'NO'}", + f" fixed minimum inversion delay: {target['fixed']['minimum_stall_for_inversion']}", + f" adversarial order inverted: {order.get('order_inverted', False)}", + f" expected provenance: {order.get('expected_provenance', [])}; observed: {order.get('observed_provenance', [])}", + "", + ] + guns = order.get("smoking_guns", {}) + if order.get("wrong_generation_observed"): + lines += [ + " CONFIRMED SMOKING GUN:", + f" old generation ST: {evidence_text(guns.get('old_generation_store'))}", + f" overwrite ST: {evidence_text(guns.get('overwrite_store'))}", + f" consumer LD: {evidence_text(guns.get('consumer_load'))}", + "", + ] + elif guns.get("consumer_load"): + lines += [ + " TRACE EVIDENCE (no wrong generation observed):", + f" old generation ST: {evidence_text(guns.get('old_generation_store'))}", + f" next generation ST: {evidence_text(guns.get('overwrite_store'))}", + f" consumer LD: {evidence_text(guns['consumer_load'])}", + "", + ] + read_before = item["smoking_guns"].get("first_read_before_produce") + if read_before: + lines += [ + " READ-BEFORE-PRODUCE SMOKING GUN:", + f" reader: {evidence_text(read_before.get('reader'))}", + f" required writer: {evidence_text(read_before.get('required_writer'))}", + "", + ] + lines += [" CONCLUSION:", f" {item['classification']}", ""] + lines += [ + "Interpretation:", + " A structural unordered relation without a dynamic counterexample is not called safe.", + " Each smoking gun names the generated JSON instruction and the executed .pim file when Raptor uses binary instructions.", + "", + f"PIMCOMP CLASSIFICATION: {report.get('pimcomp_classification', report['classification'])}", + f"RAPTOR CLASSIFICATION: {report.get('raptor_classification', 'not run')}", + f"FINAL CLASSIFICATION: {report['classification']}", + "", + f"PIMCOMP compiler semantics modified by this run: {report['pimcomp_semantics_modified']}", + ] + return "\n".join(lines) + "\n" + + +def self_check(report: dict[str, Any]) -> None: + if report["classification"] == INVALID: + raise ExperimentError("self-check rejected the reproducer") + for architecture, architecture_item in report["architectures"].items(): + for source, item in architecture_item["artifacts"].items(): + validation = item["validation"] + if item["classification"] == RAPTOR_UNAVAILABLE: + availability = item["smoking_guns"].get("availability", {}) + if not availability.get("comparison_report") or not availability.get("error"): + raise ExperimentError(f"{architecture}/{source}: missing-artifact evidence is incomplete") + continue + if not validation["valid"]: + raise ExperimentError(f"{architecture}/{source}: artifact validation failed") + for target in item["targets"]: + if not target["adversarial"]["completed"]: + raise ExperimentError(f"{architecture}/{source}: adversarial target did not complete") + order = target["adversarial"]["order"] + if order.get("wrong_generation_observed") and not order.get("order_inverted"): + raise ExperimentError(f"{architecture}/{source}: provenance mismatch lacks overwrite order proof") + if item["classification"] == READ_BEFORE_PRODUCE and not item["diagnostics"]["read_before_produce"]: + raise ExperimentError(f"{architecture}/{source}: read-before-produce lacks evidence") + if item["classification"] == GENERATION_RACE and not item["diagnostics"]["generation_races"]: + raise ExperimentError(f"{architecture}/{source}: generation-race lacks provenance proof") + if item["classification"] == PROVEN_ORDERED and item["happens_before"]["hb_unordered"]: + raise ExperimentError(f"{architecture}/{source}: ordered classification has unordered dependencies") + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out-dir", type=Path, required=True) + parser.add_argument("--batch-size", type=int, default=4) + parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--max-targets", type=int, default=2) + parser.add_argument("--randomized-seeds", type=int, default=2) + parser.add_argument("--deferral-budget", type=int, default=30000) + parser.add_argument("--threshold", type=float, default=1e-3) + parser.add_argument("--rtol", type=float, default=1e-4) + parser.add_argument("--timeout", type=float, default=1800.0) + parser.add_argument("--no-fast", action="store_true") + parser.add_argument("--self-check", action="store_true") + args = parser.parse_args() + if args.batch_size < 2: + parser.error("--batch-size must be at least 2") + if args.max_targets < 1 or args.randomized_seeds < 0 or args.deferral_budget < 1: + parser.error("target count, randomized seeds, and deferral budget are invalid") + args.out_dir = args.out_dir.resolve() + return args + + +def main() -> int: + args = parse_args() + out = args.out_dir + if out.exists() and any(out.iterdir()): + print(f"output directory must be new or empty: {out}", file=sys.stderr) + return 1 + out.mkdir(parents=True, exist_ok=True) + pimcomp_before = audit.git_identity(audit.PIMCOMP_ROOT) + manifest = audit.load_evidence() + source_contract = audit.source_contract() + report: dict[str, Any] = { + "schema": 1, + "experiment": "PIMCOMP adversarial global-memory synchronization", + "seed": args.seed, + "batch_size": args.batch_size, + "scheduler": { + "default_policy": "greedy", + "diagnostic_policies": ["bounded_target_stall", "randomized", "adversarial"], + "deferral_budget": args.deferral_budget, + }, + "source_identity": { + "repository": audit.git_identity(REPO), + "pimcomp": pimcomp_before, + "pimsim_nn": audit.git_identity(audit.PIMSIM_ROOT), + "rust_pim_simulator": audit.git_identity(audit.RUST_ROOT), + "script": {"path": relative(SCRIPT, REPO), "sha256": sha256(SCRIPT)}, + "architecture_audit_script": {"path": relative(AUDIT_SCRIPT, REPO), "sha256": sha256(AUDIT_SCRIPT)}, + }, + "architectures": {}, + "classification": INVALID, + "pimcomp_semantics_modified": False, + "errors": [], + } + try: + if not PYTHON.is_file(): + raise ExperimentError(f"Python virtual environment missing: {PYTHON}") + for architecture in ("arch-a", "arch-b", "arch-c"): + global_sync.check_prerequisites(*global_sync.architecture_configs(architecture)) + global_sync.build_simulator(args, out / "build") + model = out / "model.onnx" + global_sync.make_model(model) + report["model"] = { + "path": relative(model, out), + "sha256": sha256(model), + "description": "two connected identity padded 3x3 Conv stages", + } + run_count = max(8, args.batch_size) + input_batch, raptor_inputs, pimcomp_inputs, _ = global_sync.make_inputs(model, run_count, args.seed, out) + isolated_inputs = audit.make_identical_inputs(model, args.batch_size, out) + isolated_raptor_inputs = [ + Path(path) for path in global_sync.compare.write_input_batch_binaries( + [input_batch[0] for _ in range(args.batch_size)], out / "inputs/raptor_isolated" + ) + ] + isolated_hashes = {sha256(path) for path in isolated_inputs} + isolated_raptor_hashes = {sha256(path) for path in isolated_raptor_inputs} + if len(isolated_hashes) != 1: + raise ExperimentError("input-isolated files are not byte-identical") + if len(isolated_raptor_hashes) != 1: + raise ExperimentError("Raptor input-isolated files are not byte-identical") + _, outputs_desc = global_sync.compare.onnx_io(model) + references: list[Path] | None = None + for architecture in ("arch-a", "arch-b", "arch-c"): + architecture_out = out / architecture + throughput_config, latency_config = global_sync.architecture_configs(architecture) + compiled = global_sync.compile_artifact(args, model, architecture_out, throughput_config) + if references is None: + references = global_sync.make_references(model, input_batch, architecture_out, args) + contract = audit.classify_contract(architecture, source_contract, manifest) + raptor_available = ( + compiled["raptor_artifact"].is_dir() + and (compiled["raptor_artifact"] / "config.json").is_file() + and any(compiled["raptor_artifact"].glob("core_*.json")) + ) + raptor_result = ( + run_artifact_experiment( + architecture, "raptor", compiled["raptor_artifact"], None, + architecture_out / "raptor", raptor_inputs, isolated_raptor_inputs, references, + outputs_desc, args, manifest, throughput_config, + latency_config, out, + ) + if raptor_available + else unavailable_raptor_artifact( + architecture, manifest, throughput_config, latency_config, out, + architecture_out / "comparison/pimcomp/comparison_report.json", + compiled.get("raptor_error"), + ) + ) + artifacts = { + "pimcomp": run_artifact_experiment( + architecture, "pimcomp", compiled["artifact"], compiled["pimsim_artifact"], + architecture_out / "pimcomp", pimcomp_inputs, isolated_inputs, references, + outputs_desc, args, manifest, throughput_config, + latency_config, out, + ), + "raptor": raptor_result, + } + pimcomp_item = artifacts["pimcomp"] + item = { + "identity": manifest["architectures"][architecture]["pimcomp_identity"], + "contract_classification": contract, + # Keep the PIMCOMP fields at the architecture level for old consumers; + # the new nested records are the authoritative per-artifact evidence. + **pimcomp_item, + "contract_classification": contract, + "artifacts": artifacts, + "artifact_classifications": { + "pimcomp": artifacts["pimcomp"]["classification"], + "raptor": artifacts["raptor"]["classification"], + }, + "raptor": artifacts["raptor"], + } + report["architectures"][architecture] = item + write_json(architecture_out / "adversarial_memory_sync.json", item) + write_json(architecture_out / "pimcomp/adversarial_memory_sync.json", artifacts["pimcomp"]) + write_json(architecture_out / "raptor/adversarial_memory_sync.json", artifacts["raptor"]) + pimcomp_classes = [item["artifacts"]["pimcomp"]["classification"] for item in report["architectures"].values()] + raptor_classes = [item["artifacts"]["raptor"]["classification"] for item in report["architectures"].values()] + report["pimcomp_classification"] = aggregate_classifications(pimcomp_classes) + report["raptor_classification"] = aggregate_classifications(raptor_classes) + runnable_classes = [ + classification for classification in pimcomp_classes + raptor_classes + if classification != RAPTOR_UNAVAILABLE + ] + report["classification"] = aggregate_classifications(runnable_classes or raptor_classes) + pimcomp_after = audit.git_identity(audit.PIMCOMP_ROOT) + report["source_identity"]["pimcomp_after"] = pimcomp_after + report["pimcomp_semantics_modified"] = pimcomp_before["worktree_status"] != pimcomp_after["worktree_status"] + if args.self_check: + self_check(report) + except Exception as exc: + report["errors"].append(f"{type(exc).__name__}: {exc}") + report["classification"] = INVALID + write_json(out / "adversarial_memory_sync_report.json", report) + (out / "adversarial_memory_sync_report.md").write_text(markdown(report), encoding="utf-8") + print("=" * 60) + for architecture, item in report["architectures"].items(): + print(f"{architecture}/PIMCOMP: {item['artifacts']['pimcomp']['classification']}") + print(f"{architecture}/Raptor: {item['artifacts']['raptor']['classification']}") + print("=" * 60) + print(f"FINAL CLASSIFICATION: {report['classification']}") + print(f"JSON report: {out / 'adversarial_memory_sync_report.json'}") + print(f"Markdown report: {out / 'adversarial_memory_sync_report.md'}") + if report["errors"]: + print("Errors:", file=sys.stderr) + for error in report["errors"]: + print(f" {error}", file=sys.stderr) + return 0 if report["classification"] != INVALID else 1 + + +if __name__ == "__main__": + raise SystemExit(main())