From bd0a23186395b92c25d3dfd960099e55bd99c804 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Wed, 9 Sep 2026 17:24:57 +0200 Subject: [PATCH] fix(dispatcher): add dispatcher relay --- bec_widgets/utils/bec_dispatcher.py | 55 +++++-- tests/unit_tests/test_bec_dispatcher.py | 207 ++++++++++++++++++++++++ 2 files changed, 250 insertions(+), 12 deletions(-) diff --git a/bec_widgets/utils/bec_dispatcher.py b/bec_widgets/utils/bec_dispatcher.py index 0a0ad4be..5bd0e668 100644 --- a/bec_widgets/utils/bec_dispatcher.py +++ b/bec_widgets/utils/bec_dispatcher.py @@ -19,6 +19,7 @@ from bec_lib.service_config import ServiceConfig from qtpy.QtCore import QObject from qtpy.QtCore import Signal as pyqtSignal +from bec_widgets.utils.error_popups import SafeSlot from bec_widgets.utils.rpc_logging import elapsed_seconds, format_elapsed from bec_widgets.utils.serialization import register_serializer_extension @@ -72,6 +73,8 @@ class QtThreadSafeCallback(QObject): the owning widget alive. Non-method callables (lambdas, partials, module functions) have no owning object whose lifetime could be tracked, so they are kept alive by the wrapper itself; pass ``owner=`` to ``connect_slot`` to bind their lifetime to a widget. + Create the wrapper on the GUI thread: queued messages are delivered to its Qt slot, + which resolves the callback only when delivery occurs. """ cb_signal = pyqtSignal(dict, dict) @@ -101,7 +104,8 @@ class QtThreadSafeCallback(QObject): self.cb_owner = louie.saferef.safe_ref(owner) if owner is not None else None self._strong_cb = cb self.cb_ref = louie.saferef.safe_ref(cb) - self.cb_signal.connect(cb) + self._active = True + self.cb_signal.connect(self._deliver) self.topics = set() @property @@ -120,9 +124,33 @@ class QtThreadSafeCallback(QObject): return False return self.cb_ref == other.cb_ref and self.cb_info == other.cb_info + def invalidate(self) -> None: + """Suppress pending deliveries and release any strongly held callback.""" + self._active = False + self._strong_cb = None + + @SafeSlot(dict, dict) + def _deliver(self, msg_content: dict, metadata: dict) -> None: + """Resolve and retain a live callback for the duration of this delivery.""" + if not self._active: + return + owner = self.cb_owner() if self.cb_owner is not None else None + if self.cb_owner is not None and owner is None: + return + if isinstance(owner, QObject) and not shiboken6.isValid(owner): + return + if owner is not None and getattr(owner, "_destroyed", False): + return + + # Never connect Qt directly to a plain Python receiver's bound method: a + # queued invocation can outlive that receiver. Resolving here also holds + # the receiver alive while its callback runs. + callback = self.cb_ref() + if callback is not None: + callback(msg_content, metadata) + def __call__(self, msg_content, metadata): - if self.cb_ref() is None: - # callback has been deleted + if not self._active: return self.cb_signal.emit(msg_content, metadata) @@ -284,11 +312,12 @@ class BECDispatcher: break else: return - self.client.connector.unregister(topics, cb=connected_slot) topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics) - self._registered_slots[connected_slot].topics.difference_update(set(topics_str)) - if not self._registered_slots[connected_slot].topics: - del self._registered_slots[connected_slot] + if not connected_slot.topics.difference(topics_str): + self._release_slot(connected_slot) + else: + self.client.connector.unregister(topics, cb=connected_slot) + connected_slot.topics.difference_update(topics_str) def disconnect_topics(self, topics: str | list): """ @@ -297,16 +326,17 @@ class BECDispatcher: Args: topics(Union[str, list]): The topic(s) to disconnect from """ - self.client.connector.unregister(topics) topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics) remove_slots = [] - for connected_slot in self._registered_slots.values(): - connected_slot.topics.difference_update(set(topics_str)) - - if not connected_slot.topics: + for connected_slot in list(self._registered_slots.values()): + if not connected_slot.topics.difference(topics_str): + connected_slot.invalidate() remove_slots.append(connected_slot) + self.client.connector.unregister(topics) + for connected_slot in self._registered_slots.values(): + connected_slot.topics.difference_update(topics_str) for connected_slot in remove_slots: self._registered_slots.pop(connected_slot, None) @@ -354,6 +384,7 @@ class BECDispatcher: def _release_slot(self, qt_slot: QtThreadSafeCallback) -> None: """Unregister all topics of a slot wrapper and drop it from the registry.""" + qt_slot.invalidate() topics = list(qt_slot.topics) if topics: self.client.connector.unregister(topics, cb=qt_slot) diff --git a/tests/unit_tests/test_bec_dispatcher.py b/tests/unit_tests/test_bec_dispatcher.py index 651b8e37..e69387ab 100644 --- a/tests/unit_tests/test_bec_dispatcher.py +++ b/tests/unit_tests/test_bec_dispatcher.py @@ -1,12 +1,21 @@ # pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring +import gc +import os +import subprocess +import sys import threading import time +import weakref +from pathlib import Path from unittest import mock import pytest +import shiboken6 from bec_lib import service_config from bec_lib.messages import GUIInstructionMessage, ScanMessage from bec_lib.serialization import MsgpackSerialization +from qtpy.QtCore import QEvent, QObject, QThread +from qtpy.QtWidgets import QApplication from bec_widgets.utils.bec_dispatcher import BECDispatcher, QtRedisConnector, QtThreadSafeCallback @@ -272,3 +281,201 @@ def test_stop_cli_server_is_idempotent(bec_dispatcher): bec_dispatcher.stop_cli_server() bec_dispatcher.stop_cli_server() mock_logger.error.assert_not_called() + + +def _queue_callback(wrapper): + """Queue a real cross-thread emission without allowing GUI delivery yet.""" + thread = threading.Thread(target=wrapper, args=({"value": 1}, {"source": "worker"})) + thread.start() + thread.join(timeout=5) + assert not thread.is_alive() + + +def _deliver_after_receiver_collection(cyclic): + """Run in a child process: the original queued bound method could segfault.""" + app = QApplication([]) + calls = [] + + class Receiver: + def __init__(self): + self.state = "alive" + if cyclic: + self.cycle = self + + def receive(self, content, metadata): + calls.append("entered") + calls.append(self.state) + + receiver = Receiver() + receiver_ref = weakref.ref(receiver) + wrapper = QtThreadSafeCallback(receiver.receive) + _queue_callback(wrapper) + del receiver + gc.collect() + assert receiver_ref() is None + assert wrapper.cb is None + + # A second queued callback proves that Qt has drained the queued deliveries. + delivered = [] + sentinel = QtThreadSafeCallback(lambda *_: delivered.append(True)) + _queue_callback(sentinel) + app.processEvents() + assert delivered == [True] + assert calls == [] + print("Collected receiver was not called", flush=True) + + +@pytest.mark.parametrize("cyclic", [False, True], ids=["refcount", "cyclic-gc"]) +def test_qthreadsafe_callback_skips_collected_receiver_in_subprocess(cyclic): + """Keep native crash regressions from terminating the pytest process.""" + result = subprocess.run( + [ + sys.executable, + "-X", + "faulthandler", + "-c", + "from tests.unit_tests.test_bec_dispatcher import " + f"_deliver_after_receiver_collection; _deliver_after_receiver_collection({cyclic!r})", + ], + cwd=Path(__file__).resolve().parents[2], + env={**os.environ, "QT_QPA_PLATFORM": "offscreen", "OPHYD_CONTROL_LAYER": "dummy"}, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + assert result.returncode == 0, ( + f"Queued callback subprocess exited with {result.returncode}\n" + f"{result.stdout[-6000:]}\n{result.stderr[-6000:]}" + ) + assert "Collected receiver was not called" in result.stdout + + +def test_qthreadsafe_callback_delivers_live_receiver_on_gui_thread(qapp): + calls = [] + + class Receiver: + def receive(self, content, metadata): + calls.append((content, metadata, QThread.currentThread())) + + receiver = Receiver() + wrapper = QtThreadSafeCallback(receiver.receive) + _queue_callback(wrapper) + assert calls == [] + + qapp.processEvents() + + assert calls == [({"value": 1}, {"source": "worker"}, qapp.thread())] + + +@pytest.mark.parametrize( + "disconnect", ["disconnect_owner", "disconnect_slot", "disconnect_topics", "disconnect_all"] +) +def test_dispatcher_disconnect_cancels_queued_delivery(bec_dispatcher, qapp, disconnect): + owner = QObject() + callback = mock.Mock(spec=[]) + bec_dispatcher.connect_slot(callback, "lifetime-test", owner=owner) + wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback) + _queue_callback(wrapper) + callback.assert_not_called() + + if disconnect == "disconnect_owner": + bec_dispatcher.disconnect_owner(owner) + elif disconnect == "disconnect_slot": + bec_dispatcher.disconnect_slot(callback, "lifetime-test") + elif disconnect == "disconnect_topics": + bec_dispatcher.disconnect_topics("lifetime-test") + else: + bec_dispatcher.disconnect_all() + + # The connector may still hold an in-flight reference to the released wrapper. + _queue_callback(wrapper) + qapp.processEvents() + callback.assert_not_called() + assert wrapper not in bec_dispatcher._registered_slots + + +@pytest.mark.parametrize("disconnect", ["disconnect_slot", "disconnect_topics"]) +def test_dispatcher_partial_disconnect_preserves_delivery(bec_dispatcher, qapp, disconnect): + owner = QObject() + callback = mock.Mock(spec=[]) + bec_dispatcher.connect_slot(callback, ["removed-topic", "retained-topic"], owner=owner) + wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback) + _queue_callback(wrapper) + + if disconnect == "disconnect_slot": + bec_dispatcher.disconnect_slot(callback, "removed-topic") + else: + bec_dispatcher.disconnect_topics("removed-topic") + + qapp.processEvents() + callback.assert_called_once_with({"value": 1}, {"source": "worker"}) + assert wrapper.topics == {"retained-topic"} + _queue_callback(wrapper) + qapp.processEvents() + assert callback.call_count == 2 + + +def test_dispatcher_reconnect_does_not_reactivate_old_delivery(bec_dispatcher, qapp): + owner = QObject() + callback = mock.Mock(spec=[]) + bec_dispatcher.connect_slot(callback, "lifetime-test", owner=owner) + old_wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback) + _queue_callback(old_wrapper) + bec_dispatcher.disconnect_owner(owner) + + bec_dispatcher.connect_slot(callback, "lifetime-test", owner=owner) + new_wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback) + assert new_wrapper is not old_wrapper + _queue_callback(old_wrapper) + _queue_callback(new_wrapper) + qapp.processEvents() + + callback.assert_called_once_with({"value": 1}, {"source": "worker"}) + + +def test_dispatcher_disconnect_releases_lambda_capture(bec_dispatcher): + owner = QObject() + owner_ref = weakref.ref(owner) + bec_dispatcher.connect_slot( + lambda _content, _metadata, target=owner: target.objectName(), "lifetime-test", owner=owner + ) + wrapper = next( + slot + for slot in bec_dispatcher._registered_slots + if slot.cb_owner is not None and slot.cb_owner() is owner + ) + callback_ref = weakref.ref(wrapper.cb) + del owner + gc.collect() + assert owner_ref() is not None + + bec_dispatcher.disconnect_owner(owner_ref()) + gc.collect() + + # An in-flight wrapper must not keep the lambda or its captured owner alive. + assert callback_ref() is None + assert owner_ref() is None + + +@pytest.mark.parametrize("teardown", ["native-delete", "cleanup", "garbage-collect"]) +def test_qthreadsafe_callback_checks_lambda_owner_at_delivery(qapp, teardown): + calls = [] + owner = QObject() + owner_ref = weakref.ref(owner) + wrapper = QtThreadSafeCallback(lambda *_: calls.append(True), owner=owner) + _queue_callback(wrapper) + + if teardown == "native-delete": + owner.deleteLater() + qapp.sendPostedEvents(owner, QEvent.Type.DeferredDelete) + assert not shiboken6.isValid(owner) + elif teardown == "cleanup": + owner._destroyed = True + else: + del owner + gc.collect() + assert owner_ref() is None + + qapp.processEvents() + assert calls == []