From 8db7cee9d53deaa2f2553f2ab33810dc8e0c3f39 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 10:06:38 +0000 Subject: [PATCH] test: dense/blocked route parity gate and route capability matrix Test-only, on c6d5dc1. DenseBlockedRouteParity runs the same configuration, initial state, body/Jet schedule and tilings on D/D [None, None], D/B [None, FINE_BLOCK] and B/B [(12, 8), FINE_BLOCK], periodic and slot/reservoir, for body-free, passive circle, actuated circle (+/-) and a velocity Jet with an amplitude switch: every public and owner array must match in FP32 bits and NaN masks at steps 0, 1, 2, 8 and after re-initialize, and the D/B and B/B raw-exchange ledgers (last base step and last substep) must be equal. No tolerance. test_route_capability_matrix checks before CUDA what the dense route rejects (raw exchange on a dense body level, shifted, projected, channel, several L1 rectangles, depth >= 3 dense). The module docstring holds the supported/unsupported table and the parked cases (MRT covered elsewhere, shifted and depth >= 3 have no dense route, several bodies later). Co-Authored-By: Claude Opus 5.5 --- .../multigrid/test_refinement_blocked_base.py | 130 ++++++++++++++++++ 1 file changed, 130 insertions(+) diff --git a/tests/validation/multigrid/test_refinement_blocked_base.py b/tests/validation/multigrid/test_refinement_blocked_base.py index cfe6232..1fafa99 100644 --- a/tests/validation/multigrid/test_refinement_blocked_base.py +++ b/tests/validation/multigrid/test_refinement_blocked_base.py @@ -2,6 +2,34 @@ Run CPU and GPU classes in separate processes. GPU callers must first check resources/device availability; this file never selects or reserves a GPU. + +Dense/blocked route parity (DenseBlockedRouteParity): the same configuration, +initial state, body/Jet schedule and tilings on three storage routes, only +level_blocks varying: D/D [None, None], D/B [None, FINE_BLOCK] (dense root, +the mixed pattern of current examples) and B/B [(12, 8), FINE_BLOCK]. Gate: +every public and owner array identical in FP32 bits and NaN masks at steps +0, 1, 2, 8 and after re-initialize; D/B and B/B raw-exchange ledgers equal. +No tolerance. A mismatch is reported with route pair/level/step and diagnosed +(boundary semantics vs arithmetic ordering), never widened. + +Route capabilities (rejections checked before CUDA by test_route_capability_matrix): + + combination D/D D/B B/B + periodic / slot_reservoir root, fine circle yes yes yes (parity gate) + actuated circle, velocity Jet + amplitude yes yes yes (parity gate) + raw/window exchange on a body level rejected yes yes + shifted storage rejected rejected yes + projected explosion rejected rejected yes + channel root rejected rejected yes + several L1 rectangles rejected rejected yes + depth >= 3 with a dense level rejected rejected yes +Slot/reservoir walls are compile-time WALL_* macros on a dense root and the +d_block_wall constant on a blocked root: BC-encoding parity, bit-identical by +this gate. Dense fine bodies use CurvedLinkSoA + CurvedBoundaryKernel +(n_objects > 0), blocked ones BlockCurvedBoundary; same links and physics, +compared through the population rows. Parked here: MRT (covered by +test_refinement_mrt_gpu::test_dense_versus_blocked_mrt_two_level), shifted and +depth >= 3 (no dense route to compare), several bodies (later slice). """ from contextlib import ExitStack, contextmanager, nullcontext from dataclasses import replace @@ -20,6 +48,7 @@ from CelerisLab.lbm.descriptors import D2Q9_EX, D2Q9_EY from CelerisLab.lbm.refinement import ACTIVE, OUTER_GHOST, FixedPatch2D from CelerisLab.lbm.refinement_stepper import FixedPatchStepper, blocked_base_tables from test_refinement_blocked_fine import NX, NY, BOUNDS, config, state, circle, actuation +from test_refinement_jets import SPEED, jet BASE_BLOCKS = ((24, 32), (12, 8), (3, 4), (1, 1)) FINE_BLOCK = (16, 20) @@ -324,5 +353,106 @@ class BlockedBaseGPU(unittest.TestCase): print(f"PASS {label}: full finite bits/NaN masks at 0,1,2,8 + reset", flush=True) +ROUTES = (("D/D", [None, None]), ("D/B", [None, FINE_BLOCK]), ("B/B", [(12, 8), FINE_BLOCK])) + + +def same_summary(a, b): + """Exact equality of nested exchange summaries (dicts, sequences, arrays, numbers).""" + if isinstance(a, dict): + return a.keys() == b.keys() and all(same_summary(a[k], b[k]) for k in a) + if isinstance(a, (list, tuple)): + return len(a) == len(b) and all(same_summary(x, y) for x, y in zip(a, b)) + return np.array_equal(np.asarray(a), np.asarray(b), equal_nan=True) + + +class DenseBlockedRouteCPU(unittest.TestCase): + def test_route_capability_matrix(self): + context = Mock(activate=Mock(side_effect=AssertionError("CUDA reached"))) + periodic = config("periodic") + inner = (BOUNDS[0] + 2, BOUNDS[1] + 2, BOUNDS[2] - 2, BOUNDS[3] - 2) + second = (BOUNDS[0], BOUNDS[1], BOUNDS[0] + 2, BOUNDS[1] + 2) + cases = ( + ("raw exchange on dense body level", periodic, dict( + level_blocks=[None, None], level_objects=[(), (circle(),)], observe_raw_exchange=True)), + ("shifted D/D", replace(periodic, ddf_shifting=True), dict(level_blocks=[None, None])), + ("shifted D/B", replace(periodic, ddf_shifting=True), dict(level_blocks=[None, FINE_BLOCK])), + ("projected D/D", periodic, dict(level_blocks=[None, None], explosion="projected_linear")), + ("projected D/B", periodic, dict(level_blocks=[None, FINE_BLOCK], explosion="projected_linear")), + ("channel D/B", periodic, dict(level_blocks=[None, FINE_BLOCK], coarse_boundary="channel")), + ("several L1 rectangles", periodic, dict(refinements=[[BOUNDS, second]], level_blocks=[None, None])), + ("depth 3 dense", periodic, dict(refinements=[BOUNDS, inner], level_blocks=[None, None, None])), + ) + for label, cfg, kw in cases: + kw.setdefault("refinements", [BOUNDS]) + with self.subTest(label), self.assertRaises(ValueError): + FixedPatchStepper(cfg, context=context, **kw) + context.activate.assert_not_called() + + +class DenseBlockedRouteParity(unittest.TestCase): + """D/D vs D/B vs B/B, unshifted SRT, one circle or one velocity Jet; see module docstring.""" + + def run_routes(self, mode, objects=(), jets=(), switch=None): + from CelerisLab.cuda.context import CudaContext + cfg = config(mode) + p = FixedPatch2D.from_config(cfg, BOUNDS) + arrays = state(NY, NX, 1), state(*p.fine_shape, 2) + results = {} + with CudaContext(0) as context: + for name, blocks in ROUTES: + observe = bool(objects) and blocks[1] is not None + snapshots, ledgers = [], [] + with FixedPatchStepper(cfg, context=context, coarse_boundary=mode, refinements=[BOUNDS], + level_blocks=blocks, level_objects=[(), tuple(objects)], + level_jets=[(), tuple(jets)], + observe_raw_exchange=observe) as backend: + backend.initialize(*arrays) + for step in (0, 1, 2, 8): + if step == 8 and switch is not None: + backend.set_jet_amplitude(0, switch) + backend.step(step - backend.step_count) + snapshots.append(backend.download() + raw(backend)) + if observe and step: + ledgers.append(tuple(backend.read_raw_exchange(0, window=window) + for window in ("last_base_step", "last_substep"))) + backend.initialize(*arrays) + snapshots.append(backend.download() + raw(backend)) + results[name] = snapshots, ledgers + return results + + def check(self, label, results): + reference, _ = results["D/D"] + for name in ("D/B", "B/B"): + for step, actual, expected in zip((0, 1, 2, 8, "reset"), results[name][0], reference): + for index, (a, b) in enumerate(zip(actual, expected)): + exact(self, a, b, f"{label} {name} vs D/D step={step} array={index}") + if results["D/B"][1]: + for step, a, b in zip((1, 2, 8), results["D/B"][1], results["B/B"][1]): + self.assertTrue(same_summary(a, b), f"{label} raw exchange D/B vs B/B step={step}") + for name, (snapshots, _) in results.items(): + masses = [float(np.nansum(level.astype(np.float64))) for level in snapshots[3][:2]] + print(f"{label} {name}: step 8 mass by level {masses}", flush=True) + print(f"PASS {label}: D/D = D/B = B/B bits/NaN masks at 0,1,2,8 + reset", flush=True) + + def test_body_free(self): + for mode in ("periodic", "slot_reservoir"): + self.check(f"{mode} body-free", self.run_routes(mode)) + + def test_passive_circle(self): + for mode in ("periodic", "slot_reservoir"): + self.check(f"{mode} circle", self.run_routes(mode, objects=[circle()])) + + def test_actuated_circle(self): + for mode in ("periodic", "slot_reservoir"): + for speed in (SPEED, -SPEED): + self.check(f"{mode} actuated {speed}", + self.run_routes(mode, objects=[circle(actuation(speed))])) + + def test_velocity_jet_with_amplitude_switch(self): + for mode in ("periodic", "slot_reservoir"): + self.check(f"{mode} jet switch", + self.run_routes(mode, objects=[circle()], jets=[jet()], switch=-SPEED)) + + if __name__ == "__main__": unittest.main()