Fixes for 3.12 and for a few hangs in specific other environments

Author: oremanj

tests/python/conftest.py

@@ -115,10 +115,11 @@ def skip(rel_id):
                 "test_log_slow_callbacks"
             )
 
-        xfail(
-            "test_tasks.py::RunCoroutineThreadsafeTests::"
-            "test_run_coroutine_threadsafe_task_factory_exception"
-        )
+        if sys.version_info < (3, 12):
+            xfail(
+                "test_tasks.py::RunCoroutineThreadsafeTests::"
+                "test_run_coroutine_threadsafe_task_factory_exception"
+            )
         if sys.version_info >= (3, 8):
             xfail(
                 "test_tasks.py::RunCoroutineThreadsafeTests::"
@@ -130,7 +131,8 @@ def skip(rel_id):
                     "test_run_coroutine_threadsafe_with_timeout"
                 )
             if sys.platform == "win32":
-                xfail("test_windows_events.py::ProactorLoopCtrlC::test_ctrl_c")
+                # hangs on 3.11+, fails without hanging on 3.8-3.10
+                skip("test_windows_events.py::ProactorLoopCtrlC::test_ctrl_c")
 
         if sys.implementation.name == "pypy":
             # This test depends on the C implementation of asyncio.Future, and
@@ -141,12 +143,6 @@ def skip(rel_id):
                 "test_inherit_without_calling_super_init"
             )
             if sys.version_info < (3, 8):
-                # This fails due to a trivial difference in how pypy handles IPv6
-                # addresses
-                xfail(
-                    "test_base_events.py::BaseEventLoopWithSelectorTests::"
-                    "test_create_connection_ipv6_scope"
-                )
                 # These tests assume CPython-style immediate finalization of
                 # objects when they become unreferenced
                 for test in (
@@ -155,6 +151,18 @@ def skip(rel_id):
                     "test_start_tls_client_reg_proto_1",
                 ):
                     xfail("test_sslproto.py::SelectorStartTLSTests::{}".format(test))
+            if sys.version_info < (3, 8) or sys.version_info >= (3, 9):
+                # This fails due to a trivial difference in how pypy handles IPv6
+                # addresses (fails in a different way on each of 3.7 and 3.9)
+                xfail(
+                    "test_base_events.py::BaseEventLoopWithSelectorTests::"
+                    "test_create_connection_ipv6_scope"
+                )
+            if sys.platform == "darwin":
+                # https://foss.heptapod.net/pypy/pypy/-/issues/3964 causes infinite loops
+                for nodeid, item in by_id.items():
+                    if "sendfile" in nodeid:
+                        item.add_marker(pytest.mark.skip)
 
         if sys.version_info >= (3, 11):
             # This tries to use a mock ChildWatcher that does something unlikely.
@@ -169,3 +177,18 @@ def skip(rel_id):
             # This tries to create a new loop from within an existing one,
             # which we don't support.
             xfail("test_locks.py::ConditionTests::test_ambiguous_loops")
+
+        if sys.version_info >= (3, 12):
+            # This test sets signal handlers from within a coroutine,
+            # which doesn't work for us because SyncTrioEventLoop runs on
+            # a non-main thread.
+            xfail("test_unix_events.py::TestFork::test_fork_signal_handling")
+
+            # This test explicitly uses asyncio.tasks._c_current_task,
+            # bypassing our monkeypatch.
+            xfail("test_tasks.py::CCurrentLoopTests::test_current_task_with_implicit_loop")
+
+            # These tests assume asyncio.sleep(0) is sufficient to run all pending tasks
+            xfail("test_futures2.py::PyFutureTests::test_task_exc_handler_correct_context")
+            xfail("test_futures2.py::CFutureTests::test_task_exc_handler_correct_context")
+

tests/test_misc.py

@@ -15,16 +15,16 @@ class Seen:
 class TestMisc:
     @pytest.mark.trio
     async def test_close_no_stop(self):
-        with pytest.raises(RuntimeError):
-            async with trio_asyncio.open_loop() as loop:
+        async with trio_asyncio.open_loop() as loop:
+            triggered = trio.Event()
 
-                def close_no_stop():
+            def close_no_stop():
+                with pytest.raises(RuntimeError):
                     loop.close()
+                triggered.set()
 
-                loop.call_soon(close_no_stop)
-
-                await trio.sleep(0.1)
-                await loop.wait_closed()
+            loop.call_soon(close_no_stop)
+            await triggered.wait()
 
     @pytest.mark.trio
     async def test_too_many_stops(self):

trio_asyncio/_async.py

@@ -1,7 +1,7 @@
 import trio
 import asyncio
 
-from ._base import BaseTrioEventLoop
+from ._base import BaseTrioEventLoop, TrioAsyncioExit
 
 
 class TrioEventLoop(BaseTrioEventLoop):
@@ -32,17 +32,23 @@ def default_exception_handler(self, context):
         asynchronous loops.
 
         """
-        # TODO: add context.get('handle') to the exception
 
+        # Call the original default handler so we get the full info in the log
+        super().default_exception_handler(context)
+
+        # Also raise an exception so it can't go unnoticed
         exception = context.get('exception')
         if exception is None:
             message = context.get('message')
             if not message:
                 message = 'Unhandled error in event loop'
-            raise RuntimeError(message)
-        else:
+            exception = RuntimeError(message)
+
+        async def propagate_asyncio_error():
             raise exception
 
+        self._nursery.start_soon(propagate_asyncio_error)
+
     def stop(self, waiter=None):
         """Halt the main loop.
 
@@ -64,7 +70,7 @@ def stop(self, waiter=None):
 
         def stop_me():
             waiter.set()
-            raise StopAsyncIteration
+            raise TrioAsyncioExit("stopping trio-asyncio loop")
 
         if self._stopped.is_set():
             waiter.set()

trio_asyncio/_base.py

@@ -37,6 +37,13 @@ def clear(self):
         pass
 
 
+# Exception raised internally to stop the main loop. Must subclass
+# SystemExit or KeyboardInterrupt in order to make it through various
+# asyncio layers.
+class TrioAsyncioExit(SystemExit):
+    pass
+
+
 class _TrioSelector(_BaseSelectorImpl):
     """A selector that hooks into a ``TrioEventLoop``.
 
@@ -503,6 +510,8 @@ async def _reader_loop(self, fd, handle):
         with handle._scope:
             try:
                 while True:
+                    if handle._cancelled:
+                        break
                     await _wait_readable(fd)
                     if handle._cancelled:
                         break
@@ -554,6 +563,8 @@ async def _writer_loop(self, fd, handle):
         with handle._scope:
             try:
                 while True:
+                    if handle._cancelled:
+                        break
                     await _wait_writable(fd)
                     if handle._cancelled:
                         break
@@ -646,7 +657,7 @@ async def _main_loop(self, task_status=trio.TASK_STATUS_IGNORED):
             with trio.CancelScope(shield=True):
                 while not self._stopped.is_set():
                     await self._main_loop_one()
-        except StopAsyncIteration:
+        except TrioAsyncioExit:
             # raised by .stop_me() to interrupt the loop
             pass
         finally:
@@ -693,7 +704,7 @@ async def _main_loop_one(self, no_wait=False):
             obj._context.run(self._nursery.start_soon, obj._run, name=obj._callback)
             await obj._started.wait()
         else:
-            obj._context.run(obj._callback, *obj._args)
+            obj._run()
 
     async def _main_loop_exit(self):
         """Finalize the loop. It may not be re-entered."""
@@ -719,7 +730,7 @@ async def _main_loop_exit(self):
                     await self._main_loop_one(no_wait=True)
                 except trio.WouldBlock:
                     break
-                except StopAsyncIteration:
+                except TrioAsyncioExit:
                     pass
 
         # Kill off unprocessed work

trio_asyncio/_loop.py

@@ -1,6 +1,7 @@
 # This code implements a clone of the asyncio mainloop which hooks into
 # Trio.
 
+import os
 import sys
 import trio
 import asyncio
@@ -181,6 +182,22 @@ def set_event_loop(self, loop):
         else:
             super().set_event_loop(loop)
 
+# get_event_loop() without a running loop is deprecated in 3.12+. The logic for emitting the
+# DeprecationWarning walks the stack looking at module names in order to associate it with
+# the first caller outside asyncio. We need to pretend to be asyncio in order for that to work.
+if sys.version_info >= (3, 12):
+    __name__ = "asyncio.fake.trio_asyncio._loop"
+
+# Make sure we don't try to continue using the Trio loop after a fork()
+def _clear_state_after_fork():
+    if _in_trio_context():
+        from trio._core._run import GLOBAL_RUN_CONTEXT
+
+        del GLOBAL_RUN_CONTEXT.task
+        del GLOBAL_RUN_CONTEXT.runner
+        current_loop.set(None)
+
+os.register_at_fork(after_in_child=_clear_state_after_fork)
 
 from asyncio import events as _aio_event
 
@@ -220,39 +237,33 @@ def _new_policy_set(new_policy):
 
 #####
 
-try:
-    _orig_run_get = _aio_event._get_running_loop
-
-except AttributeError:
-    pass
-
-else:
+_orig_run_get = _aio_event._get_running_loop
 
-    def _new_run_get():
-        try:
-            task = trio.lowlevel.current_task()
-        except RuntimeError:
-            pass
-        else:
-            # Trio context. Note: NOT current_loop.get()!
-            # See comment in _TrioPolicy.get_event_loop().
-            return task.context.get(current_loop)
-        # Not Trio context
-        return _orig_run_get()
-
-    # Must override the non-underscore-prefixed get_running_loop() too,
-    # else will use the C-accelerated one which doesn't call the patched
-    # _get_running_loop()
-    def _new_run_get_or_throw():
-        result = _new_run_get()
-        if result is None:
-            raise RuntimeError("no running event loop")
-        return result
-
-    _aio_event._get_running_loop = _new_run_get
-    _aio_event.get_running_loop = _new_run_get_or_throw
-    asyncio._get_running_loop = _new_run_get
-    asyncio.get_running_loop = _new_run_get_or_throw
+def _new_run_get():
+    try:
+        task = trio.lowlevel.current_task()
+    except RuntimeError:
+        pass
+    else:
+        # Trio context. Note: NOT current_loop.get()!
+        # See comment in _TrioPolicy.get_event_loop().
+        return task.context.get(current_loop)
+    # Not Trio context
+    return _orig_run_get()
+
+# Must override the non-underscore-prefixed get_running_loop() too,
+# else will use the C-accelerated one which doesn't call the patched
+# _get_running_loop()
+def _new_run_get_or_throw():
+    result = _new_run_get()
+    if result is None:
+        raise RuntimeError("no running event loop")
+    return result
+
+_aio_event._get_running_loop = _new_run_get
+_aio_event.get_running_loop = _new_run_get_or_throw
+asyncio._get_running_loop = _new_run_get
+asyncio.get_running_loop = _new_run_get_or_throw
 
 #####
 
@@ -282,6 +293,16 @@ def _new_loop_new():
 asyncio.set_event_loop = _new_loop_set
 asyncio.new_event_loop = _new_loop_new
 
+# current_task is implemented in C in 3.12+, which creates a problem because it
+# accesses the non-monkeypatched version of _get_running_loop()
+from asyncio import current_task as _orig_current_task
+
+def _new_current_task(loop=None):
+    return _orig_current_task(loop or _new_run_get())
+
+asyncio.tasks.current_task = _new_current_task
+asyncio.current_task = _new_current_task
+
 #####
 
 

trio_asyncio/_sync.py

@@ -6,7 +6,7 @@
 import threading
 import outcome
 
-from ._base import BaseTrioEventLoop
+from ._base import BaseTrioEventLoop, TrioAsyncioExit
 
 
 async def _sync(proc, *args):
@@ -49,7 +49,7 @@ def stop(self):
 
         def do_stop():
             self._stop_pending = False
-            raise StopAsyncIteration
+            raise TrioAsyncioExit
 
 
 #        async def stop_me():
@@ -148,7 +148,7 @@ def is_done(_):
                 while result is None:
                     try:
                         await self._main_loop_one(no_wait=True)
-                    except StopAsyncIteration:
+                    except TrioAsyncioExit:
                         pass
             except trio.WouldBlock:
                 pass