Add an asyncio.TaskGroup.cancel method

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

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).

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.

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.

Ugh.

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

Sorry. Reworded.

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.

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 :)

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?

I may be mistaken, but it appears this discussion is going around in circles. Why exactly was the title changed from stop to cancel? It seems we're not even on the same page about whether the desired API should exit the async with clause without error (if all tasks were cancelled without errors) or not.

Maybe it's time for the key participants to start collaborating on a PEP? If you can get a draft PEP ready and rough consensus around it I am willing to sponsoring it. I am proposing this as an alternative to continuing the debate without making progress for another 6 months.

One suggestion for a draft PEP: make minimal use of references to Trio. We can't expect the ultimate deciders (the SC; and before them myself or whichever core dev ends up sponsoring) to understand much of the ins and outs of Trio, so motivations of the form "this is how Trio does it" are not helpful for the deciders (especially since this appears largely about cancellation, and that's an are where asyncio and Trio differ fundamentally).

You're also all welcome to keep arguing like you have been, but there has to be some kind of consensus. Currently the discussion is hard to follow unless you have read the entire issue, which is unreasonable at this point.

Why exactly was the title changed from stop to cancel?

Yury balked at stop() in the PR, so I changed it back to cancel().

Though I suggested starting a PEP when asking for the issue to be reopened, I wasn't imagining this level of controversy 😓

Why exactly was the title changed from stop to cancel?

Yury balked at stop() in the PR, so I changed it back.

But he was just asking why — not telling you to change it. You could maybe have answered “because this design doesn’t raise CancelledError out of the asynchronous with block”?

One suggestion for a draft PEP: make minimal use of references to Trio. We can't expect the ultimate deciders (the SC; and before them myself or whichever core dev ends up sponsoring) to understand much of the ins and outs of Trio, so motivations of the form "this is how Trio does it" are not helpful for the deciders (especially since this appears largely about cancellation, and that's an are where asyncio and Trio differ fundamentally).

I agree with Guido, fwiw this is the reason I don't participate much in these discussions because to me it feels more about making asyncio behave like trio or anyio etc and I don't understand much of those projects. Both projects are fundamentally different in many regards for that to happen so IMO it would be better to propose concrete semantics in terms of how things happen in asyncio rather than any other such projects.

Though I suggested starting a PEP when asking for the issue to be reopened, I wasn't imagining this level of controversy 😓

It seems there are at least two dimensions of controversy, and it also seems we're going back and forth on at least one?

Whether it should raise CancelledError or exits cleanly after all tasks are cancelled cleanly (and from the outcome of this I would derive a preference for the name -- stop() if it exits cleanly, cancel() if it raises CancelledError -- though I see in he PR that you prefer cancel() regardless; nevertheless we should decide whether it raises CancelledError or not first, IMO).
Whether the use case is important enough to add such an API at all. Connected to this is whether it can be done without too much fuss and with the correct semantics using the existing APIs; or by making an existing internal API public, which would argue that it's not really adding new functionality. But it appears we have conclusively proven that we can't just promote an existing API, and the only correct way using existing APIs seems to be by adding a dummy task that raises a unique exception, and catching that, which is hardly clean. You also appear to have shown (by looking at the popularity of the corresponding Trio API) that the use case is important -- although I'd like to see at least a few of those Trio uses for myself to understand how it's used in Trio and whether we can really draw conclusions for asyncio from that.

Maybe you can try to address these bullets separately and drive consensus that way (instead of by drafting a PEP)? You'd have to go to battle a bit with the faction (or one person?) who want the async with block to raise CanceledError. Does that desire come largely out of the desire to name the API cancel()? Or is there a practical reason, like making the coding of various realistic-sounding examples more convenient? Arguments from a pure desire of consistency are sometimes red herrings.

Finally: If I'm the only person here who believes that if the async with block can exit cleanly, then the API should be called stop(), I'll give up on that.

I'd like to see at least a few of those Trio uses for myself

The canonical example is Happy Eyeballs-style concurrency, i.e. start a few tasks that all try to produce a result, the first one that succeeds wins, all the others get cancelled, you return the result of the winning task.

This is easy enough with a taskgroup that doesn't raise an exception when getting cancelled. If you don't have that, you need a wrapper to catch the exception. That wrapper needs to distinguish "the taskgroup has been (self-)cancelled" from "the cancellation was triggered externally and thus must propagate". That's boilerplate which people can and will get wrong. (Consider what happens when, due to scheduling quirks, these coincide …)

Another example: a network connection. When you decide to disconnect, you self-cancel the whole connection handling taskgroup (sometimes these are quite complex: send task, receive task, background ping task …) and you're done. Why would you propagate that cancellation? Worse, if the network connection was to retrieve a result from the other side, again you need that wrapper.

I'd ask the other way 'round: what's the usecase of propagating that cancellation? Because frankly I don't see one, esp. given that you can do it that way simply by cancelling the current task instead.

Whether the use case is important enough to add such an API at all.

Anecdotally, at my current employer we have a lot of hardware drivers that do something similar to the following:

class MyDriver:

    async def arm(self):
        self._background_tasks = []
        self._background_tasks.append(asyncio.create_task(...))
        self._background_tasks.append(asyncio.create_task(...))
        self._background_tasks.append(asyncio.create_task(...))

    async def disarm(self):
        for t in self._background_tasks:
            t.cancel()
            with contextlib.suppress(asyncio.CancelledError):
               await t

People don't seem to use task groups because they aren't providing any ergonomic benefit - you still have to manually track all the background tasks:

class MyDriver:

    async def arm(self):
        self._background_tasks = []
        self._context = AsyncExitStack()
        self._task_group = await self._context.enter_async_context(asyncio.TaskGroup())
        self._background_tasks.append(self._task_group.create_task(...))
        self._background_tasks.append(self._task_group.create_task(...))
        self._background_tasks.append(self._task_group.create_task(...))

    async def disarm(self):
        for t in self._background_tasks:
            t.cancel()
        await self._context.aclose()

However, if there were a TaskGroup.cancel method that didn't raise (or could optionally) it'd be great to do:

class MyDriver:

    async def arm(self):
        self._context = AsyncExitStack()
        self._task_group = await self._context.enter_async_context(asyncio.TaskGroup())
        self._task_group.create_task(...)
        self._task_group.create_task(...)
        self._task_group.create_task(...)

    async def disarm(self):
        self._task_group.cancel()
        await self._context.aclose()

How do you feel about this API? agronholm/anyio#890

Anyways, with the master version of AnyIO, you would write the code like this:

from anyio import AsyncContextManagerMixin, create_task_group

class MyDriver(AsyncContextManagerMixin):

    @asynccontextmanager
    async def __asynccontextmanager__():
        async with create_task_group() as tg:
            tg.start_soon(...)
            tg.start_soon(...)
            tg.start_soon(...)
            yield
            tg.cancel_scope.cancel()

This, of course, implies that you would use this class as an async context manager.

pythongh-108951: add TaskGroup.cancel() (python#127214)

gh-108951: document addition of TaskGroup.cancel() (#149031)

Navigation Menu

Repository navigation

Add an `asyncio.TaskGroup.cancel` method #108951

Feature or enhancement

Proposal:

Has this already been discussed elsewhere?

Links to previous discussion of this feature:

Linked PRs

66 remaining items

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Add an asyncio.TaskGroup.cancel method #108951

Description

Feature or enhancement

Proposal:

Has this already been discussed elsewhere?

Links to previous discussion of this feature:

Linked PRs

Activity

gvanrossum commented on Sep 5, 2023

m-u-xyz commented on Sep 6, 2023

gvanrossum commented on Sep 6, 2023

gvanrossum commented on Sep 6, 2023

m-u-xyz commented on Sep 6, 2023

m-u-xyz commented on Sep 6, 2023

sobolevn commented on Sep 5, 2024

picnixz commented on Sep 5, 2024

66 remaining items

gvanrossum commented on Jan 2, 2025

belm0 commented on Jan 4, 2025

gvanrossum commented on Jan 4, 2025

kumaraditya303 commented on Jan 4, 2025

gvanrossum commented on Jan 4, 2025

m-u-xyz commented on Jan 4, 2025

rmorshea commented on Jun 30, 2025

agronholm commented on Jul 1, 2025

agronholm commented on Jul 1, 2025

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions

Add an `asyncio.TaskGroup.cancel` method #108951