"""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", [])))}, )