diff --git a/docs/validation/refined_dfg_timing.md b/docs/validation/refined_dfg_timing.md index b7814b8..214a5fe 100644 --- a/docs/validation/refined_dfg_timing.md +++ b/docs/validation/refined_dfg_timing.md @@ -7,18 +7,21 @@ physics-gate or production change; no force or accuracy result is produced. | Variant | Loop | Force history | |---|---|---| | A | refined `step(1)` + `read_raw_exchange` per base step (as the runner) | valid | -| B | refined `step(1)` only | timing-only | -| C | refined single `step(N)` | timing-only | +| B | refined `step(1)` only, no host reads (`observe_raw_exchange` stays on) | timing-only | +| C | refined single `step(N)` | timing-only; ledger keeps the final base interval only | | E | uniform `sim.run(<=100)` + `read_force` (as the runner) | valid | -| E' | uniform single `sim.run(2N)` | timing-only | +| E' | uniform single `sim.run(2N)` | timing-only; at most a whole-window average, different summation order | - Equal simulated duration: N refined base steps vs 2N uniform D80 steps. - Every repeat starts from the same fresh state (refined `initialize`, uniform `snapshot/restore`) and the same warm-up; construction/JIT, warm-up, downloads and parity checks are untimed. -- `step()` synchronizes after each base interval and `Simulation.run()` before returning, so the - wall timer (`time.perf_counter`) brackets completed GPU work. No CUDA events are used. -- Final populations and NaN masks must be bitwise identical to the first A (refined) or E - (uniform) repeat; any difference aborts. +- The actual stepping stream is synchronized explicitly before and after each timed loop; the + wall timer (`time.perf_counter`) is the only source. No CUDA-event span is called kernel time. +- `--max-wall-seconds` is required for GPU runs (outer timeout, checked between variants and + every 100 steps in per-step loops). +- Final populations and NaN masks must be bitwise identical to the first A repeat (A/B/C) or the + first E repeat (E/E'); refined and uniform grids are never compared. Leaf/fluid populations + must be finite. All parity/health checks are untimed. - Cases run sequentially (uniform, then refined) in one process with startup serialized; variants are interleaved within each repeat. A-B and B-C are measured comparative differences under GPU variability, not exact causal decompositions. @@ -27,7 +30,7 @@ Usage (root selects one verified idle GPU): PYTHONPATH=src python -m unittest tests/unit/test_time_refined_dfg.py python tests/validation/time_refined_dfg.py --output --plan - CUDA_VISIBLE_DEVICES= python tests/validation/time_refined_dfg.py --output + CUDA_VISIBLE_DEVICES= python tests/validation/time_refined_dfg.py --output --max-wall-seconds Defaults: 2000 base steps, 200 warm-up base steps, 3 repeats. Profiler capture and the `observe_raw_exchange=False` variant are deferred until these measurements exist. diff --git a/tests/unit/test_time_refined_dfg.py b/tests/unit/test_time_refined_dfg.py index d165500..25183b8 100644 --- a/tests/unit/test_time_refined_dfg.py +++ b/tests/unit/test_time_refined_dfg.py @@ -26,6 +26,14 @@ class TimingHarnessContract(unittest.TestCase): for name in ("A", "E"): self.assertTrue(t.VARIANTS[name]["force_history_valid"]) self.assertEqual({v["case"] for v in t.VARIANTS.values()}, {"refined", "uniform"}) + self.assertIn("final base interval", t.VARIANTS["C"]["force_note"]) + self.assertIn("whole-window", t.VARIANTS["E'"]["force_note"]) + + def test_deadline(self): + import time + with self.assertRaises(TimeoutError): + t.check_deadline(time.monotonic() - 1) + t.check_deadline(time.monotonic() + 60) def test_parity_detects_bits_and_nan_masks(self): a = np.array([[1., np.nan], [2., 3.]], np.float32) diff --git a/tests/validation/time_refined_dfg.py b/tests/validation/time_refined_dfg.py index f79376f..6d89c33 100644 --- a/tests/validation/time_refined_dfg.py +++ b/tests/validation/time_refined_dfg.py @@ -2,15 +2,19 @@ Variants over EQUAL simulated duration (one refined base step = two uniform D80 steps): A refined: step(1) + read_raw_exchange per base step (as run_refined_dfg) - B refined: step(1) only -- timing-only, no force history - C refined: one step(N) call -- timing-only, no force history + B refined: step(1) only, no host reads -- timing-only, no force history + C refined: one step(N) call -- timing-only; ledger keeps final base interval only E uniform: sim.run(<=100) + read_force (as run_refined_dfg) - E' uniform: one sim.run(2N) call -- timing-only, no force history + E' uniform: one sim.run(2N) call -- timing-only; at most a whole-window force + average (different summation order), no history +observe_raw_exchange stays True for A/B/C: B/C omit host reads only. +Parity compares A/B/C with each other and E/E' with each other, never uniform with refined. A-B and B-C are measured comparative differences, not exact causal decompositions. Each repeat starts from the same fresh initial state with the same warm-up; construction, -compilation and parity checks are outside timed regions. FixedPatchStepper.step() synchronizes -its stream after every base interval and Simulation.run() synchronizes before returning, so the -timer starts and stops with no GPU work in flight. Wall time (perf_counter) is the only source. +compilation, parity and health checks are outside timed regions. The actual stepping stream is +synchronized explicitly immediately before the timer starts and after the timed loop. Wall time +(perf_counter) is the only timing source; no CUDA-event span is reported as kernel time. An outer +--max-wall-seconds deadline is checked between variants and inside per-step loops. """ import argparse @@ -31,10 +35,13 @@ import run_refined_dfg as r # noqa: E402 (published runner: same setup functio VARIANTS = { "A": dict(case="refined", force_history_valid=True), - "B": dict(case="refined", force_history_valid=False, timing_only=True), - "C": dict(case="refined", force_history_valid=False, timing_only=True), + "B": dict(case="refined", force_history_valid=False, timing_only=True, + force_note="no host reads"), + "C": dict(case="refined", force_history_valid=False, timing_only=True, + force_note="raw ledger retains only the final base interval"), "E": dict(case="uniform", force_history_valid=True), - "E'": dict(case="uniform", force_history_valid=False, timing_only=True), + "E'": dict(case="uniform", force_history_valid=False, timing_only=True, + force_note="at most a whole-window accumulated average; different summation order"), } @@ -60,17 +67,34 @@ def same_state(reference, values): return None -def refined_loop(backend, variant, n): +def check_deadline(deadline): + if time.monotonic() >= deadline: + raise TimeoutError("timing harness exceeded --max-wall-seconds") + + +def sync_refined(backend): + with backend.ctx.activate(): + backend._stream.synchronize() + + +def sync_uniform(sim): + with sim.ctx.activate(): + sim.stream.synchronize() + + +def refined_loop(backend, variant, n, deadline): if variant == "C": backend.step(n) return - for _ in range(n): + for i in range(n): backend.step(1) if variant == "A": backend.read_raw_exchange(0, level=1, window="last_base_step") + if i % 100 == 99: + check_deadline(deadline) -def uniform_loop(sim, variant, n): +def uniform_loop(sim, variant, n, deadline): if variant == "E'": sim.run(n) return @@ -80,6 +104,7 @@ def uniform_loop(sim, variant, n): sim.run(batch) sim.read_force(0) done += batch + check_deadline(deadline) def time_refined(a, h, sched): @@ -100,12 +125,17 @@ def time_refined(a, h, sched): print(json.dumps(dict(status="READY", case="refined")), flush=True) for repeat in range(a.repeats): for variant in ("A", "B", "C"): + check_deadline(a.deadline) backend.initialize(*fresh) # resets step count; same fresh state every time - refined_loop(backend, variant, warm) + refined_loop(backend, variant, warm, a.deadline) + sync_refined(backend) start = time.perf_counter() - refined_loop(backend, variant, n) + refined_loop(backend, variant, n, a.deadline) + sync_refined(backend) seconds = time.perf_counter() - start values = [np.array(v, copy=True) for v in backend._download_raw().values] + if not all(np.isfinite(v[:, m]).all() for v, m in zip(values, backend._state)): + raise RuntimeError(f"refined {variant} repeat {repeat}: nonfinite leaf populations") mismatch = None if repeat == 0 and variant == "A": reference["A"] = values @@ -121,6 +151,7 @@ def time_refined(a, h, sched): def time_uniform(a, h, sched): from CelerisLab.simulation import Simulation + from CelerisLab.lbm.descriptors import FLUID warm, n = sched["uniform"] rows, reference = [], {} @@ -135,14 +166,20 @@ def time_uniform(a, h, sched): print(json.dumps(dict(status="READY", case="uniform")), flush=True) for repeat in range(a.repeats): for variant in ("E", "E'"): + check_deadline(a.deadline) sim.restore() if sim.stepper.step_count % 2: raise RuntimeError("uniform restore parity changed between variants") - uniform_loop(sim, variant, warm) + uniform_loop(sim, variant, warm, a.deadline) + sync_uniform(sim) start = time.perf_counter() - uniform_loop(sim, variant, n) + uniform_loop(sim, variant, n, a.deadline) + sync_uniform(sim) seconds = time.perf_counter() - start values = [sim.get_ddf()] + fluid = (sim.get_flags().reshape(values[0].shape[1:]) & FLUID) != 0 + if not np.isfinite(values[0][:, fluid]).all(): + raise RuntimeError(f"uniform {variant} repeat {repeat}: nonfinite fluid populations") mismatch = None if repeat == 0 and variant == "E": reference["E"] = values @@ -184,6 +221,7 @@ def main(): p.add_argument("--base-steps", type=int, default=2000) p.add_argument("--warmup-base-steps", type=int, default=200) p.add_argument("--repeats", type=int, default=3) + p.add_argument("--max-wall-seconds", type=float, help="Required outer timeout for GPU runs") p.add_argument("--plan", action="store_true") a = p.parse_args() if a.repeats < 1 or a.base_steps > 20000: @@ -197,6 +235,9 @@ def main(): if a.plan: print(json.dumps(plan, indent=2)) return 0 + if a.max_wall_seconds is None or not math.isfinite(a.max_wall_seconds) or a.max_wall_seconds <= 0: + p.error("GPU timing requires a finite positive --max-wall-seconds") + a.deadline = time.monotonic() + a.max_wall_seconds if not os.environ.get("CUDA_VISIBLE_DEVICES") or "," in os.environ["CUDA_VISIBLE_DEVICES"]: p.error("Root must select exactly one verified idle GPU via CUDA_VISIBLE_DEVICES") if a.output.exists() and any(a.output.iterdir()):