From 519473a61c7b319183aa54e55b9f097499dde0bf Mon Sep 17 00:00:00 2001 From: TRON Date: Sun, 18 Oct 2026 01:55:21 -0700 Subject: [PATCH] test: add explicit CUDA explosion parity smoke --- .../taylor_green/check_gpu_explosion.py | 145 ++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 tests/validation/taylor_green/check_gpu_explosion.py diff --git a/tests/validation/taylor_green/check_gpu_explosion.py b/tests/validation/taylor_green/check_gpu_explosion.py new file mode 100644 index 0000000..4b8a2e3 --- /dev/null +++ b/tests/validation/taylor_green/check_gpu_explosion.py @@ -0,0 +1,145 @@ +"""Explicit tiny CUDA explosion parity and constructor/one-interval smoke check. + +No installs or device selection defaults. Run only on the authorized GPU host. +This is operator/runtime evidence, not Taylor-Green accuracy qualification. +""" +import argparse +import hashlib +import json +import os +from pathlib import Path +import sys + +import numpy as np + + +def check_case(ctx, shifted, projected): + import pycuda.driver as cuda + from CelerisLab.config import LBMConfig + from CelerisLab.lbm.refinement_stepper import FixedPatchStepper + from CelerisLab.lbm.refinement_tables import projected_explosion_tables + from CelerisLab.lbm.refinement import ACTIVE + from test_projected_explosion import reconstruct, W + + cfg = LBMConfig(nx=16, ny=16, viscosity=.1, velocity=0, + ddf_shifting=shifted) + cfg.validate() + options = {"explosion": "projected_linear"} if projected else {} + # Omitting explosion in the constant case tests the production default. + with FixedPatchStepper(cfg, context=ctx, refinements=[[(4, 4, 12, 12)]], + level_blocks=[(4, 4), (8, 4)], **options) as run: + parent, child = run._levels + pair = run._pairs[0] + patch = run.hierarchy.pairs[0] + coarse_cells = int(np.prod(parent.layout.local_shape)) + fine_cells = int(np.prod(child.layout.local_shape)) + assert coarse_cells != fine_cells + assert pair.parent_cells == coarse_cells + assert len(pair.interpolation) == (2 if projected else 0) + donor = patch.ghost_coarse.astype(np.intp) + # Random variation must exercise the gradient, not just a constant field. + c = np.random.default_rng(11).normal(0, .001, (9, 16, 16)) + if not shifted: + c += W[:, None, None] + c = c.astype(np.float32) + legal = np.zeros(256, bool) + legal[donor] = True + c.reshape(9, -1)[:, ~legal] = np.nan + coarse = np.full(np.prod(parent.layout.blocked_shape), np.nan, np.float32) + for q in range(9): + coarse[parent.layout.slot(parent.layout.owner.ravel()[legal], q)] = c.reshape(9, -1)[q, legal] + fine = np.full(np.prod(child.layout.blocked_shape), np.nan, np.float32) + cuda.memcpy_htod(parent.current, coarse) + cuda.memcpy_htod(child.current, fine) + calls = [] + real_hop = run._hop + try: + # Execute the production explosion launch exactly; isolate it from + # subsequent streaming/collision without replacing the kernel/tables. + run._hop = lambda *args, **kwargs: calls.append(args) + run._advance(1, parent.current, block=run._block, stream=run._stream) + run._stream.synchronize() + finally: + run._hop = real_hop + assert calls == [(1, child.current, child.spare), (1, child.spare, child.current)] + cuda.memcpy_dtoh(fine, child.current) + targets = child.layout.owner.ravel()[patch.ghost_fine] + slots = np.array([child.layout.slot(targets, q) for q in range(9)]) + actual = fine[slots] + neighbors, offsets = projected_explosion_tables(patch, parent.layout) + inverse = np.full(parent.layout.nblocks * coarse_cells, -1, dtype=int) + inverse[parent.layout.owner.ravel()] = np.arange(256) + constant = c.reshape(9, -1)[:, donor].astype(float) + expected = reconstruct(c.astype(float), patch, inverse[neighbors], offsets)[0] if projected else constant + assert np.isfinite(actual).all() + if projected: + assert np.max(np.abs(expected - constant)) > 1e-5 + np.testing.assert_allclose(actual, expected, atol=6e-8, rtol=2e-7) + untouched = np.ones(fine.size, bool) + untouched[slots.ravel()] = False + assert np.isnan(fine[untouched]).all(), "explosion wrote outside target slots" + error = float(np.max(np.abs(actual - expected))) + + # Restore ordinary execution and exercise real constructor wiring, + # stream-only ghost hops, restriction, halo copies and collision once. + initial = [np.broadcast_to(W[:, None, None], (9, *k.shape)).copy() + for k in run.hierarchy.kinds] + run.initialize(*initial) + run.step(1) + for values, kind in zip(run.download(), run.hierarchy.kinds): + active = values[:, kind == ACTIVE] + assert np.isfinite(active).all() + np.testing.assert_allclose(active, np.broadcast_to(W[:, None], active.shape), + atol=3e-7, rtol=1e-6) + assert run.step_count == 1 + return dict(shifted=shifted, explosion="projected_linear" if projected else "constant_default", + parent_stride=coarse_cells, child_stride=fine_cells, + compared_slots=int(actual.size), max_abs_error=error, + poisoned_nondonors=True, untouched_slots_preserved=True, equilibrium_step=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--run-gpu", action="store_true", help="Explicitly opt into GPU execution") + args = parser.parse_args() + selected = os.environ.get("CUDA_VISIBLE_DEVICES", "") + if not args.run_gpu or not selected or "," in selected or selected == "-1": + parser.error("--run-gpu and exactly one explicit physical CUDA_VISIBLE_DEVICES are required") + if args.output.exists(): + parser.error("output already exists; choose a new receipt path") + root = args.repo.resolve() + sys.path[:0] = [str(root / "src"), str(root / "tests/unit"), + str(root / "tests/validation/taylor_green")] + import CelerisLab + if Path(CelerisLab.__file__).resolve() != root / "src/CelerisLab/__init__.py": + raise RuntimeError("Wrong imported source checkout") + from run_case import provenance + report = dict(source=provenance(root), + numpy_oracle_sha256=hashlib.sha256((root / "tests/unit/test_projected_explosion.py").read_bytes()).hexdigest(), + cuda_visible_devices=selected, + logical_device=0, completed=False, cases=[]) + args.output.parent.mkdir(parents=True, exist_ok=True) + def save(): + args.output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") + save() + try: + from CelerisLab.cuda.context import CudaContext + with CudaContext(0) as ctx: + report["device"] = ctx.device.name() + for shifted in (False, True): + for projected in (False, True): + report["cases"].append(check_case(ctx, shifted, projected)) + save() + report["completed"] = True + save() + except Exception as exc: + report["error"] = f"{type(exc).__name__}: {exc}" + save() + raise + print(json.dumps(report, indent=2)) + + +if __name__ == "__main__": + main()