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()