"""Bounded FP32 GPU / pinned FP64 CPUREF-0.3 equivalence, not accuracy validation. Run explicitly in pycuda_3_10 with PYTHONPATH=/src, one BLAS thread, and CUDA_VISIBLE_DEVICES=0 (or 1) only after checking the physical GPU is idle. At most eight coarse steps per case. No files, plots or flow arrays are saved. """ import hashlib import importlib.util from contextlib import nullcontext from dataclasses import replace from pathlib import Path import resource import sys from threading import RLock import unittest from unittest.mock import Mock, patch import numpy as np import pycuda.driver as cuda import CelerisLab from CelerisLab.config import LBMConfig from CelerisLab.cuda import compiler from CelerisLab.cuda.context import CudaContext from CelerisLab.lbm.refinement_stepper import FixedPatchStepper ROOT = Path(__file__).resolve().parents[3] POP_ATOL = 3e-6 MOMENT_ATOL = 2e-4 REFERENCE_SHA = "bba2a06a9f6517b32079ffff15e4060d8c3b7ed35f16ca134e1877226c9e4e1e" REF_PATH = Path(__file__).with_name("cpu_reference.py") assert hashlib.sha256(REF_PATH.read_bytes()).hexdigest() == REFERENCE_SHA loader = importlib.util.spec_from_file_location("pinned_patch_reference", REF_PATH) ref = importlib.util.module_from_spec(loader) sys.modules[loader.name] = ref loader.loader.exec_module(ref) def postcollision(state, spec, maps): # Observation conversion only. Never applied to production G. return tuple(ref._collide(spec, a, np.isfinite(a), tau, mask, np.zeros(mask.shape, dtype=np.int64)) for a, tau, mask in ((state.coarse_f, spec.tau_coarse, maps.coarse_active), (state.fine_f, spec.tau_fine, maps.fine_active))) def totals(arrays, maps): # Leaf sum independent of the reference's flux/budget bookkeeping helpers. c, f = arrays populations = c[:, maps.coarse_active].astype(np.float64).sum(axis=1) populations += .25 * f[:, maps.fine_active].astype(np.float64).sum(axis=1) return np.array([populations.sum(), populations @ ref.CX, populations @ ref.CY]) def level_storage(level): """Raw device storage of one blocked level as dense (9, ny, nx).""" blocked = np.empty(level.layout.blocked_shape, np.float32) cuda.memcpy_dtoh(blocked, level.current) return level.layout.to_dense(blocked) class CoupledGPU(unittest.TestCase): def test_actual_tg_rates_one_interval(self): """Algebraic rate/phase check, not a new physical Taylor--Green benchmark. The N64/Re100 TG rates are exercised on the existing small patch with a nonequilibrium state. One interval checks g=C(f), including the shifted blocked path; no accuracy gate or production source changes. """ self.assertEqual(Path(CelerisLab.__file__).resolve(), ROOT / "src/CelerisLab/__init__.py") self.assertEqual(Path(compiler._KERNEL_DIR).resolve(), ROOT / "src/CelerisLab/lbm/kernels") spec = replace(ref.make_spec(collision="srt"), tau_coarse=.5576, tau_fine=.6152) maps = ref.build_maps(spec) state = ref.make_state(spec, velocity=(.03, -.02)) rng = np.random.default_rng(1) state = ref.State(*(a * (1 + 1e-3 * rng.standard_normal(a.shape)) for a in (state.coarse_f, state.fine_f))) initial = postcollision(state, spec, maps) next_state, _ = ref.advance_coarse_once(spec, maps, state) expected = postcollision(next_state, spec, maps) arrivals = (next_state.coarse_f, next_state.fine_f) cases = (("blocked_unshifted", False, dict(level_blocks=[(16, 16), (24, 24)])), ("blocked_shifted", True, dict(level_blocks=[(4, 4), (6, 6)]))) with CudaContext(0) as context: for label, shifted, storage in cases: with self.subTest(storage=label): cfg = LBMConfig(nx=16, ny=16, viscosity=.0192, ddf_shifting=shifted) with FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)], **storage) as backend: self.assertAlmostEqual(backend.patch.tau_coarse, spec.tau_coarse) self.assertAlmostEqual(backend.patch.tau_fine, spec.tau_fine) self.assertEqual(tuple(level.shape for level in backend._levels), ((9, 16, 16), (9, 24, 24))) # Physical FP64 g is encoded by initialize before FP32 rounding. backend.initialize(*initial) before = totals(backend.download(), maps) backend.step(1) actual = backend.download() self.assertEqual(backend.step_count, 1) backend.patch.check_synchronization( 1, next_state.tick, phase="postcollision_g") for got, want, arrival in zip(actual, expected, arrivals): np.testing.assert_array_equal(np.isnan(got), np.isnan(want)) np.testing.assert_allclose(got, want, rtol=0, atol=POP_ATOL, equal_nan=True) # A raw arrival-f comparison must fail on BOTH levels. # Equilibrium-only data would make this control vacuous. with self.assertRaises(AssertionError): np.testing.assert_allclose(got, arrival, rtol=0, atol=POP_ATOL, equal_nan=True) np.testing.assert_allclose(totals(actual, maps), before, rtol=0, atol=MOMENT_ATOL) np.testing.assert_allclose(totals(actual, maps), totals(expected, maps), rtol=0, atol=MOMENT_ATOL) error = max(float(np.nanmax(np.abs(a - e))) for a, e in zip(actual, expected)) print(f"actual-TG-rate algebraic check: {label} steps=1 " f"tau=.5576/.6152 max_population={error:.9g}; " "wrong-phase controls PASS", flush=True) def test_cleanup_attempts_every_allocation(self): backend = FixedPatchStepper.__new__(FixedPatchStepper) backend.ctx = Mock(activate=nullcontext) backend._lock = RLock() backend._status = "failed" backend._stream = Mock() backend._stream.synchronize.side_effect = RuntimeError("original stream failure") first, second = Mock(), Mock() first.free.side_effect = RuntimeError("secondary free failure") backend._allocations = [first, second] backend._levels = [Mock()] with self.assertRaisesRegex(RuntimeError, "original stream failure"): backend.close() first.free.assert_called_once() second.free.assert_called_once() self.assertEqual(backend._allocations, []) self.assertEqual(backend._levels, []) backend.close() def test_partial_advance_cannot_be_published(self): spec = ref.make_spec(collision="srt") maps = ref.build_maps(spec) with CudaContext(0) as context: with FixedPatchStepper(LBMConfig(nx=16, ny=16, viscosity=.1), context=context, refinements=[(4, 4, 12, 12)], level_blocks=[(16, 16), (24, 24)]) as backend: backend.initialize(*postcollision(ref.make_state(spec), spec, maps)) coarse, fine = backend._levels def fail_after_fine(*args, **kwargs): # Ensure both queued fine hops actually completed before failing. backend._stream.synchronize() raise RuntimeError("coarse launch failed after two fine hops") with patch.object(coarse, "kernel", side_effect=fail_after_fine): with self.assertRaisesRegex(RuntimeError, "after two fine hops"): backend.step() self.assertEqual(backend.step_count, 0) for operation in (backend.download, backend.step): with self.assertRaisesRegex(RuntimeError, "failed"): operation() raw = level_storage(fine) self.assertTrue(np.isfinite(raw.ravel()[backend.patch.coalesce_fine]).all()) self.assertFalse(context.is_closed) print("partial failure: two fine hops completed, no coarse step published; cleanup PASS", flush=True) def eight_steps(self, backend, spec, maps): """Rest and seeded nonequilibrium, 8 steps against the scalar reference.""" p = backend.patch for case in ("rest", "nonequilibrium_seed1"): with self.subTest(case=case): state = ref.make_state(spec, velocity=(0, 0) if case == "rest" else (.03, -.02)) if case != "rest": rng = np.random.default_rng(1) state = ref.State(*(a * (1 + 1e-3 * rng.standard_normal(a.shape)) for a in (state.coarse_f, state.fine_f))) initial = postcollision(state, spec, maps) backend.initialize(*initial) before = totals(backend.download(), maps) max_error = max_drift = max_moment_error = max_corner = 0.0 for step in range(1, 9): state, diagnostic = ref.advance_coarse_once(spec, maps, state) backend.step() actual = backend.download() expected = postcollision(state, spec, maps) self.assertEqual(backend.step_count, step) p.check_synchronization(step, state.tick, phase="postcollision_g") for got, want in zip(actual, expected): np.testing.assert_array_equal(np.isnan(got), np.isnan(want)) error = float(np.nanmax(np.abs(got - want))) self.assertLessEqual(error, POP_ATOL) max_error = max(max_error, error) drift = float(np.max(np.abs(totals(actual, maps) - before))) moment_error = float(np.max(np.abs(totals(actual, maps) - totals(expected, maps)))) self.assertLessEqual(drift, MOMENT_ATOL) self.assertLessEqual(moment_error, MOMENT_ATOL) max_drift, max_moment_error = max(max_drift, drift), max(max_moment_error, moment_error) f = backend._levels[1] raw = level_storage(f) np.testing.assert_array_equal(np.isfinite(raw), diagnostic["hop2_valid"]) self.assertTrue(np.isfinite(raw.ravel()[p.coalesce_fine]).all()) self.assertEqual(len(diagnostic["corner_packets"]), 8) for packet in diagnostic["corner_packets"].values(): sx, sy = packet["source_coarse_xy"] self.assertTrue(maps.coarse_active[sy, sx]) values = [raw[packet["direction"], y, x] for y, x in packet["children_fine_yx"]] error = float(np.max(np.abs(np.array(values) - packet["packets"]))) self.assertLessEqual(error, POP_ATOL) max_corner = max(max_corner, error) print(f"{case}: steps=8 max_population={max_error:.9g} " f"max_leaf_drift={max_drift:.9g} max_leaf_error={max_moment_error:.9g} " f"max_corner_packet={max_corner:.9g}; PASS", flush=True) def test_bounded_equivalence_blocked(self): """The same eight-step reference and launch-failure checks on whole-level tiles.""" spec = ref.make_spec(collision="srt") maps = ref.build_maps(spec) cfg = LBMConfig(nx=16, ny=16, viscosity=.1) with CudaContext(0) as context: with FixedPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)], level_blocks=[(16, 16), (24, 24)]) as backend: self.eight_steps(backend, spec, maps) with patch.object(backend._pairs[0], "explode", side_effect=RuntimeError("injected launch failure")): with self.assertRaisesRegex(RuntimeError, "injected"): backend.step() for operation in (backend.download, backend.step): with self.assertRaisesRegex(RuntimeError, "failed"): operation() self.assertFalse(context.is_closed) if __name__ == "__main__": unittest.main(verbosity=2)