gh-101581: Add asyncio.TaskScope and let asyncio.TaskGroup subclass it

[achimnol]

This is yet another approach to embrace long-lived use cases of TaskGroup, as an alternative to #101648.

It adds a new lower-level task lifecycle management primitive, TaskScope. It provides a base scope for cancellation and tracking of child tasks, and can be extended to add specific semantics like TaskGroup which cancels all children altogether upon any first seen unhandled exception. aiotools demonstrates other new coroutine aggregation interfaces like race(), as_completed_safe(), and gather_safe() based on TaskScope. (Note: the current aiotools.Supervisor is same to TaskScope(delegate_errors=None), so TaskScope is a generalization of Supervisor.)

The implementation combines the 1st and the 2nd ideas as discussed about expressing how to handle the exceptions: delegate_errors.

• TaskScope(): Calls loop.call_exception_handler() upon unhandled exceptions
• TaskScope(delegate_errors=None): Silently ignores unhandled exceptions
  • Discouraged for end-users but useful for library writers (e.g., TaskGroup)
• TaskScope(delegate_errors=func): Calls func using the same context argument like loop.call_exception_handler()

TaskGroup is rewritten by subclassing TaskScope, and now its code highlights the specific semantics of the TaskGroup API. The rewritten TaskGroup still passes all existing test cases.

To prevent memory leaks in long-lived use cases, TaskScope just keeps a self._has_errors boolean flag only while TaskGroup keeps the full list self._errors.

class TaskGroup(taskscopes.TaskScope):
    def __init__(self):
        super().__init__(delegate_errors=None)
        self._errors = []

    def __repr__(self): ...

    def create_task(self, coro, *, name=None, context=None):
        """Create a new task in this group and return it.

        Similar to `asyncio.create_task`.
        """
        task = super().create_task(coro, name=name, context=context)
        if not task.done():
            task.add_done_callback(self._handle_completion_as_group)
        return task

    async def __aexit__(self, et, exc, tb):
        await super().__aexit__(et, exc, tb)

        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.
            try:
                me = BaseExceptionGroup('unhandled errors in a TaskGroup', self._errors)
                raise me from None
            finally:
                self._errors = None

    def _handle_completion_as_group(self, task):
        if task.cancelled():
            return
        if (exc := task.exception()) is not None:
            self._errors.append(exc)
            if not self._aborting and not self._parent_cancel_requested:
                self._abort()
                self._parent_cancel_requested = True
                self._parent_task.cancel()

• Issue: Optionally prevent child tasks from being cancelled in asyncio.TaskGroup #101581

[achimnol]

I'm work in progress to revamp the test cases copied from #101648.

[achimnol]

Under experimentation in achimnol/aiotools#58.

[1st1]

@achimnol, in general I like the solution. However, I'm wondering if instead of using inheritance we can somehow use composition for this.