diff --git a/ovos_core/__main__.py b/ovos_core/__main__.py index 425704d0899..a4fce93a66d 100644 --- a/ovos_core/__main__.py +++ b/ovos_core/__main__.py @@ -44,7 +44,7 @@ def main(alive_hook=on_alive, started_hook=on_started, ready_hook=on_ready, # Connect this process to the OpenVoiceOS message bus bus = MessageBusClient() - bus.run_in_thread() + bus_thread = bus.run_in_thread() bus.connected_event.wait() skill_manager = SkillManager(bus, watchdog, @@ -65,6 +65,28 @@ def main(alive_hook=on_alive, started_hook=on_started, ready_hook=on_ready, skill_manager.shutdown() + # Stop the messagebus websocket thread and its event dispatcher before + # the interpreter starts tearing down. `bus.run_in_thread()` spawns a + # daemon thread that keeps receiving messages and dispatching them onto + # `bus.emitter`'s internal ThreadPoolExecutor for as long as the socket + # is open. If that thread is still alive when Python begins interpreter + # shutdown, an incoming message can call `emitter.emit()` -> `executor + # .submit()` after the executor has already been torn down, raising + # `RuntimeError: cannot schedule new futures after shutdown` from a + # background thread and leaving the process unable to exit cleanly + # (systemd then waits out the stop timeout and SIGKILLs it). + # + # `bus.close()` itself is fire-and-forget: it signals the run_forever + # loop to stop and closes the socket, but does not wait for the + # receiver thread to actually exit. We join that thread here, with a + # bounded timeout, to narrow the window further. This is not a + # guarantee the thread has fully stopped by the time we return (the + # join can time out), but it removes most of the residual race where a + # buffered inbound frame is still being dispatched during teardown. + bus.close() + if bus_thread is not None: + bus_thread.join(timeout=3) + LOG.info('Skills service shutdown complete!') diff --git a/test/unittests/test_main.py b/test/unittests/test_main.py new file mode 100644 index 00000000000..901cf99a786 --- /dev/null +++ b/test/unittests/test_main.py @@ -0,0 +1,61 @@ +"""Regression test for the shutdown-hang bug where the SkillManager's +messagebus client is never closed, leaving the websocket dispatch thread +alive to race against interpreter teardown and raise +`RuntimeError: cannot schedule new futures after shutdown` +(https://github.com/OpenVoiceOS/ovos-core - shutdown hang / SIGKILL after +60s on `systemctl restart`). +""" +import unittest +from unittest.mock import MagicMock, patch + + +class TestMainShutdown(unittest.TestCase): + + @patch('ovos_core.__main__.setup_locale') + @patch('ovos_core.__main__.init_service_logger') + @patch('ovos_core.__main__.wait_for_exit_signal') + @patch('ovos_core.__main__.SkillManager') + @patch('ovos_core.__main__.MessageBusClient') + def test_bus_is_closed_before_main_returns(self, mock_bus_cls, mock_manager_cls, + mock_wait, mock_init_logger, + mock_setup_locale): + """After the exit signal arrives and the skill manager has been + shut down, `main()` must close the bus's websocket connection and + join the receiver thread so the background dispatch thread stops + before the interpreter starts tearing down. Without this, the + daemon thread spawned by `bus.run_in_thread()` keeps calling + `emitter.emit()` -> `executor.submit()` and can lose the race with + the executor/interpreter shutdown, raising `RuntimeError: cannot + schedule new futures after shutdown` and hanging the process until + SIGKILL. `bus.close()` alone is fire-and-forget and does not wait + for the thread to exit, so `main()` must also join it with a + bounded timeout. + """ + from ovos_core.__main__ import main + + mock_bus = MagicMock() + mock_bus_thread = MagicMock() + mock_bus.run_in_thread.return_value = mock_bus_thread + mock_bus_cls.return_value = mock_bus + mock_manager = MagicMock() + mock_manager_cls.return_value = mock_manager + + main() + + # shutdown must happen before the bus is closed, and both must + # happen before main() returns + mock_manager.shutdown.assert_called_once() + mock_bus.close.assert_called_once() + + # the receiver thread returned by `bus.run_in_thread()` must be + # joined with a bounded timeout after `bus.close()`, so `main()` + # doesn't block forever if the thread never stops + mock_bus_thread.join.assert_called_once() + join_args, join_kwargs = mock_bus_thread.join.call_args + timeout = join_kwargs.get('timeout', join_args[0] if join_args else None) + self.assertIsNotNone(timeout) + self.assertGreater(timeout, 0) + + +if __name__ == '__main__': + unittest.main()