From e7c1d943b19b919bc0855e6c382a70d5d44e2adf Mon Sep 17 00:00:00 2001 From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com> Date: Wed, 29 Jul 2026 07:21:26 +0200 Subject: [PATCH] Handle the loop closing between ThreadsafeProxy's `is_closed()` check and dispatch The worker thread can close the loop after `func_wrapper`'s `is_closed()` check but before `run_coroutine_threadsafe()` / `call_soon_threadsafe()`, raising `RuntimeError: Event loop is closed` at the caller -- the same race #727 suppressed in `force_stop()`. Treat close-during-dispatch like already-closed: log the existing warning and drop the call. Disconnected async proxy calls (both already-closed and closed-mid- dispatch) now return an already-completed future resolving to None instead of bare None, so `await proxy.method()` no longer raises `TypeError: object NoneType can't be used in 'await' expression`. Also raise a legible `RuntimeError` from `EventLoopThread.run_coroutine_threadsafe()` when the loop is already gone, instead of `AttributeError: 'NoneType' object has no attribute 'call_soon_threadsafe'`, closing the passed coroutine so it doesn't warn as never-awaited. Closes #740 --- bellows/thread.py | 35 +++++++++++++++++---- tests/test_thread.py | 74 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 103 insertions(+), 6 deletions(-) diff --git a/bellows/thread.py b/bellows/thread.py index a10b58d9..60c7d049 100644 --- a/bellows/thread.py +++ b/bellows/thread.py @@ -16,7 +16,12 @@ def __init__(self): def run_coroutine_threadsafe(self, coroutine): current_loop = asyncio.get_event_loop() - future = asyncio.run_coroutine_threadsafe(coroutine, self.loop) + # Snapshot: the worker thread publishes `None` when it exits + loop = self.loop + if loop is None: + coroutine.close() + raise RuntimeError("Event loop is not running") + future = asyncio.run_coroutine_threadsafe(coroutine, loop) return asyncio.wrap_future(future, loop=current_loop) def _thread_main(self, init_task): @@ -99,12 +104,26 @@ def func_wrapper(*args, **kwargs): call = functools.partial(func, *args, **kwargs) if loop == curr_loop: return call() - if loop.is_closed(): - # Disconnected + + def disconnected_result(): + # Disconnected: sync calls are dropped, async calls resolve to None LOGGER.warning("Attempted to use a closed event loop") - return + if not asyncio.iscoroutinefunction(func): + return None + future = curr_loop.create_future() + future.set_result(None) + return future + + if loop.is_closed(): + return disconnected_result() if asyncio.iscoroutinefunction(func): - future = asyncio.run_coroutine_threadsafe(call(), loop) + coro = call() + try: + future = asyncio.run_coroutine_threadsafe(coro, loop) + except RuntimeError: + # The worker thread may close the loop after our is_closed() check + coro.close() + return disconnected_result() return asyncio.wrap_future(future, loop=curr_loop) else: @@ -118,6 +137,10 @@ def check_result_wrapper(): ).format(self._obj.__class__.__name__, name) ) - loop.call_soon_threadsafe(check_result_wrapper) + try: + loop.call_soon_threadsafe(check_result_wrapper) + except RuntimeError: + # The worker thread may close the loop after our is_closed() check + return disconnected_result() return func_wrapper diff --git a/tests/test_thread.py b/tests/test_thread.py index 235507f7..fb0c89d3 100644 --- a/tests/test_thread.py +++ b/tests/test_thread.py @@ -192,6 +192,80 @@ async def test_proxy_loop_closed(): assert obj.test.call_count == 0 +async def test_proxy_loop_closed_async(): + """An async call through a proxy to a closed loop is awaitable and resolves to None.""" + loop = asyncio.new_event_loop() + obj = mock.MagicMock() + call_count = 0 + + async def magic(): + nonlocal call_count + call_count += 1 + + obj.test = magic + proxy = ThreadsafeProxy(obj, loop) + loop.close() + + assert await proxy.test() is None + assert call_count == 0 + + +@pytest.mark.filterwarnings("error::RuntimeWarning") +async def test_proxy_loop_closed_during_async_dispatch(caplog): + """The loop closing between the `is_closed()` check and dispatch is handled.""" + loop = asyncio.new_event_loop() + try: + obj = mock.MagicMock() + call_count = 0 + + async def magic(): + nonlocal call_count + call_count += 1 + + obj.test = magic + proxy = ThreadsafeProxy(obj, loop) + loop.call_soon_threadsafe = mock.Mock( + side_effect=RuntimeError("Event loop is closed") + ) + + assert await proxy.test() is None + + assert call_count == 0 + assert "Attempted to use a closed event loop" in caplog.text + finally: + loop.close() + + +async def test_proxy_loop_closed_during_sync_dispatch(caplog): + """The loop closing between the `is_closed()` check and dispatch is handled.""" + loop = asyncio.new_event_loop() + try: + obj = mock.MagicMock() + obj.test.return_value = None + proxy = ThreadsafeProxy(obj, loop) + loop.call_soon_threadsafe = mock.Mock( + side_effect=RuntimeError("Event loop is closed") + ) + + proxy.test() + + assert obj.test.call_count == 0 + assert "Attempted to use a closed event loop" in caplog.text + finally: + loop.close() + + +@pytest.mark.filterwarnings("error::RuntimeWarning") +async def test_thread_run_coroutine_threadsafe_loop_not_running(): + """A `RuntimeError` (not `AttributeError`) is raised when the loop is gone.""" + thread = EventLoopThread() + assert thread.loop is None + + with pytest.raises(RuntimeError): + # The coroutine is closed internally: no "never awaited" RuntimeWarning + thread.run_coroutine_threadsafe(asyncio.sleep(0)) + + async def test_thread_task_cancellation_after_stop(thread): loop = asyncio.get_event_loop() obj = mock.MagicMock()