-
-
Notifications
You must be signed in to change notification settings - Fork 32.1k
bpo-46752: Introduce task groups in asyncio #31270
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
Merged
Merged
Changes from all commits
Commits
Show all changes
24 commits
Select commit
Hold shift + click to select a range
7edee0b
Integrate task groups from EdgeDb
gvanrossum f495375
Make test_taskgroups.py run and pass
gvanrossum a87275a
Rename taskgroup to taskgroups in the test code
gvanrossum 4df0acc
Export TaskGroup from asyncio; remove __future__ import
gvanrossum 56db921
Only keep the newest _is_base_error() and _task_cancel()
gvanrossum 500581e
Get rid of MultiError in favor of ExceptionGroup
gvanrossum 4843e94
Add TaskGroupError to __all__
gvanrossum 63e712d
Avoid DeprecationWarning: There is no current event loop
gvanrossum d233dd1
Prevent warning "test altered the execution environment"
gvanrossum af574d5
Get rid of custom TaskGroupError
gvanrossum 299f366
Update comments explaining why test 21 doesn't work
gvanrossum 9de3c87
Add tests showing that 'plain' BaseExceptions work
gvanrossum 0e1355d
Allow creating new tasks while __aexit__ is waiting
gvanrossum 77ec0e4
Add an API to Task to manage 'cancel_requested' flag
gvanrossum 17b64b5
Add tests for .cancelling() and .uncancel()
gvanrossum 5e3f4b9
Merge remote-tracking branch 'origin/main' into taskgroups
gvanrossum 0b9bccd
📜🤖 Added by blurb_it.
blurb-it[bot] 137ebe6
Replace EdgeDb copyright with a simpler attribution
gvanrossum f693c1c
Use task.cancelling() in task repr instead of access to private attri…
asvetlov b83734c
Change the internal imports
gvanrossum de3d820
Avoid needing self.loop in test
gvanrossum 9712241
Make test 14 more robust
gvanrossum b3d4d18
Update Lib/asyncio/taskgroups.py
1st1 c1e5d64
Update Lib/test/test_asyncio/test_taskgroups.py
1st1 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,235 @@ | ||
# Adapted with permission from the EdgeDB project. | ||
|
||
|
||
__all__ = ["TaskGroup"] | ||
|
||
import itertools | ||
import textwrap | ||
import traceback | ||
import types | ||
import weakref | ||
|
||
from . import events | ||
from . import exceptions | ||
from . import tasks | ||
|
||
class TaskGroup: | ||
|
||
def __init__(self, *, name=None): | ||
if name is None: | ||
self._name = f'tg-{_name_counter()}' | ||
else: | ||
self._name = str(name) | ||
|
||
self._entered = False | ||
self._exiting = False | ||
self._aborting = False | ||
self._loop = None | ||
self._parent_task = None | ||
self._parent_cancel_requested = False | ||
self._tasks = weakref.WeakSet() | ||
self._unfinished_tasks = 0 | ||
self._errors = [] | ||
self._base_error = None | ||
self._on_completed_fut = None | ||
|
||
def get_name(self): | ||
return self._name | ||
|
||
def __repr__(self): | ||
msg = f'<TaskGroup {self._name!r}' | ||
if self._tasks: | ||
msg += f' tasks:{len(self._tasks)}' | ||
if self._unfinished_tasks: | ||
msg += f' unfinished:{self._unfinished_tasks}' | ||
if self._errors: | ||
msg += f' errors:{len(self._errors)}' | ||
if self._aborting: | ||
msg += ' cancelling' | ||
elif self._entered: | ||
msg += ' entered' | ||
msg += '>' | ||
return msg | ||
|
||
async def __aenter__(self): | ||
if self._entered: | ||
raise RuntimeError( | ||
f"TaskGroup {self!r} has been already entered") | ||
self._entered = True | ||
|
||
if self._loop is None: | ||
self._loop = events.get_running_loop() | ||
|
||
self._parent_task = tasks.current_task(self._loop) | ||
if self._parent_task is None: | ||
raise RuntimeError( | ||
f'TaskGroup {self!r} cannot determine the parent task') | ||
|
||
return self | ||
|
||
async def __aexit__(self, et, exc, tb): | ||
self._exiting = True | ||
propagate_cancellation_error = None | ||
|
||
if (exc is not None and | ||
self._is_base_error(exc) and | ||
self._base_error is None): | ||
self._base_error = exc | ||
|
||
if et is exceptions.CancelledError: | ||
if self._parent_cancel_requested: | ||
# Only if we did request task to cancel ourselves | ||
# we mark it as no longer cancelled. | ||
self._parent_task.uncancel() | ||
else: | ||
propagate_cancellation_error = et | ||
|
||
if et is not None and not self._aborting: | ||
# Our parent task is being cancelled: | ||
# | ||
# async with TaskGroup() as g: | ||
# g.create_task(...) | ||
# await ... # <- CancelledError | ||
# | ||
if et is exceptions.CancelledError: | ||
propagate_cancellation_error = et | ||
|
||
# or there's an exception in "async with": | ||
# | ||
# async with TaskGroup() as g: | ||
# g.create_task(...) | ||
# 1 / 0 | ||
# | ||
self._abort() | ||
|
||
# We use while-loop here because "self._on_completed_fut" | ||
# can be cancelled multiple times if our parent task | ||
# is being cancelled repeatedly (or even once, when | ||
# our own cancellation is already in progress) | ||
while self._unfinished_tasks: | ||
if self._on_completed_fut is None: | ||
self._on_completed_fut = self._loop.create_future() | ||
|
||
try: | ||
await self._on_completed_fut | ||
except exceptions.CancelledError as ex: | ||
if not self._aborting: | ||
# Our parent task is being cancelled: | ||
# | ||
# async def wrapper(): | ||
# async with TaskGroup() as g: | ||
# g.create_task(foo) | ||
# | ||
# "wrapper" is being cancelled while "foo" is | ||
# still running. | ||
propagate_cancellation_error = ex | ||
self._abort() | ||
|
||
self._on_completed_fut = None | ||
|
||
assert self._unfinished_tasks == 0 | ||
self._on_completed_fut = None # no longer needed | ||
|
||
if self._base_error is not None: | ||
raise self._base_error | ||
|
||
if propagate_cancellation_error is not None: | ||
# The wrapping task was cancelled; since we're done with | ||
# closing all child tasks, just propagate the cancellation | ||
# request now. | ||
raise propagate_cancellation_error | ||
|
||
if et is not None and et is not exceptions.CancelledError: | ||
self._errors.append(exc) | ||
|
||
if self._errors: | ||
# Exceptions are heavy objects that can have object | ||
# cycles (bad for GC); let's not keep a reference to | ||
# a bunch of them. | ||
errors = self._errors | ||
self._errors = None | ||
|
||
me = BaseExceptionGroup('unhandled errors in a TaskGroup', errors) | ||
gvanrossum marked this conversation as resolved.
Show resolved
Hide resolved
|
||
raise me from None | ||
|
||
def create_task(self, coro): | ||
if not self._entered: | ||
raise RuntimeError(f"TaskGroup {self!r} has not been entered") | ||
if self._exiting and self._unfinished_tasks == 0: | ||
raise RuntimeError(f"TaskGroup {self!r} is finished") | ||
task = self._loop.create_task(coro) | ||
task.add_done_callback(self._on_task_done) | ||
self._unfinished_tasks += 1 | ||
self._tasks.add(task) | ||
return task | ||
|
||
# Since Python 3.8 Tasks propagate all exceptions correctly, | ||
# except for KeyboardInterrupt and SystemExit which are | ||
# still considered special. | ||
|
||
def _is_base_error(self, exc: BaseException) -> bool: | ||
assert isinstance(exc, BaseException) | ||
return isinstance(exc, (SystemExit, KeyboardInterrupt)) | ||
|
||
def _abort(self): | ||
self._aborting = True | ||
|
||
for t in self._tasks: | ||
if not t.done(): | ||
t.cancel() | ||
|
||
def _on_task_done(self, task): | ||
self._unfinished_tasks -= 1 | ||
assert self._unfinished_tasks >= 0 | ||
|
||
if self._on_completed_fut is not None and not self._unfinished_tasks: | ||
if not self._on_completed_fut.done(): | ||
self._on_completed_fut.set_result(True) | ||
|
||
if task.cancelled(): | ||
return | ||
|
||
exc = task.exception() | ||
if exc is None: | ||
return | ||
|
||
self._errors.append(exc) | ||
if self._is_base_error(exc) and self._base_error is None: | ||
self._base_error = exc | ||
|
||
if self._parent_task.done(): | ||
# Not sure if this case is possible, but we want to handle | ||
# it anyways. | ||
self._loop.call_exception_handler({ | ||
'message': f'Task {task!r} has errored out but its parent ' | ||
f'task {self._parent_task} is already completed', | ||
'exception': exc, | ||
'task': task, | ||
}) | ||
return | ||
|
||
self._abort() | ||
if not self._parent_task.cancelling(): | ||
# If parent task *is not* being cancelled, it means that we want | ||
# to manually cancel it to abort whatever is being run right now | ||
# in the TaskGroup. But we want to mark parent task as | ||
# "not cancelled" later in __aexit__. Example situation that | ||
# we need to handle: | ||
# | ||
# async def foo(): | ||
# try: | ||
# async with TaskGroup() as g: | ||
# g.create_task(crash_soon()) | ||
# await something # <- this needs to be canceled | ||
# # by the TaskGroup, e.g. | ||
# # foo() needs to be cancelled | ||
# except Exception: | ||
# # Ignore any exceptions raised in the TaskGroup | ||
# pass | ||
# await something_else # this line has to be called | ||
# # after TaskGroup is finished. | ||
self._parent_cancel_requested = True | ||
self._parent_task.cancel() | ||
|
||
|
||
_name_counter = itertools.count(1).__next__ |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.