Асинхронные генераторы в Python: подводные камни

Асинхронный генератор в Python выглядит как обычная функция с async def и yield внутри. Пока по нему честно итерируются до конца, всё работает. Проблемы начинаются, когда цикл прерывается на середине, когда генератор оборачивают в asyncio.gather или когда сборщик мусора решает закрыть его в момент, когда event loop уже мёртв. Тогда вы получаете либо утёкшее соединение с БД, либо предупреждение RuntimeError: async generator ignored GeneratorExit, либо тихую потерю данных из finally. Разбираем на CPython 3.11–3.12, где поведение финализации стабилизировалось.

Почему обычный try/finally не спасает

Синхронный генератор при досрочном выходе из цикла закрывается детерминированно: интерпретатор бросает в него GeneratorExit, отрабатывает finally, ресурс освобождается. С асинхронным генератором это не так. Закрытие требует выполнения корутины aclose(), а корутину нельзя запустить синхронно из сборщика мусора — её должен прогнать event loop.

import asyncio

async def read_rows(pool):
    conn = await pool.acquire()
    try:
        async for row in conn.cursor("SELECT id FROM orders"):
            yield row
    finally:
        # выполнится ТОЛЬКО если генератор корректно закрыт
        await pool.release(conn)

async def main(pool):
    async for row in read_rows(pool):
        if row.id == 42:
            break   # выходим досрочно — что будет с conn?

При break объект генератора остаётся живым, пока на него есть ссылка. Соединение из finally не вернётся в пул до тех пор, пока кто-то явно не закроет генератор или его не приберёт GC. На нагруженном сервисе пул исчерпается за минуты.

Асинхронный finally — это обещание, а не гарантия. Оно выполняется, только если генератор довели до конца или явно закрыли через aclose().

Кто должен звать aclose

Начиная с Python 3.6 интерпретатор регистрирует хук финализации асинхронных генераторов. При завершении программы asyncio.run вызывает loop.shutdown_asyncgens(), который проходит по всем незакрытым генераторам и вызывает у них aclose(). Но это происходит в конце жизни цикла, а не в момент вашего break. До этого момента ресурс висит открытым.

Надёжный способ — не полагаться на финализатор, а закрывать генератор самому. С Python 3.10 в стандартной библиотеке появился contextlib.aclosing:

from contextlib import aclosing

async def main(pool):
    async with aclosing(read_rows(pool)) as gen:
        async for row in gen:
            if row.id == 42:
                break
    # на выходе из with гарантированно вызван gen.aclose()
    # соединение вернулось в пул сразу

Разница видна на замерах: без aclosing под нагрузкой в 300 параллельных запросов с досрочными выходами пул из 20 соединений исчерпывается, и запросы начинают ждать освобождения. С aclosing соединение освобождается в момент выхода из async with, и деградации нет. Точные цифры зависят от размера пула и таймаута acquire — мерить нужно на своём железе через метрики пула (asyncpg отдаёт размер и число свободных соединений).

Финализация из чужого event loop

Второй частый капкан: генератор создан в одном цикле, а сборщик мусора пытается финализировать его после того, как цикл закрыт. Тогда shutdown_asyncgens уже отработал, а GC натыкается на живой объект и получает RuntimeError: Event loop is closed или молча проглатывает исключение из finally.

Типичный сценарий — генератор, сохранённый в глобальную переменную или в атрибут долгоживущего объекта, который переживает asyncio.run:

gen = None

async def leak():
    global gen
    gen = read_rows(pool)     # ссылка утекает наружу
    await gen.__anext__()

asyncio.run(leak())
# здесь loop уже закрыт, а gen ещё жив
# GC при выходе бросит предупреждение и не выполнит finally

Лечится тем же правилом: генератор не должен переживать цикл, в котором был создан. Оборачивайте его в aclosing внутри той же корутины, а не выносите наружу.

Что теряется при gather и таймаутах

Асинхронный генератор нельзя напрямую отдать в asyncio.gather — gather работает с корутинами и задачами, а не с итераторами. Но частая ошибка — обернуть итерацию в задачу и потом её отменить по таймауту:

async def consume(gen):
    async for item in gen:
        await handle(item)

task = asyncio.create_task(consume(read_rows(pool)))
try:
    await asyncio.wait_for(task, timeout=5)
except asyncio.TimeoutError:
    pass  # задача отменена, но gen закрыт ли?

При отмене задачи CancelledError прилетает в точку await внутри генератора. Если внутри генератора есть finally, он отработает — но только если отмена дошла до aclose. Проще и надёжнее закрывать генератор явно, снова через aclosing, и передавать наружу уже обёрнутый объект.

Сводка капканов

Когда async-генератор вообще не нужен

Если вы не стримите данные, а просто собираете их в список, асинхронный генератор — лишняя сложность. Обычная async def, возвращающая список, не требует ручного закрытия и не участвует в танцах с финализацией. Генератор оправдан, когда данных много и вы не хотите держать их все в памяти: построчное чтение курсора БД, чтение из сетевого потока, обработка большого файла чанками.

• Стрим строк из БД — генератор оправдан, но всегда под aclosing.
• Пагинация внешнего API — генератор удобен, каждый yield отдаёт страницу.
• Собрать 50 записей и вернуть — не нужен генератор, верните список.

Что настроить и на что смотреть в проде

Первое — включите отладку asyncio на стейджинге через переменную окружения PYTHONASYNCIODEBUG=1 или loop.set_debug(True). В режиме отладки интерпретатор ругается на генераторы, которые финализируются без явного закрытия, и печатает место их создания. Это ловит утечки до того, как они дойдут до продакшена.

Второе — метрики пула соединений. Если используете asyncpg, следите за числом свободных соединений и временем ожидания acquire. Рост времени ожидания при неизменной нагрузке — первый признак незакрытых генераторов.

Третье — правило код-ревью: любой async for по своему генератору, внутри которого есть освобождение ресурса в finally, должен быть обёрнут в async with aclosing(...). Формулировка замечания на ревью: «генератор держит соединение, при досрочном выходе finally не выполнится — оберни в aclosing». Инвалидация ресурсов при отмене задач — отдельная тема, здесь разбираем только закрытие генераторов.