diff --git a/docs/validation/refined_dfg_window.md b/docs/validation/refined_dfg_window.md index 1998c40..9257494 100644 --- a/docs/validation/refined_dfg_window.md +++ b/docs/validation/refined_dfg_window.md @@ -56,3 +56,16 @@ Root must independently review and admit any execution. An admitted paired pilot the existing watchdog, device selection, `--gpu-gates-approved`, positive `--max-wall-seconds` and a new output directory, adding only `--exchange-window` to the selected collision's pilot command. No GPU or remote action was performed for this patch. + +## Wide fine storage allocation + +The wide runner uses `FINE_BLOCK = (240, 568)`: one block containing the whole +fine allocation, including the four-cell refinement ghost band on each side. +The active `232 x 560` region alone is not a valid substitute. Physical bounds, +resolution, numerical kernels, 256-thread launches, per-root synchronization +and all scientific gates are unchanged. The plan records this storage block. + +This is a wide-only configuration backport; no compact-placement or public-Jet +API is added. Existing source hashes remain checked, and the separate runner +hash requires a fresh completed pilot before a full run. Earlier short-layout +timings do not establish a full-run speedup or change historical failed verdicts. diff --git a/src/CelerisLab/lbm/refinement_stepper.py b/src/CelerisLab/lbm/refinement_stepper.py index 2cf63ab..dbbc1df 100644 --- a/src/CelerisLab/lbm/refinement_stepper.py +++ b/src/CelerisLab/lbm/refinement_stepper.py @@ -822,27 +822,41 @@ class FixedPatchStepper: grid=((pairs + self._block[0] - 1) // self._block[0], 1), **kw) def step(self, n=1): - """Advance G→G by n base intervals: level l takes 2**l hops per interval.""" + """Advance G→G by n base intervals: level l takes 2**l hops per interval. + + All n intervals are enqueued in order on the one stream, and the host waits + once, before returning; step(0) launches and waits for nothing. On return + the device work is complete, step_count and an open window cover all n + intervals, and the raw-exchange ledger holds the final interval only. + + Asynchronous device errors surface at that single wait, so they are not + attributed to the interval that caused them; call step(1) repeatedly to + localize one (same launches, same bits). On any failure the instance is + failed, the exchange ledger and window are invalid, and step_count keeps + its value from entry to the failing call. + """ if isinstance(n, bool) or not isinstance(n, Integral) or n < 0: raise ValueError("n must be a nonnegative integer") with self._lock, self.ctx.activate(): self._require("ready") + if n == 0: + return root = self._levels[0] kw = dict(block=self._block, stream=self._stream) try: self._status = "advancing" + self._exchange_valid = False for _ in range(n): - if self._observe_raw_exchange: - self._exchange_valid = False + if self._observe_raw_exchange: # stream-ordered: holds the last interval only for b in self._level_boundaries: if b is not None: cuda.memset_d8_async(b.interval_integral, 0, len(b.tables["cell"]) * 32, self._stream) self._hop(0, root.current, root.spare, **kw) - self._stream.synchronize() - root.current, root.spare = root.spare, root.current - self._step_count += 1 - self._exchange_valid = self._observe_raw_exchange + root.current, root.spare = root.spare, root.current # host handles only + self._stream.synchronize() + self._step_count += n + self._exchange_valid = self._observe_raw_exchange self._status = "ready" except BaseException: self._window_valid = False diff --git a/tests/unit/test_step_sync.py b/tests/unit/test_step_sync.py new file mode 100644 index 0000000..2cc04a5 --- /dev/null +++ b/tests/unit/test_step_sync.py @@ -0,0 +1,170 @@ +"""Host contract of FixedPatchStepper.step(n): one stream wait per call (mocked driver). + +All n base intervals are enqueued in order on the backend stream and the host +waits once before step returns. Launch order and buffer arguments must equal n +step(1) calls; completion metadata, raw-exchange and window ledgers, and failure +invalidation are published only after that wait. +""" +from contextlib import nullcontext +from unittest.mock import Mock +import unittest + +import numpy as np + +from CelerisLab.config import LBMConfig +from tests.unit.test_raw_exchange import body +from tests.unit.test_refinement_three_level_multipatch import host_module + +BODY_LEVEL, HOPS = 3, 8 # body on L3: 2**3 hops per base interval + + +def backend(window=False): + """Four-level host backend; every launch, memset and stream wait goes to one log.""" + host = host_module() + log, allocations = [], [] + + def allocate(size): + allocations.append(Mock(size=size)) + return allocations[-1] + + module = Mock() + module.get_global.return_value = (Mock(), 20) + module.get_function.side_effect = lambda name: Mock( + side_effect=lambda *a, **k: log.append((name, a, k))) + host.compiler.load_module.return_value = module + host.cuda.mem_alloc.side_effect = allocate + host.cuda.memcpy_dtoh.side_effect = lambda array, pointer: array.fill(0) + host.cuda.memset_d8_async.side_effect = lambda *a: log.append(("memset", a, {})) + run = host.FixedPatchStepper( + LBMConfig(nx=48, ny=48, viscosity=.02), context=Mock(activate=nullcontext, sm_arch="sm_70"), + refinements=[[(6, 6, 26, 26)], [(20, 20, 44, 44)], [(56, 56, 72, 72)]], + level_blocks=[(4, 4)] * 4, level_objects=[(), (), (), (body(),)], + observe_raw_exchange=True, observe_window_exchange=window) + run._stream.synchronize.side_effect = lambda: log.append(("sync", (), {})) + run.initialize(*[np.full(level.shape, 1 / 9, np.float32) for level in run._levels]) + log.clear() + return run, log, allocations + + +def signature(log, allocations): + """Event names with device-pointer arguments replaced by allocation index.""" + index = {id(a): i for i, a in enumerate(allocations)} + return [(name, tuple(index.get(id(a), "-") for a in args)) for name, args, _ in log] + + +def names(log): + return [name for name, _, _ in log] + + +class StepSync(unittest.TestCase): + def test_step1_waits_once_after_all_launches(self): + run, log, _ = backend() + run.step(1) + self.assertEqual(names(log).count("sync"), 1) + self.assertEqual(names(log)[-1], "sync") + self.assertEqual(names(log)[0], "memset") + self.assertEqual(names(log).count("BlockStep"), 1 + 2 + 4 + 8) + self.assertEqual(run.step_count, 1) + run.close() + + def test_step_n_equals_n_step1_launches_with_one_wait(self): + single, single_log, single_alloc = backend() + for _ in range(3): + single.step(1) + batched, batched_log, batched_alloc = backend() + batched.step(3) + expected = [e for e in signature(single_log, single_alloc) if e[0] != "sync"] + got = signature(batched_log, batched_alloc) + self.assertEqual(got[-1][0], "sync") + self.assertEqual(names(batched_log).count("sync"), 1) + self.assertEqual(got[:-1], expected) # same order, same ping-pong buffers + self.assertEqual(names(batched_log).count("memset"), 3) + self.assertEqual(batched.step_count, single.step_count) + index = [{id(a): i for i, a in enumerate(alloc)} for alloc in (single_alloc, batched_alloc)] + self.assertEqual(index[1][id(batched._levels[0].current)], + index[0][id(single._levels[0].current)]) + single.close() + batched.close() + + def test_step0_launches_and_waits_for_nothing(self): + run, log, _ = backend() + run.step(2) + log.clear() + run.step(0) + self.assertEqual(log, []) + self.assertEqual(run.step_count, 2) + value = run.read_raw_exchange(0, level=BODY_LEVEL) # step(0) keeps the ledger valid + self.assertEqual((value["time_start"], value["time_end"]), (1, 2)) + run.close() + + def test_completion_metadata_covers_all_intervals(self): + run, _, _ = backend() + with self.assertRaisesRegex(RuntimeError, "completed"): + run.read_raw_exchange(0, level=BODY_LEVEL) + run.step(3) + value = run.read_raw_exchange(0, level=BODY_LEVEL) + self.assertEqual((value["time_start"], value["time_end"], value["sample_count"]), + (2, 3, HOPS)) # step(n) retains only the final base interval + run.step(2) + self.assertEqual(run.read_raw_exchange(0, level=BODY_LEVEL)["time_end"], 5) + run.close() + + def test_window_accounting_across_batched_calls(self): + windows = [] + for calls in ((1, 3), (4,), (1, 1, 1, 1)): + run, log, _ = backend(window=True) + run.step(1) + run.start_exchange_window() + log.clear() + for n in calls: + run.step(n) + self.assertEqual(names(log).count("BlockAccumulateExchange"), 2 * 4 * HOPS) + self.assertEqual(names(log).count("sync"), len(calls)) + out = run.read_exchange_window(0, level=BODY_LEVEL) + windows.append((out["time_start"], out["time_end"], out["sample_count"])) + run.close() + self.assertEqual(windows, [(1., 5, 4 * HOPS)] * 3) + + def test_failed_wait_publishes_nothing(self): + run, _, _ = backend(window=True) + run.step(2) + run.start_exchange_window() + run._stream.synchronize.side_effect = RuntimeError("device failure") + with self.assertRaisesRegex(RuntimeError, "device failure"): + run.step(3) + self.assertEqual(run.step_count, 2) # value at entry to the failing call + self.assertFalse(run._exchange_valid) + self.assertFalse(run._window_valid) + for read in (lambda: run.read_raw_exchange(0, level=BODY_LEVEL), + lambda: run.read_exchange_window(0, level=BODY_LEVEL), run.download, + lambda: run.step(1)): + with self.assertRaisesRegex(RuntimeError, "failed"): + read() + run._stream.synchronize.side_effect = None + run.close() + + def test_launch_error_mid_batch_skips_wait_and_fails(self): + run, log, _ = backend() + run.step(1) + log.clear() + hop, calls = run._hop, [] + + def failing(index, *args, **kw): + if index == 0: + calls.append(1) + if len(calls) == 2: + raise RuntimeError("launch failure") + return hop(index, *args, **kw) + run._hop = failing + with self.assertRaisesRegex(RuntimeError, "launch failure"): + run.step(3) + self.assertNotIn("sync", names(log)) + self.assertEqual(run.step_count, 1) + self.assertFalse(run._exchange_valid) + with self.assertRaisesRegex(RuntimeError, "failed"): + run.read_raw_exchange(0, level=BODY_LEVEL) + run.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/validation/run_refined_dfg.py b/tests/validation/run_refined_dfg.py index d43902a..d5e315c 100644 --- a/tests/validation/run_refined_dfg.py +++ b/tests/validation/run_refined_dfg.py @@ -17,7 +17,7 @@ HERE = Path(__file__).resolve().parent DEFAULT_SOURCE = HERE.parents[1] BOUNDS = (40, 24, 320, 140) BASE_BLOCK = (2, 881) # Exact 166 x 881 domain; no geometric padding. -FINE_BLOCK = (8, 8) # 240 x 568 including four fine halo cells. +FINE_BLOCK = (240, 568) # Whole fine allocation, including its refinement ghost band. SPEED = 0.01