bugfix: fix cancelled tasks pending forever

When an AsyncHandle is cancelled, _main_loop_one drops it at `if obj._cancelled: return`
without running `_run`, leaving the result_future pending forever. The fix is to always
run `AsyncHandle._run`.

Author: Huan Nguyen

newsfragments/161.bugfix.rst

@@ -0,0 +1,8 @@
+When an ``AsyncHandle`` is cancelled before the ``_main_loop`` is able to 
+dequeue it (often due to heavy CPU load), ``trio-asyncio`` currently leaves 
+the underlying ``asyncio.Future`` hanging in the ``PENDING`` state forever, 
+preventing the cancellation to travel upwards back into the trio loop, 
+leading to us missing task cancellation/completion. Specifically, 
+``_main_loop_one`` greedily drops it at ``if obj._cancelled: return`` 
+without running ``_run``, leaving the ``result_future`` pending forever. 
+The fix is to always run ``AsyncHandle._run``.

tests/test_trio_asyncio.py

@@ -2,6 +2,7 @@
 import sys
 import types
 import asyncio
+import threading
 import trio
 import trio.testing
 import trio_asyncio
@@ -307,3 +308,62 @@ async def iterate_one(label, extra=""):
         }
     finally:
         sys.unraisablehook = prev_hook
+
+
+def test_async_handle_cancelled_before_dequeue():
+    """
+    Regression test for the wedge where an AsyncHandle queued onto _q_send
+    and then cancelled before _main_loop_one dequeues it would never resolve
+    its result_future.
+
+    Run trio.run in a daemon thread with a hard 5s join timeout: when the bug
+    is present, both run_aio_future and the open_loop shutdown wedge on the
+    dropped AsyncHandle, so an in-trio timeout cannot recover the test.
+    """
+
+    async def trio_main():
+        async with trio_asyncio.open_loop() as loop:
+            # Define a trio function for the AsyncHandle to wrap.
+            async def trio_proc():
+                return 42
+
+            # Queue an AsyncHandle on _q_send. trio_as_future is sync — it
+            # creates the AsyncHandle, calls _queue_handle (= _q_send.send_nowait),
+            # and returns the asyncio.Future immediately. _main_loop hasn't
+            # had a chance to run yet because we haven't yielded.
+            f = loop.trio_as_future(trio_proc)
+            assert not f.done()
+
+            # Cancel the future BEFORE yielding. wrapped_cancel marks the
+            # AsyncHandle._cancelled = True but leaves f PENDING.
+            assert f.cancel() is True
+            assert not f.done()  # the bug is here: f is PENDING despite cancel
+
+            # Now yield. _main_loop dequeues the cancelled handle. Without
+            # the fix, _main_loop_one drops it at `if obj._cancelled: return`
+            # and f stays PENDING forever. With the fix, _run is called and
+            # marks f cancelled.
+            try:
+                await loop.run_aio_future(f)
+            except asyncio.CancelledError:
+                pass
+
+            assert f.done()
+            assert f.cancelled()
+
+    error = []
+
+    def runner():
+        try:
+            trio.run(trio_main)
+        except BaseException as e:
+            error.append(e)
+
+    thread = threading.Thread(target=runner, daemon=True)
+    thread.start()
+    thread.join(timeout=5.0)
+
+    if thread.is_alive():
+        pytest.fail("AsyncHandle.result_future was never resolved within 5s")
+    if error:
+        raise error[0]

trio_asyncio/_base.py

@@ -763,18 +763,14 @@ async def _main_loop_one(self, no_wait=False):
         # Hopefully one of ours
         # but it might be a standard asyncio handle
 
-        if obj._cancelled:
-            # simply skip cancelled handlers
-            return
-
         # Don't go through the expensive nursery dance
         # if this is a sync function.
         if isinstance(obj, AsyncHandle):
             # AsyncHandle is only used to run Trio tasks, so no need to set the
             # sniffio library
             obj._context.run(self._nursery.start_soon, obj._run, name=obj._callback)
             await obj._started.wait()
-        else:
+        elif not obj._cancelled:
             prev_library, sniffio_library.name = sniffio_library.name, "asyncio"
             try:
                 obj._run()

trio_asyncio/_handles.py

@@ -161,8 +161,13 @@ def wrapped_cancel(msg=None):
 
     async def _run(self):
         self._started.set()
+
         if self._cancelled:
+            # asyncio.Handle.cancel() nulls out _callback and _args, so we
+            # cannot run the callback. We however must resolve the result_future as cancelled
             self._finished = True
+            if self._fut is not None and not self._fut.done():
+                self._fut.cancel()
             return
 
         try: