Add an `asyncio.TaskGroup.cancel` method

[smurfix]

Feature or enhancement

Proposal:

Trio/anyio have a way to cancel all tasks within a taskgroup, via taskgroup.cancel_scope.cancel().

Asyncio doesn't have cancel scopes. The way to currently do this is to cancel the task which the taskgroup is a child of, but there are problems with this approach:

• the task is a private member of the taskgroup, so the task is inaccessible through it
• what about nested taskgroups?
• also passing the task around is additional overhead and risks cancelling it when the taskgroup is no longer active

To fix this, I propose to add a taskgroup.cancelmethod which behaves like task.cancel if the taskgroup is still active (i.e. not aborting).

If this change is likely to be accepted I'm OK with writing a PR.

Has this already been discussed elsewhere?

I have already discussed this feature proposal on Discourse

Links to previous discussion of this feature:

https://discuss.python.org/t/how-to-cancel-all-tasks-created-within-same-taskgroup/30215

Linked PRs

• gh-108951: document how to terminate an asyncio.TaskGroup #123837
• [3.13] gh-108951: Document how to terminate an asyncio.TaskGroup (GH-123837) #123956
• [3.12] gh-108951: Document how to terminate an asyncio.TaskGroup (GH-123837) #123957
• gh-108951: add TaskGroup.cancel() #127214

[gvanrossum]

I am afraid that I don't understand the problem well enough. Could you explain the need for this and the desired behavior in edge cases better? A bullet like "what about nested taskgroups" makes me wonder -- what about them? I have no idea what structure your app has (and I haven't been following Trio or anyio so referencing that doesn't help me).

[smurfix]

Consider this test program.

import asyncio

async def t1():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(asyncio.sleep(1))
        tg._parent_task.cancel() # asyncio.current-task().cancel()
    assert False, "not reached"

try:
    asyncio.run(t1())
    assert False, "not reached"
except asyncio.CancelledError:
    print("Cancelled")

The cancellation always bubbles out of the taskgroup.

Here's the anyio equivalent:

import anyio

async def t2():
    async with anyio.create_task_group() as tg:
        tg.start_soon(anyio.sleep,1)
        tg.cancel_scope.cancel()
    print("after anyio cancel")

anyio.run(t2)

While asyncio doesn't have cancel scopes, that's an implementation detail. The point is that, when cancelled via its own .cancel method, the taskgroup swallows the exception (after cleaning up its subtasks of course) and simply leaves the scope.

As to code structure, let's assume I have a complex program with parts that want to be restarted separately if/when their configuration changes. With a cancel-able taskgroup, this is easy, each subpart can do

async def subsys(config):
    while True:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(wait_changed(tg.cancel, config))
            ... # now do whatever the config tells us to do

where wait_changed runs the callback (cancelling the taskgroup) as soon as the configuration starts changing, but waits for the change to stabilize before it itself exits.

This is difficult to do without TaskGroup.cancel. I'd have to raise some exception from within wait_changed instead of using a callback, and catch that outside of the taskgroup. Then I need to add a separate way to do the stabilization part, which probably means I need to create an object for the wait_changed part instead of simply starting a subtask. So these two lines I need for config-change handling blow up to ten lines of boilerplate I need to sprinkle my program with, which is kindof suboptimal.

[gvanrossum]

It feels, at least to me, that "cancel" elicits all the wrong connotations, in an asyncio context, because it sounds like it would cause a CancelledError exception to propagate out.

It looks like the functionality you are looking for is a way to stop all the tasks running in the task group but then make the task group's async with block exit "peacefully" (not raising).

If you come up with a name other than "cancel" (maybe "stop"? but up to you) and a PR that implements it, I will consider it. Try not to use Trio/anyio as guidance or model, since we should not assume asyncio users are familiar with those.

[gvanrossum]

Ugh.

Can you please refrain from this kind of negative language? It's not helping your case.

[smurfix]

Sorry. Reworded.

[smurfix]

If you come up with a name other than "cancel" (maybe "stop"? but up to you) and a PR that implements it, I will consider it.

Will do.

[sobolevn]

I had some time today to experiment with this. Here's a patch that seems to do the very basic thing:

diff --git Lib/asyncio/taskgroups.py Lib/asyncio/taskgroups.py
index f2ee9648c43..968a539f5dd 100644
--- Lib/asyncio/taskgroups.py
+++ Lib/asyncio/taskgroups.py
@@ -150,25 +150,34 @@ async def __aexit__(self, et, exc, tb):
             # cycles (bad for GC); let's not keep a reference to
             # a bunch of them.
             try:
-                me = BaseExceptionGroup('unhandled errors in a TaskGroup', self._errors)
-                raise me from None
+                raise BaseExceptionGroup(
+                    'unhandled errors in a TaskGroup',
+                    self._errors
+                ) from None
+            except* _TaskGroupStop:
+                pass
             finally:
                 self._errors = None
 
+        # Direct `.stop()` call in `TaskGroup` body, like:
+        #
+        # async with TaskGroup() as tg:
+        #     if some_condition:
+        #         tg.stop()  <- here
+        if issubclass(et, _TaskGroupStop):
+            return True
+
+    def stop(self):
+        """Stop all the tasks in the task group."""
+        self._check_state()
+        raise _TaskGroupStop()
+
     def create_task(self, coro, *, name=None, context=None):
         """Create a new task in this group and return it.
 
         Similar to `asyncio.create_task`.
         """
-        if not self._entered:
-            coro.close()
-            raise RuntimeError(f"TaskGroup {self!r} has not been entered")
-        if self._exiting and not self._tasks:
-            coro.close()
-            raise RuntimeError(f"TaskGroup {self!r} is finished")
-        if self._aborting:
-            coro.close()
-            raise RuntimeError(f"TaskGroup {self!r} is shutting down")
+        self._check_state(coro)
         if context is None:
             task = self._loop.create_task(coro, name=name)
         else:
@@ -184,6 +193,19 @@ def create_task(self, coro, *, name=None, context=None):
             task.add_done_callback(self._on_task_done)
         return task
 
+    def _check_state(self, coro=None):
+        error = None
+        if not self._entered:
+            error = RuntimeError(f"TaskGroup {self!r} has not been entered")
+        if self._exiting and not self._tasks:
+            error = RuntimeError(f"TaskGroup {self!r} is finished")
+        if self._aborting:
+            error = RuntimeError(f"TaskGroup {self!r} is shutting down")
+        if error is not None:
+            if coro is not None:
+                coro.close()
+            raise error
+
     # Since Python 3.8 Tasks propagate all exceptions correctly,
     # except for KeyboardInterrupt and SystemExit which are
     # still considered special.
@@ -250,3 +272,7 @@ def _on_task_done(self, task):
             self._abort()
             self._parent_cancel_requested = True
             self._parent_task.cancel()
+
+
+class _TaskGroupStop(Exception):
+    """Exception to stop all the tasks in a task group."""

It allows to stop the TaskGroup gracefully:

• from tasks
• from the TaskGroup's body

I used this code to test the concept:

import asyncio
from asyncio import TaskGroup

from random import random

async def job(i, group=None):
    while True:
        print(f"Task {i}")
        await asyncio.sleep(1)
        if random() < 0.4:
            print(f"Task {i} cancelling others")
            group.stop()
            return

async def main():
    async with TaskGroup() as group:
        group.create_task(job(1, group))
        group.create_task(job(2, group))
        # TODO: try this as well
        # group.stop()

asyncio.run(main())

It seems to work for the general case, but I am not confident enough in it to submit a PR. Maybe someone else can improve it :)

[picnixz]

Playing the devil's advocate but should we care of a try-except that could trap this exception?

async with TaskGroup() as group: 
    group.create_task(job(1, group))
	try:
		group.stop()
	except:
		pass

Alternatively, can we use an event + waiter task to cancel the tasks in the group? or would this solution too tricky to implement?