diff --git a/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py b/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py index 7d788d74..cfd8307c 100644 --- a/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py +++ b/bec_widgets/widgets/control/device_manager/components/ophyd_validation/ophyd_validation.py @@ -9,11 +9,13 @@ In addition, it allows to configure the test parameters. Mostly relevant for ADBase integrations. """ +import atexit import queue import weakref from typing import Any from uuid import uuid4 +import shiboken6 from bec_lib.atlas_models import Device as DeviceModel from bec_lib.logger import bec_logger from qtpy import QtCore, QtWidgets @@ -134,6 +136,57 @@ class DeviceTest(QtCore.QRunnable): ) +_RETIRED_POOL_POLL_MS = 200 +# Thread pools of shut-down ThreadPoolManagers that still run device tests (see _retire_pool). +_retired_pools: list[QtCore.QThreadPool] = [] + + +def _retire_pool(pool: QtCore.QThreadPool) -> None: + """ + Take over a thread pool that still runs device tests from its (closing) owner. + + A running StaticDeviceTest cannot be interrupted, and destroying a QThreadPool blocks until + its runnables return. The pool is therefore re-parented to the application, so destroying + its former parent does not freeze the GUI thread, and deleted once it is idle. + + Args: + pool (QtCore.QThreadPool): The thread pool to retire. + """ + pool.setParent(QtCore.QCoreApplication.instance()) + _retired_pools.append(pool) + _delete_pool_when_idle(pool) + + +def _delete_pool_when_idle(pool: QtCore.QThreadPool) -> None: + """ + Delete a retired thread pool once none of its runnables is running any more. + + Polled from the GUI thread, so the QThreadPool destructor never has to wait for a test. + + Args: + pool (QtCore.QThreadPool): The thread pool to delete. + """ + if shiboken6.isValid(pool) and pool.activeThreadCount() > 0: + QtCore.QTimer.singleShot(_RETIRED_POOL_POLL_MS, pool, lambda: _delete_pool_when_idle(pool)) + return + _retired_pools[:] = [p for p in _retired_pools if p is not pool] + if shiboken6.isValid(pool): + pool.deleteLater() + + +@atexit.register +def _wait_for_retired_pools() -> None: + """ + At interpreter exit, wait for device tests that are still running in retired pools. + + Registered after PySide's own atexit cleanup and therefore run before it: PySide destroys + the Python-owned QObjects (e.g. the DeviceTestResult a running test emits on) in its cleanup. + """ + for pool in list(_retired_pools): + if shiboken6.isValid(pool): + pool.waitForDone() + + class ThreadPoolManager(QtCore.QObject): """ Manager wrapping QThreadPool to expose a queue for jobs. @@ -159,6 +212,7 @@ class ThreadPoolManager(QtCore.QObject): self.poll_interval_ms = poll_interval_ms self._timer.setInterval(self.poll_interval_ms) self._active_tests: dict[str, weakref.ReferenceType[DeviceTest]] = {} + self._shut_down = False def start_polling(self): """Start the polling timer.""" @@ -172,16 +226,26 @@ class ThreadPoolManager(QtCore.QObject): def _emit_device_validation_started(self, device_name: str): """Emit device validation started signal.""" + if self._shut_down: + return self.device_validation_started.emit(device_name) def _emit_device_validated( self, device_config: dict, config_status: int, connection_status: int, error_message: str ): """Emit device validated signal.""" + if self._shut_down: + return self.device_validated.emit(device_config, config_status, connection_status, error_message) def submit(self, device_name: str, device_test: DeviceTest): """Queue a job for execution.""" + if self._shut_down: + logger.debug( + f"Thread pool manager is shut down, dropping device test of {device_name}." + ) + device_test.cancel() + return device_test.signals.device_validation_started.connect(self._emit_device_validation_started) device_test.signals.device_validated.connect(self._emit_device_validated) self._queue.put((device_name, device_test)) @@ -225,8 +289,45 @@ class ThreadPoolManager(QtCore.QObject): with self._queue.mutex: return [device_name for device_name, _ in list(self._queue.queue)] + def shutdown(self) -> None: + """ + Stop starting device tests and drop all pending results, without blocking. + + The poll timer is stopped, queued (not yet started) tests are discarded and running tests + are cancelled; their result signals are disconnected so late results are dropped. A test + that is already running cannot be interrupted: if any is still running, the thread pool + is retired (handed over to the application and deleted once idle), so destroying the + parent does not block the GUI thread in the QThreadPool destructor until the tests return. + """ + if self._shut_down: + return + self._shut_down = True + self.stop_polling() + with self._queue.mutex: + queued = [runnable for _, runnable in self._queue.queue] + self._queue.queue.clear() + running = [ref() for ref in self._active_tests.values()] + self._active_tests.clear() + for runnable in queued + running: + if runnable is None: + continue + runnable.cancel() + for signal in ( + runnable.signals.device_validated, + runnable.signals.device_validation_started, + ): + try: + signal.disconnect() + except RuntimeError: # nothing left to disconnect + pass + self.pool.clear() + if self.pool.activeThreadCount() > 0: + _retire_pool(self.pool) + def _process_queue(self): """Start new jobs if there is capacity. Runs with specified poll interval.""" + if self._shut_down: + return while not self._queue.empty() and len(self._active_tests) < self.pool.maxThreadCount(): device_name, runnable = self._queue.get() runnable.signals.device_validated.connect(self._on_task_finished) @@ -371,6 +472,17 @@ class OphydValidation(BECWidget, QtWidgets.QWidget): if self.thread_pool_manager: self.thread_pool_manager.start_polling() + def cleanup(self): + """ + Stop all validations without blocking, then clean up the widget. + + Queued device tests are dropped and results of running ones are discarded; a test that + is already running cannot be interrupted and finishes in the background. + """ + if self.thread_pool_manager is not None: + self.thread_pool_manager.shutdown() + super().cleanup() + def _create_list_widget_with_label(self, label_text: str) -> BECList: """Setup the running validations section.""" widget = QtWidgets.QWidget() @@ -607,7 +719,9 @@ class OphydValidation(BECWidget, QtWidgets.QWidget): # and therefore must not have the signal emitted immediately in the same event loop iteration. # Otherwise, the dialog would block signal processing. QtCore.QTimer.singleShot( - 0, lambda: self.multiple_validations_completed.emit(devices_already_in_session) + 0, + self, + lambda: self.multiple_validations_completed.emit(devices_already_in_session), ) def cancel_validation(self, device_name: str) -> None: @@ -725,8 +839,10 @@ class OphydValidation(BECWidget, QtWidgets.QWidget): self, widget: ValidationListItem, connect: bool, force_connect: bool, timeout: float ) -> None: """Delayed submission of device test to ensure UI updates.""" + # The list item is the context object: the submission is dropped if the item (or this + # widget, its ancestor) is deleted before the event loop runs it. QtCore.QTimer.singleShot( - 0, lambda: self._submit_test(widget, connect, force_connect, timeout) + 0, widget, lambda: self._submit_test(widget, connect, force_connect, timeout) ) def _submit_test( diff --git a/tests/unit_tests/test_device_manager_components.py b/tests/unit_tests/test_device_manager_components.py index b8de3332..c2d04998 100644 --- a/tests/unit_tests/test_device_manager_components.py +++ b/tests/unit_tests/test_device_manager_components.py @@ -1,17 +1,19 @@ """Unit tests for device_manager_components module.""" +import time from threading import Event from typing import Generator from unittest import mock import pytest +import shiboken6 import yaml from bec_lib.atlas_models import Device as DeviceModel from ophyd_devices.interfaces.device_config_templates.ophyd_templates import ( OPHYD_DEVICE_TEMPLATES, EpicsMotorDeviceConfigTemplate, ) -from ophyd_devices.utils.static_device_test import TestResult +from ophyd_devices.utils.static_device_test import StaticDeviceTest, TestResult from qtpy import QtCore, QtGui, QtWidgets from bec_widgets.tests.client_mocks import mocked_client @@ -1330,6 +1332,122 @@ class TestOphydValidation: assert blocker.args[2] == connection_status assert blocker.args[3] == msg + #################### + ### Teardown + #################### + + def test_thread_pool_manager_shutdown_drops_queued_and_running_tests( + self, thread_pool_manager, qtbot + ): + """shutdown() stops polling, cancels queued and running tests and drops late results.""" + manager: ThreadPoolManager = thread_pool_manager._pool_manager + manager.pool.setMaxThreadCount(1) + running_test, queued_test = mock.MagicMock(), mock.MagicMock() + manager.submit(device_name="running_device", device_test=running_test) + manager.submit(device_name="queued_device", device_test=queued_test) + manager.start_polling() + with mock.patch.object(manager.pool, "start"): + manager._process_queue() + assert manager.get_active_tests() == ["running_device"] + assert manager.get_scheduled_tests() == ["queued_device"] + + manager.shutdown() + + assert not manager._timer.isActive() + assert manager.get_active_tests() == [] + assert manager.get_scheduled_tests() == [] + for device_test in (running_test, queued_test): + device_test.cancel.assert_called_once() + device_test.signals.device_validated.disconnect.assert_called_once_with() + device_test.signals.device_validation_started.disconnect.assert_called_once_with() + # Results that were already on their way and new submissions are dropped + with ( + qtbot.assertNotEmitted(manager.device_validated), + qtbot.assertNotEmitted(manager.device_validation_started), + ): + manager._emit_device_validation_started("running_device") + manager._emit_device_validated({"name": "running_device"}, 1, 1, "late result") + late_test = mock.MagicMock() + manager.submit(device_name="late_device", device_test=late_test) + late_test.cancel.assert_called_once() + assert manager.get_scheduled_tests() == [] + + def test_ophyd_validation_close_mid_run_stops_queue_without_blocking( + self, mocked_client, qtbot + ): + """ + Closing the widget while device tests run must not start the queued tests, and deleting + it must not block the GUI thread until the running tests (which cannot be interrupted) + return. Their results are dropped and the thread pool is deleted once idle. + """ + release = Event() + started = [] + + def blocking_test(tester, connect=False, force_connect=False, timeout_per_device=30): + name = next(iter(tester.config)) + started.append(name) + release.wait(timeout=3) + return [TestResult(name=name, success=False, message="", config_is_valid=True)] + + configs = [ + {"name": f"dev_{i}", "deviceClass": "ophyd.Signal", "enabled": True} for i in range(6) + ] + with ( + mock.patch.object(StaticDeviceTest, "run_with_list_output", blocking_test), + mock.patch.object(OphydValidation, "_is_device_in_redis_session", return_value=False), + ): + widget = OphydValidation(client=mocked_client) + manager = widget.thread_pool_manager + pool = manager.pool + n_workers = pool.maxThreadCount() + results = [] + widget.validation_completed.connect(lambda *args: results.append(args)) + try: + widget.change_device_configs(configs, added=True) + qtbot.waitUntil(lambda: len(started) == n_workers) + + widget.close() + observed = { + "poll_timer_active": manager._timer.isActive(), + "queued_after_close": manager.get_scheduled_tests(), + } + qtbot.wait(300) + start = time.monotonic() + widget.deleteLater() + QtWidgets.QApplication.sendPostedEvents(None, QtCore.QEvent.DeferredDelete) + delete_s = time.monotonic() - start + finally: + release.set() + qtbot.waitUntil(lambda: not shiboken6.isValid(pool), timeout=5000) + observed.update( + {"delete_blocked": delete_s >= 1.0, "started": len(started), "results": results} + ) + assert observed == { + "poll_timer_active": False, + "queued_after_close": [], + "delete_blocked": False, + "started": n_workers, + "results": [], + }, f"deleting the widget took {delete_s:.2f} s" + del widget + + def test_ophyd_validation_deferred_submit_dropped_when_deleted(self, mocked_client, qtbot): + """A test submission still pending in the event loop must not run on a deleted widget.""" + with ( + mock.patch.object(OphydValidation, "_is_device_in_redis_session", return_value=False), + mock.patch.object(OphydValidation, "_submit_test") as mock_submit_test, + ): + widget = OphydValidation(client=mocked_client) + widget.change_device_configs( + [{"name": "dev_0", "deviceClass": "ophyd.Signal", "enabled": True}], added=True + ) + widget.close() + widget.deleteLater() + QtWidgets.QApplication.sendPostedEvents(None, QtCore.QEvent.DeferredDelete) + qtbot.wait(50) + mock_submit_test.assert_not_called() + del widget + class TestDeviceConfigTemplate: