differential: a model's checks share its cap, and a kill keeps what finished

.github/workflows/nightly-parity.yml

@@ -619,10 +619,14 @@ jobs:
       # cap that finish in 22-90 s cold on an M3 Max (BIOMD0000000505: 48 s there),
       # and the slowest that finished took 248 s; a timeout that flips with runner
       # speed would alert as a regression.
+      # --timeout 1500, not 900: MODEL1006230049, 0053 and 0077 took 689-894 s
+      # across five runs on the same commit's checks, so a cap of 900 (its deadline
+      # 810, #1321) flipped their rows with runner speed. The cap costs only the
+      # few models that use it.
       - name: Run differential_run.py
         env:
           BNGSIM_CODEGEN_CACHE_DIR: ${{ runner.temp }}/codegen
-        run: uv run --no-sync python parity_checks/differential/differential_run.py --workers 4 --timeout 900 --out "$RUNNER_TEMP/differential.json" || true
+        run: uv run --no-sync python parity_checks/differential/differential_run.py --workers 4 --timeout 1500 --out "$RUNNER_TEMP/differential.json" || true
       - name: Compare
         env:
           GH_TOKEN: ${{ github.token }}

parity_checks/differential/_differential.py

@@ -38,9 +38,10 @@
 Each returns a :class:`Verdict`: ``pass``, ``diff`` (a finding, with its size),
 ``unsupported`` (the path declined by type: not a wrong answer), ``skip`` (the
 path does not apply here, said why), ``exception`` (a path raised where the base
-ran: a finding) or ``no_base`` (the model does not load or its plain run does
+ran: a finding), ``no_base`` (the model does not load or its plain run does
 not complete, so there is nothing to compare against: the parity suites own
-that).
+that) or ``timeout`` (not finished within the model's cap, or a compile past
+bngsim's own limit: that row only).
 
 The oracle has to be careful
 ----------------------------
@@ -69,6 +70,7 @@
 import os
 import subprocess
 import sys
+import time
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -93,6 +95,10 @@
 #: Comma-decimal locales, in the order tried.
 COMMA_LOCALES = ("de_DE.UTF-8", "de_DE.utf8", "de_DE", "fr_FR.UTF-8", "fr_FR.utf8", "fr_FR")
 
+#: The ``time.monotonic()`` every run of the model being checked must finish
+#: by, which :func:`check_all` sets: None outside it, or with no cap.
+_DEADLINE: float | None = None
+
 CHECKS = (
     "sens_runs",
     "sens_vs_plain",
@@ -109,7 +115,7 @@
 
 @dataclass
 class Verdict:
-    status: str  # pass | diff | unsupported | skip | exception | no_base
+    status: str  # pass | diff | unsupported | skip | exception | no_base | timeout
     value: float | None = None
     note: str = ""
     extra: dict = field(default_factory=dict)
@@ -165,9 +171,22 @@ def sens_params(model) -> list[str]:
     return [p for p in model.primary_param_names if p not in excluded][:MAX_SENS_PARAMS]
 
 
+def _left() -> float | None:
+    """The seconds to :data:`_DEADLINE`, a run's ``timeout``: None for none. Past
+    it no run starts (``SimulationTimeout``)."""
+    if _DEADLINE is None:
+        return None
+    left = _DEADLINE - time.monotonic()
+    if left <= 0:
+        import bngsim
+
+        raise bngsim.SimulationTimeout("the model's cap is spent")
+    return left
+
+
 def run(job: Job, **kw):
     """A run of a fresh model (a run leaves its state on the Model it was given):
-    the Result and the Simulator."""
+    the Result and the Simulator. It has until :data:`_DEADLINE`."""
     import bngsim
 
     sim = bngsim.Simulator(load(job.path), method="ode", **{**job.sim, **kw})
@@ -177,6 +196,7 @@ def run(job: Job, **kw):
         rtol=job.rtol,
         atol=job.atol,
         max_steps=job.max_steps,
+        timeout=_left(),
     )
     return result, sim
 
@@ -279,7 +299,33 @@ def _declined(e: Exception) -> bool:
     )
 
 
+#: The words bngsim's compile timeout is reported in (``_codegen``), where
+#: ``codegen=True``'s refusal carries it in its text and not as its cause.
+_COMPILE_TIMED_OUT = "Codegen compilation timed out"
+
+
+def _timed_out(e: BaseException) -> bool:
+    """A run stopped by the model's cap, or a compile by bngsim's own limit
+    (BIOMD0000000385's 4.2 MB of C past 600 s): a timeout, whatever wraps it,
+    and never a crash."""
+    import bngsim
+
+    seen, depth = e, 0
+    while seen is not None and depth < 16:
+        if isinstance(seen, bngsim.SimulationTimeout | subprocess.TimeoutExpired):
+            return True
+        seen, depth = seen.__cause__ or seen.__context__, depth + 1
+    return _COMPILE_TIMED_OUT in str(e)
+
+
 def _verdict_of(e: Exception) -> Verdict:
+    if _timed_out(e):
+        # A cap reached in this check: its own row, not every row.
+        if _COMPILE_TIMED_OUT in str(e):
+            why = f"the compile timed out: {type(e).__name__}: {_first_line(e)}"
+        else:
+            why = f"not finished within the model's cap: {_first_line(e)}"
+        return Verdict("timeout", None, why)
     status = "unsupported" if _declined(e) or isinstance(e, _Declined) else "exception"
     return Verdict(status, None, f"{type(e).__name__}: {_first_line(e)}")
 
@@ -302,6 +348,9 @@ def check_sens(base: Base) -> dict[str, Verdict]:
     try:
         result, sim = run(base.job, sensitivity_params=params)
     except Exception as e:  # noqa: BLE001 - that it did not complete is the verdict
+        if _timed_out(e):
+            v = _verdict_of(e)
+            return {"sens_runs": v, "sens_vs_plain": v}
         # Where the plain run completes and the sensitivity run does not: a
         # refusal by name, or a derivative that does not exist at the start
         # (a power of a species at 0), or a crash. The rule records it; the
@@ -576,7 +625,11 @@ def _states(job: Job) -> list[tuple[float, np.ndarray]]:
         if t > job.t_start:
             try:
                 bngsim.Simulator(model, method="ode", **job.sim).run(
-                    t_span=(job.t_start, t), n_points=2, rtol=job.rtol, atol=job.atol
+                    t_span=(job.t_start, t),
+                    n_points=2,
+                    rtol=job.rtol,
+                    atol=job.atol,
+                    timeout=_left(),
                 )
             except bngsim.SimulationError:
                 # A state this one-interval run does not reach, where the base's
@@ -647,12 +700,21 @@ def check_locale(base: Base, name: str | None, timeout: float = 600.0) -> Verdic
 
     def go():
         here = str(Path(__file__).resolve().parent)
-        out = subprocess.run(
-            [sys.executable, "-c", _LOCALE_PROBE, name, json.dumps(base.job.as_dict()), here],
-            capture_output=True,
-            text=True,
-            timeout=timeout,
-        )
+        # The model's cap, where less of it is left than this run's own limit.
+        limit = timeout if _DEADLINE is None else min(timeout, _left())
+        try:
+            out = subprocess.run(
+                [sys.executable, "-c", _LOCALE_PROBE, name, json.dumps(base.job.as_dict()), here],
+                capture_output=True,
+                text=True,
+                timeout=limit,
+            )
+        except subprocess.TimeoutExpired as e:
+            if limit < timeout:  # the model's cap: TIMEOUT, this row only
+                raise
+            # Its own limit, where the run under C finished: a crash, as it read
+            # before the cap. A run the locale stalls is this check's finding.
+            return Verdict("exception", None, f"{type(e).__name__}: {_first_line(e)}")
         if out.returncode != 0:
             last = (out.stderr.strip().splitlines() or ["?"])[-1][:300]
             return Verdict("diff", None, f"runs under C, fails under {name}: {last}")
@@ -661,19 +723,67 @@ def go():
     return _guard(go)
 
 
-def check_all(job: Job, locale_name: str | None, checks=CHECKS) -> dict[str, Verdict]:
+def check_all(
+    job: Job,
+    locale_name: str | None,
+    checks=CHECKS,
+    deadline: float | None = None,
+    on_verdict=None,
+) -> dict[str, Verdict]:
     """Every check of ``checks`` on one model. A base that does not run is no
-    reference: each check is then an exception on the base."""
+    reference: each check is then an exception on the base. Every run has
+    until ``deadline`` (a ``time.monotonic()``): a check that has not finished
+    by then is TIMEOUT, its own row only, where the runner's cap would take
+    every row of the model (MODEL1006230049, 0053 and 0077 take 716-751 s of
+    the nightly's 900 s, so any check added to every model tipped them over).
+    ``on_verdict(check, verdict)`` hears each as it is reached, so a caller
+    killed at the cap still has those."""
+    global _DEADLINE
+    _DEADLINE = deadline
+    try:
+        return _check_all(job, locale_name, checks, on_verdict or (lambda c, v: None))
+    finally:
+        _DEADLINE = None
+
+
+class _Heard(dict):
+    """The verdicts so far, each told to ``on_verdict`` as it is set."""
+
+    def __init__(self, on_verdict):
+        super().__init__()
+        self.on_verdict = on_verdict
+
+    def __setitem__(self, check, verdict):
+        super().__setitem__(check, verdict)
+        self.on_verdict(check, verdict)
+
+    def update(self, other):
+        for check, verdict in other.items():
+            self[check] = verdict
+
+
+def _check_all(job: Job, locale_name: str | None, checks, on_verdict) -> dict[str, Verdict]:
     try:
         base = Base(job)
     except Exception as e:  # noqa: BLE001
         # No base, no reference: not a finding of this suite's (the parity suites
         # own a model that does not load or run), and not a pass either.
         v = Verdict("no_base", None, f"the base run: {type(e).__name__}: {str(e)[:300]}")
-        return {c: v for c in checks}
-    out: dict[str, Verdict] = {}
+        return {c: _verdict_of(e) if _timed_out(e) else v for c in checks}
+    out: dict[str, Verdict] = _Heard(on_verdict)
+
+    def spent() -> Verdict | None:
+        # No check starts past the deadline: one that builds and compiles
+        # Simulators without running them (backend_as_asked) has no run to bound.
+        try:
+            _left()
+        except Exception as e:  # noqa: BLE001 - SimulationTimeout, the cap spent
+            return _verdict_of(e)
+        return None
+
     if {"sens_runs", "sens_vs_plain"} & set(checks):
-        out.update(check_sens(base))
+        v = spent()
+        out.update({"sens_runs": v, "sens_vs_plain": v} if v else check_sens(base))
     one = {
         "ajac_vs_fdjac": check_ajac,
         "klu_vs_dense": check_klu,
@@ -685,7 +795,7 @@ def check_all(job: Job, locale_name: str | None, checks=CHECKS) -> dict[str, Ver
     }
     for name, fn in one.items():
         if name in checks:
-            out[name] = fn(base)
+            out[name] = spent() or fn(base)
     if "locale" in checks:
-        out["locale"] = check_locale(base, locale_name)
+        out["locale"] = spent() or check_locale(base, locale_name)
     return {c: out[c] for c in checks if c in out}

parity_checks/differential/differential_run.py

@@ -15,7 +15,8 @@
     the path does not apply here (no KLU, no MIR),
       or the model does not load or run at all ....... SKIP
     it raised where the base ran .................... EXCEPTION
-    the model's cap was exceeded .................... TIMEOUT
+    it did not finish within the model's cap, or its
+      compile within bngsim's own limit ............. TIMEOUT
 
 The baseline (``nightly/baselines/differential.json``) holds tonight's known
 findings, so a night alerts on what changed: a model whose paths agreed and do
@@ -39,9 +40,12 @@
 from __future__ import annotations
 
 import argparse
+import atexit
 import datetime as _dt
 import json
+import shutil
 import sys
+import tempfile
 import time
 from pathlib import Path
 
@@ -60,6 +64,9 @@
 RR_JOBS = HERE.parent / "rr_parity" / "ode_jobs.json"
 REFERENCE_ENGINE = "bngsim_paths"
 DEFAULT_TIMEOUT = 600.0
+#: The share of a model's cap its checks may use before the pool kills it: the
+#: rest is the margin for the run that ends past the deadline, and the reply.
+CAP_SHARE = 0.9
 DEFAULT_MAX_SPECIES = 200
 
 _OUTCOME = {
@@ -129,6 +136,25 @@ def list_jobs(listing: Path, span: tuple[float, float]) -> list[D.Job]:
     return [_job_of(Path(x).resolve(), span) for x in listing.read_text().split() if x]
 
 
+def _row(v) -> dict:
+    return {"status": v.status, "value": v.value, "note": v.note, "extra": v.extra}
+
+
+def _keep(spec: dict, check: str, v) -> None:
+    """Each verdict as it is reached, to ``spec["partial"]``: a worker the pool
+    kills at the cap (one compiling C past the deadline) leaves the runner every
+    row it finished, and only the rest TIMEOUT."""
+    path = spec.get("partial")
+    if not path:
+        return
+    path = Path(path)
+    rows = json.loads(path.read_text()) if path.is_file() else {}
+    rows[check] = _row(v)
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(rows, default=str))
+    tmp.replace(path)
+
+
 def _worker(spec: dict, q) -> None:
     """One model's checks in a child process; puts one result dict on ``q``."""
     t0 = time.perf_counter()
@@ -150,11 +176,17 @@ def _worker(spec: dict, q) -> None:
             res["status"] = "over"
             res["why"] = f"{n} species, over the cap of {spec['max_species']}"
         else:
-            verdicts = D.check_all(job, spec["locale"], tuple(spec["checks"]))
-            res["rows"] = {
-                c: {"status": v.status, "value": v.value, "note": v.note, "extra": v.extra}
-                for c, v in verdicts.items()
-            }
+            # Short of the cap, at which the pool kills the worker and every row
+            # of the model with it: past the deadline a check is its own TIMEOUT.
+            deadline = time.monotonic() + spec["timeout"] * CAP_SHARE - (time.perf_counter() - t0)
+            verdicts = D.check_all(
+                job,
+                spec["locale"],
+                tuple(spec["checks"]),
+                deadline,
+                lambda c, v: _keep(spec, c, v),
+            )
+            res["rows"] = {c: _row(v) for c, v in verdicts.items()}
             res["status"] = "done"
     except Exception as e:  # noqa: BLE001 - a model that does not load is its rows' EXCEPTION
         res["status"] = "exception"
@@ -172,9 +204,17 @@ def to_results(spec: dict, r: dict, stamp: dict) -> list[JobResult]:
     rows = r.get("rows")
     if rows is None:  # the model as a whole: over the cap, raised, timed out or died
         status = {"over": "skip"}.get(r.get("status"), r.get("status", "exception"))
-        note = r.get("why") or (f"cap of {spec['timeout']:.0f} s" if status == "timeout" else "")
+        note = r.get("why") or {
+            "timeout": f"cap of {spec['timeout']:.0f} s",
+            "dead": f"the worker died (exit code {r.get('exitcode')})",
+        }.get(status, "")
+        # What a worker killed at the cap, or one that died or raised, finished
+        # stands (_keep): only the rest take the model's outcome.
+        kept = spec.get("partial") and Path(spec["partial"])
+        done = json.loads(kept.read_text()) if kept and kept.is_file() else {}
         rows = {
-            c: {"status": status, "value": None, "note": note, "extra": {}} for c in spec["checks"]
+            c: done.get(c) or {"status": status, "value": None, "note": note, "extra": {}}
+            for c in spec["checks"]
         }
     out = []
     for check, row in rows.items():
@@ -248,6 +288,9 @@ def main(argv=None) -> int:
     if args.models:
         subs = [s.strip() for s in args.models.split(",") if s.strip()]
         unique = [j for j in unique if any(s in j.model_id for s in subs)]
+    # Each model's verdicts as its worker reaches them (_keep), for one the pool kills.
+    kept = Path(tempfile.mkdtemp(prefix="differential-rows-"))
+    atexit.register(shutil.rmtree, kept, ignore_errors=True)
     specs = [
         {
             "key": j.model_id,
@@ -256,8 +299,9 @@ def main(argv=None) -> int:
             "locale": locale,
             "max_species": args.max_species,
             "timeout": args.timeout,
+            "partial": str(kept / f"{i}.json"),
         }
-        for j in unique
+        for i, j in enumerate(unique)
     ]
     print(f"differential: {len(specs)} models x {len(checks)} checks; locale {locale}")
     t0 = time.perf_counter()