diff --git a/BlocksScreen/BlocksScreen.py b/BlocksScreen/BlocksScreen.py index 0b425d28..1d1c5454 100644 --- a/BlocksScreen/BlocksScreen.py +++ b/BlocksScreen/BlocksScreen.py @@ -6,6 +6,7 @@ import socket import subprocess # nosec B404 import sys +import threading import typing from pathlib import Path @@ -163,7 +164,9 @@ def _handler(__signum: int, __frame: typing.Any) -> None: def on_quit() -> None: - logging.info("Final exit cleanup") + # Daemon threads alive here can emit into widgets sip destroys at exit (the stop SEGV). + _alive = [t.name for t in threading.enumerate() if t is not threading.main_thread()] + logging.info("Final exit cleanup, threads still alive: %s", _alive) LogManager.shutdown() diff --git a/BlocksScreen/lib/moonrakerComm.py b/BlocksScreen/lib/moonrakerComm.py index bb2e3144..f2314213 100644 --- a/BlocksScreen/lib/moonrakerComm.py +++ b/BlocksScreen/lib/moonrakerComm.py @@ -1,6 +1,7 @@ # Moonraker api import json import logging +import socket import threading import typing @@ -57,6 +58,8 @@ def __init__(self, parent: QtCore.QObject) -> None: self.daemon = True self.connected = False + # Set once on shutdown: ws callbacks must not touch Qt while widgets are destroyed. + self._closing = False self._reconnect_count = 0 self._klippy_state: str | None = None self.callback_table: dict = {} @@ -109,7 +112,7 @@ def try_connection(self): def reconnect(self) -> bool: """Watchdog tick, retries forever per Moonraker's client startup sequence""" # Non-blocking: a tick and a Qt-thread call must not open two sockets - if not self._connect_lock.acquire(blocking=False): + if self._closing or not self._connect_lock.acquire(blocking=False): return False try: with self._state_lock: @@ -171,13 +174,24 @@ def connect(self) -> bool: return True def wb_disconnect(self) -> None: - """Websocket disconnect""" + """Stop the timers and the websocket thread before the widget tree is destroyed""" + self._closing = True + self.query_klippy_status_timer.stopTimer() if self._retry_timer is not None: self._retry_timer.stopTimer() if self._wst is not None and self.ws is not None: - self.ws.close() + self.ws.keep_running = False + # ws.close() frees the fd under the dispatcher's epoll (it sleeps ping_timeout); SHUT_RD wakes it + _raw = getattr(self.ws.sock, "sock", None) + try: + if _raw is not None: + _raw.shutdown(socket.SHUT_RD) + except OSError: + self.ws.close() if self._wst.is_alive(): self._wst.join(timeout=self.timeout + 1) + if self._wst.is_alive(): + logger.warning("Websocket thread still alive after shutdown join") logger.info("Websocket closed") def on_error(self, *args) -> None: @@ -196,6 +210,10 @@ def on_close(self, *args) -> None: # First argument is ws, second is close status code, third is close message if self.ws is None: return + if self._closing: + with self._state_lock: + self.connected = False + return _close_status_code = args[1] if len(args) == 3 else None _close_message = args[2] if len(args) == 3 else None with self._state_lock: @@ -242,6 +260,8 @@ def on_open(self, *args) -> None: TypeError: When QApplication.instance `is` None """ _ws = args[0] if len(args) == 1 else None + if self._closing: + return with self._state_lock: self.connected = True self._reconnect_count = 0 @@ -266,6 +286,8 @@ def on_message(self, *args) -> None: Raises: TypeError: Raised when events cannot be sent because QApplication.instance is None """ + if self._closing: + return _message = ( args[1] if len(args) == 2 else args[0] ) # First argument is ws second is message diff --git a/BlocksScreen/lib/updater_worker.py b/BlocksScreen/lib/updater_worker.py index 42a7e0e1..bb4a28ed 100644 --- a/BlocksScreen/lib/updater_worker.py +++ b/BlocksScreen/lib/updater_worker.py @@ -478,6 +478,11 @@ def _cancel_and_stop() -> None: _log.warning( "shutdown: loop already closed (crashed during restart window)" ) + # Bounded: a callback still running at interpreter exit can emit into destroyed widgets. + if self._thread.is_alive() and self._thread is not threading.current_thread(): + self._thread.join(timeout=2.0) + if self._thread.is_alive(): + _log.warning("shutdown: UpdaterAsyncLoop still alive after 2 s") def _submit(self, coro: Coroutine[Any, Any, None]) -> None: """Submit a coroutine to the asyncio loop, emit daemon_unavailable if loop is closed.""" diff --git a/tests/lib/test_moonraker_comm_unit.py b/tests/lib/test_moonraker_comm_unit.py index 6ac45ce1..dd32923f 100644 --- a/tests/lib/test_moonraker_comm_unit.py +++ b/tests/lib/test_moonraker_comm_unit.py @@ -1,7 +1,13 @@ """Unit tests for moonrakerComm.""" +import base64 +import hashlib +import importlib import json +import socket +import sys import threading +import time from types import SimpleNamespace from unittest.mock import MagicMock @@ -282,6 +288,101 @@ def test_send_while_disconnected_never_calls_back(self, ws, qtbot): callback.assert_not_called() +class TestShutdown: + def test_disconnect_wakes_dispatcher_and_stops_poll(self, ws): + ws.ws = MagicMock() + ws._wst = MagicMock(is_alive=MagicMock(return_value=False)) + ws.wb_disconnect() + ws.ws.sock.sock.shutdown.assert_called_once_with(socket.SHUT_RD) + ws.ws.close.assert_not_called() # teardown on the ws thread sends the close frame + assert ws.ws.keep_running is False + ws.query_klippy_status_timer.stopTimer.assert_called_once() + + def test_disconnect_falls_back_to_close(self, ws): + ws.ws = MagicMock() + ws.ws.sock.sock.shutdown.side_effect = OSError("not connected") + ws._wst = MagicMock(is_alive=MagicMock(return_value=False)) + ws.wb_disconnect() + ws.ws.close.assert_called_once() + + def test_callbacks_silent_after_disconnect(self, ws, monkeypatch, qtbot): + ws.wb_disconnect() + connect = MagicMock() + monkeypatch.setattr(ws, "connect", connect) + emitted = [] + ws.connected_signal.connect(lambda: emitted.append("open")) + ws.connection_lost[str].connect(emitted.append) + ws.server_components_signal.connect(emitted.append) + ws.ws = MagicMock() + ws.request_table[1] = ("server.info", {}, None) + ws.on_open(None) + ws.on_message(None, json.dumps({"id": 1, "result": {"components": []}})) + ws.on_close(None, 1000, "bye") + assert ws.reconnect() is False + qtbot.wait(20) + assert emitted == [] + assert ws.connected is False + connect.assert_not_called() + + def test_real_socket_thread_exits_within_join(self, ws, qtbot, monkeypatch): + # Unfixed, ~40% of closes left run_forever in epoll for ping_timeout (10 s). + monkeypatch.delitem(sys.modules, "websocket", raising=False) # conftest stub + monkeypatch.setattr( + moonrakerComm, "websocket", importlib.import_module("websocket") + ) + port = _serve_ws() + ws._host, ws._port = "127.0.0.1", port + ws.moonRest.get_oneshot_token.return_value = "t" + for _ in range(8): + ws._closing = False + assert ws.connect() is True + qtbot.waitUntil(lambda: ws.connected, timeout=3000) + start = time.monotonic() + ws.wb_disconnect() + assert not ws._wst.is_alive() + assert time.monotonic() - start < 2 + ws.connected = False + + +def _serve_ws() -> int: + """Minimal RFC 6455 server: handshake, then answer a close frame like Moonraker.""" + srv = socket.create_server(("127.0.0.1", 0)) + + def _client(conn: socket.socket) -> None: + with conn: + req = b"" + while b"\r\n\r\n" not in req: + req += conn.recv(4096) + key = next( + line.split(b":", 1)[1].strip() + for line in req.split(b"\r\n") + if line.lower().startswith(b"sec-websocket-key") + ) + accept = base64.b64encode(hashlib.sha1(key + _WS_GUID).digest()) # nosec B324 + conn.sendall( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n" + b"Connection: Upgrade\r\nSec-WebSocket-Accept: " + accept + b"\r\n\r\n" + ) + while (hdr := conn.recv(2)) and len(hdr) == 2: + length = hdr[1] & 0x7F + conn.recv(4 + length) + if hdr[0] & 0x0F == 8: + conn.sendall(b"\x88\x02\x03\xe8") + return + + def _accept() -> None: + while True: + threading.Thread( + target=_client, args=(srv.accept()[0],), daemon=True + ).start() + + threading.Thread(target=_accept, daemon=True).start() + return srv.getsockname()[1] + + +_WS_GUID = b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + + @pytest.fixture def api(): """MoonAPI over a stub socket; it parents itself to ws, so ws must be a QObject.""" diff --git a/tests/lib/test_updater_worker_unit.py b/tests/lib/test_updater_worker_unit.py index 2d0fc35a..ff428648 100644 --- a/tests/lib/test_updater_worker_unit.py +++ b/tests/lib/test_updater_worker_unit.py @@ -2,6 +2,7 @@ import asyncio import sys +import threading from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -38,6 +39,7 @@ def _make_worker(): w._escalated = False w._init_lock = asyncio.Lock() w._system_bus = MagicMock() + w._thread = threading.Thread(target=lambda: None) return w @@ -544,6 +546,12 @@ def test_shutdown_cancels_owner_watch(self, worker): owner_task.cancel.assert_called_once() listener.cancel.assert_called_once() + def test_shutdown_joins_loop_thread(self, worker): + worker._thread = threading.Thread(target=worker._loop.run_forever, daemon=True) + worker._thread.start() + worker.shutdown() + assert not worker._thread.is_alive() + class TestReplayBusy: def test_replays_true_only(self, worker, qtbot):