step(n) docstring: raw-exchange ledger holds the final interval only

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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:
+                    if self._observe_raw_exchange:  # stream-ordered: holds the last interval only
-                        self._exchange_valid = False
                         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  # host handles only
-                    root.current, root.spare = root.spare, root.current
+                self._stream.synchronize()
-                    self._step_count += 1
+                self._step_count += n
-                    self._exchange_valid = self._observe_raw_exchange
+                self._exchange_valid = self._observe_raw_exchange
                 self._status = "ready"
             except BaseException:
                 self._window_valid = False

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()

tests/validation/multigrid/test_refinement_three_level.py

@@ -80,12 +80,11 @@ class ThreeLevelCPU(unittest.TestCase):
         f, g = run._levels[1:]
         b = g.boundary
         events, feeds = [], []
-        advance, gather = run._advance, run._gather_halo
+        advance, gather = run._advance_child, run._gather_halo
-        def record_feed(index, buffer, **kw):  # index 2: L1 hop feeds L2
+        def record_feed(buffer, **kw):
-            if index == 2:
+            self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
-                self.assertIs(buffer, (f.current, f.spare)[len(feeds)])
+            feeds.append(buffer)
-                feeds.append(buffer)
+            advance(buffer, **kw)
-            advance(index, buffer, **kw)
         def record_halo(level, buffer, **kw):
             if level is g:
                 events.append(("halo", buffer))
@@ -96,7 +95,7 @@ class ThreeLevelCPU(unittest.TestCase):
             self.assertIs(kind, g.kind)
             events.append(("step", src))
         g.kernel.side_effect = record_step
-        with patch.object(run, "_advance", side_effect=record_feed), \
+        with patch.object(run, "_advance_child", side_effect=record_feed), \
                 patch.object(run, "_gather_halo", side_effect=record_halo):
             run.step()
         self.assertEqual(feeds, [f.current, f.spare])
@@ -106,8 +105,8 @@ class ThreeLevelCPU(unittest.TestCase):
         self.assertGreater(len(np.unique(b.tables["block"])), 1)
         # Negative control: reusing L1's old source at the midpoint must fail.
         feeds.clear()
-        with patch.object(run, "_advance", side_effect=lambda index, buffer, **kw:
+        with patch.object(run, "_advance_child", side_effect=lambda buffer, **kw:
-                          record_feed(index, f.current if index == 2 else buffer, **kw)):
+                          record_feed(f.current, **kw)):
             with self.assertRaises(AssertionError):
                 run.step()
         run.close()
@@ -173,12 +172,11 @@ class ThreeLevelGPU(unittest.TestCase):
             p1, p2 = run.patch, run.child_patch
             run.initialize(state(16, 16, 1), state(*p1.fine_shape, 2), state(*p2.fine_shape, 3))
             f = run._levels[1]
-            feeds, original = [], run._advance
+            feeds, original = [], run._advance_child
-            def record(index, buffer, **kw):  # index 2: L1 hop feeds L2
+            def record(buffer, **kw):
-                if index == 2:
+                feeds.append(int(buffer))
-                    feeds.append(int(buffer))
+                original(buffer, **kw)
-                original(index, buffer, **kw)
+            with patch.object(run, "_advance_child", side_effect=record):
-            with patch.object(run, "_advance", side_effect=record):
                 cur, spare = int(f.current), int(f.spare)
                 run.step(1)
             self.assertEqual(feeds, [cur, spare])  # current G, then the new G of hop 1