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/4] 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/4] 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/4] 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() From 7fb32e78738cff6289f4d3991dde3ec21bff1f05 Mon Sep 17 00:00:00 2001 From: zigpy-review-bot <286747149+zigpy-review-bot@users.noreply.github.com> Date: Sat, 22 Aug 2026 03:45:18 +0200 Subject: [PATCH 4/4] Treat a stopping worker loop as unusable, not merely a closed one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `loop.is_closed()` only flips once the loop has actually been closed, so it cannot answer the question the proxy is really asking: "will this loop still run what I hand it?". Between `EventLoopThread.force_stop()` and `loop.close()` the answer is no, but the loop still looks fine: `asyncio.run_coroutine_threadsafe()` accepts the coroutine without raising, the loop stops before running it, and closing drops the pending task — so the caller (`await self._gw.disconnect()`, which has no timeout) waits on a future that is never resolved. Unlike the `TypeError` and `RuntimeError` members of this family, that failure is completely silent: no exception, no log line, no retry. `force_stop()` now publishes `stopping = True` before it schedules anything — an explicit flag, because every loop-state check is racy here: `is_closed()` is `False` for the whole window and `is_running()` is `True` until `run_forever()` returns, so both only narrow it. `ThreadsafeProxy` takes the owning `EventLoopThread` (`uart._connect()` passes it through) and treats `stopping` exactly like an already-closed loop: sync calls dropped, async calls resolved to `None`, with a warning naming the shutdown. The check sits before the coroutine is constructed, so nothing leaks. `start()` clears the flag, and `EventLoopThread.run_coroutine_threadsafe()` applies the same guard. Reported by jetliuzhe on the PR, with a self-contained reproduction; verified against bellows itself (a proxy call dispatched right after `force_stop()` never resolves, with `Task was destroyed but it is pending!` at teardown). Three new tests, all failing on the unpatched code: `test_proxy_loop_stopping_async` pins the no-hang behavior end-to-end on a real worker thread (wedged so the window is reliably open, with a callee that cannot finish inside it, as `Gateway.disconnect()` cannot); `test_proxy_loop_stopping_sync` pins the dropped sync call; `test_thread_run_coroutine_threadsafe_stopping` pins the `RuntimeError` and the closed coroutine. `test_proxy_stopping_ignored_without_thread` and `test_thread_start_clears_stopping` pin the unchanged no-thread path and the restart path. --- bellows/thread.py | 33 +++++++++++-- bellows/uart.py | 6 +-- tests/test_thread.py | 112 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 144 insertions(+), 7 deletions(-) diff --git a/bellows/thread.py b/bellows/thread.py index c2abfb11..8eb22c7c 100644 --- a/bellows/thread.py +++ b/bellows/thread.py @@ -14,12 +14,17 @@ class EventLoopThread: def __init__(self): self.loop = None self.thread_complete = None + # Published by `force_stop()` before it schedules anything. `loop.is_closed()` + # only becomes `True` once the loop has actually been closed, so it cannot answer + # "will this loop still run what I hand it?" during the shutdown window: between + # `force_stop()` and `loop.close()` the loop accepts work it will never run. + self.stopping = False def run_coroutine_threadsafe(self, coroutine): current_loop = asyncio.get_event_loop() # Snapshot: the worker thread publishes `None` when it exits loop = self.loop - if loop is None: + if loop is None or self.stopping: coroutine.close() raise RuntimeError("Event loop is not running") try: @@ -47,6 +52,9 @@ def _thread_main(self, init_task): async def start(self): current_loop = asyncio.get_event_loop() if self.loop is not None and not self.loop.is_closed(): + # Note this returns a thread that is still stopping, if one is: reusing a loop + # that is winding down is not safe, and spawning a second thread while the + # first is in its `finally` would have it close the new loop out from under us return executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix=__name__) @@ -56,6 +64,8 @@ async def start(self): async def init_task(): current_loop.call_soon_threadsafe(thread_started_future.set_result, None) + self.stopping = False + # Use current loop so current loop has a reference to the long-running thread # as one of its tasks thread_complete = current_loop.run_in_executor( @@ -67,6 +77,10 @@ async def init_task(): return thread_complete def force_stop(self): + # Published before anything is scheduled below: from here on the loop may stop at + # any moment, so work handed to it may never run. + self.stopping = True + loop = self.loop if loop is None or loop.is_closed(): return @@ -91,9 +105,12 @@ class ThreadsafeProxy: using that object's methods are done on a particular event loop """ - def __init__(self, obj, obj_loop): + def __init__(self, obj, obj_loop, loop_thread=None): self._obj = obj self._obj_loop = obj_loop + # The `EventLoopThread` running `obj_loop`, when there is one. Only it knows that a + # stop has been requested; the loop object itself still looks perfectly usable. + self._loop_thread = loop_thread def __getattr__(self, name): func = getattr(self._obj, name) @@ -111,9 +128,9 @@ def func_wrapper(*args, **kwargs): if loop == curr_loop: return call() - def disconnected_result(): + def disconnected_result(message="Attempted to use a closed event loop"): # Disconnected: sync calls are dropped, async calls resolve to None - LOGGER.warning("Attempted to use a closed event loop") + LOGGER.warning(message) if not inspect.iscoroutinefunction(func): return None future = curr_loop.create_future() @@ -122,6 +139,14 @@ def disconnected_result(): if loop.is_closed(): return disconnected_result() + if self._loop_thread is not None and self._loop_thread.stopping: + # The loop is still open, but it has been asked to stop: anything handed + # to it from here on may never run, and `run_coroutine_threadsafe()` will + # not complain. Without this branch the caller waits on a future that is + # never resolved -- a silent, unbounded hang. + return disconnected_result( + "Attempted to use an event loop that is shutting down" + ) if inspect.iscoroutinefunction(func): coro = call() try: diff --git a/bellows/uart.py b/bellows/uart.py index af274dc8..ed1497f6 100644 --- a/bellows/uart.py +++ b/bellows/uart.py @@ -105,7 +105,7 @@ async def reset(self): return await self._reset_future -async def _connect(config, api): +async def _connect(config, api, thread=None): loop = asyncio.get_event_loop() connection_done_future = loop.create_future() @@ -129,7 +129,7 @@ async def _connect(config, api): await gateway.wait_until_connected() - thread_safe_protocol = ThreadsafeProxy(gateway, loop) + thread_safe_protocol = ThreadsafeProxy(gateway, loop, thread) return thread_safe_protocol, connection_done_future @@ -140,7 +140,7 @@ async def connect(config, api, use_thread=True): await thread.start() try: protocol, connection_done = await thread.run_coroutine_threadsafe( - _connect(config, api) + _connect(config, api, thread) ) except Exception: thread.force_stop() diff --git a/tests/test_thread.py b/tests/test_thread.py index 51c3b0d6..db24fe10 100644 --- a/tests/test_thread.py +++ b/tests/test_thread.py @@ -2,6 +2,7 @@ from asyncio import timeout as asyncio_timeout import inspect import threading +import time from unittest import mock import pytest @@ -300,3 +301,114 @@ async def wait_forever(): # This will stall forever without the patch async with asyncio_timeout(1): await proxy.wait_forever() + + +@pytest.mark.filterwarnings("error::RuntimeWarning") +async def test_proxy_loop_stopping_async(thread, caplog): + """An async call dispatched after `force_stop()` resolves instead of hanging.""" + obj = mock.MagicMock() + call_count = 0 + worker_loop = thread.loop + + async def magic(): + nonlocal call_count + call_count += 1 + # Like a real `Gateway.disconnect()`, this cannot finish within the window: the + # loop stops with the task still pending, and closing drops it + await worker_loop.create_future() + + obj.test = magic + proxy = ThreadsafeProxy(obj, worker_loop, thread) + + # Wedge the worker thread so the stop cannot complete while we dispatch: the loop is + # then reliably in the window this pins, stopping but not yet closed + worker_loop.call_soon_threadsafe(lambda: time.sleep(0.5)) + thread.force_stop() + + async with asyncio_timeout(1): + # `run_coroutine_threadsafe()` accepts work for a stopping loop without raising, + # so without the patch this awaits a future that is never resolved + assert await proxy.test() is None + + assert call_count == 0 + assert "Attempted to use an event loop that is shutting down" in caplog.text + + +async def test_proxy_loop_stopping_sync(caplog): + """A sync call dispatched to a stopping loop is dropped, not silently queued.""" + loop = asyncio.new_event_loop() + try: + thread = EventLoopThread() + thread.loop = loop + thread.stopping = True + + obj = mock.MagicMock() + obj.test.return_value = None + proxy = ThreadsafeProxy(obj, loop, thread) + + proxy.test() + + assert obj.test.call_count == 0 + assert "Attempted to use an event loop that is shutting down" in caplog.text + finally: + loop.close() + + +async def test_proxy_stopping_ignored_without_thread(thread): + """A proxy with no `EventLoopThread` behaves exactly as before.""" + obj = mock.MagicMock() + obj.test.return_value = None + proxy = ThreadsafeProxy(obj, thread.loop) + + # A proxy that was not given the thread has no way to see this, and must not guess + thread.stopping = True + proxy.test() + thread.stopping = False + + await yield_other_thread(thread) + + assert obj.test.call_count == 1 + + +@pytest.mark.filterwarnings("error::RuntimeWarning") +async def test_thread_run_coroutine_threadsafe_stopping(): + """Handing a coroutine to a stopping loop raises instead of hanging.""" + thread = EventLoopThread() + thread.loop = asyncio.new_event_loop() + try: + thread.force_stop() + assert thread.stopping + assert not thread.loop.is_closed() + + coro = asyncio.sleep(0) + with pytest.raises(RuntimeError): + thread.run_coroutine_threadsafe(coro) + + # The coroutine is closed internally: no "never awaited" RuntimeWarning + assert inspect.getcoroutinestate(coro) == inspect.CORO_CLOSED + finally: + thread.loop.close() + + +async def test_thread_start_clears_stopping(): + """A restarted thread is usable again: `start()` clears the stopping flag.""" + thread = EventLoopThread() + thread_complete = await thread.start() + thread.force_stop() + assert thread.stopping + + async with asyncio_timeout(1): + await thread_complete + + await thread.start() + assert not thread.stopping + result = await thread.run_coroutine_threadsafe( + asyncio.sleep(0, mock.sentinel.result) + ) + assert result is mock.sentinel.result + + thread.force_stop() + async with asyncio_timeout(1): + await thread.thread_complete + [t.join(1) for t in threading.enumerate() if "bellows" in t.name] + assert [t for t in threading.enumerate() if "bellows" in t.name] == []