"""Measure whole native processes; reference comparison remains a separate step.""" from __future__ import annotations import argparse import hashlib import importlib.util import json import math import os from pathlib import Path import platform import shutil import subprocess import time ROOT = Path(__file__).resolve().parents[2] CASES = ( ("hexa_nonlinearstatic", ROOT / "reference/hexa/hexa_nonlinearstatic.inp", 2.0), ("hexa_dynnonlinear", ROOT / "reference/hexa/hexa_dynnonlinear.inp", 100.0), ("shellNL-dyn-meshfine", ROOT / "reference/shellNL-dyn-meshfine/shellNL-dyn-meshfine.inp", 100.0), ) THREAD_KEYS = ("MKL_NUM_THREADS", "MKL_DYNAMIC", "MKL_THREADING_LAYER", "MKL_CBWR", "TBB_NUM_THREADS", "OMP_NUM_THREADS", "OMP_DYNAMIC", "OMP_PROC_BIND", "OMP_PLACES", "KMP_AFFINITY", "KMP_BLOCKTIME", "FESA_NUM_THREADS") def sha256(path): if not path.is_file(): return None 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 publish(path, value): """Expose only a fully serialized record; retain the last closed report.""" temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n", encoding="utf-8") os.replace(temporary, path) def command_output(command): try: result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, errors="replace", timeout=30, check=False) return {"command": command, "exit": result.returncode, "output": (result.stdout + result.stderr).strip()} except (OSError, subprocess.TimeoutExpired) as error: return {"command": command, "exit": None, "error": str(error)} def collect_environment(exe): """Capture actual build/runtime evidence outside every timed interval.""" build = next((p for p in exe.parents if (p / "CMakeCache.txt").is_file()), None) cache = (build / "CMakeCache.txt").read_text(errors="replace") if build else None result = { "source_commit": command_output(["git", "rev-parse", "HEAD"]), "source_status": command_output(["git", "status", "--porcelain"]), "measurement_tool_sha256": sha256(Path(__file__)), "comparator_sha256": {name: sha256(ROOT / "scripts/verification" / name) for name in ("compare_hex8.py", "compare_hht_dynamic.py")}, "source_diff_sha256": hashlib.sha256(command_output(["git", "diff", "HEAD", "--", "include", "src", "cmake", "CMakeLists.txt", "CMakePresets.json"])["output"].encode()).hexdigest(), "exe": str(exe), "exe_sha256": sha256(exe), "os": platform.platform(), "cpu": platform.processor(), "logical_processors": os.cpu_count(), "thread_environment": {key: os.environ.get(key) for key in THREAD_KEYS}, "related_environment": {key: value for key, value in sorted(os.environ.items()) if key.startswith(("MKL_", "TBB_", "OMP_", "KMP_", "FESA_"))}, "runtime_path": os.environ.get("PATH"), "thread_settings_note": "Environment requests; TBB uses runtime default (TBB_NUM_THREADS is not an enforced cap). Actual thread use requires profile evidence.", "build_type_flags_dependencies": cache, "build_files": {str(p): sha256(p) for p in ((build / "CMakeCache.txt", build / "build.ninja") if build else ())}, "runtime_dlls": {p.name: sha256(p) for p in sorted(exe.parent.glob("*.dll"))}, "compiler": command_output(["cl"]), "vtune": command_output([os.environ.get("VTUNE_EXE", "vtune"), "-version"]), } if os.name == "nt": import ctypes class PowerStatus(ctypes.Structure): _fields_ = [("ac_line", ctypes.c_ubyte), ("battery_flag", ctypes.c_ubyte), ("battery_percent", ctypes.c_ubyte), ("reserved", ctypes.c_ubyte), ("battery_life", ctypes.c_ulong), ("battery_full_life", ctypes.c_ulong)] power = PowerStatus() obtained = bool(ctypes.windll.kernel32.GetSystemPowerStatus(ctypes.byref(power))) result["system_power"] = {"obtained": obtained, "ac_line_status": power.ac_line if obtained else None, "battery_flag": power.battery_flag if obtained else None} result["power_scheme"] = command_output(["powercfg", "/getactivescheme"]) machine = command_output([ "powershell", "-NoProfile", "-Command", "ConvertTo-Json -Depth 4 @{boot=(Get-CimInstance Win32_OperatingSystem).LastBootUpTime.ToString('o'); cpu=(Get-CimInstance Win32_Processor | Select-Object Name,NumberOfCores,NumberOfLogicalProcessors); battery=(Get-CimInstance Win32_Battery | Select-Object BatteryStatus); runtime=(Get-ChildItem -LiteralPath '" + str(exe.parent).replace("'", "''") + "' -Filter *.dll | ForEach-Object { @{name=$_.Name;version=$_.VersionInfo.FileVersion} })}"]) if machine.get("exit") == 0: machine["output"] = json.loads(machine["output"]) result["machine_power_runtime_versions"] = machine return result def validate_result(case, scratch, result): """Reuse established inventory/finite prechecks without claiming comparison.""" name = "compare_hex8" if case.startswith("hexa_") else "compare_hht_dynamic" spec = importlib.util.spec_from_file_location(name, ROOT / f"scripts/verification/{name}.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) if case.startswith("hexa_"): model = module.input_inventory(scratch, case) values = module.read_results(result, model) module.check_result_values(values, model) return {**values["evidence"], "physical_values_checked": True} nodes, identity = module.input_inventory(scratch) _, evidence = module.read_results(result, nodes, identity) import h5py with h5py.File(result, "r") as file: if int(file["metadata"].attrs["schema_version"]) != 2: raise ValueError("Shell performance requires schema v2") source = file["metadata"].attrs["source_input_identity"] if isinstance(source, bytes): source = source.decode() if source != f"path={scratch.absolute().as_posix()};content_identity={identity}": raise ValueError("Shell result scratch path identity mismatch") return evidence def run_once(exe, original, case, directory, index, timeout=None): directory.mkdir(parents=True, exist_ok=False) scratch = directory / original.name original_before = sha256(original) shutil.copyfile(original, scratch) scratch_before = sha256(scratch) if original_before != scratch_before: raise ValueError("Input copy differs from source bytes") result = scratch.with_name(scratch.stem + "_results.h5") native_path, stdout_path = directory / "native.json", directory / "stdout.txt" command = [str(exe), str(scratch)] state, native_exit, error, process = "completed", None, None, None with stdout_path.open("wb") as output: start = time.perf_counter() try: process = subprocess.Popen(command, cwd=directory, stdout=output, stderr=subprocess.STDOUT) native_exit = process.wait(timeout=timeout) except OSError as exception: state, error = "launch-failed", str(exception) except (KeyboardInterrupt, subprocess.TimeoutExpired) as exception: state = "interrupted" if isinstance(exception, KeyboardInterrupt) else "timed-out" error = str(exception) if process is not None: process.kill() process.communicate() native_exit = process.returncode finally: elapsed = time.perf_counter() - start native = { "case": case, "command": command, "cwd": str(directory), "native_exit": native_exit, "execution_state": state, "original_sha256": original_before, "scratch_sha256": scratch_before, "result_sha256": sha256(result), "source_unchanged": sha256(original) == original_before, "scratch_unchanged": sha256(scratch) == scratch_before, "result_exists": result.is_file(), "original_sha256_after": sha256(original), "scratch_sha256_after": sha256(scratch), "exe_sha256": sha256(exe), "elapsed_seconds": elapsed, } publish(native_path, native) valid = (state == "completed" and native_exit == 0 and native["result_exists"] and native["source_unchanged"] and native["scratch_unchanged"] and math.isfinite(elapsed) and elapsed >= 0) evidence, validation_state = None, "not-run" if valid: try: evidence = validate_result(case, scratch, result) validation_state = "passed" except KeyboardInterrupt: valid, validation_state = False, "interrupted" error = "Result validation interrupted after native completion" except Exception as exception: valid, validation_state, error = False, "failed", str(exception) row = {**native, "kind": "measurement", "measurement_index": index, "original_input": str(original), "scratch_input": str(scratch), "result": str(result), "native_record": str(native_path), "stdout_log": str(stdout_path), "native_record_sha256": sha256(native_path), "solver_log": str(scratch.with_name(scratch.stem + "_log.txt")), "comparison_status": "pending", "comparison_report": None, "execution_valid": valid, "result_precheck": evidence, "result_validation_state": validation_state, "error": error} publish(directory / "run.json", row) return row def measure(exe, output_dir, cases=CASES): if output_dir.exists() and any(output_dir.iterdir()): raise FileExistsError(f"Nonempty artifact directory: {output_dir}") output_dir.mkdir(parents=True, exist_ok=True) environment = collect_environment(exe) report = {"schema": "fesa-performance-v1", "environment": environment, "cases": [], "protocol": {"warmup_runs": 0, "measurement_runs": 1, "cache_preparation": "no dedicated warmup or cache reset"}, "execution_complete": False} path = output_dir / "measurements.json" publish(path, report) for case, original, target in cases: case_report = {"case": case, "target_seconds": target, "runs": [], "all_measured_runs_within_target": False} report["cases"].append(case_report) if collect_environment(exe) != environment: report["error"] = "Source, binary, build or environment changed during measurement" publish(path, report) return 1 row = run_once(exe, original, case, output_dir / case / "run-1", 1) native_valid = row["execution_valid"] row["execution_valid"], row["within_target"] = False, False row["environment_validation_state"] = "pending" case_report["runs"].append(row) publish(Path(row["native_record"]).parent / "run.json", row) publish(path, report) try: unchanged = collect_environment(exe) == environment except KeyboardInterrupt: row["environment_validation_state"] = "interrupted" row["error"] = "Environment validation interrupted after native completion" except Exception as error: row["environment_validation_state"] = "failed" row["error"] = f"Environment validation failed after native completion: {error}" else: row["environment_validation_state"] = "passed" if unchanged else "changed" row["execution_valid"] = native_valid and unchanged row["within_target"] = row["execution_valid"] and row["elapsed_seconds"] <= target if not unchanged: row["error"] = "Source, binary, build or environment changed during native execution" publish(Path(row["native_record"]).parent / "run.json", row) case_report["all_measured_runs_within_target"] = row["within_target"] publish(path, report) print(f"{case} measurement 1: {row['elapsed_seconds']:.6f}s exit={row['native_exit']} valid={row['execution_valid']}", flush=True) if not row["execution_valid"]: return 1 report["execution_complete"] = True publish(path, report) return 0 def link_comparisons(output_dir): """Attach closed comparator reports by hashes without changing raw timing.""" path = output_dir / "measurements.json" report = json.loads(path.read_text(encoding="utf-8")) for case in report["cases"]: for row in case["runs"]: comparison_path = Path(row["native_record"]).parent / "comparison.json" if not comparison_path.is_file(): continue comparison = json.loads(comparison_path.read_text(encoding="utf-8")) identities = comparison.get("identities", {}) native_path = Path(row["native_record"]) native = json.loads(native_path.read_text(encoding="utf-8")) if (sha256(native_path) != row["native_record_sha256"] or row["elapsed_seconds"] != native["elapsed_seconds"] or sha256(Path(row["scratch_input"])) != row["scratch_sha256"] or sha256(Path(row["result"])) != row["result_sha256"] or identities.get("input", {}).get("sha256") != row["scratch_sha256"] or identities.get("results", {}).get("sha256") != row["result_sha256"] or comparison.get("case") != case["case"]): raise ValueError("Comparison does not identify the recorded native run/result") if case["case"].startswith("hexa_") and comparison.get("blocking_passed") and ( comparison.get("execution", {}).get("record_sha256") != row["native_record_sha256"]): raise ValueError("Hex8 comparator did not verify this execution record") row["comparison_status"] = "passed" if ( row["execution_valid"] and comparison.get("precheck", {}).get("passed") and comparison.get("blocking_passed") and comparison.get("verdict") == "passed" ) else "failed" row["comparison_report"] = str(comparison_path) row["comparison_report_sha256"] = sha256(comparison_path) report["comparisons_complete"] = (report["execution_complete"] and all( row["comparison_status"] != "pending" for case in report["cases"] for row in case["runs"])) report["all_comparisons_passed"] = (report["comparisons_complete"] and all( row["comparison_status"] == "passed" for case in report["cases"] for row in case["runs"])) publish(path, report) return 0 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--exe", type=Path) parser.add_argument("--output-dir", required=True, type=Path) parser.add_argument("--link-comparisons", action="store_true", help="Attach existing per-run comparison.json by native/input/result hashes.") args = parser.parse_args() if args.link_comparisons: try: return link_comparisons(args.output_dir.resolve()) except (OSError, ValueError, KeyError) as error: print(f"comparison link failed: {error}", flush=True) return 1 if args.exe is None: parser.error("--exe is required for measurement") if not args.exe.is_file(): parser.error("--exe must be an existing executable") if args.output_dir.exists() and (not args.output_dir.is_dir() or any(args.output_dir.iterdir())): parser.error("--output-dir must be new or empty") try: return measure(args.exe.resolve(), args.output_dir.resolve()) except (OSError, ValueError, KeyboardInterrupt) as error: print(f"measurement failed: {error}", flush=True) return 1 if __name__ == "__main__": raise SystemExit(main())