Optionally prevent child tasks from being cancelled in asyncio.TaskGroup
#101581
Comments
|
Yeah, I think this is reasonable. We need to bikeshed on the name of the flag, and I would like to ensure that when the awaiting task (i.e., the |
Yes, it's quite hard to figure out a name that's descriptive enough but reasonably short. May
In the local clone of my fork I've already defined new tests for the new intended behaviour and I've copied the ones you've pointed out (setting the new flag to prevent child tasks cancellation). Should I open a PR now or it'd be better to wait for the flag name to have been found? |
|
|
|
I consider |
|
Maybe |
|
Oh! |
Fine with me. We all agree on |
SGTM |
|
So, since there is still more discussion on Discourse, I'd like to get a review from @achimnol, who (I hope) is collecting all the various possible solutions, e.g. Supervisor (also yours) and PersistentTaskGroup; there's a somewhat recent (post-US-PyCon) writeup in Discourse: https://discuss.python.org/t/asyncio-tasks-and-exception-handling-recommended-idioms/23806/9 |
|
Any updates on this? :-) |
|
This is by far the cleanest solution I've found: It will be possible to implement with
I tried very hard to get something working with aiotools, but the exception_handler doesn't count as retrieving the exceptions, so it gives "Future exception was never retrieved". import asyncio
from types import TracebackType
import aiotools
MULTIPLE_EXCEPTIONS_MESSAGE = "Multiple exceptions occurred"
class WaitTaskGroup(aiotools.PersistentTaskGroup):
_exceptions: list[BaseException]
def __init__(
self,
*,
name: str | None = None,
) -> None:
async def exception_handler(
exc_type: type[BaseException], # noqa: ARG001
exc_obj: BaseException,
exc_tb: TracebackType, # noqa: ARG001
) -> None:
self._exceptions.append(exc_obj)
super().__init__(name=name, exception_handler=exception_handler)
async def __aenter__(self) -> "WaitTaskGroup":
self._exceptions = []
return await super().__aenter__()
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_obj: BaseException | None,
exc_tb: TracebackType | None,
) -> bool | None:
ret = await super().__aexit__(exc_type, exc_obj, exc_tb)
# exception_handler is awaited in parent class
if self._exceptions:
for t in self._tasks:
# this doesn't fix "Future exception was never retrieved"
try:
await t
except BaseException:
pass
raise BaseExceptionGroup(
MULTIPLE_EXCEPTIONS_MESSAGE,
self._exceptions,
)
return retYou can see how annoying it is to deal with the unretrieved exceptions here: import asyncio
from types import TracebackType
import aiotools
MULTIPLE_EXCEPTIONS_MESSAGE = "Multiple exceptions occurred"
async def get_string() -> str:
return "hello"
async def get_int() -> int:
return 1
# async def raise_exception() -> typing.Never: # also works
async def raise_exception(s: str, *, fail: bool = True) -> None:
if fail:
raise Exception(s)
async def slow_print(s: str) -> None:
await asyncio.sleep(1)
print(s)
async def test_persistent_task_group(
*,
fail: bool = True,
) -> tuple[str, int, None, None]:
exceptions = []
async def exception_handler(
exc_type: type[BaseException], # noqa: ARG001
exc_obj: BaseException,
exc_tb: TracebackType, # noqa: ARG001
) -> None:
exceptions.append(exc_obj)
async with aiotools.PersistentTaskGroup(exception_handler=exception_handler) as tg:
t1 = tg.create_task(get_string())
t2 = tg.create_task(get_int())
t3 = tg.create_task(raise_exception("PersistentTaskGroup", fail=fail))
t4 = tg.create_task(slow_print("PersistentTaskGroup"))
t5 = tg.create_task(raise_exception("PersistentTaskGroup2", fail=fail))
if exceptions:
for t in (t1, t2, t3, t4, t5):
# Tasks need to be awaited before ExceptionGroup is raised
# see: https://github.com/aio-libs/aiojobs/blob/master/aiojobs/_scheduler.py#L241
try:
await t
except BaseException:
pass
raise BaseExceptionGroup(
MULTIPLE_EXCEPTIONS_MESSAGE,
exceptions,
)
return t1.result(), t2.result(), t3.result(), t4.result()I tried to raise the exceptions from Ts = TypeVarTuple("Ts")
def cast_not_exception[*Ts](*args: tuple[*Ts | BaseException]) -> tuple[*Ts]:
return cast("tuple[*Ts]", args)Overriding import asyncio
class UnabortableTaskGroup(asyncio.TaskGroup):
_abort = lambda self: None
async def raise_exception() -> None:
await asyncio.sleep(0.1)
raise Exception("cancel")
async def slow_print(s: str) -> None:
await asyncio.sleep(1)
print(s)
async def test_my_task_group() -> None:
async with MyTaskGroup() as tg:
tg.create_task(slow_print("MyTaskGroup"))
async def test_unabortable_task_group() -> None:
async with UnabortableTaskGroup() as tg:
tg.create_task(slow_print("UnabortableTaskGroup"))
async def test_cancel() -> None:
async with asyncio.TaskGroup() as tg:
# UnabortableTaskGroup is the only one that prints
tg.create_task(test_my_task_group())
tg.create_task(test_unabortable_task_group())
tg.create_task(raise_exception()) |
Feature or enhancement
Add a flag to
asyncio.TaskGroupto control whether, if a child task crashes, other should be cancelled or not.Pitch
Currently
asyncio.TaskGroupalways cancels all child tasks if one fails. Even if that's the most common use case, I'd like to be able to switch from the current behaviour to prevent child tasks cancellation when one failure occurs and raise anExceptionGroupwith all exceptions raised (only after other tasks completed).async withblock still have to cause child tasks to be canceled.SystemExitandKeyboardInterruptraised in a child task still cancel other tasks.Example usage:
Looking at
asyncio.TaskGroupsource code, it seems that it could be achieved by adding theabort_on_first_exception(or whatever it should be named) flag to the__init__method and then modify_on_task_doneas follow:If it's reasonable to have it in
asyncio, I'll be glad to submit a PR for it.Previous discussion
Post on discuss
Linked PRs
asyncio.TaskScopeand letasyncio.TaskGroupsubclass it #105011The text was updated successfully, but these errors were encountered: