test(D1): refinement GPU physics also runs on whole-level blocked tiles
- actual-TG-rate case: the unshifted leg uses [(16, 16), (24, 24)]; - partial-advance: blocked storage, fine owner storage read through the layout; - new test_bounded_equivalence_blocked: the same eight-step scalar-reference checks (rest + seeded nonequilibrium, synchronization, leaf drift/moments, hop-2 validity mask, coalesce sources, 8 corner packets) and launch-failure publication guard on blocked tiles, via the shared eight_steps helper. The dense audit build (CELERIS_REFINEMENT_AUDIT counters) stays in test_bounded_equivalence_and_ownership until D2 retires it with the dense kernels. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
3716b16555
commit
452eaf32de
@@ -52,6 +52,17 @@ def totals(arrays, maps):
|
||||
return np.array([populations.sum(), populations @ ref.CX, populations @ ref.CY])
|
||||
|
||||
|
||||
def level_storage(level):
|
||||
"""Raw device storage of one level as dense (9, ny, nx), dense or blocked."""
|
||||
if level.layout is None:
|
||||
raw = np.empty(level.shape, np.float32)
|
||||
cuda.memcpy_dtoh(raw, level.current)
|
||||
return raw
|
||||
blocked = np.empty(level.layout.blocked_shape, np.float32)
|
||||
cuda.memcpy_dtoh(blocked, level.current)
|
||||
return level.layout.to_dense(blocked)
|
||||
|
||||
|
||||
def global_array(module, name, shape):
|
||||
out = np.empty(shape, dtype=np.uint32)
|
||||
ptr, size = module.get_global(name)
|
||||
@@ -82,7 +93,7 @@ class CoupledGPU(unittest.TestCase):
|
||||
expected = postcollision(next_state, spec, maps)
|
||||
arrivals = (next_state.coarse_f, next_state.fine_f)
|
||||
|
||||
cases = (("dense_unshifted", False, {}),
|
||||
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:
|
||||
@@ -148,7 +159,8 @@ class CoupledGPU(unittest.TestCase):
|
||||
maps = ref.build_maps(spec)
|
||||
with CudaContext(0) as context:
|
||||
with PeriodicPatchStepper(LBMConfig(nx=16, ny=16, viscosity=.1),
|
||||
context=context, refinements=[(4, 4, 12, 12)]) as backend:
|
||||
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
|
||||
|
||||
@@ -164,12 +176,94 @@ class CoupledGPU(unittest.TestCase):
|
||||
for operation in (backend.download, backend.step):
|
||||
with self.assertRaisesRegex(RuntimeError, "failed"):
|
||||
operation()
|
||||
raw = np.empty(fine.shape, np.float32)
|
||||
cuda.memcpy_dtoh(raw, fine.current)
|
||||
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, audit):
|
||||
"""Rest and seeded nonequilibrium, 8 steps against the scalar reference.
|
||||
|
||||
audit=True also checks the dense CELERIS_REFINEMENT_AUDIT ownership counters.
|
||||
"""
|
||||
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)
|
||||
if audit: # dense audit build only
|
||||
for level in backend._levels:
|
||||
for name in ("collision_counts", "explosion_counts", "incoming_counts"):
|
||||
ptr, size = level.module.get_global(name)
|
||||
cuda.memset_d8(ptr, 0, size)
|
||||
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)
|
||||
c, f = backend._levels
|
||||
if audit:
|
||||
np.testing.assert_array_equal(global_array(c.module, "collision_counts", (16, 16)),
|
||||
step * maps.coarse_active)
|
||||
np.testing.assert_array_equal(global_array(f.module, "collision_counts", (24, 24)),
|
||||
2 * step * maps.fine_active)
|
||||
np.testing.assert_array_equal(global_array(f.module, "explosion_counts", (24, 24)),
|
||||
step * ~maps.fine_active)
|
||||
owners = np.zeros((9, 16, 16), np.uint32)
|
||||
owners.ravel()[p.coalesce_coarse] = step
|
||||
np.testing.assert_array_equal(global_array(c.module, "incoming_counts", owners.shape), owners)
|
||||
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}; ownership/counts 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 PeriodicPatchStepper(cfg, context=context, refinements=[(4, 4, 12, 12)],
|
||||
level_blocks=[(16, 16), (24, 24)]) as backend:
|
||||
self.eight_steps(backend, spec, maps, audit=False)
|
||||
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)
|
||||
|
||||
def test_bounded_equivalence_and_ownership(self):
|
||||
self.assertEqual(Path(CelerisLab.__file__).resolve(), ROOT / "src/CelerisLab/__init__.py")
|
||||
self.assertEqual(Path(compiler._KERNEL_DIR).resolve(), ROOT / "src/CelerisLab/lbm/kernels")
|
||||
@@ -198,64 +292,7 @@ class CoupledGPU(unittest.TestCase):
|
||||
for level in backend._levels)
|
||||
self.assertLess(device_bytes + audit_bytes, 128 * 1024)
|
||||
print(f"device_arrays={device_bytes} audit_arrays={audit_bytes} bytes", flush=True)
|
||||
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)
|
||||
for level in backend._levels:
|
||||
for name in ("collision_counts", "explosion_counts", "incoming_counts"):
|
||||
ptr, size = level.module.get_global(name)
|
||||
cuda.memset_d8(ptr, 0, size)
|
||||
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)
|
||||
c, f = backend._levels
|
||||
np.testing.assert_array_equal(global_array(c.module, "collision_counts", (16, 16)),
|
||||
step * maps.coarse_active)
|
||||
np.testing.assert_array_equal(global_array(f.module, "collision_counts", (24, 24)),
|
||||
2 * step * maps.fine_active)
|
||||
np.testing.assert_array_equal(global_array(f.module, "explosion_counts", (24, 24)),
|
||||
step * ~maps.fine_active)
|
||||
owners = np.zeros((9, 16, 16), np.uint32)
|
||||
owners.ravel()[p.coalesce_coarse] = step
|
||||
np.testing.assert_array_equal(global_array(c.module, "incoming_counts", owners.shape), owners)
|
||||
raw = np.empty(f.shape, np.float32)
|
||||
cuda.memcpy_dtoh(raw, f.current)
|
||||
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}; ownership/counts PASS", flush=True)
|
||||
self.eight_steps(backend, spec, maps, audit=True)
|
||||
|
||||
# Fail before any additional time step; no stale synchronized export.
|
||||
with patch.object(backend._pairs[0], "explode", side_effect=RuntimeError("injected launch failure")):
|
||||
|
||||
Reference in New Issue
Block a user