diff --git a/apps/studio/src/orisys_studio/scene3d/coordinator.py b/apps/studio/src/orisys_studio/scene3d/coordinator.py index 4847e83..6e77c24 100644 --- a/apps/studio/src/orisys_studio/scene3d/coordinator.py +++ b/apps/studio/src/orisys_studio/scene3d/coordinator.py @@ -20,13 +20,25 @@ Staleness is about *identity*, not timing: a frame from another generation, anot session or another source than the active scene is dropped with ``stale_frame`` and counted, and an old result can never overwrite a new scene because the composer's result carries the identity it was built from and is checked again on arrival. + +**A withdrawn display claim is an epoch, not a coincidence.** ``Future.cancel()`` +cannot stop a function that has already started, so "stop", "hide", "new session" and +"new scene" each bump an epoch that the running compose captured when it started. The +work itself only *returns* its frame; the GUI thread accepts it, and only if the epoch +and the identity it carries are still the current ones. Without that, a compose that +was already running when the operator pressed stop finished a moment later and put the +old measurement back on screen while the view said "live" again — the exact bug the +epoch exists to prevent. The slot stays occupied until the work really finishes, so +repeated hiding and showing cannot queue up composes behind a slow one. """ from __future__ import annotations import threading -from concurrent.futures import Executor, Future -from typing import Any, Optional +import time +from concurrent.futures import CancelledError, Executor, Future +from dataclasses import dataclass +from typing import Any, Callable, Optional from orisys_app_shared.scene3d.contracts import ( Diagnostic, @@ -46,6 +58,25 @@ from orisys_app_shared.scene3d.contracts import ( #: counter is informational, never a reason to throttle acquisition. MAX_INFLIGHT_COMPOSES = 1 +#: How long a measurement may fail to arrive before the display stops claiming it is +#: live. One dropped frame is not a reason to blank the picture; a stream of them is. +DEFAULT_STALE_GRACE_S = 1.0 + + +@dataclass(frozen=True) +class _Inflight: + """One compose task, and the display claim it was started under. + + Kept as a whole rather than as a bare future so that "which claim does this + result belong to?" can be asked *before* the result is published, and so that a + future which refused to cancel still occupies the single compose slot. + """ + + future: Future + epoch: int + identity: FrameIdentity + started: float + class ThreeDCoordinator: """Owns the two mailboxes, the compose worker and the scene generation.""" @@ -56,28 +87,42 @@ class ThreeDCoordinator: composer: Any, executor: Executor, max_fps: int = 30, + *, + clock: Callable[[], float] = time.monotonic, + stale_grace_s: float = DEFAULT_STALE_GRACE_S, ) -> None: self._widget = widget self._composer = composer self._executor = executor + self._clock = clock + self._stale_grace_s = max(0.0, float(stale_grace_s)) self._max_fps = max(1, int(max_fps)) self._incoming: LatestFrameMailbox[SurfaceFieldFrame] = LatestFrameMailbox() self._composed: LatestFrameMailbox[RenderFrame] = LatestFrameMailbox() self._lock = threading.Lock() self._state: SceneState = "disabled" self._generation = 0 + #: Which display claim the frames belong to. Bumped whenever the claim is + #: withdrawn: a stop, a hide, a new session, a new scene. A compose result + #: whose epoch is not this one is thrown away instead of presented. + self._epoch = 0 self._assets: Optional[SceneAssets] = None self._source_id = "" self._session_id: Optional[int] = None + #: The newest request id seen in this epoch: a producer that falls behind + #: (a seek UI, a queue that reorders) must not put an older frame on screen. + self._last_request_id = 0 self._visible = False self._settings = DisplaySettings(max_fps=self._max_fps) - self._inflight: Optional[Future] = None + self._inflight: Optional[_Inflight] = None self._stale_frames = 0 self._invalid_frames = 0 self._last_diagnostic: Optional[Diagnostic] = None self._closed = False self._settings_dirty = False self._last_composed_identity: Optional[FrameIdentity] = None + #: What is on screen and when it got there, for the pane's frame counter. + self._last_presented: Optional[tuple[FrameIdentity, float]] = None #: The field frame currently on screen, so a gain change can redraw exactly #: it rather than a newer instant. self._last_field_frame: Optional[SurfaceFieldFrame] = None @@ -92,6 +137,15 @@ class ThreeDCoordinator: def generation(self) -> int: return self._generation + @property + def epoch(self) -> int: + """The display claim results are accepted under (see :class:`_Inflight`).""" + return self._epoch + + def last_presented(self) -> Optional[tuple[FrameIdentity, float]]: + """What is on screen, and the monotonic time it was handed to the widget.""" + return self._last_presented + @property def superseded_frames(self) -> int: """Frames the display never showed. Never a recording drop.""" @@ -121,6 +175,28 @@ class ThreeDCoordinator: self._state = state # --------------------------------------------------------------- lifetime + def _invalidate(self) -> None: + """Withdraw the current display claim: nothing in flight may change the view. + + Called by every transition that means "what was on screen is no longer being + claimed": stop, hide, a new session, a new scene, close. The epoch is what + makes a compose that is *already running* harmless — it will finish, its + result will be recognised as belonging to a withdrawn claim, and it will be + dropped. The single compose slot is deliberately **not** freed here: a future + that refused to cancel still owns it until it really finishes, so repeatedly + hiding and showing cannot build a queue behind a slow composer. + """ + self._epoch += 1 + self._incoming.clear() + self._composed.clear() + self._last_request_id = 0 + self._last_composed_identity = None + self._last_field_frame = None + self._settings_dirty = False + item = self._inflight + if item is not None: + item.future.cancel() + def activate(self, assets: SceneAssets, source_id: str) -> int: """Adopt a scene and a data source; returns the new generation. @@ -129,16 +205,11 @@ class ThreeDCoordinator: """ if self._closed: raise error("asset_rejected", "this coordinator has been closed") - self._cancel_inflight() - self._incoming.clear() - self._composed.clear() + self._invalidate() self._assets = assets self._source_id = str(source_id) self._session_id = None self._generation += 1 - self._last_composed_identity = None - self._last_field_frame = None - self._settings_dirty = False self._widget.set_scene(assets) if self._visible: self._widget.present(self._composer.rest_frame(assets, self._identity(0))) @@ -155,9 +226,7 @@ class ThreeDCoordinator: self._widget.setVisible(visible) if not visible: self._visible = False - self._incoming.clear() - self._composed.clear() - self._cancel_inflight() + self._invalidate() self._set_state("suspended") return self._visible = True @@ -179,14 +248,9 @@ class ThreeDCoordinator: """ if self._closed: return - self._cancel_inflight() - self._incoming.clear() - self._composed.clear() + self._invalidate() self._session_id = None self._generation += 1 - self._last_composed_identity = None - self._last_field_frame = None - self._settings_dirty = False if self._assets is not None: # The reference model, not the last measurement: the new session has not # produced anything yet, and an old picture would be a claim about it. @@ -226,6 +290,19 @@ class ThreeDCoordinator: scene=str(session_id), ) return "stale" + newest_request = self._last_request_id + if identity.request_id > 0 and identity.request_id < newest_request: + # Request ids are monotonic within a session, so an older one is a frame + # the producer has already superseded (a seek, a reordered queue). Showing + # it would move the picture backwards in time. + self._stale_frames += 1 + self._last_diagnostic = diagnostic( + "stale_frame", + "a frame with an older request id was dropped", + frame=str(identity.request_id), + scene=str(newest_request), + ) + return "stale" problem = self._reject_reason(frame, assets) if problem is not None: self._invalid_frames += 1 @@ -234,6 +311,8 @@ class ThreeDCoordinator: with self._lock: if session_id is None: self._session_id = identity.session_id + if identity.request_id > 0: + self._last_request_id = max(self._last_request_id, identity.request_id) return self._incoming.offer(frame) @staticmethod @@ -294,6 +373,7 @@ class ThreeDCoordinator: self._widget.set_superseded_frames(self.superseded_frames) self._widget.present(composed) self._last_composed_identity = composed.identity + self._last_presented = (composed.identity, self._clock()) self._set_state("live" if composed.valid.any() else "no_data") elif self._state in ("ready", "loading"): # Nothing has arrived yet: show the reference model rather than a blank @@ -327,10 +407,7 @@ class ThreeDCoordinator: """ if self._closed or self._assets is None or not self._visible: return - self._incoming.clear() - self._cancel_inflight() - self._last_field_frame = None - self._last_composed_identity = None + self._invalidate() self._widget.present(self._composer.rest_frame(self._assets, self._identity(0))) self._set_state("no_data") @@ -339,9 +416,7 @@ class ThreeDCoordinator: if self._closed: return self._closed = True - self._cancel_inflight() - self._incoming.clear() - self._composed.clear() + self._invalidate() self._set_state("disabled") # ------------------------------------------------------------------ internal @@ -360,6 +435,10 @@ class ThreeDCoordinator: identity.source_id == self._source_id and identity.generation == self._generation and (self._session_id is None or identity.session_id == self._session_id) + and not ( + identity.request_id > 0 + and identity.request_id < self._last_request_id + ) ) def _start_compose(self, frame: SurfaceFieldFrame) -> None: @@ -367,44 +446,45 @@ class ThreeDCoordinator: settings = self._settings if assets is None: return - generation = self._generation + epoch = self._epoch # Kept so a display-settings change can recompose *this* frame instead of # jumping to a newer one. self._last_field_frame = frame - def work() -> None: - try: - composed = self._composer.compose(assets, frame, settings) - except Exception as exc: # noqa: BLE001 - a failed compose is not a crash - self._last_diagnostic = getattr(exc, "diagnostic", None) or diagnostic( - "field_unavailable", f"compose failed: {type(exc).__name__}: {exc}" - ) - return - if composed.identity.generation != generation: - # The scene changed while this was running: the result is for a - # scene that no longer exists. - self._stale_frames += 1 - return - self._composed.offer(composed) + def work() -> RenderFrame: + # No mailbox write here: the result is accepted on the GUI thread, where + # the current epoch can be compared without racing the operator's stop. + return self._composer.compose(assets, frame, settings) - self._inflight = self._executor.submit(work) + self._inflight = _Inflight( + future=self._executor.submit(work), + epoch=epoch, + identity=frame.identity, + started=self._clock(), + ) def _collect_inflight(self) -> None: - future = self._inflight - if future is None: - return - if not future.done(): + item = self._inflight + if item is None or not item.future.done(): return self._inflight = None try: - future.result() + composed = item.future.result() + except CancelledError: + # The work never started, so there is nothing to report: the epoch bump + # that cancelled it is already the reason the display moved on. + return except Exception as exc: # noqa: BLE001 self._last_diagnostic = getattr(exc, "diagnostic", None) or diagnostic( - "field_unavailable", f"compose raised: {type(exc).__name__}: {exc}" + "field_unavailable", f"compose failed: {type(exc).__name__}: {exc}" ) - - def _cancel_inflight(self) -> None: - future = self._inflight - self._inflight = None - if future is not None: - future.cancel() + return + if composed is None: + return + if item.epoch != self._epoch or not self._identity_is_current(composed.identity): + # A result from a claim that has been withdrawn (stop, hide, new session, + # new scene, older request). Presenting it would put a measurement back on + # screen for something that is no longer being measured. + self._stale_frames += 1 + return + self._composed.offer(composed) diff --git a/apps/studio/tests/test_scene3d_coordinator.py b/apps/studio/tests/test_scene3d_coordinator.py index 47b61bf..4680dfd 100644 --- a/apps/studio/tests/test_scene3d_coordinator.py +++ b/apps/studio/tests/test_scene3d_coordinator.py @@ -10,7 +10,8 @@ from __future__ import annotations import threading import time -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import Future, ThreadPoolExecutor +from dataclasses import replace import numpy as np import pytest @@ -415,3 +416,227 @@ def test_a_frame_already_queued_is_dropped_when_the_model_comes_back(scene, pool "a frame that was in flight when the pass stopped was presented anyway" ) coordinator.close() + + +# ------------------------------------------- a withdrawn claim, and the work behind it +class BlockingComposer(SceneComposer): + """A composer that has *entered* compose and waits for permission to finish. + + This is the shape the audit reproduced: the work is already running when the + operator stops the display, so ``Future.cancel()`` cannot stop it and only the + epoch can make its result harmless. + """ + + def __init__(self) -> None: + super().__init__() + self.entered = threading.Event() + self.release = threading.Event() + + def compose(self, assets, frame, settings): + self.entered.set() + assert self.release.wait(5.0), "the test never released the blocking composer" + return super().compose(assets, frame, settings) + + +class ImmediateExecutor: + """Runs the work inline: makes the schedule deterministic without threads.""" + + def submit(self, function): + future: Future = Future() + try: + future.set_result(function()) + except BaseException as exc: # noqa: BLE001 - mirrored into the future + future.set_exception(exc) + return future + + +class ManualExecutor: + """Futures that only run when the test says so (and honour a cancel first).""" + + def __init__(self) -> None: + self.pending: list[tuple[Future, object]] = [] + + def submit(self, function): + future: Future = Future() + self.pending.append((future, function)) + return future + + def run_next(self) -> None: + """Run the oldest task that was not cancelled, like a real worker would.""" + while self.pending: + future, function = self.pending.pop(0) + if future.cancelled(): + continue + if future.set_running_or_notify_cancel(): + try: + future.set_result(function()) + except BaseException as exc: # noqa: BLE001 - mirrored into the future + future.set_exception(exc) + return + + +def _start_blocked_compose(scene, pool, source="replay-A", value=5.0): + """Set up a coordinator whose first compose is running and blocked.""" + composer = BlockingComposer() + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, composer, pool) + coordinator.set_visible(True) + coordinator.activate(scene, source) + coordinator.submit(_frame(scene, source=source, value=value)) + coordinator.tick() + assert composer.entered.wait(3.0), "the compose never started" + assert coordinator._inflight is not None + return composer, widget, coordinator + + +def test_a_compose_running_when_the_pass_stopped_can_never_come_back(scene, pool): + """R1: "stopped, then the old worker finished and the pane said live again".""" + composer, widget, coordinator = _start_blocked_compose(scene, pool) + epoch = coordinator.epoch + + coordinator.show_rest() + assert coordinator.state == "no_data" + assert not widget.frames[-1].valid.any(), "stop did not clear the measurement" + + composer.release.set() + _drain(coordinator, ticks=40, sleep=0.01) + + assert coordinator.epoch == epoch + 1 + assert coordinator._inflight is None, "the finished work still owns the slot" + assert not widget.frames[-1].valid.any(), ( + "the stopped pass put its measurement back on screen" + ) + assert coordinator.state != "live" + coordinator.close() + + +def test_hiding_while_a_compose_runs_does_not_let_it_paint_later(scene, pool): + composer, widget, coordinator = _start_blocked_compose(scene, pool) + coordinator.set_visible(False) + assert coordinator.state == "suspended" + composer.release.set() + # Hidden: the tick that would present the result does not even run. + for _ in range(10): + coordinator.tick() + time.sleep(0.005) + coordinator.set_visible(True) + _drain(coordinator, ticks=40, sleep=0.01) + assert not widget.frames[-1].valid.any(), ( + "a compose started before the view was hidden painted after it came back" + ) + assert coordinator.state != "live" + coordinator.close() + + +def test_a_compose_running_for_the_old_session_is_dropped_by_begin_session(scene, pool): + composer, widget, coordinator = _start_blocked_compose(scene, pool) + coordinator.begin_session() + assert widget.frames[-1].scalar_descriptor.quantity == "unavailable" + composer.release.set() + _drain(coordinator, ticks=40, sleep=0.01) + assert not widget.frames[-1].valid.any(), ( + "the previous session's compose was presented into the new session" + ) + coordinator.close() + + +def test_a_compose_cancelled_before_it_started_is_not_an_error(scene): + """A future that never ran is not a failure, and it does not hold the slot.""" + executor = ManualExecutor() + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, SceneComposer(), executor) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + coordinator.submit(_frame(scene, value=4.0)) + coordinator.tick() + assert len(executor.pending) == 1 + + coordinator.show_rest() + assert executor.pending[0][0].cancelled() + coordinator.tick() + assert coordinator.last_diagnostic is None, "a cancelled compose was reported as a failure" + assert coordinator._inflight is None, "the cancelled work still owns the slot" + + coordinator.submit(_frame(scene, frame_id=2, value=4.0)) + coordinator.tick() + executor.run_next() + coordinator.tick() + assert coordinator.state == "live" + assert np.allclose(np.asarray(widget.frames[-1].scalar), 4.0) + coordinator.close() + + +def test_repeated_start_and_stop_never_leaves_an_old_measurement(scene, pool): + """Twenty stop/start cycles: one compose at a time, and never a reanimated one.""" + composer = CountingComposer(delay_s=0.005) + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, composer, pool) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + for index in range(20): + coordinator.begin_session() + coordinator.submit( + _frame( + scene, + frame_id=index, + value=float(index + 1), + generation=coordinator.generation, + ) + ) + coordinator.tick() + coordinator.show_rest() + assert not widget.frames[-1].valid.any(), f"cycle {index} kept a measurement" + _drain(coordinator, ticks=60, sleep=0.01) + assert not widget.frames[-1].valid.any(), "a stopped pass painted after the fact" + assert coordinator.state != "live" + assert composer.max_in_flight == 1 + coordinator.close() + + +def test_an_older_request_id_cannot_move_the_picture_backwards(scene): + """R7: request ids are monotonic within a session, and the pane must obey.""" + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, SceneComposer(), ImmediateExecutor()) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + newer = _frame(scene, frame_id=20, value=2.0) + newer = replace(newer, identity=replace(newer.identity, request_id=2)) + older = _frame(scene, frame_id=10, value=9.0) + older = replace(older, identity=replace(older.identity, request_id=1)) + + assert coordinator.submit(newer) == "accepted" + coordinator.tick() + coordinator.tick() + assert widget.frames[-1].identity.request_id == 2 + + assert coordinator.submit(older) == "stale" + assert coordinator.last_diagnostic is not None + assert coordinator.last_diagnostic.code == "stale_frame" + coordinator.tick() + coordinator.tick() + assert widget.frames[-1].identity.request_id == 2, "an older frame was shown" + assert np.allclose(np.asarray(widget.frames[-1].scalar), 2.0) + assert coordinator.stale_frames == 1 + coordinator.close() + + +def test_a_result_from_an_older_request_is_dropped_on_arrival(scene): + """The same rule on the way out: the composer's result carries the request id.""" + executor = ManualExecutor() + widget = FakeWidget() + coordinator = ThreeDCoordinator(widget, SceneComposer(), executor) + coordinator.set_visible(True) + coordinator.activate(scene, "live-A") + old = _frame(scene, frame_id=10, value=7.0) + old = replace(old, identity=replace(old.identity, request_id=1)) + coordinator.submit(old) + coordinator.tick() # starts the compose for request 1, and it stays pending + newer = _frame(scene, frame_id=20, value=3.0) + newer = replace(newer, identity=replace(newer.identity, request_id=2)) + assert coordinator.submit(newer) == "accepted" + executor.run_next() # request 1's compose finishes late + _drain(coordinator, ticks=6, sleep=0.0) + assert not np.allclose(np.asarray(widget.frames[-1].scalar), 7.0), ( + "a result for a superseded request was presented" + ) + coordinator.close()