asyncio в Python без сюрпризов: event loop, задачи, отмена и таймауты

asyncio — это не “магия, которая делает код быстрее”. Это механизм конкурентного исполнения: один поток событий координирует множество операций ввода‑вывода и “пропускает” ожидания, пока другие задачи работают. Когда всё сделано правильно, получаются предсказуемые таймауты, корректная отмена и отсутствие утечек. Когда сделано неправильно — возникают призраки: задачи, которые не завершаются, отмена, которую “поглотили”, и таймауты, которые выглядят как сработавшие, но на деле оставляют фоновые операции живыми.

Ниже разберём, как устроен event loop, как корректно создавать задачи, как отменять их безопасно и как правильно сочетать отмену с таймаутами. Постараюсь опираться на факты и типичные сценарии из практики.

Как работает asyncio: event loop и ожидание

Event loop как диспетчер

В центре asyncio находится event loop — объект, который:

1. планирует выполнение корутин (через Task);
2. ожидает события ввода‑вывода (socket, файлы, сигналы, таймеры);
3. возобновляет корутины, когда операция готова;
4. управляет отменой и завершением задач.

Важно: asyncio в “классическом” варианте живёт в одном потоке. Конкурентность достигается за счёт того, что корутины не блокируют поток. Если вы сделаете в async‑коде блокирующий вызов (например, долгий CPU‑труд или синхронный запрос), вы “заморозите” цикл событий и потеряете смысл конкурентности.

Корутинная модель: suspend/resume

Корутина — это функция, которая может “приостанавливаться” на await и “возобновляться” позже. У await всегда есть конкретный объект ожидания:

• await some_coroutine()
• await asyncio.sleep(...)
• await reader.read(...) / await writer.drain()
• await some_future

Понимание “точек приостановки” критично для дебага: если где-то не стоит await, выполнение может блокировать; если неправильно обработать исключения, корутина может завершиться раньше, чем вы ожидаете.

Task vs coroutine

Корутину нельзя “просто так” запускать как самостоятельную сущность. Обычно вы создаёте Task:

import asyncio

async def worker():
    await asyncio.sleep(1)
    return 42

async def main():
    task = asyncio.create_task(worker())  # запускаем конкурентно
    result = await task  # ждём результат

asyncio.run(main())

• asyncio.create_task(coro) ставит задачу в цикл событий.
• Без Task корутина существует как объект и не планируется автоматически.
• await task — это ожидание завершения задачи, но задача уже живёт и может выполнять параллельно другим задачам.

Создание задач без сюрпризов

create_task: когда это уместно

Создавайте Task, когда у вас есть обязательная конкурентность и вы планируете управлять жизненным циклом (ждать, отменять, собирать результаты).

Частая ошибка — начать корутину “в фоне”, но забыть превратить её в Task:

async def main():
    coro = some_io()
    # ошибка: coro не выполняется, пока не будет await или create_task

“Фоновая задача” и отсутствие ссылок

Если вы создадите Task и потеряете на неё ссылку, она всё равно может жить, но вы потеряете возможность правильно обработать исключения и отмену. Тогда возможны сообщения вроде “Task exception was never retrieved”.

Правильнее — хранить задачу и решать, что с ней делать.

Сбор нескольких задач: gather и её нюансы

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

Пример “собрать все результаты”:

results = await asyncio.gather(*tasks)

При этом:

• если одна задача падает исключением, gather по умолчанию пробросит исключение наружу;
• остальные задачи продолжат выполняться или будут отменены — зависит от контекста отмены и версии asyncio, но общий смысл: вы должны понимать, что исключение — это сигнал “сборщик прекращает ожидание”.

Более управляемый вариант — собирать с флагом return_exceptions=True, чтобы получить результаты и исключения как элементы списка.

results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        # обработать
        ...

Это полезно, когда задачи независимы: одна из них провалилась, но вам нужно собрать информацию о всех.

TaskGroup: управляемая конкурентность (Python 3.11+)

Если вы на Python 3.11+, предпочтительнее рассмотреть asyncio.TaskGroup — он делает то, что обычно приходится вручную: управляет жизненным циклом задач и отменой внутри блока.

Схема:

import asyncio

async def job(x: int) -> int:
    await asyncio.sleep(0.1)
    return x * 2

async def main():
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(job(i)) for i in range(10)]
        # tg гарантирует, что задачи будут завершены/отменены корректно при ошибках

asyncio.run(main())

• При исключении одной задачи остальные будут отменены.
• Исключение будет сгруппировано.
• Это снижает шанс утечек и “полузавершённых” фоновых задач.

Безопасная отмена: что реально происходит

Отмена — это не “убить поток”, а запрос на прекращение

Отмена в asyncio реализована через отмену задачи:

task.cancel()

Это не мгновенное “стоп”. Это отправка CancelledError в корутину при ближайшей точке управления (обычно на await). Дальше корутина может:

1. не обрабатывать CancelledError — тогда она корректно завершится отменой;
2. обработать и “проглотить” отмену — тогда задача может не завершиться и продолжить работу;
3. обработать частично, но потом вернуть управление — итог зависит от того, что вы сделали с отменой.

Правильный паттерн обработки CancelledError

Если вам нужно выполнить очистку (закрыть соединение, снять логику, отменить внутренние ожидания), оборачивайте ожидания так:

import asyncio

async def cancellable_resource():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Здесь мы в точке отмены. Обычно важно не "продолжать работу".
        # Очистка:
        # await conn.close()
        raise  # критично: пробросить отмену дальше

Почему raise критично:

• иначе задача продолжит жить и отмена будет “скрыта”;
• outer-код (кто отменял) не получит ожидаемое завершение;
• можно получить утечки и странные состояния.

Щадящая отмена с этапами

В реальных системах отмену иногда делают в два шага: сначала “попросить” завершиться, затем “дожать” таймаутом.

Например:

• на отмену вы выставляете флаг и прекращаете новые операции;
• после таймаута — отменяете задачу принудительно.

Пример “попросить” + таймаут:

import asyncio

async def worker(stop_event: asyncio.Event):
    try:
        while not stop_event.is_set():
            await asyncio.sleep(0.2)
    except asyncio.CancelledError:
        # Если отменили принудительно — очистка и проброс отмены
        raise

async def main():
    stop_event = asyncio.Event()
    task = asyncio.create_task(worker(stop_event))

    await asyncio.sleep(1)
    stop_event.set()  # мягкое завершение

    try:
        await asyncio.wait_for(task, timeout=2)
    except asyncio.TimeoutError:
        task.cancel()
        await task  # дождаться финальной отмены (иначе фон останется)

asyncio.run(main())

Заметьте:

• после task.cancel() мы всегда ждём await task;
• иначе можно пропустить CancelledError и получить неочевидное состояние.

Отмена и asyncio.sleep

asyncio.sleep — удобный пример, потому что он отменяемый: при отмене будет поднят CancelledError. Это важно учитывать при построении циклов ожидания.

Таймауты: wait_for и как не оставить задачу “живой”

Базовый таймаут

asyncio.wait_for(coro, timeout) отменяет корутину/задачу при истечении таймаута.

import asyncio

async def slow():
    await asyncio.sleep(10)
    return "ok"

async def main():
    try:
        result = await asyncio.wait_for(slow(), timeout=1)
    except asyncio.TimeoutError:
        result = None

Здесь wait_for создаёт ожидание и при таймауте инициирует отмену. Но дальше есть два подводных камня.

Подводный камень №1: “проглоченная” отмена внутри slow()

Если slow() перехватывает CancelledError и не пробрасывает его, отмена не завершит выполнение. Тогда ваш таймаут не даст ожидаемого результата в общем “жизненном цикле” — внешне будет TimeoutError, но внутри может продолжаться работа.

Пример плохого поведения:

async def bad_slow():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        # Ошибка: отмену не пробрасываем
        return "still running?"

Это приводит к тому, что отмена превращается в “контрольный сигнал”, который программа игнорирует.

Правильно — после очистки raise.

Подводный камень №2: таймаут на задачу ≠ корректная отмена всей логики

Иногда вы ждёте task напрямую, но в середине уже есть внутренние ожидания (например, несколько операций в цикле, параллельные запросы и т.д.). Таймаут должен приводить к остановке всей цепочки.

Если вы используете TaskGroup, то отмена из‑вне отменит все задачи внутри группы. Если нет — вам придётся делать это вручную.

Практика: управляйте жизненным циклом явно

Паттерн “создать задачу → отменить → дождаться”

Самый безопасный и предсказуемый паттерн:

1. создаём Task;
2. запускаем таймаут ожидания;
3. при таймауте отменяем задачу;
4. обязательно await task чтобы завершить отмену и получить финальный результат/исключение.

import asyncio

async def fetch(url: str):
    await asyncio.sleep(2)  # имитация
    return f"data from {url}"

async def fetch_with_timeout(url: str, timeout: float):
    task = asyncio.create_task(fetch(url))
    try:
        return await asyncio.wait_for(task, timeout=timeout)
    except asyncio.TimeoutError:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        return None

async def main():
    print(await fetch_with_timeout("https://example.com", timeout=1))

asyncio.run(main())

Зачем ловить CancelledError после await task:

• в большинстве случаев оно ожидаемо после cancel();
• если этого не сделать, оно может всплыть выше и “сломать” логику обработки таймаута.

Параллельные операции: отмена “по событию” и кооперация

Если у вас сервис, который делает много шагов, предпочтительнее не только отменять Task, но и передавать “сигнал остановки” внутрь. Это превращает отмену в кооперативное завершение: корутины корректно освобождают ресурсы, прекращают новые операции и закрывают соединения.

Общий подход:

• создайте asyncio.Event или asyncio.CancelledError как механизм;
• при отмене — выставьте флаг и/или прокиньте отмену в дочерние задачи.

Отслеживание исключений и предотвращение утечек

Исключения в фоновых задачах

Если задача в фоне падает исключением и никто не делает await или не собирает результат, вы получите сообщения об “unretrieved exception”.

Например, так делать нельзя:

async def main():
    asyncio.create_task(buggy())
    await asyncio.sleep(1)  # не ждём buggy()

Варианты исправления:

• собрать задачи через await/gather;
• хранить ссылки и дождаться;
• использовать TaskGroup, где исключения структурированы.

“Потерянная” отмена

Ещё один источник проблем — когда внешняя отмена инициируется, но внутренняя корутина:

• глушит CancelledError,
• продолжает работать,
• создаёт новые задачи без привязки к общему контуру остановки.

Если вы строите систему “сверху вниз”, стремитесь к тому, чтобы все дочерние операции:

• были либо частью TaskGroup,
• либо имели общий stop_event/контекст отмены,
• либо корректно реагировали на CancelledError и пробрасывали его после cleanup.

Сложные случаи: таймаут на блокирующие точки, вложенные wait_for и защита

Вложенный wait_for: не делайте случайно “таймаутов на таймаут”

Когда вы используете wait_for в разных уровнях стека, легко получить ситуацию, где:

• верхний уровень отменяет корутину,
• нижний уровень ловит и глушит CancelledError,
• а затем верхний уровень ошибочно думает, что работа корректно завершилась.

Решение: договоритесь о правилах.

• Внутренние функции после cleanup должны raise отмену.
• Обработку TimeoutError делайте в том месте, где вы принимаете решение “что считать ошибкой”.

Защита от бесконечных ожиданий

Помните правило: каждое ожидание ввода‑вывода должно иметь верхнюю границу. Иначе “зависание” сервиса со временем превращается в вечные задачи.

Практический подход:

• таймаут на сетевые операции (connect/read/write);
• таймаут на “получить результат” и “дожать отмену”;
• минимально необходимая длительность внутренних циклов ожидания.

Компоненты “как из карты”: event loop + задачи + отмена + таймауты

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

Пример: конкурентные запросы с отменой и глобальным таймаутом

import asyncio
from typing import Any

async def query(i: int) -> dict[str, Any]:
    # имитация работы с потенциальной зависимостью
    await asyncio.sleep(0.3 + (i % 3) * 0.2)
    return {"i": i, "value": i * 10}

async def run_queries(n: int, timeout: float) -> list[dict[str, Any]] | None:
    results: list[dict[str, Any]] = []
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(query(i)) for i in range(n)]
            # wait_for вокруг всего блока:
            # при таймауте сработает отмена TaskGroup
            # (TaskGroup обязан отменить дочерние задачи)
            async def collect():
                nonlocal results
                for t in tasks:
                    results.append(await t)

            await asyncio.wait_for(collect(), timeout=timeout)
        return results
    except asyncio.TimeoutError:
        return None

async def main():
    data = await run_queries(10, timeout=1.0)
    print("data:", data)

asyncio.run(main())

Что здесь важно:

• TaskGroup структурирует дочерние задачи.
• Глобальный таймаут оборачивает “коллектор” (ожидание результатов).
• При таймауте инициируется отмена; дочерние задачи должны корректно реагировать на отмену.
• Вы возвращаете None вместо того, чтобы оставлять фоновые задачи “в неизвестности”.

Типичные ошибки, которые ломают конкурентность

1) Блокирующий код внутри async

Если внутри корутины есть синхронная операция, которая занимает секунды, event loop не сможет обслужить другие задачи. Симптомы: “всё зависает”, “таймауты не срабатывают как ожидается”, “CPU растёт, но I/O нет”.

Решение: выносить в thread/process executor или использовать async‑библиотеки.

2) Глушение CancelledError

Это самая частая “причина утечек” при отмене. Если вы перехватили CancelledError, почти всегда нужно:

• сделать cleanup,
• затем raise.

Исключение — если вы уверены в логике “отмена превращается в штатное завершение” и вы контролируете это на верхнем уровне. В общем случае — не надо.

3) Создание фоновых задач без политики завершения

Если задача живёт “самостоятельно”, у неё должен быть:

• механизм остановки,
• или стратегия “await + отмена при выходе”.

4) Отмена без ожидания завершения

task.cancel() — лишь запрос. Пока не сделаете await task, задача может остаться в процессе очистки, и вы потеряете контроль.

Как выбрать стратегию: gather, TaskGroup, ручная отмена

• asyncio.gather: когда у вас простой параллельный “собрать результаты и обработать исключения”.
• TaskGroup: когда у вас структурированная конкурентность и важна “коррект