KOKO\Mimi
307 lines
16 KiB
Python
307 lines
16 KiB
Python
"""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())
|