TaskGroup、取消与超时

目标：能够用 TaskGroup 管理一组共同成败的子任务，并通过实验解释 sibling cancellation、cleanup 和 ExceptionGroup。

手动 create_task() 后忘记等待、保存引用或处理失败，很容易让任务生命周期脱离当前操作。结构化并发的核心约束是：父作用域拥有子任务，退出作用域前必须得到明确结果或明确失败。

asyncio.TaskGroup 是异步 context manager：

tg.create_task() 创建属于该组的 Task；
正常退出时等待全部任务；
某个子任务出现非 CancelledError 失败时，取消其余未完成任务；
等待取消清理完成后，把失败组成 ExceptionGroup 抛出。

完整示例：看见 sibling 被取消

保存为 taskgroup_lab.py：

from __future__ import annotations

import asyncio


async def worker(name: str, delay: float, events: list[str]) -> str:
    events.append(f"start:{name}")
    try:
        await asyncio.sleep(delay)
        events.append(f"done:{name}")
        return name
    finally:
        events.append(f"cleanup:{name}")


async def fail(events: list[str]) -> None:
    await asyncio.sleep(0.01)
    events.append("fail:boom")
    raise ValueError("boom")


async def main() -> None:
    events: list[str] = []
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(worker("slow", 0.2, events))
            group.create_task(fail(events))
    except* ValueError as group:
        assert [str(exc) for exc in group.exceptions] == ["boom"]

    print(events)
    assert "done:slow" not in events
    assert "cleanup:slow" in events


if __name__ == "__main__":
    asyncio.run(main())

运行：

uv run python taskgroup_lab.py

预期输出形如：

['start:slow', 'fail:boom', 'cleanup:slow']

前两个 start 的精确相对位置可能受调度影响，但 done:slow 不会出现，cleanup:slow 必须出现。取消在 worker 的下一个 await 点注入 CancelledError；finally 仍会执行。

为什么通常不要吞掉 CancelledError

TaskGroup 和 asyncio.timeout() 都依赖取消实现结构化控制。如果 coroutine 捕获 CancelledError 后假装成功，父作用域可能无法正确收束。需要清理时使用：

try:
    await operation()
finally:
    await close_resource()

若确实捕获取消，完成清理后通常应 raise 继续传播。

引导实验：默认 gather 的差异

把 TaskGroup 改为：

await asyncio.gather(worker("slow", 0.2, events), fail(events))

默认 gather 会把第一个异常传播给等待者，但不会因为这个异常自动取消其他 awaitable。不要简单称其为“旧而错误”；当你明确希望收集彼此独立的结果，或使用 return_exceptions=True 时 gather 仍有用途。共同成败的任务树更适合 TaskGroup。

独立练习

实现 run_pair(limit)：在一个整体 timeout 内并发执行两个 worker。超时时两个任务都要 cleanup，并向调用方抛 TimeoutError。

参考实现

async def run_pair(limit: float, events: list[str]) -> tuple[str, str]:
    async with asyncio.timeout(limit):
        async with asyncio.TaskGroup() as group:
            first = group.create_task(worker("a", 0.05, events))
            second = group.create_task(worker("b", 0.05, events))
        return first.result(), second.result()


async def check_timeout() -> None:
    events: list[str] = []
    try:
        await run_pair(0.01, events)
    except TimeoutError:
        pass
    else:
        raise AssertionError("timeout expected")
    assert "cleanup:a" in events
    assert "cleanup:b" in events

排错路线