Clear EventLoopThread.loop before closing it

bellows/thread.py

@@ -27,8 +27,11 @@
             self.loop.run_until_complete(init_task)
             self.loop.run_forever()
         finally:
-            self.loop.close()
-            self.loop = None
+            # Publish `None` before closing: `self.loop` then never names a closed loop, and
+            # this thread no longer writes `self.loop` after a concurrent `start()` may have
+            # replaced it.
+            loop, self.loop = self.loop, None
+            loop.close()
 
     async def start(self):
         current_loop = asyncio.get_event_loop()
@@ -100,7 +103,7 @@
                 # Disconnected
                 LOGGER.warning("Attempted to use a closed event loop")
                 return
             if asyncio.iscoroutinefunction(func):
                 future = asyncio.run_coroutine_threadsafe(call(), loop)
                 return asyncio.wrap_future(future, loop=curr_loop)
             else:

tests/test_thread.py

@@ -29,6 +29,37 @@ def mockrun(task):
     assert loopmock.close.call_count == 1
 
 
+async def test_thread_clears_loop_before_closing(monkeypatch):
+    """`self.loop` is cleared before the loop is closed, so it never names a closed loop."""
+    thread = EventLoopThread()
+    real_new_event_loop = asyncio.new_event_loop
+    loop_at_close = mock.sentinel.close_not_called
+
+    def new_event_loop():
+        loop = real_new_event_loop()
+        real_close = loop.close
+
+        def close():
+            nonlocal loop_at_close
+            loop_at_close = thread.loop
+            real_close()
+
+        loop.close = close
+        return loop
+
+    monkeypatch.setattr(asyncio, "new_event_loop", new_event_loop)
+
+    thread_complete = await thread.start()
+    thread.force_stop()
+    async with asyncio_timeout(1):
+        await thread_complete
+
+    assert loop_at_close is None
+    assert thread.loop is None
+
+    [t.join(1) for t in threading.enumerate() if "bellows" in t.name]
+
+
 class ExceptionCollector:
     def __init__(self):
         self.exceptions = []