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 1/3] 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() From 8a0b5caab7135fe6d7c924015fb19a6a3db8f63d 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:36:03 +0200 Subject: [PATCH 2/3] Use `inspect.iscoroutinefunction` instead of the deprecated asyncio one `asyncio.iscoroutinefunction()` is deprecated since Python 3.14 and slated for removal in 3.16; `inspect.iscoroutinefunction()` is a drop-in replacement for every case here (coroutine functions, plain callables, `functools.partial`, `AsyncMock`). --- bellows/thread.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/bellows/thread.py b/bellows/thread.py index 60c7d049..fb840def 100644 --- a/bellows/thread.py +++ b/bellows/thread.py @@ -2,6 +2,7 @@ from concurrent.futures import ThreadPoolExecutor import contextlib import functools +import inspect import logging LOGGER = logging.getLogger(__name__) @@ -108,7 +109,7 @@ def func_wrapper(*args, **kwargs): def disconnected_result(): # Disconnected: sync calls are dropped, async calls resolve to None LOGGER.warning("Attempted to use a closed event loop") - if not asyncio.iscoroutinefunction(func): + if not inspect.iscoroutinefunction(func): return None future = curr_loop.create_future() future.set_result(None) @@ -116,7 +117,7 @@ def disconnected_result(): if loop.is_closed(): return disconnected_result() - if asyncio.iscoroutinefunction(func): + if inspect.iscoroutinefunction(func): coro = call() try: future = asyncio.run_coroutine_threadsafe(coro, loop) From 690562937261b8c9f0d12345aa4a8deca68ec43d Mon Sep 17 00:00:00 2001 From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:23:39 +0200 Subject: [PATCH 3/3] Close the coroutine when the loop closes between snapshot and dispatch `EventLoopThread.run_coroutine_threadsafe`'s `None` guard covers the already-exited worker, but when the worker closes the loop after the snapshot, `asyncio.run_coroutine_threadsafe()` raises `RuntimeError` correctly while the passed coroutine leaks as never-awaited. Close it before re-raising, mirroring the proxy's async dispatch path. --- bellows/thread.py | 7 ++++++- tests/test_thread.py | 15 +++++++++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/bellows/thread.py b/bellows/thread.py index fb840def..c2abfb11 100644 --- a/bellows/thread.py +++ b/bellows/thread.py @@ -22,7 +22,12 @@ def run_coroutine_threadsafe(self, coroutine): if loop is None: coroutine.close() raise RuntimeError("Event loop is not running") - future = asyncio.run_coroutine_threadsafe(coroutine, loop) + try: + future = asyncio.run_coroutine_threadsafe(coroutine, loop) + except RuntimeError: + # The worker thread may close the loop after our None check + coroutine.close() + raise return asyncio.wrap_future(future, loop=current_loop) def _thread_main(self, init_task): diff --git a/tests/test_thread.py b/tests/test_thread.py index fb0c89d3..51c3b0d6 100644 --- a/tests/test_thread.py +++ b/tests/test_thread.py @@ -1,5 +1,6 @@ import asyncio from asyncio import timeout as asyncio_timeout +import inspect import threading from unittest import mock @@ -266,6 +267,20 @@ async def test_thread_run_coroutine_threadsafe_loop_not_running(): thread.run_coroutine_threadsafe(asyncio.sleep(0)) +@pytest.mark.filterwarnings("error::RuntimeWarning") +async def test_thread_run_coroutine_threadsafe_loop_closed_mid_dispatch(): + """The coroutine is closed when the loop closes between snapshot and dispatch.""" + thread = EventLoopThread() + thread.loop = asyncio.new_event_loop() + thread.loop.close() + + coro = asyncio.sleep(0) + with pytest.raises(RuntimeError): + thread.run_coroutine_threadsafe(coro) + + assert inspect.getcoroutinestate(coro) == inspect.CORO_CLOSED + + async def test_thread_task_cancellation_after_stop(thread): loop = asyncio.get_event_loop() obj = mock.MagicMock()