diff --git a/docs/validation/refined_dfg_timing.md b/docs/validation/refined_dfg_timing.md index b424750..16557c3 100644 --- a/docs/validation/refined_dfg_timing.md +++ b/docs/validation/refined_dfg_timing.md @@ -42,3 +42,17 @@ Usage (root selects one verified idle GPU): 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. + +## A versus observation-window ledger (`tests/validation/time_window_dfg.py`) + +Measurement-only comparison on the same shifted-MRT refined setup: A (per-base `step(1)` + +`read_raw_exchange`, summed per 80-base-step interval), W (`start_exchange_window`, `step(80)`, +one window read) and W1 (window with 80 x `step(1)`, isolating the read pattern from call +batching). Same fresh state, warm-up and simulated duration; populations must be bitwise equal +to the first A repeat and per-interval total impulses within the documented engineering +tolerance, all outside timing. The candidate source is recorded (checkout revision, dirty flag, +file hashes); frozen refined-DFG manifests/results are not used or changed. + + PYTHONPATH=src python -m unittest tests/unit/test_time_window_dfg.py + python tests/validation/time_window_dfg.py --output --plan + CUDA_VISIBLE_DEVICES= python tests/validation/time_window_dfg.py --output --max-wall-seconds diff --git a/tests/unit/test_time_window_dfg.py b/tests/unit/test_time_window_dfg.py new file mode 100644 index 0000000..5189fb9 --- /dev/null +++ b/tests/unit/test_time_window_dfg.py @@ -0,0 +1,38 @@ +"""CPU contract of the A-versus-window timing harness (no CUDA).""" +from pathlib import Path +import sys +import unittest + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation")) +import time_window_dfg as w # noqa: E402 + + +class WindowTimingContract(unittest.TestCase): + def test_interval_and_variants(self): + self.assertEqual(w.INTERVAL, 80) + self.assertEqual(w.VARIANTS, ("A", "W", "W1")) + + def test_components(self): + obs = dict(total=dict(impulse=np.array([1., 2.]), angular_impulse=3., mass_into_fluid=4.)) + np.testing.assert_array_equal(w.components(obs), [1., 2., 3., 4.]) + + def test_interval_comparison(self): + ref = np.array([[1., -1., 0., 0.]]) + scale = np.array([[10., 10., 1., 1.]]) + self.assertEqual(w.compare_intervals(ref, scale, ref.copy()), 0.) + near = ref + np.array([[5e-9, 0., 0., 0.]]) + self.assertLessEqual(w.compare_intervals(ref, scale, near), 1.) + far = ref + np.array([[1e-7, 0., 0., 0.]]) + self.assertGreater(w.compare_intervals(ref, scale, far), 1.) + zero = np.zeros((1, 4)) + self.assertGreater(w.compare_intervals(zero, zero, zero + 1e-12), 1.) # floor, no div-by-0 + + def test_candidate_identity_records_files(self): + identity = w.candidate_identity(Path(__file__).resolve().parents[2]) + self.assertIn("checkout_revision", identity) + self.assertEqual(set(identity["sha256"]), set(w.CANDIDATE_FILES)) + + +if __name__ == "__main__": unittest.main() diff --git a/tests/validation/time_window_dfg.py b/tests/validation/time_window_dfg.py new file mode 100644 index 0000000..b90a9cd --- /dev/null +++ b/tests/validation/time_window_dfg.py @@ -0,0 +1,183 @@ +"""Measurement-only A-versus-window timing for the matched shifted-MRT refined DFG setup. + +Candidate source (window ledger) is recorded by checkout revision, dirty flag and file hashes; +the frozen refined-DFG manifests/results are NOT used or modified (they pin pre-window source). +Equal simulated duration, same fresh state and warm-up per variant, one observation interval = +80 root base steps (the runner's dense-phase interval): + A per base step: step(1) + read_raw_exchange, summed per interval (current runner workload) + W per interval: start_exchange_window(); step(80); read_exchange_window() + W1 per interval: start_exchange_window(); 80 x step(1); read_exchange_window() +All variants keep observe_raw_exchange=True; the backend is built once with +observe_window_exchange=True (a closed window adds no launch). Populations (bitwise) and +per-interval total impulses are compared OUTSIDE timing; the impulse tolerance is a +conservative engineering tolerance (see INTERVAL_RTOL), the rigorous substep-scaled comparison +being the capability test in tests/validation/multigrid/test_exchange_window.py. +Wall time (perf_counter) with explicit stream synchronization is the only timing source. +""" + +import argparse +import hashlib +import json +import math +import os +from pathlib import Path +import statistics +import sys +import time + +import numpy as np + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) +import time_refined_dfg as t # noqa: E402 (reuses run_refined_dfg setup via t.r) + +r = t.r +INTERVAL = 80 +VARIANTS = ("A", "W", "W1") +# Interval impulses differ only by FP64 summation order (per link over time vs per step over +# links). Engineering tolerance: 1e-9 of the interval's summed absolute per-step totals. +INTERVAL_RTOL = 1e-9 +CANDIDATE_FILES = ("src/CelerisLab/lbm/refinement_stepper.py", "src/CelerisLab/lbm/raw_exchange.py", + "src/CelerisLab/lbm/refinement_tables.py", "src/CelerisLab/lbm/kernels/blocks.cu") + + +def components(obs): + o = obs["total"] + return np.array([o["impulse"][0], o["impulse"][1], o["angular_impulse"], o["mass_into_fluid"]]) + + +def candidate_identity(source): + from source_provenance import checkout_provenance + return dict(checkout_provenance(source), + sha256={f: hashlib.sha256((source / f).read_bytes()).hexdigest() + for f in CANDIDATE_FILES}) + + +def run_variant(backend, variant, n_intervals, deadline): + """Returns per-interval totals and, for A, per-interval summed |per-step totals|.""" + totals, scales = [], [] + for _ in range(n_intervals): + if variant == "A": + impulse, scale = np.zeros(4), np.zeros(4) + for _ in range(INTERVAL): + backend.step(1) + c = components(backend.read_raw_exchange(0, level=1, window="last_base_step")) + impulse += c + scale += np.abs(c) + totals.append(impulse) + scales.append(scale) + else: + backend.start_exchange_window() + if variant == "W": + backend.step(INTERVAL) + else: + for _ in range(INTERVAL): + backend.step(1) + totals.append(components(backend.read_exchange_window(0))) + t.check_deadline(deadline) + return np.array(totals), (np.array(scales) if scales else None) + + +def compare_intervals(reference, scales, totals): + """Max violation ratio; <= 1 passes. Scale includes a floor so zero cancels cannot divide.""" + bound = INTERVAL_RTOL * scales + 1e-300 + return float(np.max(np.abs(totals - reference) / bound)) + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--source", type=Path, default=r.DEFAULT_SOURCE) + p.add_argument("--output", type=Path, required=True) + 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) + p.add_argument("--plan", action="store_true") + a = p.parse_args() + if a.base_steps <= 0 or a.base_steps % INTERVAL or a.base_steps > 20000 or a.repeats < 1: + p.error(f"base-steps must be a positive multiple of {INTERVAL} (<= 20000); repeats >= 1") + a.source = a.source.resolve() + identity = candidate_identity(a.source) + plan = dict(collision="MRT", shifted_storage="as run_refined_dfg MRT config", + candidate_source=identity, frozen_manifests_used=False, interval_base_steps=INTERVAL, + base_steps=a.base_steps, warmup_base_steps=a.warmup_base_steps, repeats=a.repeats, + variants=VARIANTS, tstar_window=a.base_steps * r.SPEED / 40.0, + timer="time.perf_counter wall with explicit stream sync", + excluded="construction/JIT, warm-up, health, downloads, parity/impulse comparison") + 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") + 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()): + p.error("Output must be empty/new") + a.output.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + a.max_wall_seconds + h, _ = r.helpers(a.source) + + from contextlib import ExitStack + from CelerisLab.cuda.context import CudaContext + from CelerisLab.lbm.refinement_stepper import FixedPatchStepper + from CelerisLab.lbm.channel import channel_masks + from CelerisLab.lbm.refinement import ACTIVE + from CelerisLab.lbm.descriptors import SOLID + + rows, ref = [], {} + n_intervals = a.base_steps // INTERVAL + with ExitStack() as stack: + cfg, pair, body, host = r.host_geometry(a.source, h, "MRT") + context = stack.enter_context(CudaContext(0)) + backend = stack.enter_context(FixedPatchStepper( + cfg, r.BOUNDS, context=context, coarse_boundary="channel", base_block=r.BASE_BLOCK, + fine_block=r.FINE_BLOCK, fine_objects=(body,), observe_raw_exchange=True, + observe_window_exchange=True)) + masks = [channel_masks(pair.coarse_shape)[0] & ~pair.covered, + (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)] + fresh = (r.initial_state(pair.coarse_shape, (0.0, 0.0), 1.0, root=True), + r.initial_state(pair.fine_shape, pair.fine_origin, 0.5)) + print(json.dumps(dict(status="CONSTRUCTED", case="refined")), flush=True) + for repeat in range(a.repeats): + for variant in VARIANTS: + t.check_deadline(deadline) + backend.initialize(*fresh) + for _ in range(a.warmup_base_steps): + backend.step(1) + warm = t.refined_health(backend, masks) + if repeat == 0 and variant == "A": + print(json.dumps(dict(status="WARM_READY", case="refined")), flush=True) + t.sync_refined(backend) + start = time.perf_counter() + totals, scales = run_variant(backend, variant, n_intervals, deadline) + t.sync_refined(backend) + seconds = time.perf_counter() - start + end = t.refined_health(backend, masks) + values = [np.array(v, copy=True) for v in backend._download_raw().values] + if variant == "A" and repeat == 0: + ref.update(values=values, totals=totals, scales=scales) + parity, ratio = "identical", 0.0 + else: + parity = t.same_state(ref["values"], values) or "identical" + ratio = compare_intervals(ref["totals"], ref["scales"], totals) + rows.append(dict(variant=variant, repeat=repeat, seconds=seconds, + intervals=n_intervals, population_parity_vs_A=parity, + interval_impulse_violation_ratio=ratio, + health_after_warmup=warm, health_after_window=end)) + if parity != "identical" or not ratio <= 1: + raise RuntimeError(f"{variant} repeat {repeat}: parity {parity}, impulse ratio {ratio}") + summary = {v: dict(median_seconds=statistics.median(x["seconds"] for x in rows if x["variant"] == v), + seconds=[x["seconds"] for x in rows if x["variant"] == v]) for v in VARIANTS} + summary.update(W_over_A=summary["W"]["median_seconds"] / summary["A"]["median_seconds"], + W1_over_A=summary["W1"]["median_seconds"] / summary["A"]["median_seconds"], + interpretation="Measured comparative timings under GPU variability; W also " + "batches step calls, W1 isolates the read pattern. Not a " + "benchmark-throughput or causal claim.") + (a.output / "window_timing.json").write_text(json.dumps(dict(plan=plan, rows=rows, summary=summary), + indent=2)) + print(json.dumps(summary, indent=2), flush=True) + return 0 + + +if __name__ == "__main__": + sys.exit(main())