Compare commits

..
Author SHA1 Message Date
ClaudeandClaude Opus 5.5 1cbdc479cf test_block_step_sync: locate block_stepper without importing it
The module-level import of CelerisLab.lbm.block_stepper loaded
pycuda.driver into the shared test process, unlike the other CPU host
tests (which load mocked copies via AST). find_spec gives the source path
without the import. Test content and assertions unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:39:52 +00:00
ClaudeandClaude Opus 5.5 f3f1e3d588 PeriodicBlockStepper.step: mid-batch launch failure test; async-error docstring note
Review follow-up (non-blocking items): a launch failure on the second of
three steps fails closed without the batch wait, keeps step_count at its
entry value and blocks download; the docstring now states, as for
FixedPatchStepper, that async faults surface at the single wait.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:31:09 +00:00
ClaudeandClaude Opus 5.5 3648ad70ef PeriodicBlockStepper.step(n): wait on the stream once per call
Same change as the reviewed FixedPatchStepper end-sync (cherry-picked
ad1d82c+512dcc9 above): the n steps are enqueued in order and the host
waits once; step(0) launches nothing; step_count advances after the wait
and keeps its entry value on failure. Launches, buffer roles and bits are
those of n step(1) calls. Mocked-driver host test covers the single
wait, launch/role equality with n step(1), step(0) and failure.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:25:16 +00:00
ClaudeandClaude Opus 5.5 5399c9c0a6 step(n) docstring: raw-exchange ledger holds the final interval only
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:24:36 +00:00
ClaudeandClaude Opus 5.5 54be33e006 FixedPatchStepper.step(n): wait on the stream once per call
Enqueue all n base intervals in order on the one stream and synchronize
once before returning (step(0): no launch, no wait). step_count, the raw
exchange ledger and an open window are published only after that wait;
any failure leaves step_count at its entry value and invalidates both
ledgers. Async device errors are no longer attributed to an interval;
the docstring says to use step(1) loops to localize one. Launch order and
buffers are unchanged. Adds a mocked-driver host test of the contract.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:24:36 +00:00
6 changed files with 309 additions and 187 deletions
+14 -4
View File
@@ -204,11 +204,21 @@ class PeriodicBlockStepper:
raise
def step(self, n=1):
"""Advance n steps: halo refresh, then core pull/collide, per step."""
"""Advance n steps: halo refresh, then core pull/collide, per step.
All n steps are enqueued in order on the one stream and the host waits
once before returning (same launches and bits as n step(1) calls).
Asynchronous device errors surface at that single wait, so they are not
attributed to the step that caused them; call step(1) repeatedly to
localize one. On any failure the instance is failed and step_count keeps
its entry value.
"""
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
kw = dict(block=self._block, stream=self._stream)
try:
self._status = "advancing"
@@ -224,9 +234,9 @@ class PeriodicBlockStepper:
self._step(self._current, self._spare, np.uint32(self.layout.halo),
self._tables.get("flags", np.uintp(0)), np.uintp(0), # no kind
grid=self._grid, **kw)
self._stream.synchronize()
self._current, self._spare = self._spare, self._current
self._step_count += 1
self._current, self._spare = self._spare, self._current # host handles only
self._stream.synchronize()
self._step_count += n
self._status = "ready"
except BaseException:
self._status = "failed"
+21 -7
View File
@@ -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
-101
View File
@@ -1,101 +0,0 @@
"""Synthetic checks of the annulus mass ledger. Host-only, stdlib; not solver evidence."""
import random
import sys
from pathlib import Path
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
import annulus_ledger as al # noqa: E402
NX = NY = 24
# Annulus 4 <= r <= 8 around (12, 12); dyadic values keep every sum exact.
A = {(x, y) for x in range(NX) for y in range(NY) if 16 <= (x - 12) ** 2 + (y - 12) ** 2 <= 64}
DELTA = 0.375
def owner(seed=1):
rng = random.Random(seed)
return [[[rng.randint(1, 1024) / 1024 for _ in range(NX)] for _ in range(NY)] for _ in range(9)]
def collide(P, source=None):
"""Mass-conserving stand-in: rotate populations within each cell (+ optional source)."""
D = [[[P[(q + 1) % 9][y][x] for x in range(NX)] for y in range(NY)] for q in range(9)]
if source:
(x, y), s = source
D[0][y][x] += s
return D
def link(interior):
"""(x, y, q) with x in A and x - c_q inside (seam) or outside (edge) A."""
for (x, y) in sorted(A):
for q in range(1, 9):
src = ((x - al.CX[q]) % NX, (y - al.CY[q]) % NY)
if (src in A) == interior:
return x, y, q
raise AssertionError
class Ledger(unittest.TestCase):
def setUp(self):
self.O = owner()
self.P = al.pull(self.O, NX, NY)
def run_ledger(self, P, D):
return al.ledger(self.O, P, D, A, NX, NY)
def test_ideal_pull_has_no_h_and_closes(self):
r = self.run_ledger(self.P, collide(self.P))
self.assertEqual((r["H_seam"], r["H_edge"], r["C"], r["residual"]), (0, 0, 0, 0))
self.assertEqual(r["dM"], -r["Q_exact"])
self.assertNotEqual(r["Q_exact"], 0)
def test_collision_source_is_c(self):
x, y = sorted(A)[3]
r = self.run_ledger(self.P, collide(self.P, ((x, y), DELTA)))
self.assertEqual((r["C"], r["H"], r["residual"]), (DELTA, 0, 0))
def test_interior_seam_perturbation(self):
x, y, q = link(interior=True)
self.P[q][y][x] += DELTA
r = self.run_ledger(self.P, collide(self.P))
self.assertEqual((r["H_seam"], r["H_edge"], r["residual"]), (DELTA, 0, 0))
# Invisible to any flux through the edge of A; only H explains the mass change.
self.assertEqual(al.staged_flux(self.O, self.P, A, NX, NY), r["Q_exact"])
self.assertEqual(r["dM"] + r["Q_exact"], DELTA)
def test_edge_perturbation_and_staged_flux_double_count(self):
x, y, q = link(interior=False)
self.P[q][y][x] += DELTA
r = self.run_ledger(self.P, collide(self.P))
self.assertEqual((r["H_seam"], r["H_edge"], r["residual"]), (0, DELTA, 0))
staged = al.staged_flux(self.O, self.P, A, NX, NY)
self.assertEqual(staged, r["Q_exact"] - DELTA)
# Staged flux + full H counts the staging change twice.
self.assertEqual(r["dM"] + staged - r["C"] - r["H"], -DELTA)
# Staged flux alone closes only without the edge part of H.
self.assertEqual(r["dM"] + staged - r["C"] - r["H_seam"], 0)
def test_owner_snapshot_after_staging_hides_the_change(self):
# Snapshot taken after staging wrote the same slot into the owner image.
x, y, q = link(interior=False)
sx, sy = (x - al.CX[q]) % NX, (y - al.CY[q]) % NY
self.P[q][y][x] += DELTA
late = [[row[:] for row in plane] for plane in self.O]
late[q][sy][sx] += DELTA
r = al.ledger(late, self.P, collide(self.P), A, NX, NY)
self.assertEqual((r["H"], r["residual"]), (0, 0))
def test_wrong_pull_direction_is_flagged(self):
# Indexing error (push instead of pull): H is nonzero with no physical cause.
bad = [[[self.O[q][(y + al.CY[q]) % NY][(x + al.CX[q]) % NX] for x in range(NX)]
for y in range(NY)] for q in range(9)]
r = self.run_ledger(bad, collide(bad))
self.assertNotEqual(r["H"], 0)
self.assertEqual(r["residual"], 0)
if __name__ == "__main__":
unittest.main()
+104
View File
@@ -0,0 +1,104 @@
"""Host contract of PeriodicBlockStepper.step(n): one stream wait per call (mocked driver)."""
import ast
from contextlib import nullcontext
import importlib.util
from pathlib import Path
import types
from unittest.mock import Mock
import unittest
import numpy as np
from CelerisLab.config import LBMConfig
from CelerisLab.lbm.blocks import BlockLayout2D
def host_module():
"""block_stepper with pycuda/compiler replaced by mocks; no GPU context."""
# Locate the source without importing it: importing block_stepper loads pycuda.driver,
# which CPU-only import-boundary tests (test_build_contracts) must never see.
path = Path(importlib.util.find_spec("CelerisLab.lbm.block_stepper").origin)
tree = ast.parse(path.read_text())
tree.body = [node for node in tree.body if not (
isinstance(node, ast.Import) and any(a.name == "pycuda.driver" for a in node.names)
or isinstance(node, ast.ImportFrom) and node.module == "cuda")]
module = types.ModuleType("CelerisLab.lbm._test_host_block_stepper")
module.__package__ = "CelerisLab.lbm"
module.__dict__.update(cuda=Mock(), compiler=Mock())
exec(compile(tree, str(path), "exec"), module.__dict__)
return module
def backend():
host, log = host_module(), []
host.cuda.mem_get_info.return_value = (1 << 40, 1 << 40)
host.cuda.mem_alloc.side_effect = lambda size: Mock(size=size) # distinct buffers
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[:2])))
host.compiler.load_module.return_value = module
cfg = LBMConfig(nx=16, ny=16, viscosity=.1)
run = host.PeriodicBlockStepper(cfg, BlockLayout2D.build(16, 16, 8, 8),
context=Mock(activate=nullcontext, sm_arch="sm_70"))
run._stream.synchronize.side_effect = lambda: log.append(("sync", ()))
run.initialize(np.full((9, 16, 16), 1 / 9))
log.clear()
return run, log
class BlockStepSync(unittest.TestCase):
def test_step_n_waits_once_and_matches_n_single_steps(self):
many, log_many = backend()
many.step(3)
single, log_single = backend()
for _ in range(3):
single.step(1)
names = [name for name, _ in log_many]
self.assertEqual(names.count("sync"), 1)
self.assertEqual(names[-1], "sync")
self.assertEqual(names.count("BlockStep"), 3)
# Same launch order and buffer roles as three step(1) calls.
strip = lambda log: [(n, tuple(id(x) for x in a)) for n, a in log if n != "sync"]
roles = lambda run, log: [(n, tuple({id(run._current): "c", id(run._spare): "s"}.get(x, "-")
for x in ids)) for n, ids in strip(log)]
self.assertEqual([n for n, _ in strip(log_many)], [n for n, _ in strip(log_single)])
self.assertEqual(roles(many, log_many), roles(single, log_single))
self.assertIsNot(many._current, many._spare)
self.assertEqual((many.step_count, single.step_count), (3, 3))
def test_step_zero_launches_and_waits_for_nothing(self):
run, log = backend()
run.step(0)
self.assertEqual((log, run.step_count), ([], 0))
def test_failure_keeps_entry_step_count(self):
run, log = backend()
run.step(1)
run._stream.synchronize.side_effect = RuntimeError("device fault")
with self.assertRaises(RuntimeError):
run.step(2)
self.assertEqual((run.step_count, run._status), (1, "failed"))
def test_launch_failure_mid_batch_fails_closed_without_waiting(self):
run, log = backend()
calls = []
step = run._step
def failing(*args, **kwargs):
calls.append(1)
if len(calls) == 2:
raise RuntimeError("launch failure")
return step(*args, **kwargs)
run._step = failing
with self.assertRaises(RuntimeError):
run.step(3)
self.assertNotIn("sync", [name for name, _ in log])
self.assertEqual((run.step_count, run._status), (0, "failed"))
with self.assertRaises(RuntimeError):
run.download()
if __name__ == "__main__":
unittest.main()
+170
View File
@@ -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()
-75
View File
@@ -1,75 +0,0 @@
"""Per-hop mass ledger of a fluid cell set A (host-only, stdlib, D2Q9 pull scheme).
Diagnostic definition only; no solver code, no GPU. Arrays are indexed [q][y][x] on a
periodic nx x ny lattice; A is a set of (x, y) ACTIVE cells.
O owner populations before any staging (canonical owner slots, start of the hop)
P populations each cell actually pulls: P[q](x) is what x reads for direction q
(after halo copy, BFL/coalesce staging, ...)
D populations written by the cell after collision
With M0 = sum_{A,q} O and M1 = sum_{A,q} D (both owner snapshots of A):
M1 - M0 + Q_exact = C + H (exact)
C = sum_{x in A, q} (D_q(x) - P_q(x)) collision source
H = sum_{x in A, q} (P_q(x) - O_q(x - c_q)) actual minus ideal owner input,
split into H_seam (x - c_q in A: internal pulls, e.g. tile seams) and
H_edge (x - c_q not in A: inputs crossing the edge of A)
Q_exact = sum_{y in A, y+c_q not in A} O_q(y) - sum_{y not in A, y+c_q in A} O_q(y)
owner-based net outward flux (outgoing minus incoming)
Proof: sum_{A,q} O_q(x - c_q) - M0 = -Q_exact (relabel y = x - c_q per direction), and
sum_{A,q} D = C + H + sum_{A,q} O_q(x - c_q). Q_exact must use O, not staged values: a
flux built from staged inputs already contains H_edge, so adding the full H counts it
twice (see staged_flux). The edge of A must not cross cut links, otherwise O at the solid
end enters Q_exact.
"""
CX = (0, 1, -1, 0, 0, 1, -1, 1, -1)
CY = (0, 0, 0, 1, -1, 1, -1, -1, 1)
def _src(x, y, q, nx, ny):
return (x - CX[q]) % nx, (y - CY[q]) % ny
def pull(O, nx, ny):
"""Ideal pulled input: P[q](x) = O[q](x - c_q)."""
return [[[O[q][_src(x, y, q, nx, ny)[1]][_src(x, y, q, nx, ny)[0]]
for x in range(nx)] for y in range(ny)] for q in range(9)]
def ledger(O, P, D, A, nx, ny):
A = set(A)
out = dict(dM=0.0, C=0.0, H_seam=0.0, H_edge=0.0, Q_exact=0.0)
for (x, y) in A:
for q in range(9):
out["dM"] += D[q][y][x] - O[q][y][x]
out["C"] += D[q][y][x] - P[q][y][x]
sx, sy = _src(x, y, q, nx, ny)
key = "H_seam" if (sx, sy) in A else "H_edge"
out[key] += P[q][y][x] - O[q][sy][sx]
if (sx, sy) not in A:
out["Q_exact"] -= O[q][sy][sx] # incoming owner population
tx, ty = (x + CX[q]) % nx, (y + CY[q]) % ny
if (tx, ty) not in A:
out["Q_exact"] += O[q][y][x] # outgoing owner population
out["H"] = out["H_seam"] + out["H_edge"]
out["residual"] = out["dM"] + out["Q_exact"] - out["C"] - out["H"]
return out
def staged_flux(O, P, A, nx, ny):
"""Outward flux with incoming terms from the staged inputs P (the wrong combination)."""
A = set(A)
q_out = 0.0
for (x, y) in A:
for q in range(9):
sx, sy = _src(x, y, q, nx, ny)
if (sx, sy) not in A:
q_out -= P[q][y][x]
tx, ty = (x + CX[q]) % nx, (y + CY[q]) % ny
if (tx, ty) not in A:
q_out += O[q][y][x]
return q_out