asyncio честно: приостанавливаемые фреймы, await как единственная точка переключения и блокирующий вызов, морозящий всех

Корутины — приостанавливаемые фреймы, а не потоки

Прежде чем диагностировать 26-секундный p99 из вступления — или писать async-код, который не воспроизводит его случайно, — нужна механическая картина: что на самом деле создаёт async def и что на самом деле делает await.

async def не определяет фоновую работу; он определяет функцию, чей стековый фрейм можно приостановить и возобновить. Вызов её не исполняет ничего — он возвращает объект корутины, замороженный фрейм с локальными переменными и указателем инструкции (забудете await — не выполнится ничего, и при сборке мусора прилетит RuntimeWarning: coroutine 'fetch' was never awaited — классика молча пропущенного шага). await expr делает две вещи: исполняет expr, пока тот не завершится или не приостановится, и — при приостановке — пробрасывает её вверх по всей цепочке await до цикла событий. Приостановка упирается в Future: awaitable, который ещё не готов. Цепочка важна механически: когда handler await-ит fetch, который await-ит sock_recv, все три фрейма замирают как одно связанное целое, а когда сокет приносит данные, цикл возобновляет нижний фрейм, и значения текут обратно вверх. Вытеснения в этой истории нет нигде — корутина исполняется, пока сама не await-нет нечто приостанавливающееся. Это свойство — суперсила asyncio (между await не нужны блокировки: код между двумя точками await атомарен относительно других корутин) и его заряженный пистолет.

import asyncio

async def fetch(url: str) -> bytes:
    reader, writer = await asyncio.open_connection("example.com", 80)
    writer.write(f"GET {url} HTTP/1.0\r\nHost: example.com\r\n\r\n".encode())
    await writer.drain()
    body = await reader.read(-1)     # frame suspends here until data arrives
    writer.close()
    return body

coro = fetch("/")        # ничего ещё не запущено — просто замороженный фрейм
# asyncio.run(coro)      # цикл доводит его до завершения

Цикл: очередь готовых, селектор и никакой пощады

Цикл событий — маленький честный механизм: очередь готовых колбэков и OS-селектор (epoll на Linux, kqueue на BSD/macOS). На каждой итерации он (1) исполняет все готовые колбэки — возобновить корутину до следующего await это один колбэк, (2) спрашивает селектор, какие сокеты стали готовы, с таймаутом до ближайшего таймера, (3) ставит пробуждения в очередь. Это весь двигатель, и из него следует правило, которое нарушил инцидент из вступления: шаг 1 обязан быть быстрым. Цикл не может опрашивать сокеты и запускать таймеры, пока исполняется колбэк, поэтому корутина, считающая две секунды — или зовущая requests.get, time.sleep, синхронный драйвер БД, bcrypt.hashpw или читающая большой файл простым open().read(), — останавливает мир: каждый таймер опаздывает, ни одно соединение не принимается, и p99 всех остальных запросов раздувается ровно на эту длительность. Инструменты обнаружения встроены: PYTHONASYNCIODEBUG=1 (или loop.set_debug(True)) логирует любой колбэк дольше 100 мс через loop.slow_callback_duration. Лекарство — делегирование, а не героизм: await loop.run_in_executor(None, blocking_fn, arg) исполняет функцию в дефолтном ThreadPoolExecutor и приостанавливает корутину до завершения; asyncio.to_thread(blocking_fn, arg) (3.9+) — то же без церемоний. Честная сноска из урока про GIL: поточный экзекьютор помогает, только когда заблокированный вызов отпускает GIL (I/O, большинство C-вызовов); для чистопайтоновского CPU-счёта передавайте первым аргументом ProcessPoolExecutor.

import asyncio, hashlib
from concurrent.futures import ProcessPoolExecutor

async def handler(password: bytes, big_doc: bytes):
    loop = asyncio.get_running_loop()
    # блокирующий I/O или C-вызов, отпускающий GIL → хватит потока
    rate = await asyncio.to_thread(fetch_rate_sync, "EUR")
    # чистая CPU-работа → пул процессов, иначе GIL сводит поток на нет
    digest = await loop.run_in_executor(cpu_pool, hashlib.pbkdf2_hmac,
                                        "sha256", password, b"salt", 600_000)
    return rate, digest

cpu_pool = ProcessPoolExecutor(max_workers=4)

Task против корутины: где на самом деле начинается конкурентность

await some_coro() — это последовательно: сахар для вызова функции, который может приостановиться. Конкурентность начинается только когда вы отдаёте корутину циклу как независимую единицу: asyncio.create_task(coro) заворачивает фрейм в Task, планирует старт на следующую итерацию цикла и сразу возвращается. Task — подкласс Future, который ведёт свою корутину и запоминает результат или исключение. Эта разница и кусается на код-ревью: три последовательных await трёх HTTP-вызовов стоят сумму их латентностей; три задачи, await-нутые после создания, — максимум. asyncio.gather(*coros) упаковывает паттерн: заворачивает каждый аргумент в Task, исполняет конкурентно и возвращает результаты в порядке аргументов; по умолчанию первое исключение пробрасывается сразу, пока выжившие задачи продолжают работать без присмотра, а return_exceptions=True вместо этого отдаёт исключения как значения для поэлементного разбора. Отсюда два продакшен-правила. Первое: голый create_task, ссылку на который вы выбросили, может быть собран сборщиком мусора прямо в полёте — цикл держит лишь слабую ссылку; храните ссылку или используйте структурные инструменты следующего урока. Второе: поведение gather «работаем дальше при ошибке» — замаскированная утечка ресурсов для fan-out с побочными эффектами; тоже тема следующего урока.