diff --git a/src/CelerisLab/lbm/stepper.py b/src/CelerisLab/lbm/stepper.py index d141688..facfe54 100644 --- a/src/CelerisLab/lbm/stepper.py +++ b/src/CelerisLab/lbm/stepper.py @@ -20,6 +20,9 @@ class LBMStepper: self.cfg = cfg self._esopull = (cfg.streaming == "esopull") + # Set by the owner when a step would be wrong (translating circles + # excluded from the packed links before the moving kernels exist). + self.blocked: str | None = None if self._esopull: self.step_fn = module.get_function("EsoPullStep") @@ -70,6 +73,8 @@ class LBMStepper: stream: cuda.Stream | None = None, ): """Advance *n* time steps.""" + if n > 0 and self.blocked is not None: + raise RuntimeError(self.blocked) f = self.field launch_kw = {} if stream is not None: diff --git a/src/CelerisLab/simulation.py b/src/CelerisLab/simulation.py index bd4ed22..fc8f7d1 100644 --- a/src/CelerisLab/simulation.py +++ b/src/CelerisLab/simulation.py @@ -27,6 +27,10 @@ from .config import LBMConfig, BodyConfig, load_lbm_config, load_body_config from .cuda.context import CudaContext, context_bound from .cuda import compiler from .lbm.descriptors import FLUID + +# S1: translating circles are declared and validated, but no moving kernel exists yet. +MOVING_BLOCKED = ("translating circles need the moving kernels (not implemented); " + "positive-step runs are rejected") from .lbm.field import LBMField from .lbm.stepper import LBMStepper from .body.objects import SimObject @@ -144,6 +148,7 @@ class Simulation: from .body.motion import MovingCircle obj.motion = MovingCircle((cx, cy), radius, translation, on_failure=self._fail_moving) + self.stepper.blocked = MOVING_BLOCKED elif type_lower == "sensor": from .body.geometry.circle import CircleGeometry cx, cy = float(center[0]), float(center[1]) @@ -511,6 +516,7 @@ class Simulation: self.field, self._module, self.lbm_cfg, ) self.stepper._step_count = _prev_step_count + self.stepper.blocked = MOVING_BLOCKED if self._moving_bodies() else None self._assert_object_count_contract(expected_count=expected_n_objects) @@ -537,6 +543,7 @@ class Simulation: # change. Compact-list construction publishes no outputs on failure. next_flags = self.bodies.build_flags(self.field.build_channel_flags()) self._validate_moving_bodies() + self.stepper.blocked = MOVING_BLOCKED if self._moving_bodies() else None if self.bodies.jets and self.bodies.count == 0: raise ValueError("Jets need bodies with cut links") self.bodies.build_compact_lists_for(self.bodies.objects, next_flags) @@ -670,8 +677,7 @@ class Simulation: if steps < 0 or checkpoint_interval < 0: raise ValueError("steps and checkpoint_interval must be nonnegative") if steps > 0 and self._moving_bodies(): - raise RuntimeError("translating circles need the moving kernels (not implemented); " - "positive-step runs are rejected") + raise RuntimeError(MOVING_BLOCKED) self._assert_runtime_contracts() # Discard any uncommitted body edits before stepping. if self.bodies.has_pending_edit(): diff --git a/tests/unit/test_moving_bodies.py b/tests/unit/test_moving_bodies.py index 7059b42..3f26682 100644 --- a/tests/unit/test_moving_bodies.py +++ b/tests/unit/test_moving_bodies.py @@ -103,27 +103,70 @@ def moving_simulation(monkeypatch, *, moving=True): center=(50., 40.), radius=10.) writes = [] sim._failed, sim._initialized = None, True - sim.bodies = SimpleNamespace(objects=[obj], count=1, jets=[], + sim.bodies = SimpleNamespace(objects=[obj], count=1, jets=[], action_gpu=None, obs_gpu=None, has_pending_edit=lambda: False, _upload_action_async=lambda s: writes.append("action"), - zero_obs_async=lambda s: writes.append("obs")) - sim.stepper = SimpleNamespace(step=lambda *a, **k: writes.append("step")) + zero_obs_async=lambda s: writes.append("obs"), + increment_obs_steps=lambda n: writes.append("count"), + download_obs_full_async=lambda s: writes.append("download")) + sim.stepper = SimpleNamespace(step=lambda n, **k: writes.append(("step", n)), blocked=None) + sim._assert_runtime_contracts = lambda: None return sim, writes +STREAM = SimpleNamespace(synchronize=lambda: None) + + def test_positive_steps_rejected_before_device_writes(monkeypatch): sim, writes = moving_simulation(monkeypatch) sim.bodies.objects[0].motion.set_target((0., .1), (0., .01)) - for call in (lambda: sim.run(1), lambda: sim.run(5), lambda: sim.step(), - lambda: sim.run(3, checkpoint_interval=1)): + sim.bodies.objects[0].motion.applied = (0., 0.) + for call in (lambda: sim.run(1, stream=STREAM), lambda: sim.run(5, stream=STREAM), + lambda: sim.step(), lambda: sim.run(3, stream=STREAM, checkpoint_interval=1)): with pytest.raises(RuntimeError, match="moving kernels"): call() with pytest.raises(RuntimeError, match="moving kernels"): # nominal at rest as well sim.bodies.objects[0].motion.pending = None sim.run(1) + sim.bodies.objects[0].state = SimpleNamespace(omega=.02) # spin does not lift the guard + with pytest.raises(RuntimeError, match="moving kernels"): + sim.run(2, stream=STREAM) assert writes == [] and sim.bodies.objects[0].motion.applied == (0., 0.) +def test_run_zero_keeps_pending_and_static_runs_proceed(monkeypatch): + sim, writes = moving_simulation(monkeypatch) + motion = sim.bodies.objects[0].motion + motion.set_target((0., .1), (0., .01)) + sim.run(0, stream=STREAM) # existing action/obs work only; no motion upload or commit + assert motion.pending == ((0., .1), (0., .01)) and motion.applied == (0., 0.) + assert ("step", 0) in writes + static, writes = moving_simulation(monkeypatch, moving=False) + static.save_checkpoint = lambda *a: writes.append("checkpoint") + static.run(3, stream=STREAM) + static.run(4, stream=STREAM, checkpoint_interval=2) + assert writes.count(("step", 3)) == 1 and writes.count(("step", 2)) == 2 + + +def test_direct_stepper_entry_is_guarded(): + from CelerisLab.lbm.stepper import LBMStepper + launches = [] + module = SimpleNamespace(get_function=lambda name: lambda *a, **k: launches.append(name)) + cfg = SimpleNamespace(streaming="double_buffer", threads_per_block=128, is_d2q9=True, + nx=64, ny=32, nz=1) + field = SimpleNamespace(ddf_gpu="ddf", temp_gpu="temp", n_curved=0) + stepper = LBMStepper(field, module, cfg) + assert stepper.blocked is None + stepper.blocked = "translating circles need the moving kernels" + for n in (1, 3): + with pytest.raises(RuntimeError, match="moving kernels"): + stepper.step(n, action_gpu=None, obs_gpu=None) + assert launches == [] and (field.ddf_gpu, field.temp_gpu) == ("ddf", "temp") + assert stepper.step_count == 0 + stepper.step(0, action_gpu=None, obs_gpu=None) # zero steps stay a no-op + assert launches == [] and stepper.step_count == 0 + + def test_moving_body_rejects_topology_and_checkpoint(monkeypatch): sim, _ = moving_simulation(monkeypatch) for call in (sim.sync_bodies, sim.save_checkpoint, lambda: sim.load_checkpoint("x.h5")): @@ -155,6 +198,7 @@ def body(x, y, r, translation=None, **kw): def routed_simulation(monkeypatch): sim = make_simulation(monkeypatch) sim._failed, sim._initialized, sim.lbm_cfg = None, False, Cfg() + sim.stepper = SimpleNamespace(blocked=None) sim.bodies = ObjectManager(nx=120, ny=80, nz=1, nq=9, cfg=Cfg()) return sim @@ -164,6 +208,7 @@ def test_add_body_and_set_body_route_targets(monkeypatch): bid = sim.add_body("circle", (50., 40.), 10., translation=Translation((-2., 2., -3., 3.), .1)) still = sim.add_body("circle", (100., 40.), 5.) motion = sim.bodies.get(bid).motion + assert sim.stepper.blocked is not None # declaring a moving circle blocks direct steps sim.set_body(bid, offset=(0., .3), vy=.05) sim.set_body(bid, offset=(0., .4), vy=.05) # last wins for kwargs in (dict(offset=(0., .6)), dict(vy=.05), dict(vx=.05), @@ -229,14 +274,18 @@ def test_initialize_validates_before_mutation(monkeypatch): def test_moving_circle_excluded_from_static_links(): - still = [body(100., 40., 5.)] - mgr = ObjectManager(nx=120, ny=80, nz=1, nq=9, cfg=Cfg()) - reference = mgr.build_compact_lists_for(still) - with_moving = mgr.build_compact_lists_for( - still + [body(50., 40., 10., Translation((0., 0., -1., 1.), .1))]) - for a, b in zip(reference[:8], with_moving[:8]): + def manager(*objects): + mgr = ObjectManager(nx=120, ny=80, nz=1, nq=9, cfg=Cfg()) + ids = [mgr.add(obj) for obj in objects] + return mgr, ids + reference_mgr, _ = manager(body(100., 40., 5.), body(20., 40., 6.)) + mixed_mgr, ids = manager(body(100., 40., 5.), body(20., 40., 6.), + body(55., 40., 10., Translation((0., 0., -1., 1.), .1))) + reference = reference_mgr.build_compact_lists_for(reference_mgr.objects) + mixed = mixed_mgr.build_compact_lists_for(mixed_mgr.objects) + for a, b in zip(reference[:8], mixed[:8]): # static link columns byte-identical assert np.asarray(a).tobytes() == np.asarray(b).tobytes() - assert len(reference[0]) > 0 + assert len(reference[0]) > 0 and ids[2] not in set(np.asarray(mixed[3]).tolist()) def test_failed_moving_step_marks_simulation_failed(monkeypatch): @@ -250,7 +299,12 @@ def test_failed_moving_step_marks_simulation_failed(monkeypatch): with pytest.raises(RuntimeError, match="sync failed"): motion.advance(lambda *_: None, lambda: None, fail) assert sim._failed is not None and motion.applied == (0., 0.) - for call in (lambda: sim.run(0), lambda: sim.run(1), sim.step, sim.sync_bodies): + for call in (lambda: sim.run(0), lambda: sim.run(1), sim.step, sim.sync_bodies, + sim.save_checkpoint, lambda: sim.load_checkpoint("x.h5")): with pytest.raises(RuntimeError, match="Simulation failed"): call() - sim.close() # close still works; no re-entry afterwards + with pytest.raises(RuntimeError, match="unknown"): # no retry on the failed circle + motion.advance(lambda *_: None, lambda: None, lambda: None) + sim.close() # close still works + with pytest.raises(Exception): # no re-entry after close + sim.run(1)