testCases/runParallelSmokeTests.py
#!/usr/bin/env python3Compare short serial and two-rank MPI solver runs and restarts.
import argparse
import hashlib
import math
import os
import select
import signal
import shutil
import subprocess
import sys
import tempfile
import time
import uuid
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
RUNNER = REPO_ROOT / "runSimulation.sh"
SMOKE_PARAMS = REPO_ROOT / "testCases" / "smoke.params"
BUILD_FILES = ("bretherton", "bretherton.c", "build.meta")
INVOCATION_TAG = "BRETHERTON_PARALLEL_SMOKE_INVOCATION"
TERMINAL_PREFIXES = (
"SUCCESS:", "INCOMPLETE_TMAX:", "INCOMPLETE_OUTLET:", "HARDFAIL_",
"ERROR: Runtime parameters must be finite."
)
def decode_output(output: str | bytes | None) -> str:Normalize partial subprocess output for an on-disk receipt.
if output is None:
return ""
if isinstance(output, bytes):
return output.decode("utf-8", errors="replace")
return output
def invocation_processes(process_group: int, invocation_tag: str) -> set[int]:Find non-zombie processes by process group or inherited unique tag.
members = set()
expected_tag = f"{INVOCATION_TAG}={invocation_tag}".encode()
for entry in Path("/proc").iterdir():
if not entry.name.isdigit():
continue
process_id = int(entry.name)
try:
status_lines = (entry / "status").read_text(encoding="utf-8").splitlines()
process_state = next(
line.split()[1] for line in status_lines if line.startswith("State:")
)
if process_state == "Z":
continue
in_process_group = os.getpgid(process_id) == process_group
try:
process_environment = (entry / "environ").read_bytes().split(b"\0")
has_invocation_tag = expected_tag in process_environment
except PermissionError:
has_invocation_tag = False
if in_process_group or has_invocation_tag:
members.add(process_id)
except (FileNotFoundError, ProcessLookupError, PermissionError, StopIteration):
continue
return members
def path_is_within(candidate: Path, root: Path) -> bool:Return whether candidate is root or one of its descendants.
try:
candidate.relative_to(root)
except ValueError:
return False
return True
def open_verified_group(
process_group: int, work_root: Path, invocation_tag: str
) -> tuple[list[tuple[int, int]], list[str]]:Open pidfds and prove ownership of every process-group member.
initial_members = invocation_processes(process_group, invocation_tag)
evidence = [f"process_group={process_group}",
f"initial_members={sorted(initial_members)}"]
if not initial_members:
raise RuntimeError("no live process-group members were available for ownership proof")
expected_uid = os.getuid()
expected_tag = f"{INVOCATION_TAG}={invocation_tag}".encode()
handles: list[tuple[int, int]] = []
try:
for process_id in sorted(initial_members):
process_fd = os.pidfd_open(process_id)
handles.append((process_id, process_fd))
process_root = Path(f"/proc/{process_id}")
process_uid = process_root.stat().st_uid
process_cwd = Path(os.readlink(process_root / "cwd")).resolve()
process_environment = (process_root / "environ").read_bytes().split(b"\0")
if process_uid != expected_uid:
raise RuntimeError(
f"pid {process_id} has uid {process_uid}, expected {expected_uid}"
)
if not path_is_within(process_cwd, work_root):
raise RuntimeError(
f"pid {process_id} cwd {process_cwd} is outside {work_root}"
)
if expected_tag not in process_environment:
raise RuntimeError(f"pid {process_id} lacks the invocation tag")
evidence.append(
f"pid={process_id} uid={process_uid} cwd={process_cwd} tag=verified"
)
final_members = invocation_processes(process_group, invocation_tag)
evidence.append(f"final_members={sorted(final_members)}")
if final_members != initial_members:
raise RuntimeError(
"process-group membership changed during ownership proof: "
f"{sorted(initial_members)} -> {sorted(final_members)}"
)
return handles, evidence
except Exception:
for _, process_fd in handles:
os.close(process_fd)
raise
def signal_verified_group(
handles: list[tuple[int, int]], process_group: int, invocation_tag: str
) -> list[str]:Terminate only the processes held by previously verified pidfds.
evidence = []
try:
for process_id, process_fd in handles:
try:
signal.pidfd_send_signal(process_fd, signal.SIGTERM)
evidence.append(f"pid={process_id} signal=TERM")
except ProcessLookupError:
evidence.append(f"pid={process_id} exited_before_TERM")
pending_fds = {process_fd for _, process_fd in handles}
term_deadline = time.monotonic() + 5.0
while pending_fds and time.monotonic() < term_deadline:
readable, _, _ = select.select(
list(pending_fds), [], [], max(0.0, term_deadline - time.monotonic())
)
pending_fds.difference_update(readable)
for process_id, process_fd in handles:
if process_fd not in pending_fds:
continue
try:
signal.pidfd_send_signal(process_fd, signal.SIGKILL)
evidence.append(f"pid={process_id} signal=KILL")
except ProcessLookupError:
evidence.append(f"pid={process_id} exited_before_KILL")
deadline = time.monotonic() + 5.0
while (invocation_processes(process_group, invocation_tag) and
time.monotonic() < deadline):
time.sleep(0.05)
survivors = invocation_processes(process_group, invocation_tag)
evidence.append(f"survivors={sorted(survivors)}")
if survivors:
raise RuntimeError(f"verified process-group members survived: {sorted(survivors)}")
return evidence
finally:
for _, process_fd in handles:
os.close(process_fd)
def write_params(destination: Path, case_no: int, **updates: str) -> None:Write a smoke parameter file with selected key replacements.
values = {"CaseNo": str(case_no), **{key: str(value) for key, value in updates.items()}}
lines = []
replaced = set()
for line in SMOKE_PARAMS.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#") and "=" in stripped:
key = stripped.split("=", 1)[0].strip()
if key in values:
lines.append(f"{key}={values[key]}")
replaced.add(key)
continue
lines.append(line)
lines.extend(f"{key}={value}" for key, value in values.items() if key not in replaced)
destination.write_text("\n".join(lines) + "\n", encoding="utf-8")
def read_params(source: Path) -> dict[str, str]:Read the flat key/value subset used by the smoke parameter files.
values = {}
for line in source.read_text(encoding="utf-8").splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#") and "=" in stripped:
key, value = stripped.split("=", 1)
values[key.strip()] = value.strip()
return values
def incomplete_message(params: Path) -> str:Return the solver’s exact expected tmax terminal diagnostic.
values = read_params(params)
return (
f"INCOMPLETE_TMAX: case {int(values['CaseNo'])} reached tmax without "
"the requested stationarity milestone: "
f"Ca {float(values['Ca']):.6g}, La {float(values['La']):.6g}, "
f"muR {float(values['muR']):.6g}, rhoR {float(values['rhoR']):.6g}."
)
def invoke(params: Path, output_root: Path, extra: list[str], timeout: int,
*, expected_outcome: str | None = None) -> str:Run the case runner and return its combined diagnostic stream.
command = ["bash", str(RUNNER), str(params), *extra]
work_root = output_root.parent.resolve(strict=True)
invocation_tag = uuid.uuid4().hex
environment = os.environ.copy()
environment["OUTPUT_ROOT"] = str(output_root)
environment[INVOCATION_TAG] = invocation_tag
stage = "build" if "--build-only" in extra else "run"
outcome = expected_outcome or ("build" if stage == "build" else "incomplete")
expected_status = 0 if outcome == "build" else 1
receipt = output_root.parent / f"{output_root.name}-{params.stem}-{stage}.log"
process = subprocess.Popen(
command,
cwd=work_root,
env=environment,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
start_new_session=True,
)
try:
output, _ = process.communicate(timeout=timeout + 30)
except subprocess.TimeoutExpired as timeout_error:
partial_output = decode_output(timeout_error.output)
timeout_evidence = [
"",
"# Harness timeout ownership evidence",
f"invocation_tag={invocation_tag}",
f"launcher_pid={process.pid}",
f"work_root={work_root}",
]
try:
handles, ownership_evidence = open_verified_group(
process.pid, work_root, invocation_tag
)
timeout_evidence.extend(ownership_evidence)
except Exception as ownership_error:
timeout_evidence.append(f"ownership_proof=FAILED: {ownership_error}")
receipt.write_text(
partial_output + "\n" + "\n".join(timeout_evidence) + "\n",
encoding="utf-8",
)
raise RuntimeError(
"runner timed out; ownership proof failed, so no process was signalled; "
f"evidence retained in {receipt}"
) from timeout_error
try:
timeout_evidence.extend(
signal_verified_group(handles, process.pid, invocation_tag)
)
output, _ = process.communicate(timeout=5)
except Exception as cleanup_error:
timeout_evidence.append(f"verified_cleanup=FAILED: {cleanup_error}")
receipt.write_text(
partial_output + "\n" + "\n".join(timeout_evidence) + "\n",
encoding="utf-8",
)
raise RuntimeError(
"runner timed out; verified cleanup did not complete; "
f"evidence retained in {receipt}"
) from timeout_error
receipt.write_text(
decode_output(output) + "\n" + "\n".join(timeout_evidence) + "\n",
encoding="utf-8",
)
raise RuntimeError(
f"runner exceeded {timeout + 30} seconds; its verified processes were "
f"terminated and evidence was retained in {receipt}"
) from timeout_error
receipt.write_text(output, encoding="utf-8")
if process.returncode != expected_status:
raise RuntimeError(
f"command returned {process.returncode}, expected {expected_status}: "
f"{' '.join(command)}\n"
f"{output}"
)
terminal_lines = [
line for line in output.splitlines()
if line.startswith(TERMINAL_PREFIXES)
]
if outcome == "build":
if terminal_lines:
raise AssertionError(f"build emitted a terminal solver status: {terminal_lines}")
elif outcome == "incomplete":
expected = incomplete_message(params)
if terminal_lines != [expected]:
raise AssertionError(
f"expected exact tmax incomplete diagnostic {expected!r}, "
f"found {terminal_lines!r}"
)
elif outcome == "outlet":
prefix = "INCOMPLETE_OUTLET: front tip reached the outlet buffer at t="
if (len(terminal_lines) != 1 or
not terminal_lines[0].startswith(prefix) or
not terminal_lines[0].endswith(".")):
raise AssertionError(
f"expected one outlet incomplete diagnostic, found {terminal_lines!r}"
)
try:
outlet_time = float(terminal_lines[0][len(prefix):-1])
except ValueError as error:
raise AssertionError(
f"outlet diagnostic has invalid time: {terminal_lines[0]!r}"
) from error
if not math.isfinite(outlet_time):
raise AssertionError(
f"outlet diagnostic has nonfinite time: {terminal_lines[0]!r}"
)
elif outcome == "domain-mismatch":
expected = (
"HARDFAIL_DOMAIN_MISMATCH: restart L0=16 differs from requested "
"Ldomain=15; a dump cannot be extended or shrunk in place."
)
if terminal_lines != [expected]:
raise AssertionError(
f"expected exact domain-mismatch diagnostic {expected!r}, "
f"found {terminal_lines!r}"
)
elif outcome == "invalid-parameters":
expected = "ERROR: Runtime parameters must be finite."
if terminal_lines != [expected]:
raise AssertionError(
f"expected exact finite-parameter diagnostic {expected!r}, "
f"found {terminal_lines!r}"
)
else:
raise ValueError(f"unknown expected outcome: {outcome}")
if "CFL must be <=" in output:
raise AssertionError(f"VOF CFL warning in {receipt}")
return output
def copy_build(source_case: Path, destination_case: Path) -> None:Copy a verified case-local build without copying runtime output.
destination_case.mkdir(parents=True, exist_ok=True)
for name in BUILD_FILES:
shutil.copy2(source_case / name, destination_case / name)
def sha256_file(path: Path) -> str:Hash a potentially large dump without retaining it in memory.
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 last_metrics(case_dir: Path) -> list[float]:Return the final numerical log row and require a single header.
log_files = list(case_dir.glob("c*-log"))
if len(log_files) != 1:
raise AssertionError(f"expected one case log in {case_dir}, found {log_files}")
lines = log_files[0].read_text(encoding="utf-8").splitlines()
if sum(line.startswith("# CaseNo") for line in lines) != 1:
raise AssertionError(f"expected one log header in {log_files[0]}")
rows = [line for line in lines if line and not line.startswith("#")]
if not rows:
raise AssertionError(f"no numerical rows in {log_files[0]}")
return [float(value) for value in rows[-1].split()[:8]]
def assert_metrics_close(serial: list[float], mpi: list[float], label: str) -> None:Compare state columns while allowing reduction-order roundoff.
for index, name in ((2, "t"), (3, "ke"), (4, "volume"),
(5, "front"), (6, "rear"), (7, "film")):
if not math.isclose(serial[index], mpi[index], rel_tol=5e-4, abs_tol=1e-7):
raise AssertionError(
f"{label} {name} mismatch: serial={serial[index]:.9g}, "
f"mpi={mpi[index]:.9g}"
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--rankfile", required=True, type=Path,
help="Open MPI rankfile containing at least ranks 0 and 1")
parser.add_argument("--work-root", type=Path,
help="retain test cases beneath this directory")
parser.add_argument("--timeout", type=int, default=600,
help="timeout in seconds for each runner invocation")
args = parser.parse_args()
if (sys.platform != "linux" or not hasattr(os, "pidfd_open") or
not hasattr(signal, "pidfd_send_signal")):
parser.error("guarded timeout handling requires Linux pidfd support")
rankfile = args.rankfile.expanduser().resolve(strict=True)
if args.timeout < 1:
parser.error("--timeout must be positive")
temporary_root = None
if args.work_root:
work_root = args.work_root.expanduser().resolve()
if work_root.exists() and any(work_root.iterdir()):
parser.error("--work-root must be empty; retain old test evidence separately")
work_root.mkdir(parents=True, exist_ok=True)
else:
temporary_root = Path(tempfile.mkdtemp(prefix="bretherton-parallel-smoke-"))
work_root = temporary_root
serial_output = work_root / "serial-cases"
mpi_output = work_root / "mpi-cases"
mpi_checkpoint_output = work_root / "mpi-checkpoint-cases"
serial_regrid_output = work_root / "serial-regrid-cases"
mpi_regrid_output = work_root / "mpi-regrid-cases"
domain_mismatch_output = work_root / "domain-mismatch-cases"
invalid_parameters_output = work_root / "invalid-parameter-cases"
stop_output_root = work_root / "mpi-stop-cases"
params_root = work_root / "params"
serial_output.mkdir(parents=True, exist_ok=True)
mpi_output.mkdir(parents=True, exist_ok=True)
mpi_checkpoint_output.mkdir(parents=True, exist_ok=True)
serial_regrid_output.mkdir(parents=True, exist_ok=True)
mpi_regrid_output.mkdir(parents=True, exist_ok=True)
domain_mismatch_output.mkdir(parents=True, exist_ok=True)
invalid_parameters_output.mkdir(parents=True, exist_ok=True)
stop_output_root.mkdir(parents=True, exist_ok=True)
params_root.mkdir(parents=True, exist_ok=True)
serial_params = params_root / "serial.params"
mpi_params = params_root / "mpi.params"
write_params(serial_params, 9901)
write_params(mpi_params, 9901)
invoke(serial_params, serial_output, ["--build-only"], args.timeout)
invalid_parameter_cases = (
(9905, {"tmax": "nan"}),
(9906, {"Ca": "inf"}),
)
for case_no, updates in invalid_parameter_cases:
params = params_root / f"invalid-{case_no}.params"
write_params(params, case_no, **updates)
copy_build(serial_output / "9901",
invalid_parameters_output / str(case_no))
invoke(params, invalid_parameters_output, ["--no-build"], args.timeout,
expected_outcome="invalid-parameters")
if (invalid_parameters_output / str(case_no) / "restart").exists():
raise AssertionError(
f"invalid-parameter case {case_no} created a restart"
)
invoke(serial_params, serial_output, ["--no-build"], args.timeout)
invoke(mpi_params, mpi_output,
["--ranks", "2", "--rankfile", str(rankfile), "--build-only"],
args.timeout)
invoke(mpi_params, mpi_output,
["--ranks", "2", "--rankfile", str(rankfile),
"--mpi-timeout", str(args.timeout), "--no-build"],
args.timeout)
serial_fresh_metrics = last_metrics(serial_output / "9901")
mpi_fresh_metrics = last_metrics(mpi_output / "9901")
assert_metrics_close(serial_fresh_metrics, mpi_fresh_metrics, "fresh run")
for case_dir in (serial_output / "9901", mpi_output / "9901"):
if (not (case_dir / "restart").is_file() or
not list((case_dir / "intermediate").glob("snapshot-*"))):
raise AssertionError(f"missing restart or snapshot output in {case_dir}")
# Preserve one identical t=0.05 seed before the restart checks advance and
# overwrite their case-local dumps.
seed_front = serial_fresh_metrics[5]
copy_build(serial_output / "9901", serial_regrid_output / "9903")
copy_build(mpi_output / "9901", mpi_regrid_output / "9903")
for destination in (serial_regrid_output, mpi_regrid_output):
shutil.copy2(serial_output / "9901" / "restart",
destination / "9903" / "restart")
copy_build(serial_output / "9901", domain_mismatch_output / "9904")
shutil.copy2(serial_output / "9901" / "restart",
domain_mismatch_output / "9904" / "restart")
mpi_checkpoint_params = params_root / "mpi-checkpoint-restart.params"
write_params(mpi_checkpoint_params, 9901, tmax="0.06")
copy_build(mpi_output / "9901", mpi_checkpoint_output / "9901")
shutil.copy2(mpi_output / "9901" / "restart",
mpi_checkpoint_output / "9901" / "restart")
shutil.copy2(mpi_output / "9901" / "c9901-log",
mpi_checkpoint_output / "9901" / "c9901-log")
mpi_checkpoint_restart_output = invoke(
mpi_checkpoint_params,
mpi_checkpoint_output,
["--ranks", "2", "--rankfile", str(rankfile),
"--mpi-timeout", str(args.timeout), "--no-build"],
args.timeout,
)
if (
"Restart file found - simulation will resume from checkpoint."
not in mpi_checkpoint_restart_output
):
raise AssertionError("MPI checkpoint restart was not detected by the runner")
mpi_checkpoint_restart_metrics = last_metrics(
mpi_checkpoint_output / "9901"
)
if mpi_checkpoint_restart_metrics[2] <= mpi_fresh_metrics[2]:
raise AssertionError(
"MPI checkpoint restart did not advance beyond its checkpoint time"
)
serial_restart_params = params_root / "serial-restart.params"
mpi_restart_params = params_root / "mpi-restart.params"
write_params(serial_restart_params, 9901, tmax="0.06")
write_params(mpi_restart_params, 9901, tmax="0.06")
shutil.copy2(serial_output / "9901" / "restart",
mpi_output / "9901" / "restart")
invoke(serial_restart_params, serial_output, ["--no-build"], args.timeout)
invoke(mpi_restart_params, mpi_output,
["--ranks", "2", "--rankfile", str(rankfile),
"--mpi-timeout", str(args.timeout), "--no-build"],
args.timeout)
assert_metrics_close(last_metrics(serial_output / "9901"),
last_metrics(mpi_output / "9901"), "restart")
assert_metrics_close(last_metrics(mpi_output / "9901"),
mpi_checkpoint_restart_metrics,
"serial-owned/MPI-owned checkpoint restart")
serial_regrid_params = params_root / "serial-regrid.params"
mpi_regrid_params = params_root / "mpi-regrid.params"
regrid_updates = {
"MAXlevel": "10", "filmMinLevel": "10", "regridBurnR": "0.01",
"freshFront": f"{seed_front:.17g}", "tmax": "0.06"
}
write_params(serial_regrid_params, 9903, **regrid_updates)
write_params(mpi_regrid_params, 9903, **regrid_updates)
serial_regrid_text = invoke(serial_regrid_params, serial_regrid_output,
["--no-build"], args.timeout)
mpi_regrid_text = invoke(
mpi_regrid_params, mpi_regrid_output,
["--ranks", "2", "--rankfile", str(rankfile),
"--mpi-timeout", str(args.timeout), "--no-build"], args.timeout
)
for output in (serial_regrid_text, mpi_regrid_text):
if ("Restart file found - simulation will resume from checkpoint."
not in output or
"# restart observation burn:" not in output or
"distance=0.01" not in output or "MAXlevel=10" not in output):
raise AssertionError(
"regrid case did not report restart, level, and burn provenance"
)
assert_metrics_close(last_metrics(serial_regrid_output / "9903"),
last_metrics(mpi_regrid_output / "9903"),
"regridded restart")
domain_mismatch_params = params_root / "domain-mismatch.params"
write_params(domain_mismatch_params, 9904, Ldomain="15", tmax="0.06")
mismatch_restart = domain_mismatch_output / "9904" / "restart"
restart_hash = sha256_file(mismatch_restart)
invoke(domain_mismatch_params, domain_mismatch_output,
["--no-build"], args.timeout, expected_outcome="domain-mismatch")
if sha256_file(mismatch_restart) != restart_hash:
raise AssertionError("domain-mismatch rejection modified its restart dump")
stop_params = params_root / "mpi-collective-stop.params"
write_params(stop_params, 9902, Ldomain="7")
copy_build(mpi_output / "9901", stop_output_root / "9902")
stop_output = invoke(
stop_params,
stop_output_root,
["--ranks", "2", "--rankfile", str(rankfile),
"--mpi-timeout", str(args.timeout), "--no-build"],
args.timeout,
expected_outcome="outlet",
)
if "INCOMPLETE_OUTLET: front tip reached the outlet buffer" not in stop_output:
raise AssertionError("MPI early-stop case did not exercise the outlet guard")
last_metrics(stop_output_root / "9902")
print("PASS: finite-parameter rejection, serial/MPI fresh run, cross-mode, "
"checkpoint and regridded restarts, domain rejection, outputs, and "
"collective early stop")
print(f"work root: {work_root}")
if temporary_root is not None:
shutil.rmtree(temporary_root)
return 0
if __name__ == "__main__":
raise SystemExit(main())