261 lines
14 KiB
Python
261 lines
14 KiB
Python
"""Isolated native numerical execution; Python only handles process I/O."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import queue
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
from app.simulation.config import SolveIVPConfig
|
|
from app.simulation.results import GenericSimulationResult
|
|
from .transport import NativeSeriesJson, read_indexed_result
|
|
from .build import NativeBuild, build_native
|
|
from .compiler import NativeCapabilityError, compile_native_program
|
|
from .result_storage import ResultArchive, ResultQuotaError
|
|
from .property_warnings import format_property_warning
|
|
|
|
_WORKER_SHUTDOWN_GRACE_SECONDS = 5.0
|
|
|
|
|
|
def execute_native(build: NativeBuild, config: SolveIVPConfig, sample_step: float, *,
|
|
run_dir: Path, record_samples=True, cancel_check=None,
|
|
progress_callback=None, warning_callback=None, activity_tracker=None, timeout=300.0, raw_series=False) -> dict:
|
|
# Reject unsupported calls before allocating quota or evicting any history.
|
|
if config.method not in ("RK45", "BDF"):
|
|
raise NativeCapabilityError(f"Native v1 does not support method {config.method}.")
|
|
if not isinstance(config.atol, (int, float)) or config.atol != 1e-8 or config.first_step is not None:
|
|
raise NativeCapabilityError("Native uses generated per-state absolute tolerances and automatic initial step; custom config.atol/first_step are not supported.")
|
|
if any((run_dir / name).exists() for name in ("result.json", "cancel.request", "result-index.json")):
|
|
raise ValueError("Native execution requires a fresh run directory.")
|
|
archive = None
|
|
if record_samples:
|
|
metadata = {key: build.manifest[key] for key in ("buildKey", "stateKeys", "variables")
|
|
if key in build.manifest}
|
|
if "stateKeys" in metadata:
|
|
# Algebraic-only models carry one private constant integrator state.
|
|
metadata["stateColumns"] = ["time", *(metadata["stateKeys"] or [None])]
|
|
if "variables" in metadata:
|
|
metadata["outputColumns"] = ["time", *([v["key"] for v in metadata["variables"]] or [""])]
|
|
metadata["simulation"] = {"method": config.method, "start": config.t_start,
|
|
"stop": config.t_stop, "sampleStep": sample_step,
|
|
"maxStep": config.max_step, "rtol": config.rtol}
|
|
archive = ResultArchive(metadata)
|
|
try:
|
|
payload = _execute_native(build, config, sample_step, run_dir=run_dir,
|
|
record_samples=record_samples, cancel_check=cancel_check,
|
|
progress_callback=progress_callback, warning_callback=warning_callback, activity_tracker=activity_tracker,
|
|
timeout=timeout, raw_series=raw_series, archive=archive)
|
|
if archive is not None:
|
|
payload["resultStorage"] = archive.finish(payload)
|
|
return payload
|
|
except BaseException as exc:
|
|
if archive is not None and not archive.closed:
|
|
try:
|
|
archive.finish(error=str(exc))
|
|
except Exception as storage_error:
|
|
exc.add_note(f"Could not finalize result archive: {storage_error}")
|
|
raise
|
|
finally:
|
|
if archive is not None:
|
|
archive.close()
|
|
|
|
|
|
def _execute_native(build: NativeBuild, config: SolveIVPConfig, sample_step: float, *,
|
|
run_dir: Path, record_samples=True, cancel_check=None,
|
|
progress_callback=None, warning_callback=None, activity_tracker=None, timeout=300.0, raw_series=False,
|
|
archive: ResultArchive | None = None) -> dict:
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
output = run_dir / "result.json"
|
|
cancel_path = run_dir / "cancel.request"
|
|
index_path = run_dir / "result-index.json"
|
|
command = [str(build.executable), "--method", config.method,
|
|
"--start", str(config.t_start), "--stop", str(config.t_stop),
|
|
"--sample-step", str(sample_step), "--max-step", str(config.max_step),
|
|
"--rtol", str(config.rtol), "--timeout", str(timeout),
|
|
"--cancel-file", str(cancel_path.resolve()), "--output", str(output.resolve())]
|
|
if raw_series:
|
|
command.extend(["--result-index", str(index_path.resolve())])
|
|
if not record_samples:
|
|
command.append("--solve-only")
|
|
if archive is not None:
|
|
command.extend(["--sample-file", str(archive.path / "states.bin"),
|
|
"--output-block-file", str(archive.path / "outputs.bin"), "--storage-control"])
|
|
creationflags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0
|
|
started = time.perf_counter()
|
|
# A fast worker may finish before the monitor's first iteration. Honor a
|
|
# cancellation already requested during preparation before spawning it.
|
|
cancelled_at = started if cancel_check is not None and cancel_check() else None
|
|
if cancelled_at is not None:
|
|
cancel_path.write_text("cancel\n", encoding="ascii")
|
|
process = subprocess.Popen(command, cwd=build.executable.parent,
|
|
stdin=subprocess.PIPE if archive is not None else subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
|
|
text=True, encoding="utf-8", errors="replace", creationflags=creationflags)
|
|
messages: queue.Queue[str] = queue.Queue(maxsize=128)
|
|
reader_stop = threading.Event()
|
|
def read_stderr():
|
|
assert process.stderr is not None
|
|
for line in process.stderr:
|
|
while not reader_stop.is_set():
|
|
try:
|
|
messages.put(line, timeout=.1)
|
|
break
|
|
except queue.Full:
|
|
continue
|
|
if reader_stop.is_set():
|
|
break
|
|
reader = threading.Thread(target=read_stderr, daemon=True)
|
|
reader.start()
|
|
last_time = config.t_start
|
|
last_nfev = last_accepted = 0
|
|
forced_cancel = False
|
|
property_warnings = []
|
|
try:
|
|
if archive is not None:
|
|
archive.worker_started(process.pid)
|
|
if activity_tracker is not None:
|
|
activity_tracker.start_integration(config.t_start)
|
|
with (run_dir / "worker.log").open("w", encoding="utf-8") as log:
|
|
while process.poll() is None or not messages.empty() or reader.is_alive():
|
|
now = time.perf_counter()
|
|
if cancel_check is not None and cancel_check() and cancelled_at is None:
|
|
cancel_path.write_text("cancel\n", encoding="ascii")
|
|
cancelled_at = now
|
|
cancellation_expired = (cancelled_at is not None and
|
|
now-cancelled_at > _WORKER_SHUTDOWN_GRACE_SECONDS)
|
|
deadline_expired = timeout > 0 and now-started > timeout+_WORKER_SHUTDOWN_GRACE_SECONDS
|
|
if process.poll() is None and (cancellation_expired or deadline_expired):
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
if cancelled_at is not None:
|
|
forced_cancel = True
|
|
break
|
|
raise RuntimeError("Native worker time limit expired; the unresponsive process was terminated before it returned results.")
|
|
try:
|
|
line = messages.get(timeout=0.05)
|
|
except queue.Empty:
|
|
continue
|
|
log.write(line)
|
|
try:
|
|
event = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if event.get("phase") == "storage-reserve" and archive is not None:
|
|
reply = "ok\n"
|
|
try:
|
|
archive.reserve(event.get("bytes"))
|
|
except ResultQuotaError as exc:
|
|
reply = "full\n"
|
|
log.write(f"Result storage reservation failed: {exc}\n")
|
|
except OSError as exc:
|
|
reply = "io\n"
|
|
log.write(f"Result storage I/O failed: {exc}\n")
|
|
try:
|
|
process.stdin.write(reply)
|
|
process.stdin.flush()
|
|
except (BrokenPipeError, OSError):
|
|
if process.poll() is None:
|
|
raise
|
|
continue
|
|
if event.get("phase") == "property-warning":
|
|
warning = format_property_warning(event["warning"])
|
|
property_warnings.append(warning)
|
|
if warning_callback:
|
|
warning_callback(warning)
|
|
continue
|
|
if event.get("phase") == "integrating":
|
|
last_time = max(last_time, min(config.t_stop, float(event["time"])))
|
|
last_nfev = int(event["nfev"])
|
|
last_accepted = int(event["acceptedSteps"])
|
|
if progress_callback:
|
|
progress_callback((last_time-config.t_start)/(config.t_stop-config.t_start), "integrating")
|
|
if activity_tracker is not None:
|
|
activity_tracker.record_native_progress(last_time, int(event["nfev"]), int(event["acceptedSteps"]))
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
reader_stop.set()
|
|
reader.join(timeout=2)
|
|
if process.stdin is not None:
|
|
process.stdin.close()
|
|
if process.stderr is not None:
|
|
process.stderr.close()
|
|
if forced_cancel:
|
|
# A killed worker cannot publish trustworthy result artifacts. Retain
|
|
# only its last progress report; do not invent a trajectory or totals.
|
|
payload = {
|
|
"success": False, "status": "cancelled",
|
|
"message": "Simulation cancelled; the unresponsive worker was terminated. No complete result artifact is available; time and counters are the last reported progress.",
|
|
"backend": "native-c", "method": config.method,
|
|
"simulatedUntil": last_time, "nfev": last_nfev, "acceptedSteps": last_accepted,
|
|
"series": {}, "final": {}, "finalState": [],
|
|
"solverControl": {"reason": "cancelled", "operation": "worker-termination",
|
|
"forcedTermination": True, "workerExitCode": process.returncode,
|
|
"resultAvailable": False, "statisticsComplete": False},
|
|
}
|
|
else:
|
|
if not output.is_file():
|
|
raise RuntimeError(f"Native worker exited with code {process.returncode} without results; see {run_dir / 'worker.log'}.")
|
|
if process.returncode not in (0, 2):
|
|
raise RuntimeError(f"Native worker failed with exit code {process.returncode}.")
|
|
try:
|
|
payload = (read_indexed_result(output, index_path) if raw_series
|
|
else json.loads(output.read_text(encoding="utf-8")))
|
|
except OSError as exc:
|
|
raise RuntimeError(f"Cannot read native worker result artifacts: {exc}") from exc
|
|
payload["propertyWarnings"] = (
|
|
property_warnings if forced_cancel else
|
|
[format_property_warning(w) for w in payload.get("propertyWarnings", [])])
|
|
payload["processWallSeconds"] = time.perf_counter()-started
|
|
payload["buildKey"] = build.manifest["buildKey"]
|
|
payload["cacheHit"] = build.cache_hit
|
|
payload["buildSeconds"] = build.seconds
|
|
payload["buildDetails"] = build.details
|
|
if activity_tracker is not None:
|
|
activity_tracker.record_native_progress(payload["simulatedUntil"], payload["nfev"], payload["acceptedSteps"])
|
|
return payload
|
|
|
|
|
|
def simulate_native(network, config, *, sample_step, progress_callback=None,
|
|
warning_callback=None, cancel_check=None, activity_tracker=None, raw_series=False):
|
|
if config.method not in ("RK45", "BDF"):
|
|
raise NativeCapabilityError(f"Native v1 does not support method {config.method}.")
|
|
if progress_callback:
|
|
progress_callback(0.0, "native-generation")
|
|
program = compile_native_program(network)
|
|
build = build_native(program, progress_callback=(
|
|
(lambda phase: progress_callback(0.0, phase)) if progress_callback else None
|
|
))
|
|
try:
|
|
with tempfile.TemporaryDirectory(prefix="native-simulation-") as directory:
|
|
data = execute_native(build, config, sample_step, run_dir=Path(directory),
|
|
cancel_check=cancel_check, progress_callback=progress_callback,
|
|
warning_callback=warning_callback,
|
|
activity_tracker=activity_tracker, raw_series=raw_series)
|
|
finally:
|
|
# Keep the executable pinned through process exit and result reading;
|
|
# only then may cache eviction reclaim this completed model version.
|
|
build.close()
|
|
totals = {target: data[source] for source, target in (
|
|
("nfev", "nfev"), ("njev", "njev"), ("nlu", "nlu"),
|
|
("acceptedSteps", "acceptedStepCount"), ("rejectedSteps", "rejectedStepCount"),
|
|
("stateTransitions", "stateTransitionCount"), ("solverStarts", "solverStartCount"),
|
|
) if source in data}
|
|
if progress_callback:
|
|
fraction = (data["simulatedUntil"]-config.t_start)/(config.t_stop-config.t_start)
|
|
progress_callback(fraction, "complete" if data["success"] else data["status"])
|
|
return GenericSimulationResult(
|
|
success=data["success"], status=data["status"], message=data["message"],
|
|
simulated_until=data["simulatedUntil"], requested_stop_time=config.t_stop,
|
|
variables=program.variables, series=data["series"], final=data["final"],
|
|
diagnostics={"backend": "native-c", "propertyWarnings": data.get("propertyWarnings", []), "native": {k: v for k, v in data.items()
|
|
if k not in ("series", "final", "finalState")}, "integration": {"method": config.method, "rtol": config.rtol, "totals": totals},
|
|
"stateCount": len(program.state_keys), "sampleCount": (data["series"].sample_count if isinstance(data["series"], NativeSeriesJson)
|
|
else len(data["series"].get("time", [])))},
|
|
)
|