Handle the loop closing between ThreadsafeProxy's is_closed() check and dispatch

bellows/thread.py

@@ -2,6 +2,7 @@
 from concurrent.futures import ThreadPoolExecutor
 import contextlib
 import functools
+import inspect
 import logging
 
 LOGGER = logging.getLogger(__name__)
@@ -16,7 +17,17 @@ 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")
+        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):
@@ -99,12 +110,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 asyncio.iscoroutinefunction(func):
-                future = asyncio.run_coroutine_threadsafe(call(), loop)
+                if not inspect.iscoroutinefunction(func):
+                    return None
+                future = curr_loop.create_future()
+                future.set_result(None)
+                return future
+
+            if loop.is_closed():
+                return disconnected_result()
+            if inspect.iscoroutinefunction(func):
+                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 +143,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

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
 
@@ -192,6 +193,94 @@ 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))
+
+
+@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()