asyncio: отмена задач, таймауты и утечка корутин

Сервис на aiohttp начал держать по несколько тысяч фоновых задач, хотя обрабатывал десятки запросов. В логах — тишина, память растёт, asyncio.all_tasks() показывает висящие корутины, которые давно должны были завершиться. Причина типовая: клиент отвалился по таймауту, обработчик вернул ответ, но запущенные внутри задачи никто не отменил. Разберём, что реально делает task.cancel(), почему таймаут не гарантирует освобождение ресурсов и где корутины утекают чаще всего. Материал для Python 3.11+ — поведение таймаутов и групп задач между версиями заметно менялось.

Отмена — это не «стоп», а исключение

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

Важнее другое: CancelledError — обычное исключение¹. Его можно поймать. И вот тут начинаются утечки: если код перехватил его в широком except Exception или except BaseException и не пробросил дальше, задача не отменится, а продолжит работать как ни в чём не бывало.

# Python 3.11+
import asyncio

async def worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        # так делать нельзя: отмена проглочена
        print("поймали отмену и продолжаем жить")
        await asyncio.sleep(10)

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    task.cancel()
    await task  # задача НЕ отменится вовремя

asyncio.run(main())

Правильный шаблон — сделать уборку и пробросить исключение дальше:

async def worker():
    conn = await open_connection()
    try:
        await do_work(conn)
    except asyncio.CancelledError:
        # освобождаем ресурс и отдаём отмену наверх
        await conn.close()
        raise
    finally:
        # finally выполнится и при отмене
        release_local_state()

Если вы поймали CancelledError и не сделали raise — вы отменили отмену. Задача останется в цикле событий, а тот, кто ждал её завершения, получит успешный результат вместо отменённой задачи.

Отмена во время await не мгновенна

Между task.cancel() и фактическим выбросом исключения проходит хотя бы один оборот цикла событий. Поэтому проверять task.cancelled() сразу после cancel() бессмысленно — нужно дождаться завершения через await task, обернув его в try/except:

task.cancel()
try:
    await task
except asyncio.CancelledError:
    pass  # ожидаемо: задачу мы и отменяли

Таймауты: три инструмента и разница между ними

В Python 3.11 появился asyncio.timeout() — контекстный менеджер, который считается предпочтительным. Старый asyncio.wait_for() остался, но ведёт себя иначе. Их легко перепутать, и разница влияет на то, что произойдёт с задачей после срабатывания таймаута.

Начиная с Python 3.11 asyncio.TimeoutError стал псевдонимом встроенного TimeoutError. В коде под 3.11+ можно ловить просто TimeoutError. Пример с новым API:

import asyncio

async def fetch_slow():
    await asyncio.sleep(5)
    return "готово"

async def main():
    try:
        async with asyncio.timeout(1):
            result = await fetch_slow()
    except TimeoutError:
        # к этому моменту fetch_slow уже отменена и завершена
        print("не уложились в секунду")

asyncio.run(main())

Где таймаут не спасает

Таймаут отменяет только то, что находится под управлением цикла событий. Он не прервёт:

• Синхронный блокирующий вызов — тяжёлый requests.get() или CPU-цикл. Цикл событий заблокирован, отмена не дойдёт до кода, пока он не вернёт управление.
• Задачу, которая глотает CancelledError — см. выше. Тогда timeout() будет ждать её завершения дольше собственного лимита.
• Внешние ресурсы — TCP-соединение или процесс на другой стороне таймаут в Python не закрывает сам, если вы не сделали это в except/finally.

Как корутины утекают

Утечка корутины — это задача или незапущенная корутина, которую никто не ждёт и не отменяет. Ресурсы остаются занятыми, память растёт, а при завершении программы вы видите предупреждения Task was destroyed but it is pending! или coroutine was never awaited.

Fire-and-forget без ссылки

Классика — создать задачу и не сохранить ссылку на неё:

# так задача может быть собрана GC до завершения
asyncio.create_task(send_metrics())

Сборщик мусора вправе уничтожить задачу, на которую нет сильной ссылки. Держите ссылку в множестве и убирайте по завершении:

background_tasks = set()

def spawn(coro):
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)
    return task

Незапущенная корутина

Вызов асинхронной функции без await и без create_task создаёт объект корутины, который ничего не делает:

process_order(order)          # корутина создана, но не запущена
await process_order(order)    # правильно

Python выдаст RuntimeWarning: coroutine 'process_order' was never awaited. Не игнорируйте это предупреждение — оно означает, что часть логики не выполнилась.

gather без обработки отмены

Если одна из задач в asyncio.gather() падает, остальные по умолчанию продолжают работать — они не отменяются автоматически. При gather(..., return_exceptions=False) исключение всплывёт наверх, но соседние задачи останутся висеть. Для группового запуска с корректной отменой в 3.11+ есть asyncio.TaskGroup:

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(load_user())
        tg.create_task(load_orders())
    # если одна упала — остальные отменяются,
    # выход из блока дождётся завершения всех

TaskGroup при ошибке в любой задаче отменяет остальные и собирает исключения в ExceptionGroup. Это и есть штатный способ не оставлять висящие задачи.

Как найти утечку

Не гадайте — измеряйте. Периодически логируйте число живых задач:

import asyncio

async def monitor():
    while True:
        pending = [t for t in asyncio.all_tasks() if not t.done()]
        print(f"живых задач: {len(pending)}")
        await asyncio.sleep(10)

Если число растёт линейно с нагрузкой и не падает после обработки запросов — где-то не отменяются фоновые задачи. Включите режим отладки: запуск с переменной окружения PYTHONASYNCIODEBUG=1 или asyncio.run(main(), debug=True). В этом режиме asyncio предупреждает о задачах, уничтоженных в состоянии pending, и о корутинах без await.

Что сделать дальше

• Пройдитесь по коду и найдите каждый except, который может поймать CancelledError — убедитесь, что он делает raise.
• Замените ручные gather без отмены соседей на TaskGroup, если целитесь в 3.11+.
• Все fire-and-forget задачи заведите через реестр со ссылкой и add_done_callback.
• В проде повесьте метрику числа pending-задач и следите за её ростом — это самый ранний признак утечки.

Отдельная тема — грациозное завершение (graceful shutdown) с отменой всех задач по сигналу; здесь мы её не разворачивали.

¹ CancelledError — исключение, которым asyncio сигнализирует задаче об отмене. С Python 3.8 оно наследуется от BaseException, а не Exception, именно чтобы случайный except Exception его не проглатывал.