Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

gh-108951: add TaskGroup.cancel() #127214

Open
wants to merge 10 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 19 additions & 47 deletions Doc/library/asyncio-task.rst
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,25 @@ and reliable way to wait for all tasks in the group to finish.

Close the given coroutine if the task group is not active.

.. method:: stop()

Stop the task group.

:meth:`~asyncio.Task.cancel` will be called on any tasks in the group that
aren't yet done, as well as the parent (body) of the group. This will
cause the task group context manager to exit *without*
:exc:`asyncio.CancelledError` being raised.

If :meth:`stop` is called before entering the task group, the group will be
stopped upon entry. This is useful for patterns where one piece of
code passes an unused :class:`asyncio.TaskGroup` instance to another in order to have
the ability to stop anything run within the group.

:meth:`stop` is idempotent and may be called after the task group has
already exited.

.. versionadded:: 3.14
belm0 marked this conversation as resolved.
Show resolved Hide resolved

Example::

async def main():
Expand Down Expand Up @@ -414,53 +433,6 @@ reported by :meth:`asyncio.Task.cancelling`.
Improved handling of simultaneous internal and external cancellations
and correct preservation of cancellation counts.

Terminating a Task Group
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These docs make sense for older versions.

Copy link
Contributor

@graingert graingert Nov 24, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably recommending a backport module on PyPI would be better

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These docs were just added in September, and backported to 3.13 and 3.12.

It's my understanding that the deletion here wouldn't affect the docs of previous versions.

As for this PR, I'd expected it to be backported as far back as is allowed by policy.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@belm0 are you interested in applying this change and any previous changes to my taskgroup backport?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is new API, so we won't backport it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm talking about backporting to pypi

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, sure. PyPI is off limits :)

------------------------

While terminating a task group is not natively supported by the standard
library, termination can be achieved by adding an exception-raising task
to the task group and ignoring the raised exception:

.. code-block:: python

import asyncio
from asyncio import TaskGroup

class TerminateTaskGroup(Exception):
"""Exception raised to terminate a task group."""

async def force_terminate_task_group():
"""Used to force termination of a task group."""
raise TerminateTaskGroup()

async def job(task_id, sleep_time):
print(f'Task {task_id}: start')
await asyncio.sleep(sleep_time)
print(f'Task {task_id}: done')

async def main():
try:
async with TaskGroup() as group:
# spawn some tasks
group.create_task(job(1, 0.5))
group.create_task(job(2, 1.5))
# sleep for 1 second
await asyncio.sleep(1)
# add an exception-raising task to force the group to terminate
group.create_task(force_terminate_task_group())
except* TerminateTaskGroup:
pass

asyncio.run(main())

Expected output:

.. code-block:: text

Task 1: start
Task 2: start
Task 1: done

Sleeping
========

Expand Down
34 changes: 34 additions & 0 deletions Lib/asyncio/taskgroups.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ def __init__(self):
self._errors = []
self._base_error = None
self._on_completed_fut = None
self._stop_on_enter = False

def __repr__(self):
info = ['']
Expand All @@ -62,6 +63,8 @@ async def __aenter__(self):
raise RuntimeError(
f'TaskGroup {self!r} cannot determine the parent task')
self._entered = True
if self._stop_on_enter:
self.stop()

return self

Expand Down Expand Up @@ -147,6 +150,10 @@ async def _aexit(self, et, exc):
# If there are no pending cancellations left,
# don't propagate CancelledError.
propagate_cancellation_error = None
# If Cancelled would actually be raised out of the TaskGroup,
# suppress it-- this is significant when using stop().
if not self._errors:
return True

# Propagate CancelledError if there is one, except if there
# are other errors -- those have priority.
Expand Down Expand Up @@ -273,3 +280,30 @@ def _on_task_done(self, task):
self._abort()
self._parent_cancel_requested = True
self._parent_task.cancel()

def stop(self):
"""Stop the task group

`cancel()` will be called on any tasks in the group that aren't yet
done, as well as the parent (body) of the group. This will cause the
task group context manager to exit *without* a Cancelled exception
being raised.

If `stop()` is called before entering the task group, the group will be
stopped upon entry. This is useful for patterns where one piece of
code passes an unused TaskGroup instance to another in order to have
the ability to stop anything run within the group.

`stop()` is idempotent and may be called after the task group has
already exited.
"""
if not self._entered:
self._stop_on_enter = True
return
if self._exiting and not self._tasks:
return
if not self._aborting:
self._abort()
if self._parent_task and not self._parent_cancel_requested:
self._parent_cancel_requested = True
self._parent_task.cancel()
65 changes: 65 additions & 0 deletions Lib/test/test_asyncio/test_taskgroups.py
Copy link
Contributor Author

@belm0 belm0 Dec 15, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

copying comment from @graingert

(please make all comments on the code so that there can be a thread and Resolve button)

can you test with eager tasks as well as regular tasks?

I think something like this:

class TestTaskGroupLazy(IsolatedAsyncioTestCase):
    loop_factory = asyncio.EventLoop


class TestTaskGroupEager(TestTaskGroupLazy):
    @staticmethod
    def loop_factory():
        loop = asyncio.EventLoop()
        loop.set_task_factory(asyncio.eager_task_factory)
        return loop

if you find the existing tests fail in eager tasks then probably just add the eager tests for your newly added tests.

Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,12 @@

import sys
import gc

belm0 marked this conversation as resolved.
Show resolved Hide resolved
import asyncio
import contextvars
import contextlib
from asyncio import taskgroups
import math
import unittest
import warnings

Expand Down Expand Up @@ -997,6 +999,69 @@ class MyKeyboardInterrupt(KeyboardInterrupt):
self.assertIsNotNone(exc)
self.assertListEqual(gc.get_referrers(exc), no_other_refs())

async def test_taskgroup_stop_children(self):
belm0 marked this conversation as resolved.
Show resolved Hide resolved
async with asyncio.TaskGroup() as tg:
tg.create_task(asyncio.sleep(math.inf))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe these tasks should look like this?

async def task(results, num):
    results.append(num)
    await asyncio.sleep(math.inf)
    results.append(-num)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So we can assert what was in results

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For this particular test, I chose a different test approach, which is to wrap in asyncio.timeout().

For the other tests using count, I'm not sure it's much value, since the test code is only a few lines and there is only one possible path through it. So count result of 0, 1, or 2 each have deterministic meaning that's obvious by looking at the code.

tg.create_task(asyncio.sleep(math.inf))
await asyncio.sleep(0)
tg.stop()

async def test_taskgroup_stop_body(self):
count = 0
async with asyncio.TaskGroup() as tg:
tg.stop()
count += 1
await asyncio.sleep(0)
count += 1
self.assertEqual(count, 1)

async def test_taskgroup_stop_idempotent(self):
count = 0
async with asyncio.TaskGroup() as tg:
tg.stop()
tg.stop()
count += 1
await asyncio.sleep(0)
count += 1
self.assertEqual(count, 1)

async def test_taskgroup_stop_after_exit(self):
async with asyncio.TaskGroup() as tg:
await asyncio.sleep(0)
tg.stop()

async def test_taskgroup_stop_before_enter(self):
tg = asyncio.TaskGroup()
tg.stop()
count = 0
async with tg:
count += 1
await asyncio.sleep(0)
count += 1
self.assertEqual(count, 1)

async def test_taskgroup_stop_before_exception(self):
async def raise_exc(parent_tg: asyncio.TaskGroup):
parent_tg.stop()
raise RuntimeError

with self.assertRaises(ExceptionGroup):
async with asyncio.TaskGroup() as tg:
tg.create_task(raise_exc(tg))
await asyncio.sleep(1)

async def test_taskgroup_stop_after_exception(self):
async def raise_exc(parent_tg: asyncio.TaskGroup):
try:
raise RuntimeError
finally:
parent_tg.stop()

with self.assertRaises(ExceptionGroup):
async with asyncio.TaskGroup() as tg:
tg.create_task(raise_exc(tg))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What will happen if some tasks cancels itself? How would this interact with .stop()?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you mean the case where a child task calls stop() on its parent TaskGroup, or something else?

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cancellations (and thus taskgroup stops) happen when the next await … actually yields to the asyncio loop. Who the caller of the cancel or stop operation is doesn't matter.

await asyncio.sleep(1)


if __name__ == "__main__":
unittest.main()
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add :meth:`~asyncio.TaskGroup.stop`.
Loading