от граблей с контекстом к структурированной асинхронности

Модуль asyncio появился в Python 3.4 как экспериментальная попытка принести асинхронный ввод-вывод в стандартную библиотеку. Однако путь от первых версий до Python 3.12+ был вымощен постоянно ломающимися контрактами, неявными состояниями и утечками контекста.

Ниже собрана консолидация главных архитектурных ошибок asyncio, хронология их устранения и итоговый паттерн, к которому пришла экосистема Python.

---

1. Главные «ляпы» и хрупкие абстракции (Python 3.5 – 3.10)

Главная уязвимость высокоуровневых языков с неявным контекстом — иллюзия того, что рантайм берет на себя управление системными ресурсами без side-эффектов.

Неявный глобальный контекст и привязка к Event Loop

Примитивы синхронизации (Semaphore, Lock, Event), созданные на уровне модуля, неявно опрашивали «текущий» событийный цикл via get_event_loop(). Если основной цикл запускался позже (например, через asyncio.run()), объект оказывался привязан к чужому или уже уничтоженному циклу.

Ломающиеся контракты между минорными версиями

  • Python 3.8: Создание asyncio.Semaphore() вне корутины приводило к мгновенной привязке к первичному циклу. При попытке использовать его внутри asyncio.run() вылетала ошибка:
RuntimeError: got Future attached to a different loop
  • Python 3.10: Привязку сделали «ленивой» (при первом acquire()), из-за чего тот же самый код вдруг переставал падать.

Хаос с параметром loop=

До Python 3.8 почти во все функции (Queue(loop=loop), asyncio.sleep(1, loop=loop), gather(*aws, loop=loop)) требовалось или разрешалось прокидывать loop. В Python 3.8 этот параметр объявили *deprecated*, а в Python 3.10 — жестко вырезали, вызвав массовый падеж устаревших библиотек с TypeError.

Неконтролируемая отмена задач (Task Leaks)

Использование asyncio.gather() или asyncio.create_task() без явного отслеживания приводило к «зависшим» фоновым задачам. Если одна из подзадач падала с исключением, остальные продолжали выполнять I/O в фоновом режиме, забивая ресурсы.

---

2. К чему пришли в Python 3.11–3.12+

В современной экосистеме Python концепция asyncio прошла через кардинальную чистку:

1. Уход от get_event_loop(): Больше никаких скрытых спавнов «мусорных» циклов. Инициализация ресурсов происходит строго внутри асинхронного контекста.
2. Structural Concurrency (Структурная асинхронность): Появление asyncio.TaskGroup и ExceptionGroup (идеи, позаимствованные из фреймворка Trio). Если одна задача внутри группы падает — все смежные задачи гарантированно отменяются, а ошибки агрегируются.
3. Eager Task Execution (Python 3.12): Задачи, готовые завершиться без реального I/O ожидания, выполняются мгновенно, не засоряя очередь Event Loop'а.

---

3. Практический пример: Было vs Стало

❌ Как делать НЕЛЬЗЯ (Паттерн из Python 3.6–3.8)

Этот код содержит сразу три архитектурные мины: глобальное состояние, устаревший gather и явный манипулятор цикла.

import asyncio

# ЛЯП 1: Глобальная инициализация примитива синхронизации.
# Привязывается к несуществующему/чужому loop при импорте модуля!
SEMAPHORE = asyncio.Semaphore(2)

async def worker(work_id: int):
    # ЛЯП 2: Неявное обращение к глобальному семафору
    async with SEMAPHORE:
        if work_id == 2:
            raise RuntimeError(f"Сбой в работе воркера {work_id}")
        await asyncio.sleep(0.1)
        print(f"Воркер {work_id} завершил работу")

async def main():
    # ЛЯП 3: gather() заставит остальные воркеры работать,
    # даже если один из них упал, упустив контроль над ресурсами.
    tasks = [worker(i) for i in range(5)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    # Старый синтаксис управления циклом
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()

---

✅ Как НУЖНО делать (Современный паттерн Python 3.11+)

Код лаконичен, безопасен по отношению к памяти и гарантирует предсказуемое завершение всех фоновых процессов.

import asyncio
from typing import NoReturn

class ServiceRunner:
    """Контекст и ресурсы инициализируются строго внутри асинхронной среды."""
    def __init__(self, max_concurrency: int) -> None:
        self._max_concurrency = max_concurrency
        self._semaphore: asyncio.Semaphore | None = None

    async def _worker(self, work_id: int) -> None:
        assert self._semaphore is not None
        
        async with self._semaphore:
            if work_id == 2:
                # Ремарка: Демонстрация штатного сбоя задачи
                raise ValueError(f"Фатальная ошибка воркера {work_id}")
            
            await asyncio.sleep(0.1)
            print(f"✓ Воркер {work_id} успешно обработал данные")

    async def run() -> None:
        # Ремарка: Семафор создается строго внутри запущенного event loop
        self._semaphore = asyncio.Semaphore(self._max_concurrency)

        # Ремарка: Structural Concurrency в действии.
        # Если worker(2) упадет, TaskGroup автоматически отменит worker(0,1,3,4)
        # и корректно освободит ресурсы без утечек.
        async with asyncio.TaskGroup() as tg:
            for i in range(5):
                tg.create_task(self._worker(i))

async def main() -> None:
    runner = ServiceRunner(max_concurrency=2)
    try:
        await runner.run()
    except* ValueError as eg:
        # Ремарка: Перехват ошибок через ExceptionGroup (Python 3.11+)
        print(f"x Перехвачена группа ошибок TaskGroup: {eg.exceptions}")

if __name__ == "__main__":
    # Единая точечная входная страна: изолирует жизненный цикл приложения
    asyncio.run(main())

---

Главные архитектурные выводы

1. Явное лучше неявного: Примитивы I/O и синхронизации должны создаваться строго внутри функций/классов, исполняемых в контексте запущенного Event Loop (main()).
2. Передача контекста по ссылке: Если компоненту нужен семафор, пул соединений или логика отмены — передавайте их явно через аргументы конструктора или метода.
3. Используйте TaskGroup вместо gather: Это снимает проблему необработанных фоновых задач (unhandled background tasks) и делает поведение приложения предсказуемым при возникновении любых Exception.