Merge pull request #81 from oremanj/loop-exit

Author: pquentin

newsfragments/80.bugfix.rst

@@ -0,0 +1,4 @@
+Previously, cancelling the context surrounding an :func:`open_loop`
+block might cause a deadlock in some cases. The ordering of operations
+during loop teardown has been improved, so this shouldn't happen
+anymore.

trio_asyncio/_base.py

@@ -687,8 +687,19 @@ async def _main_loop(self, task_status=trio.TASK_STATUS_IGNORED):
         sniffio.current_async_library_cvar.set("asyncio")
 
         try:
-            while not self._stopped.is_set():
-                await self._main_loop_one()
+            # The shield here ensures that if the context surrounding
+            # the loop is cancelled, we keep processing callbacks
+            # until we reach the callback inserted by stop().
+            # That's important to maintain the asyncio invariant
+            # that everything you schedule before stop() will run
+            # before the loop stops. In order to be safe against
+            # deadlocks, it's important that the surrounding
+            # context ensure that stop() gets called upon a
+            # cancellation. (open_loop() does this indirectly
+            # by calling _main_loop_exit().)
+            with trio.CancelScope(shield=True):
+                while not self._stopped.is_set():
+                    await self._main_loop_one()
         except StopAsyncIteration:
             # raised by .stop_me() to interrupt the loop
             pass
@@ -745,16 +756,27 @@ async def _main_loop_exit(self):
         if self._closed:
             return
 
-        self.stop()
-        await self.wait_stopped()
-
-        while True:
-            try:
-                await self._main_loop_one(no_wait=True)
-            except trio.WouldBlock:
-                break
-            except StopAsyncIteration:
-                pass
+        with trio.CancelScope(shield=True):
+            # wait_stopped() will return once _main_loop() exits.
+            # stop() inserts a callback that will cause such, and
+            # _main_loop() doesn't block except to wait for new
+            # callbacks to be added, so this should be deadlock-proof.
+            self.stop()
+            await self.wait_stopped()
+
+            # Drain all remaining callbacks, even those after an initial
+            # call to stop(). This avoids deadlocks in some cases if
+            # work is submitted to the loop after the shutdown process
+            # starts. TODO: figure out precisely what this helps with,
+            # maybe find a better way. test_wrong_context_manager_order
+            # deadlocks if we remove it for now.
+            while True:
+                try:
+                    await self._main_loop_one(no_wait=True)
+                except trio.WouldBlock:
+                    break
+                except StopAsyncIteration:
+                    pass
 
         # Kill off unprocessed work
         self._cancel_fds()

trio_asyncio/_loop.py

@@ -389,11 +389,6 @@ async def async_main(*args):
     """
 
     # TODO: make sure that there is no asyncio loop already running
-
-    def _main_loop_exit(self):
-        super()._main_loop_exit()
-        self._thread = None
-
     async with trio.open_nursery() as nursery:
         loop = TrioEventLoop(queue_len=queue_len)
         old_loop = current_loop.set(loop)
@@ -404,14 +399,11 @@ def _main_loop_exit(self):
             await yield_(loop)
         finally:
             try:
-                await loop.stop().wait()
+                await loop._main_loop_exit()
             finally:
-                try:
-                    await loop._main_loop_exit()
-                finally:
-                    loop.close()
-                    nursery.cancel_scope.cancel()
-                    current_loop.reset(old_loop)
+                loop.close()
+                nursery.cancel_scope.cancel()
+                current_loop.reset(old_loop)
 
 
 def run(proc, *args, queue_len=None):