orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(studio-3d): 取消 epoch —— 停止/隐藏/换会话之后，已在跑的后台合成不能再把旧测量放回屏幕

Browse Source 
- What:
  1. `ThreeDCoordinator` 新增 **epoch**（显示声明代次）与 `_Inflight` 记录：`work()` 不再直接写
     `_composed` mailbox，只 `return` 合成好的帧；结果在 GUI 线程的 `_collect_inflight()` 里被接受，
     且必须同时满足 `item.epoch == self._epoch` 与 `_identity_is_current(identity)` 才会入队。
     `activate()`/`begin_session()`/`set_visible(False)`/`show_rest()`/`close()` 统一走新的
     `_invalidate()`：epoch+1、清空两块 mailbox、清 `_last_request_id`/`_last_composed_identity`/
     `_last_field_frame`/`_settings_dirty`，并**尝试**取消 future。
  2. **未成功取消的 future 仍占用唯一合成槽**：`_inflight` 只在 `future.done()` 时由 `_collect_inflight()`
     释放，所以反复隐藏/显示不会在线程池里排起第二个合成；`CancelledError` 单独吞掉（取消不是失败，
     不产生诊断），其它异常仍记成 `field_unavailable` 诊断（文案由 "compose raised" 回到 "compose failed"）。
  3. **request_id 单调**：`submit()` 拒绝小于本 epoch 已见值的 request_id（>0 时）并计 `stale` + 给
     `stale_frame` 诊断；`_identity_is_current()` 同样比较，所以"晚到的旧结果"也进不了屏幕。
  4. 新增 `ThreeDCoordinator.epoch` 与 `last_presented()`（帧号 + 单调时刻，供后续面板帧计数用）；
     构造参数新增关键字 `clock`（默认 `time.monotonic`）与 `stale_grace_s`（默认 1.0，本提交尚未使用）。
- Why: 审查报告 P1/R1 的第一条。`Future.cancel()` 无法中断已经开始运行的函数，而旧代码在
  `_cancel_inflight()` 里就把 `_inflight` 置空并只比较 generation —— 于是"停止显示"之后旧 work 结束仍然
  `offer` 进 `_composed`，下一次 tick 取到它并把状态改回 live（审查用可控后台任务复现：
  `valid_after_show_rest=false` → 放行旧 compose → `valid_after_old_worker_finishes=true`、`state=live`）。
  同一个洞也适用于"停止重算后定时器仍继续 tick"的场景。把接受点移到 GUI 线程后，"发布前校验"和
  "接收时校验"是同一处代码，epoch 让"这一刻的显示声明"成为可比较的事实。
- Verification:
  · `test_scene3d_coordinator.py` +7（24 通过）：合成已进入后被 `show_rest`/`set_visible(False)`/
    `begin_session` 撤销 → 旧结果永不 present 且 state 不为 live、epoch 递增、槽位在 work 真正结束后释放；
    `ManualExecutor` 复现"提交后立即取消" → `CancelledError` 不产生诊断且槽位可复用；
    20 轮 begin_session→submit→tick→show_rest 循环里最后一个画面上永远没有有效测量、`max_in_flight==1`；
    request_id 逆序在入口被拒（`stale` + `stale_frame`）且画面停在较新的请求上，
    "较旧请求的结果晚到"也在接收侧被丢弃。
  · 审查复现脚本对照（`.repro/probe_stage6_r1.py` → `probe_stage6_r1.json`）：
    `stopped_frame_returns` 由 `{valid_after_old_worker_finishes: true, state: "live"}` 变为
    `{false, "no_data", epoch_bumped: true, slot_free_again: true}`；
    `request_id_not_checked` 由 `{old_submit: "accepted", shown_request: 1}` 变为
    `{old_submit: "stale", shown_request: 2, diagnostic: "stale_frame"}`。
  · apps/studio 全量 pytest 退出码 0（含真 GL 冒烟用例）。
- Not in this change: `stale_grace_s`/`clock` 只落地参数与 `last_presented()` 的记录，**失效状态的迁移
  （`stale` 状态、宽限到期回参考模型）留给下一提交**；adapter 侧的字段级有效性、设备几何身份、
  资产加载请求号、二维后端统一均未动；审查报告里那句"用 `coordinator._inflight` 当 Future 用"的复现
  脚本无法再原样运行（槽位现在是 `_Inflight` 记录），改用上面那份等价探针。
This commit is contained in:
ykcheungab
2026-09-14 20:53:38 +08:00
parent 6de6b8b703
commit d06e116c15
2 changed files with 359 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/studio/src/orisys_studio/scene3d/coordinator.py
+133 -53
View File
@@ -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)
apps/studio/tests/test_scene3d_coordinator.py
+226 -1
View File
@@ -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()