From 4bcf5ef10b5e3df2946f7d5b367f57cdef9b9f0a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 31 Oct 2026 01:46:11 +0000 Subject: [PATCH] run: one fail-closed device transaction with translating circles Split BodyManager action upload into host-only _prepare_action_upload (validation + pinned staging) and _upload_prepared_action_async (H2D); _upload_action_async stays the checked wrapper for direct callers. Simulation.run prepares outside the failure boundary, then runs action upload, obs/diagnostic zeroing, step group, readback and sync as one transaction: with translating circles or attached moving tables any exception fails every circle and the Simulation (also for run(0) and an empty registry). The inner _run_moving catch is removed (its only caller is run). Stationary runs keep their failure policy; pending/hold-consume and run(0) success behaviour unchanged. No kernel change. Co-Authored-By: Claude Opus 5.5 --- docs/moving_bodies.md | 6 +-- src/CelerisLab/body/manager.py | 15 ++++++- src/CelerisLab/simulation.py | 64 ++++++++++++++++----------- tests/unit/test_moving_bodies.py | 75 +++++++++++++++++++++++++++++++- 4 files changed, 128 insertions(+), 32 deletions(-) diff --git a/docs/moving_bodies.md b/docs/moving_bodies.md index ffab1f0..48ddeb7 100644 --- a/docs/moving_bodies.md +++ b/docs/moving_bodies.md @@ -170,9 +170,9 @@ acts as a sliding wall. One moving step consumes one target. | `run(1)` without pending | hold: `c_old == c_new`, no cover/uncover, links at `applied`, translational `u_w = 0` | | `run(n > 1)` with pending | `ValueError` before mutation (single-step-only moving API) | | `run(n > 1)`, no pending | allowed: hold for n steps, translational `u_w = 0`, spin only | -| `run(0)` | no motion upload, no commit; `pending` kept | -| validation failure | `ValueError`, nothing mutated, retry allowed | -| upload/launch/sync failure after enqueue | simulation unusable; `applied` and `pending` are not reported as device truth; no re-entry | +| `run(0)` | no motion upload, no commit; `pending` kept; action upload and obs/diagnostic zeroing still run | +| validation failure (incl. action validation and pinned staging in `run`) | `ValueError`/error before any device work, nothing mutated, retry allowed | +| failure of any device work in `run` (action upload, obs/diagnostic zeroing, record upload, launch, readback, sync), also for `run(0)` | simulation unusable, every translating circle failed; `applied` and `pending` are not reported as device truth; no re-entry | Displacement norm: Euclidean per-step `|c_new - c_old|_2 ≤ 0.5` lattice cells (declared bound; CANDIDATE value). diff --git a/src/CelerisLab/body/manager.py b/src/CelerisLab/body/manager.py index 188f6da..65a0d79 100644 --- a/src/CelerisLab/body/manager.py +++ b/src/CelerisLab/body/manager.py @@ -629,10 +629,17 @@ class ObjectManager: cuda.memcpy_dtoh_async(self.obs_pinned, self.obs_gpu, stream) def _upload_action_async(self, stream: cuda.Stream) -> None: - """Async H2D of host action array to action_gpu. + """Checked async H2D of host action array to action_gpu.""" + if self._prepare_action_upload(): + self._upload_prepared_action_async(stream) + + def _prepare_action_upload(self) -> bool: + """Validate the action array and stage it in pinned memory; no device work. Single gate for every action path (setters or direct ``action`` edits): surface-actuated and Jet-driven bodies must keep zero motion slots. + Returns False when there is nothing to upload. Restaging assumes no + copy from ``_action_pinned`` is in flight (``run`` ends with a sync). """ slot = 3*(self.cfg.dim if self.cfg else 2) for obj in self.objects: @@ -643,10 +650,14 @@ class ObjectManager: self.validate_jet_motion(obj.obj_id, {"vx": self.action[base], "vy": self.action[base+1], "omega": self.action[base+slot-1]}) if self.action_gpu is None or self.action.size == 0: - return + return False if self._action_pinned is None or self._action_pinned.size != self.action.size: self._action_pinned = cuda.pagelocked_empty(self.action.size, np.float32) self._action_pinned[:] = self.action + return True + + def _upload_prepared_action_async(self, stream: cuda.Stream) -> None: + """Enqueue H2D of the staged ``_action_pinned`` (see ``_prepare_action_upload``).""" cuda.memcpy_htod_async(self.action_gpu, self._action_pinned, stream) @property diff --git a/src/CelerisLab/simulation.py b/src/CelerisLab/simulation.py index a7e1195..b14a997 100644 --- a/src/CelerisLab/simulation.py +++ b/src/CelerisLab/simulation.py @@ -216,6 +216,12 @@ class Simulation: self._failed = f"moving circle step failed: {exc!r}" self.stepper.blocked = self._failed # direct stepper calls are refused as well + def _fail_moving_run(self, exc: BaseException) -> None: + # Device work was attempted with translating circles (or their tables) present. + for obj in self._moving_bodies(): + obj.motion.fail(exc) + self._fail_moving(exc) # also when tables are attached but no circle is registered + def _attach_moving(self) -> None: # Direct stepper calls stay blocked until the moving tables exist. tables = getattr(self, "_moving_tables", None) @@ -734,9 +740,20 @@ class Simulation: if stream is None: stream = self.stream - # Async upload action - if upload_act and self.bodies.count > 0: - self.bodies._upload_action_async(stream) + # Host-only action validation and staging: a rejection here stays retryable. + staged = upload_act and self.bodies.count > 0 and self.bodies._prepare_action_upload() + watched = bool(moving) or self.stepper.moving is not None + try: + self._run_steps(steps, stream, staged, sync_obs, zero_obs, checkpoint_interval) + except BaseException as exc: + if watched: # device work was attempted: motion state is unknown + self._fail_moving_run(exc) + raise + + def _run_steps(self, steps, stream, staged, sync_obs, zero_obs, checkpoint_interval): + """Device part of ``run``: action upload, obs zeroing, step group, sync.""" + if staged: + self.bodies._upload_prepared_action_async(stream) # Zero obs segments before step group if zero_obs: @@ -783,33 +800,28 @@ class Simulation: def _run_moving(self, steps, stream, sync_obs): """Steps with translating circles: hold, or one advancing step that commits each pending target only after synchronization and a zero ``no_donor`` count - (docs/moving_bodies.md 4a, 5). Any failure leaves the Simulation unusable.""" + (docs/moving_bodies.md 4a, 5). ``run`` fails the Simulation on any exception.""" tables = self.stepper.moving circles = [(obj.motion, obj.obj_id) for obj in self._moving_bodies()] pending = any(motion.pending is not None for motion, _ in circles) - try: - advancing = tables.upload_records(circles, stream) - self.stepper._step(steps, action_gpu=self.bodies.action_gpu, - obs_gpu=self.bodies.obs_gpu, stream=stream, advance=advancing) - self.bodies.increment_obs_steps(steps) - if sync_obs: - self.bodies.download_obs_full_async(stream) - stream.synchronize() + advancing = tables.upload_records(circles, stream) + self.stepper._step(steps, action_gpu=self.bodies.action_gpu, + obs_gpu=self.bodies.obs_gpu, stream=stream, advance=advancing) + self.bodies.increment_obs_steps(steps) + if sync_obs: + self.bodies.download_obs_full_async(stream) + stream.synchronize() + if advancing: + no_donor = tables.download_diagnostics()[:, tables.DIAG.index("no_donor")] + if np.any(no_donor > 0): + raise RuntimeError("translating circle refill found no donor cell") + if pending: # a target equal to the applied offset is consumed as a hold step + for motion, _ in circles: + motion.commit() if advancing: - no_donor = tables.download_diagnostics()[:, tables.DIAG.index("no_donor")] - if np.any(no_donor > 0): - raise RuntimeError("translating circle refill found no donor cell") - if pending: # a target equal to the applied offset is consumed as a hold step - for motion, _ in circles: - motion.commit() - if advancing: - tables.upload_records(circles, stream) # first hold: old = new = applied, u = 0 - stream.synchronize() - self._attach_moving() - except BaseException as exc: - for motion, _ in circles: # device geometry/state unknown after any failure - motion.fail(exc) - raise + tables.upload_records(circles, stream) # first hold: old = new = applied, u = 0 + stream.synchronize() + self._attach_moving() def step(self, n: int = 1): """Advance *n* steps (convenience for interactive use).""" diff --git a/tests/unit/test_moving_bodies.py b/tests/unit/test_moving_bodies.py index 1c4f418..9f9119e 100644 --- a/tests/unit/test_moving_bodies.py +++ b/tests/unit/test_moving_bodies.py @@ -113,7 +113,8 @@ def moving_simulation(monkeypatch, *, moving=True): sim._failed, sim._initialized = None, True 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"), + _prepare_action_upload=lambda: True, + _upload_prepared_action_async=lambda s: writes.append("action"), 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")) @@ -361,6 +362,78 @@ def test_failed_advance_fails_simulation(monkeypatch, failure): sim.run(1, stream=STREAM) +def raising(log, name, site): + def call(*a, **k): + log.append(name) + if name == site: + raise RuntimeError(f"{name} failed") + return call + + +def two_circle_simulation(monkeypatch, site): + # No owner hooks: run itself must fail the Simulation and every circle. + sim, writes, first = advancing_simulation(monkeypatch) + second = SimpleNamespace(motion=circle(), obj_id=1, center=(80., 40.), radius=10.) + sim.bodies.objects.append(second) + sim.bodies._upload_prepared_action_async = raising(writes, "action", site) + sim.bodies.zero_obs_async = raising(writes, "obs", site) + sim.bodies.download_obs_full_async = raising(writes, "download", site) + sim.stepper.moving.zero_diagnostics = raising(writes, "zero_diag", site) + step = raising(writes, "step", site) + sim.stepper._step = lambda n, advance=False, **k: step() + return sim, (first, second.motion), SimpleNamespace(synchronize=raising(writes, "sync", site)) + + +@pytest.mark.parametrize("steps,site", [(n, s) for n in (0, 1, 3) + for s in ("action", "obs", "zero_diag", "step", "download", "sync") + if not (n == 0 and s == "step")]) # run(0) launches nothing +def test_device_failure_in_run_fails_every_circle(monkeypatch, steps, site): + sim, motions, stream = two_circle_simulation(monkeypatch, site) + if steps < 2: + motions[0].set_target((0., .25), (0., .05)) + with pytest.raises(RuntimeError, match=f"{site} failed"): + sim.run(steps, stream=stream) + assert all(m.failed for m in motions) and sim._failed is not None + assert sim.stepper.blocked == sim._failed + assert motions[0].applied == (0., 0.) # nothing committed before the first sync + with pytest.raises(RuntimeError, match="Simulation failed"): + sim.run(1, stream=STREAM) + + +def test_action_rejection_stays_retryable_before_device_work(monkeypatch): + sim, writes, motion = advancing_simulation(monkeypatch) + motion.set_target((0., .25), (0., .05)) + + def reject(): + raise ValueError("bad action") + sim.bodies._prepare_action_upload = reject + for steps in (0, 1): + with pytest.raises(ValueError, match="bad action"): + sim.run(steps, stream=STREAM) + assert writes == [] and sim._failed is None and sim.stepper.blocked is None + assert not motion.failed and motion.pending is not None + sim.bodies._prepare_action_upload = lambda: True + sim.run(1, stream=STREAM) + assert motion.applied == (0., .25) and motion.pending is None + + +def test_attached_tables_without_circles_fail_closed(monkeypatch): + sim, writes, _ = advancing_simulation(monkeypatch) + sim.bodies.objects = [] + sim.stepper.moving.zero_diagnostics = raising(writes, "zero_diag", "zero_diag") + with pytest.raises(RuntimeError, match="zero_diag failed"): + sim.run(0, stream=STREAM) + assert sim._failed is not None and sim.stepper.blocked == sim._failed + + +def test_stationary_run_failure_policy_unchanged(monkeypatch): + sim, writes = moving_simulation(monkeypatch, moving=False) + sim.bodies.zero_obs_async = raising(writes, "obs", "obs") + with pytest.raises(RuntimeError, match="obs failed"): + sim.run(2, stream=STREAM) + assert sim._failed is None and sim.stepper.blocked is None + + def test_failed_hold_step_blocks_the_direct_stepper(monkeypatch): # No target is pending, so only the failure latch can refuse direct steps. from CelerisLab.lbm.stepper import LBMStepper