Asyncio в Python: TaskGroup, таймауты и безопасная отмена параллельных задач

Параллельное выполнение сетевых запросов в Python часто приводит к скрытым дефектам: зависшим фоновым корутинам, исчерпанию пула файловых дескрипторов и некорректно обработанным исключениям. Долгое время стандартом для группировки параллельных вызовов была функция asyncio.gather. Однако управление отменой и безопасная очистка ресурсов при её использовании требовали сложного шаблонного кода.

Начиная с Python 3.11, стандартная библиотека поддерживает концепцию структурированной конкурентности (Structured Concurrency). Контекстные менеджеры asyncio.TaskGroup и asyncio.timeout делают выполнение параллельных сценариев детерминированным, гарантируя своевременное завершение задач и закрытие ресурсов при сбоях.

---

Структурированная конкурентность: как работает asyncio.TaskGroup

Главная проблема неструктурированной конкурентности — задачи-сироты («orphaned tasks»). Если при использовании asyncio.gather одна из задач завершается с ошибкой, остальные продолжают молча выполняться в цикле событий, расходуя память, соединения и процессорное время.

asyncio.TaskGroup решает эту проблему за счет предсказуемого жизненного цикла:

1. Гарантия завершения: Блок async with asyncio.TaskGroup() не завершится до тех пор, пока не завершатся все созданные внутри него задачи.
2. Каскадная отмена: Если одна из задач группы выбрасывает необработанное исключение (отличное от asyncio.CancelledError), TaskGroup автоматически инициирует отмену всех остальных задач группы.
3. Группировка исключений: Множественные ошибки упаковываются в ExceptionGroup и перехватываются специализированным синтаксисом except*.

import asyncio

async def worker(task_id: int, delay: float, fail: bool = False) -> None:
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"Сбой в задаче {task_id}")
    print(f"Задача {task_id} успешно завершена")

async def run_group() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker(1, 0.2))
            tg.create_task(worker(2, 0.1, fail=True))
            tg.create_task(worker(3, 0.5))
    except* ValueError as eg:
        for exc in eg.exceptions:
            print(f"Перехвачена ошибка: {exc}")

В этом примере задача 2 вызывает исключение через 0.1 секунды. TaskGroup немедленно отменяет задачу 3 (рассчитанную на 0.5 секунды), дожидается ее фактической остановки и выбрасывает ExceptionGroup, содержащий ValueError.

---

Ограничение параллелизма с помощью asyncio.Semaphore

Запуск сотен параллельных сетевых запросов без контроля приводит к перегрузке целевого сервиса (ошибки 429 Too Many Requests) и исчерпанию системных сокетов.

Для ограничения одновременных операций применяется asyncio.Semaphore. Этот примитив синхронизации управляет внутренним счетчиком: метод acquire() уменьшает его, а release() — увеличивает. Если доступных слотов нет, входящие корутины приостанавливаются.

Рекомендуемый паттерн — использование семафора через асинхронный контекстный менеджер:

async def fetch_with_limit(url: str, semaphore: asyncio.Semaphore) -> None:
    async with semaphore:
        # В этой секции одновременно выполняется строго заданное число корутин
        await asyncio.sleep(0.1)

При таком подходе счетчик семафора восстанавливается автоматически даже при возникновении ошибки или отмене задачи внутри блока.

---

Управление временем: контекстный менеджер asyncio.timeout

До Python 3.11 для ограничения времени выполнения использовалась функция asyncio.wait_for. Её недостаток заключался в создании неявной задачи-обертки asyncio.Task, что усложняло отладку и трассировку стека.

Контекстный менеджер asyncio.timeout(delay) работает иначе:

• Он не создает отдельную задачу, а регистрирует таймер для текущей.
• При истечении таймера текущая задача получает внутренний CancelledError.
• На выходе из блока async with asyncio.timeout(...) контекстный менеджер перехватывает этот CancelledError и трансформирует его в TimeoutError.

Важно: Исключение TimeoutError необходимо перехватывать снаружи блока async with asyncio.timeout(). Перехват внутри контекста не позволит менеджеру корректно обработать истечение времени.

try:
    async with asyncio.timeout(2.0):
        await asyncio.sleep(5.0)
except TimeoutError:
    print("Операция превысила установленный лимит времени")

---

Механика отмены и освобождение ресурсов (CancelledError)

Отмена задач в asyncio носит кооперативный характер. Вызов метода task.cancel() не прерывает выполнение байткода принудительно, а планирует генерацию asyncio.CancelledError в ближайшей точке ожидания — там, где корутина вызывает await.

Правила обработки отмены:

1. Не подавляйте CancelledError без проброса: Если блок перехватывает asyncio.CancelledError для логирования или промежуточной очистки, исключение обязательно нужно пробросить дальше через raise. В противном случае TaskGroup зависнет в ожидании задачи.
2. Используйте try / finally: Освобождение сетевых соединений, файловых дескрипторов и блокировок должно выполняться в секции finally.
3. Учитывайте иерархию исключений: CancelledError наследуется от BaseException, чтобы его случайно не заглушил общий блок except Exception.

async def process_resource() -> None:
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("Получен сигнал отмены...")
        raise
    finally:
        print("Освобождение локальных ресурсов завершено")

---

Комплексный практический пример: клиент пакетных запросов

Ниже представлен готовый к промышленному использованию сценарий: клиент отправляет пакет HTTP-запросов, ограничивает параллелизм до 3 одновременных вызовов, устанавливает гранулярные таймауты и гарантированно закрывает пул соединений.

import asyncio
import aiohttp


async def fetch(
    session: aiohttp.ClientSession,
    url: str,
    semaphore: asyncio.Semaphore,
) -> dict | None:
    async with semaphore:
        try:
            async with asyncio.timeout(2.0):
                async with session.get(url) as response:
                    response.raise_for_status()
                    return await response.json()

        except TimeoutError:
            print(f"[Таймаут] Запрос к {url} превысил лимит")
            raise

        except asyncio.CancelledError:
            print(f"[Отмена] Задача для {url} была прервана")
            raise

        except aiohttp.ClientError as err:
            print(f"[HTTP Ошибка] {url}: {err}")
            raise

        finally:
            # Гарантированная очистка локального контекста задачи
            pass


async def main(urls: list[str]) -> None:
    semaphore = asyncio.Semaphore(3)
    client_timeout = aiohttp.ClientTimeout(total=10.0)

    # Context manager сессии управляет базовым пулом соединений
    async with aiohttp.ClientSession(timeout=client_timeout) as session:
        try:
            async with asyncio.TaskGroup() as tg:
                tasks = [
                    tg.create_task(fetch(session, url, semaphore))
                    for url in urls
                ]

            results = [t.result() for t in tasks]
            print(f"Успешно получено ответов: {len(results)}")

        except* TimeoutError:
            print("Группа: одна или несколько задач завершились по таймауту")

        except* aiohttp.ClientError as eg:
            for exc in eg.exceptions:
                print(f"Группа: сетевой сбой — {exc}")


if __name__ == "__main__":
    test_urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/500",  # Провоцирует ClientResponseError
        "https://httpbin.org/delay/3",
    ]

    # asyncio.run закрывает генераторы, пулы потоков и цикл событий
    asyncio.run(main(test_urls))

В этом коде ошибка 500 на третьем URL инициирует aiohttp.ClientResponseError. TaskGroup сразу же отменяет долгий запрос к /delay/3, не дожидаясь истечения его таймера, после чего сессия безопасно освобождает сетевой пул.

---

Тестирование сценариев отмены и остановки

Для проверки устойчивости логики к задержкам и отменам создается локальный HTTP-сервер на базе aiohttp.web и тестовых утилит библиотеки.

import asyncio
from aiohttp import web
from aiohttp.test_utils import TestServer


async def slow_handler(request: web.Request) -> web.Response:
    # Имитация длительной серверной обработки
    await asyncio.sleep(5)
    return web.json_response({"status": "ok"})


async def test_cancellation_flow() -> None:
    app = web.Application()
    app.router.add_get("/slow", slow_handler)

    server = TestServer(app)
    await server.start_server()

    try:
        url = str(server.make_url("/slow"))

        async with aiohttp.ClientSession() as session:
            timeout_caught = False
            try:
                async with asyncio.timeout(0.2):
                    async with session.get(url) as response:
                        await response.json()
            except TimeoutError:
                timeout_caught = True

            assert timeout_caught, "Таймаут не был инициирован вовремя"
            print("Тест пройден: соединение корректно разорвано по таймауту")

    finally:
        # Гарантированная остановка тестового сервера
        await server.close()


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

Тест подтверждает, что клиент прерывает сетевое чтение ровно через 0.2 секунды, не дожидаясь ответа сервера, и освобождает выделенные дескрипторы.

---

Чек-лист: правила безопасного асинхронного кода

Область	Рекомендация	Назначение
Группировка задач	asyncio.TaskGroup вместо gather	Исключение зависших задач-сирот при сбоях
Ограничение нагрузки	async with asyncio.Semaphore(N)	Защита от перегрузки API и лимитов ОС
Таймауты	async with asyncio.timeout(sec)	Гранулярный контроль времени без лишних оберток
Сигнал отмены	Проброс raise при перехвате CancelledError	Корректный graceful shutdown приложения
Управление пулом	async with aiohttp.ClientSession()	Предотвращение утечек сокетов и памяти
Точка входа	Запуск через asyncio.run()	Гарантированное закрытие event loop и генераторов

---

FAQ (Часто задаваемые вопросы)

Чем TaskGroup безопаснее asyncio.gather при возникновении ошибок?

При ошибке в одной из корутин asyncio.gather сразу передает исключение вызывающему коду, но остальные запущенные задачи оставляет выполняться в фоне. TaskGroup автоматически отправляет сигнал отмены всем задачам группы и не выходит из блока контекста, пока каждая из них полностью не завершит работу и очистку.

Нужно ли вручную вызывать session.close() для aiohttp.ClientSession?

Если сессия открыта через async with aiohttp.ClientSession() as session:, метод close() вызывается автоматически. Ручной вызов await session.close() необходим только при создании долгоживущей сессии вне контекстного менеджера (например, в свойствах класса сервиса при старте приложения).

Почему TimeoutError нельзя перехватывать внутри блока async with asyncio.timeout()?

Внутри контекста менеджер инициирует исключение CancelledError. В TimeoutError оно трансформируется только в момент выхода из контекста async with. Перехват TimeoutError внутри блока не сработает, так как на этом этапе исключение еще имеет исходный тип отмены.

Что произойдет, если заглушить asyncio.CancelledError в задаче?

Если корутина перехватит CancelledError и не выполнит повторный raise, цикл событий посчитает задачу штатно завершенной. Это нарушит согласованность работы TaskGroup или алгоритм плановой остановки сервиса, приведя к зависанию процесса.

Можно ли динамически добавлять задачи в работающий TaskGroup?

Да. Внутри блока async with asyncio.TaskGroup() as tg: метод tg.create_task(...) можно вызывать многократно из любых вложенных корутин, которым передан экземпляр tg. Менеджер не закроет контекст, пока не завершатся все зарегистрированные задачи.

---

Заключение

Инструменты структурированной конкурентности в Python 3.11+ устраняют архитектурные проблемы асинхронного программирования. Совместное использование TaskGroup, asyncio.timeout и Semaphore позволяет создавать отказоустойчивые сервисы, исключая утечки соединений, зависание фоновых процессов и неконтролируемые каскадные сбои.

Источники

• Coroutines and tasks
• Synchronization Primitives
• Runners
• Testing
• Testing — aiohttp 3.14.3 documentation
• Testing — aiohttp 3.7.4 documentation