Fixed TaskGroup and CancelScope exit issues on asyncio

.github/workflows/test.yml

@@ -61,14 +61,14 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest]
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", pypy-3.10]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", pypy-3.10]
         include:
         - os: macos-latest
-          python-version: "3.8"
+          python-version: "3.9"
         - os: macos-latest
           python-version: "3.12"
         - os: windows-latest
-          python-version: "3.8"
+          python-version: "3.9"
         - os: windows-latest
           python-version: "3.12"
     runs-on: ${{ matrix.os }}

docs/versionhistory.rst

@@ -5,6 +5,7 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- Dropped support for Python 3.8
 - Improved the performance of ``anyio.Lock`` and ``anyio.Semaphore`` on asyncio (even up
   to 50 %)
 - Added the ``fast_acquire`` parameter to ``anyio.Lock`` and ``anyio.Semaphore`` to
@@ -29,6 +30,12 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Fixed ``to_process.run_sync()`` failing to initialize if ``__main__.__file__`` pointed
   to a file in a nonexistent directory
   (`#696 <https://github.com/agronholm/anyio/issues/696>`_)
+- Fixed 100% CPU use on asyncio while waiting for an exiting task group to finish while
+  said task group is within a cancelled cancel scope
+  (`#695 <https://github.com/agronholm/anyio/issues/695>`_)
+- Fixed cancel scopes on asyncio not reraising ``CancelledError`` on exit while the
+  enclosing cancel scope has been effectively cancelled
+  (`#698 <https://github.com/agronholm/anyio/issues/698>`_)
 - Fixed ``AssertionError: feed_data after feed_eof`` on asyncio when a subprocess is
   closed early, before its output has been read
   (`#490 <https://github.com/agronholm/anyio/issues/490>`_)
@@ -42,6 +49,8 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
   arrives in an exception group)
 - Fixed support for Linux abstract namespaces in UNIX sockets that was broken in v4.2
   (#781 <https://github.com/agronholm/anyio/issues/781>_; PR by @tapetersen)
+- Fixed asyncio task groups not yielding control to the event loop at exit if there were
+  no child tasks to wait on
 
 **4.4.0**
 

pyproject.toml

@@ -19,14 +19,13 @@ classifiers = [
     "Typing :: Typed",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.8",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
 ]
-requires-python = ">= 3.8"
+requires-python = ">= 3.9"
 dependencies = [
     "exceptiongroup >= 1.0.2; python_version < '3.11'",
     "idna >= 2.8",
@@ -128,7 +127,7 @@ show_missing = true
 [tool.tox]
 legacy_tox_ini = """
 [tox]
-envlist = pre-commit, py38, py39, py310, py311, py312, py313, pypy3
+envlist = pre-commit, py39, py310, py311, py312, py313, pypy3
 skip_missing_interpreters = true
 minversion = 4.0.0
 

src/anyio/_backends/_asyncio.py

@@ -20,9 +20,18 @@
 )
 from asyncio.base_events import _run_until_complete_cb  # type: ignore[attr-defined]
 from collections import OrderedDict, deque
-from collections.abc import AsyncIterator, Iterable
+from collections.abc import (
+    AsyncGenerator,
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Collection,
+    Coroutine,
+    Iterable,
+    Sequence,
+)
 from concurrent.futures import Future
-from contextlib import suppress
+from contextlib import AbstractContextManager, suppress
 from contextvars import Context, copy_context
 from dataclasses import dataclass
 from functools import partial, wraps
@@ -42,15 +51,7 @@
 from typing import (
     IO,
     Any,
-    AsyncGenerator,
-    Awaitable,
-    Callable,
-    Collection,
-    ContextManager,
-    Coroutine,
     Optional,
-    Sequence,
-    Tuple,
     TypeVar,
     cast,
 )
@@ -444,23 +445,48 @@ def __exit__(
 
         host_task_state.cancel_scope = self._parent_scope
 
-        # Restart the cancellation effort in the closest directly cancelled parent
-        # scope if this one was shielded
+        # We only swallow the exception iff it was an AnyIO CancelledError, either
+        # directly as exc_val or inside an exception group and there are no cancelled
+        # parent cancel scopes visible to us here
+        not_swallowed_exceptions = 0
+        swallow_exception = False
+        if exc_val is not None:
+            for exc in iterate_exceptions(exc_val):
+                if self._cancel_called and isinstance(exc, CancelledError):
+                    if not (swallow_exception := self._uncancel(exc)):
+                        not_swallowed_exceptions += 1
+                else:
+                    not_swallowed_exceptions += 1
+
+        # Restart the cancellation effort in the closest visible, cancelled parent
+        # scope if necessary
         self._restart_cancellation_in_parent()
+        return swallow_exception and not not_swallowed_exceptions
 
-        if self._cancel_called and exc_val is not None:
-            for exc in iterate_exceptions(exc_val):
-                if isinstance(exc, CancelledError):
-                    self._cancelled_caught = self._uncancel(exc)
-                    if self._cancelled_caught:
-                        break
+    @property
+    def _effectively_cancelled(self) -> bool:
+        cancel_scope: CancelScope | None = self
+        while cancel_scope is not None:
+            if cancel_scope._cancel_called:
+                return True
 
-            return self._cancelled_caught
+            if cancel_scope.shield:
+                return False
 
-        return None
+            cancel_scope = cancel_scope._parent_scope
+
+        return False
+
+    @property
+    def _parent_cancellation_is_visible_to_us(self) -> bool:
+        return (
+            self._parent_scope is not None
+            and not self.shield
+            and self._parent_scope._effectively_cancelled
+        )
 
     def _uncancel(self, cancelled_exc: CancelledError) -> bool:
-        if sys.version_info < (3, 9) or self._host_task is None:
+        if self._host_task is None:
             self._cancel_calls = 0
             return True
 
@@ -469,10 +495,31 @@ def _uncancel(self, cancelled_exc: CancelledError) -> bool:
             while self._cancel_calls:
                 self._cancel_calls -= 1
                 if self._host_task.uncancel() <= self._cancelling:
-                    return True
+                    break
+
+        while True:
+            if (
+                cancelled_exc.args
+                and isinstance(cancelled_exc.args[0], str)
+                and cancelled_exc.args[0].startswith("Cancelled by cancel scope ")
+            ):
+                # Only swallow the cancellation exception if it's an AnyIO cancel
+                # exception and there are no other cancel scopes down the line pending
+                # cancellation
+                self._cancelled_caught = (
+                    self._effectively_cancelled
+                    and not self._parent_cancellation_is_visible_to_us
+                )
+                return self._cancelled_caught
 
-        self._cancel_calls = 0
-        return f"Cancelled by cancel scope {id(self):x}" in cancelled_exc.args
+            # Sometimes third party frameworks catch a CancelledError and raise a new
+            # one, so as a workaround we have to look at the previous ones in
+            # __context__ too for a matching cancel message
+            if isinstance(cancelled_exc.__context__, CancelledError):
+                cancelled_exc = cancelled_exc.__context__
+                continue
+
+            return False
 
     def _timeout(self) -> None:
         if self._deadline != math.inf:
@@ -496,19 +543,16 @@ def _deliver_cancellation(self, origin: CancelScope) -> bool:
         should_retry = False
         current = current_task()
         for task in self._tasks:
+            should_retry = True
             if task._must_cancel:  # type: ignore[attr-defined]
                 continue
 
             # The task is eligible for cancellation if it has started
-            should_retry = True
             if task is not current and (task is self._host_task or _task_started(task)):
                 waiter = task._fut_waiter  # type: ignore[attr-defined]
                 if not isinstance(waiter, asyncio.Future) or not waiter.done():
                     origin._cancel_calls += 1
-                    if sys.version_info >= (3, 9):
-                        task.cancel(f"Cancelled by cancel scope {id(origin):x}")
-                    else:
-                        task.cancel()
+                    task.cancel(f"Cancelled by cancel scope {id(origin):x}")
 
         # Deliver cancellation to child scopes that aren't shielded or running their own
         # cancellation callbacks
@@ -546,17 +590,6 @@ def _restart_cancellation_in_parent(self) -> None:
 
             scope = scope._parent_scope
 
-    def _parent_cancelled(self) -> bool:
-        # Check whether any parent has been cancelled
-        cancel_scope = self._parent_scope
-        while cancel_scope is not None and not cancel_scope._shield:
-            if cancel_scope._cancel_called:
-                return True
-            else:
-                cancel_scope = cancel_scope._parent_scope
-
-        return False
-
     def cancel(self) -> None:
         if not self._cancel_called:
             if self._timeout_handle:
@@ -663,38 +696,43 @@ async def __aexit__(
         exc_val: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> bool | None:
-        ignore_exception = self.cancel_scope.__exit__(exc_type, exc_val, exc_tb)
         if exc_val is not None:
             self.cancel_scope.cancel()
             if not isinstance(exc_val, CancelledError):
                 self._exceptions.append(exc_val)
 
-        cancelled_exc_while_waiting_tasks: CancelledError | None = None
-        while self._tasks:
-            try:
-                await asyncio.wait(self._tasks)
-            except CancelledError as exc:
-                # This task was cancelled natively; reraise the CancelledError later
-                # unless this task was already interrupted by another exception
-                self.cancel_scope.cancel()
-                if cancelled_exc_while_waiting_tasks is None:
-                    cancelled_exc_while_waiting_tasks = exc
+        try:
+            if self._tasks:
+                with CancelScope() as wait_scope:
+                    while self._tasks:
+                        try:
+                            await asyncio.wait(self._tasks)
+                        except CancelledError as exc:
+                            # Shield the scope against further cancellation attempts,
+                            # as they're not productive (#695)
+                            wait_scope.shield = True
+                            self.cancel_scope.cancel()
+                            if exc_val is None:
+                                exc_val = exc
+            else:
+                # If there are no child tasks to wait on, run at least one checkpoint
+                # anyway
+                await AsyncIOBackend.cancel_shielded_checkpoint()
 
-        self._active = False
-        if self._exceptions:
-            raise BaseExceptionGroup(
-                "unhandled errors in a TaskGroup", self._exceptions
-            )
+            self._active = False
+            if self._exceptions:
+                raise BaseExceptionGroup(
+                    "unhandled errors in a TaskGroup", self._exceptions
+                )
+            elif exc_val:
+                raise exc_val
+        except BaseException as exc:
+            if self.cancel_scope.__exit__(type(exc), exc, exc.__traceback__):
+                return True
 
-        # Raise the CancelledError received while waiting for child tasks to exit,
-        # unless the context manager itself was previously exited with another
-        # exception, or if any of the  child tasks raised an exception other than
-        # CancelledError
-        if cancelled_exc_while_waiting_tasks:
-            if exc_val is None or ignore_exception:
-                raise cancelled_exc_while_waiting_tasks
+            raise
 
-        return ignore_exception
+        return self.cancel_scope.__exit__(exc_type, exc_val, exc_tb)
 
     def _spawn(
         self,
@@ -730,7 +768,7 @@ def task_done(_task: asyncio.Task) -> None:
                     if not isinstance(exc, CancelledError):
                         self._exceptions.append(exc)
 
-                    if not self.cancel_scope._parent_cancelled():
+                    if not self.cancel_scope._effectively_cancelled:
                         self.cancel_scope.cancel()
                 else:
                     task_status_future.set_exception(exc)
@@ -806,7 +844,7 @@ async def start(
 # Threads
 #
 
-_Retval_Queue_Type = Tuple[Optional[T_Retval], Optional[BaseException]]
+_Retval_Queue_Type = tuple[Optional[T_Retval], Optional[BaseException]]
 
 
 class WorkerThread(Thread):
@@ -955,22 +993,22 @@ class Process(abc.Process):
     _stderr: StreamReaderWrapper | None
 
     async def aclose(self) -> None:
-        with CancelScope(shield=True):
+        with CancelScope(shield=True) as scope:
             if self._stdin:
                 await self._stdin.aclose()
             if self._stdout:
                 await self._stdout.aclose()
             if self._stderr:
                 await self._stderr.aclose()
 
-        try:
-            await self.wait()
-        except BaseException:
-            self.kill()
-            with CancelScope(shield=True):
+            scope.shield = False
+            try:
                 await self.wait()
-
-            raise
+            except BaseException:
+                scope.shield = True
+                self.kill()
+                await self.wait()
+                raise
 
     async def wait(self) -> int:
         return await self._process.wait()
@@ -2015,9 +2053,7 @@ def has_pending_cancellation(self) -> bool:
 
         if task_state := _task_states.get(task):
             if cancel_scope := task_state.cancel_scope:
-                return cancel_scope.cancel_called or (
-                    not cancel_scope.shield and cancel_scope._parent_cancelled()
-                )
+                return cancel_scope._effectively_cancelled
 
         return False
 
@@ -2101,7 +2137,7 @@ async def _call_in_runner_task(
     ) -> T_Retval:
         if not self._runner_task:
             self._send_stream, receive_stream = create_memory_object_stream[
-                Tuple[Awaitable[Any], asyncio.Future]
+                tuple[Awaitable[Any], asyncio.Future]
             ](1)
             self._runner_task = self.get_loop().create_task(
                 self._run_tests_and_fixtures(receive_stream)
@@ -2463,7 +2499,7 @@ async def connect_tcp(
         cls, host: str, port: int, local_address: IPSockAddrType | None = None
     ) -> abc.SocketStream:
         transport, protocol = cast(
-            Tuple[asyncio.Transport, StreamProtocol],
+            tuple[asyncio.Transport, StreamProtocol],
             await get_running_loop().create_connection(
                 StreamProtocol, host, port, local_addr=local_address
             ),
@@ -2642,7 +2678,7 @@ def current_default_thread_limiter(cls) -> CapacityLimiter:
     @classmethod
     def open_signal_receiver(
         cls, *signals: Signals
-    ) -> ContextManager[AsyncIterator[Signals]]:
+    ) -> AbstractContextManager[AsyncIterator[Signals]]:
         return _SignalReceiver(signals)
 
     @classmethod