298 lines
11 KiB
Python
298 lines
11 KiB
Python
"""Process supervisor for lifecycle management of C++ pipeline binaries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from typing import Iterable
|
|
from typing import Sequence
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ManagedProcess:
|
|
"""Metadata and subprocess handle for one managed child process."""
|
|
|
|
name: str
|
|
command: list[str]
|
|
allow_clean_exit: bool
|
|
handle: subprocess.Popen[str]
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProcessExitReport:
|
|
"""Structured report for one exited managed process."""
|
|
|
|
name: str
|
|
command: list[str]
|
|
working_directory: Path
|
|
return_code: int
|
|
stdout: str
|
|
stderr: str
|
|
expected_clean_exit: bool
|
|
|
|
@property
|
|
def level(self) -> str:
|
|
"""Return log level appropriate for this exit report."""
|
|
return "INFO" if self.expected_clean_exit else "ERROR"
|
|
|
|
def format(self) -> str:
|
|
"""Render human-readable multiline exit report."""
|
|
if self.expected_clean_exit:
|
|
headline = f"Process `{self.name}` completed normally with code {self.return_code}."
|
|
elif self.return_code == 0:
|
|
headline = f"Process `{self.name}` exited unexpectedly with code 0."
|
|
else:
|
|
headline = f"Process `{self.name}` exited with code {self.return_code}."
|
|
|
|
lines = [
|
|
headline,
|
|
f"Command: {shlex.join(self.command)}",
|
|
f"Working directory: {self.working_directory}",
|
|
]
|
|
if self.stderr:
|
|
lines.append(f"stderr:\n{self.stderr}")
|
|
if self.stdout:
|
|
lines.append(f"stdout:\n{self.stdout}")
|
|
if not self.stderr and not self.stdout:
|
|
lines.append("stdout/stderr: none")
|
|
return "\n".join(lines)
|
|
|
|
|
|
class ProcessSupervisor:
|
|
"""Start, monitor, and stop pipeline subprocesses."""
|
|
|
|
def __init__(self, project_root: Path, readiness_timeout_s: float = 3.0) -> None:
|
|
"""Create supervisor bound to repository root directory."""
|
|
self._project_root = project_root
|
|
self._readiness_timeout_s = readiness_timeout_s
|
|
self._processes: dict[str, ManagedProcess] = {}
|
|
|
|
def is_running(self) -> bool:
|
|
"""Return whether acquisition-side processes are alive."""
|
|
return self._is_alive("data_preprocessor") or self._is_alive("sweep_orchestrator")
|
|
|
|
def is_processor_running(self) -> bool:
|
|
"""Return whether data processor process is alive."""
|
|
return self._is_alive("data_processor")
|
|
|
|
def start(self, config_path: Path, *, allow_clean_orchestrator_exit: bool = False) -> None:
|
|
"""Start required pipeline binaries and wait until they are ready."""
|
|
if self.is_running():
|
|
raise RuntimeError("Acquisition processes are already running")
|
|
|
|
acquisition_command = self._acquisition_command(config_path)
|
|
command_specs = {
|
|
"data_processor": [
|
|
str(self._project_root / "build/bin/data_processor"),
|
|
"--config",
|
|
str(config_path),
|
|
],
|
|
"data_preprocessor": [
|
|
str(self._project_root / "build/bin/data_preprocessor"),
|
|
"--config",
|
|
str(config_path),
|
|
],
|
|
"sweep_orchestrator": acquisition_command,
|
|
}
|
|
processor_was_running = self.is_processor_running()
|
|
required_processes: list[str] = ["data_preprocessor"]
|
|
if not allow_clean_orchestrator_exit:
|
|
required_processes.append("sweep_orchestrator")
|
|
if not processor_was_running:
|
|
required_processes.insert(0, "data_processor")
|
|
|
|
try:
|
|
if not processor_was_running:
|
|
self._spawn("data_processor", command_specs["data_processor"], allow_clean_exit=False)
|
|
|
|
self._spawn("data_preprocessor", command_specs["data_preprocessor"], allow_clean_exit=False)
|
|
self._spawn(
|
|
"sweep_orchestrator",
|
|
command_specs["sweep_orchestrator"],
|
|
allow_clean_exit=allow_clean_orchestrator_exit,
|
|
)
|
|
self._wait_until_ready(required_processes)
|
|
except Exception:
|
|
if processor_was_running:
|
|
self.stop()
|
|
else:
|
|
self.stop_all()
|
|
raise
|
|
|
|
def stop(self) -> None:
|
|
"""Stop acquisition-side processes, keep processor process intact."""
|
|
self._stop_processes(["sweep_orchestrator", "data_preprocessor"])
|
|
|
|
def stop_orchestrator(self) -> None:
|
|
"""Stop orchestrator process only."""
|
|
self._stop_processes(["sweep_orchestrator"])
|
|
|
|
def stop_preprocessor(self) -> None:
|
|
"""Stop preprocessor process only."""
|
|
self._stop_processes(["data_preprocessor"])
|
|
|
|
def stop_all(self) -> None:
|
|
"""Stop all managed processes."""
|
|
self._stop_processes(["sweep_orchestrator", "data_preprocessor", "data_processor"])
|
|
|
|
def _spawn(self, name: str, command: list[str], *, allow_clean_exit: bool) -> None:
|
|
"""Spawn one process unless same process is already alive."""
|
|
existing = self._processes.get(name)
|
|
if existing is not None and existing.handle.poll() is None:
|
|
return
|
|
|
|
try:
|
|
handle = subprocess.Popen(
|
|
command,
|
|
cwd=self._project_root,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
except OSError as exc:
|
|
command_text = shlex.join(command)
|
|
raise RuntimeError(
|
|
f"Failed to spawn {name} with command `{command_text}` from `{self._project_root}`: "
|
|
f"{type(exc).__name__}: {exc}"
|
|
) from exc
|
|
self._processes[name] = ManagedProcess(
|
|
name=name,
|
|
command=command,
|
|
allow_clean_exit=allow_clean_exit,
|
|
handle=handle,
|
|
)
|
|
|
|
def _acquisition_command(self, config_path: Path) -> list[str]:
|
|
"""Return acquisition producer command selected by radar.model."""
|
|
radar_model = self._read_radar_model(config_path)
|
|
if radar_model == "librevna_multi":
|
|
return [
|
|
sys.executable,
|
|
"-m",
|
|
"python_app.scripts.multi_device_raw_producer",
|
|
"--config",
|
|
str(config_path),
|
|
]
|
|
return [
|
|
str(self._project_root / "build/bin/sweep_orchestrator"),
|
|
"--config",
|
|
str(config_path),
|
|
]
|
|
|
|
@staticmethod
|
|
def _read_radar_model(config_path: Path) -> str:
|
|
"""Read `radar.model` cheaply without constructing the full config model."""
|
|
payload = json.loads(config_path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
return "librevna"
|
|
radar_payload = payload.get("radar")
|
|
if not isinstance(radar_payload, dict):
|
|
return "librevna"
|
|
return str(radar_payload.get("model", "librevna"))
|
|
|
|
def _stop_processes(self, names: Iterable[str]) -> None:
|
|
"""Gracefully terminate processes, then force-kill on timeout."""
|
|
ordered_names = list(names)
|
|
for name in ordered_names:
|
|
process = self._processes.get(name)
|
|
if process is None:
|
|
continue
|
|
if process.handle.poll() is None:
|
|
process.handle.terminate()
|
|
|
|
deadline = time.monotonic() + 2.0
|
|
for name in ordered_names:
|
|
process = self._processes.get(name)
|
|
if process is None:
|
|
continue
|
|
if process.handle.poll() is not None:
|
|
continue
|
|
|
|
timeout = max(0.0, deadline - time.monotonic())
|
|
try:
|
|
process.handle.wait(timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
process.handle.kill()
|
|
process.handle.wait(timeout=1.0)
|
|
|
|
self._drop_exited()
|
|
|
|
def _drop_exited(self) -> None:
|
|
"""Remove exited process entries from internal map."""
|
|
exited_names = [name for name, process in self._processes.items() if process.handle.poll() is not None]
|
|
for name in exited_names:
|
|
self._processes.pop(name, None)
|
|
|
|
def _is_alive(self, name: str) -> bool:
|
|
"""Return `True` when named process handle exists and is running."""
|
|
process = self._processes.get(name)
|
|
if process is None:
|
|
return False
|
|
return process.handle.poll() is None
|
|
|
|
def collect_exit_reports(self) -> list[ProcessExitReport]:
|
|
"""Collect reports for managed processes that have exited."""
|
|
reports: list[ProcessExitReport] = []
|
|
exited_names: list[str] = []
|
|
|
|
for name, process in self._processes.items():
|
|
return_code = process.handle.poll()
|
|
if return_code is None:
|
|
continue
|
|
|
|
stderr = ""
|
|
stdout = ""
|
|
if process.handle.stdout is not None:
|
|
stdout = process.handle.stdout.read().strip()
|
|
if process.handle.stderr is not None:
|
|
stderr = process.handle.stderr.read().strip()
|
|
|
|
reports.append(
|
|
ProcessExitReport(
|
|
name=process.name,
|
|
command=list(process.command),
|
|
working_directory=self._project_root,
|
|
return_code=int(return_code),
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
expected_clean_exit=bool(process.allow_clean_exit and int(return_code) == 0),
|
|
)
|
|
)
|
|
exited_names.append(name)
|
|
|
|
for name in exited_names:
|
|
self._processes.pop(name, None)
|
|
return reports
|
|
|
|
def _wait_until_ready(self, required_processes: Sequence[str]) -> None:
|
|
"""Wait until all required processes are alive or timeout/crash occurs."""
|
|
deadline = time.monotonic() + self._readiness_timeout_s
|
|
|
|
while time.monotonic() < deadline:
|
|
exit_reports = self.collect_exit_reports()
|
|
unexpected_reports = [report for report in exit_reports if not report.expected_clean_exit]
|
|
if unexpected_reports:
|
|
raise RuntimeError("; ".join(report.format() for report in unexpected_reports))
|
|
if all(self._is_alive(process_name) for process_name in required_processes):
|
|
return
|
|
time.sleep(0.05)
|
|
|
|
names = ", ".join(required_processes)
|
|
raise RuntimeError(
|
|
f"Timed out waiting for processes to start: {names}. "
|
|
f"Alive processes: {self.pids() or 'none'}"
|
|
)
|
|
|
|
def pids(self) -> dict[str, int]:
|
|
"""Return PID mapping for currently alive managed processes."""
|
|
return {
|
|
process.name: process.handle.pid
|
|
for process in self._processes.values()
|
|
if process.handle.poll() is None and process.handle.pid is not None
|
|
}
|