Блокирующий вызов останавливает всё

Самая коварная ошибка в asyncio выглядит невинно: кто-то добавил в корутину обычный time.sleep. Программа не падает, исключений нет, результаты в итоге правильные. Просто всё, что вы строили предыдущие уроки — конкурентные задачи, пачки через gather, — молча превращается в последовательное исполнение. Сегодня докажем это на таймстампах и выведем главное правило асинхронного кода.

Чтобы правило было очевидным, вспомним, что такое событийный цикл физически. Это обычный цикл while в одном потоке: выбрать готовые колбэки, выполнить их по очереди, проверить таймеры, подождать сеть, повторить. «Задачи» — не потоки и не процессы, а объекты в этом цикле: пока цикл исполняет один шаг одной корутины, весь остальной мир asyncio стоит и ждёт своей очереди. Никакой параллельности в чистом asyncio нет — есть быстрые переключения в точках await. Значит, всё, что не переключается, монополизирует поток — и с этим не поспоришь никакой архитектурой.

Эксперимент: три задачи с time.sleep

Запустим через gather три задачи, каждая спит две секунды и печатает таймстампы старта и финиша. Ожидание по опыту прошлых уроков: все стартуют вместе, финишируют вместе, суммарно 2 секунды. Смотрим, что происходит на самом деле (таймстампы — целые секунды от старта программы):

import asyncio
import time

async def job(name, t0):
    print(name, "старт:", round(time.monotonic() - t0), "с")
    time.sleep(2)                     # блокирует весь цикл, а не только задачу
    print(name, "финиш:", round(time.monotonic() - t0), "с")

async def main():
    t0 = time.monotonic()
    await asyncio.gather(job("Альфа", t0), job("Бета", t0), job("Гамма", t0))

asyncio.run(main())

Альфа старт: 0 с
Альфа финиш: 2 с
Бета старт: 2 с
Бета финиш: 4 с
Гамма старт: 4 с
Гамма финиш: 6 с

Картинка жестокая и предельно ясная. Все три задачи получили слот — gather честно их запланировал. Но Альфа, начав, провалится в time.sleep — а это не await: управление циклу не возвращается. Процессор стоит внутри одного сна, Бета и Гамма не могут даже напечатать свой старт. Прошло 2 секунды — Альфа проснулась, напечатала финиш, и только теперь очередь дошла до Беты. Итого 6 секунд вместо 2: блокирующий вызов заморозил весь событийный цикл, не только свою задачу.

На продуктиве эта картина проявляется одинаково, где бы ни жил цикл: бот перестаёт отвечать на все апдейты разом, веб-сервер не принимает новые соединения, websocket-клиенты отваливаются по своим таймаутам. Причём заметьте масштаб бедствия: заблокировала один хендлер — пострадали все, даже те, кто к этому хендлеру не обращался. В синхронном сервере с потоком на соединение медленный запрос бил только своё соединение; в asyncio цена ошибки выросла до «весь сервис». Именно поэтому в async-фреймворках синхронные вызовы в хендлерах — первый подозреваемый при любом массовом подвисании.

Тот же код с asyncio.sleep

Меняем одну строку — time.sleep(2) на await asyncio.sleep(2) — и повторяем эксперимент:

import asyncio
import time

async def job(name, t0):
    print(name, "старт:", round(time.monotonic() - t0), "с")
    await asyncio.sleep(2)            # ожидание без блокировки
    print(name, "финиш:", round(time.monotonic() - t0), "с")

async def main():
    t0 = time.monotonic()
    await asyncio.gather(job("Альфа", t0), job("Бета", t0), job("Гамма", t0))

asyncio.run(main())

Альфа старт: 0 с
Бета старт: 0 с
Гамма старт: 0 с
Альфа финиш: 2 с
Бета финиш: 2 с
Гамма финиш: 2 с

Три старта в одну секунду, три финиша в одну секунду, суммарно 2 секунды. Разница между примерами — одно слово await, но за ним стоит принцип: await — единственный способ корутины сказать циклу «я жду, работай без меня». time.sleep такой фразы не знает: он блокирует поток целиком, а цикл событий живёт в том же потоке.

Справедливости ради: await asyncio.sleep не бесплатен — он ставит таймер в служебную кучу цикла, а цикл тратит чуть-чуть времени на проверку этих таймеров. Но масштаб несопоставим: тысячи параллельных sleep живут в одной куче без единого потока, тогда как тысяча time.sleep потребовала бы тысячу потоков с их стеками. В этом и заключалась ставка создателей asyncio: ожидание должно стоить почти ноль, потому что в сетевых сервисах ожидание — основной режим работы.

import asyncio
import time

async def fake_fetch():
    time.sleep(2)                     # «загрузка», которая не уступает управление
    return "данные"

async def job(name, t0):
    print(name, "старт:", round(time.monotonic() - t0), "с")
    await fake_fetch()
    print(name, "финиш:", round(time.monotonic() - t0), "с")

async def main():
    t0 = time.monotonic()
    await asyncio.gather(job("Запрос-1", t0), job("Запрос-2", t0))

asyncio.run(main())

Запрос-1 старт: 0 с
Запрос-1 финиш: 2 с
Запрос-2 старт: 2 с
Запрос-2 финиш: 4 с

Пульс: что видит задача, пока цикл мёртв

Таймстампы убедительны, но покажем заморозку изнутри. Запустим фоновый heartbeat — задачу, которая раз в секунду печатает тик. Рядом заблокируем цикл на три секунды и посмотрим, сколько тиков дозвонится:

import asyncio
import time

async def heartbeat(t0):
    while True:
        await asyncio.sleep(1)
        print("тик:", round(time.monotonic() - t0), "с")

async def blocker(t0):
    print("блокирую цикл на 3 секунды...")
    time.sleep(3)                     # цикл мёртв всё это время
    print("цикл ожил на", round(time.monotonic() - t0), "с")

async def main():
    t0 = time.monotonic()
    hb = asyncio.create_task(heartbeat(t0))
    await asyncio.sleep(0)            # даём heartbeat начать
    await blocker(t0)
    hb.cancel()

asyncio.run(main())

блокирую цикл на 3 секунды...
цикл ожил на 3 с

Ни одного тика. Heartbeat был запущен, его таймеры сработали на первой и второй секундах — но цикл, который должен их разбудить, был мёртв: он не спал, а честно исполнял time.sleep в чужой корутине. Проснулся — а отменять уже нечего, пульс не прозвучал. Именно так выглядит «подтормаживающий» asyncio-сервис на проде: бот не отвечает, веб-сервер не принимает — а в профиле единственный виновник: синхронный вызов где-то в хендлере.

Этот приём — измеритель lag цикла — стоит запомнить и как диагностику: фоновая задача, которая раз в секунду просыпается и сравнивает ожидаемое время с фактическим. Задержка в десятки миллисекунд — норма; секунды — где-то в цикле живёт блокировка, и метрика растёт ровно на время её работы. В больших проектах такой монитор превращают в метрику с алертом, и вопросы «почему всё висло» решаются просмотром графика, а не ночными раскопками.

Для сравнения — заменим в блокере time.sleep на await asyncio.sleep(3):

import asyncio
import time

async def heartbeat(t0):
    while True:
        await asyncio.sleep(1)
        print("тик:", round(time.monotonic() - t0), "с")

async def pause_async(t0):
    print("неблокирующая пауза на 3 секунды...")
    await asyncio.sleep(3)
    print("цикл жил и на", round(time.monotonic() - t0), "с")

async def main():
    t0 = time.monotonic()
    hb = asyncio.create_task(heartbeat(t0))
    await asyncio.sleep(0)            # даём heartbeat начать
    await pause_async(t0)
    hb.cancel()

asyncio.run(main())

неблокирующая пауза на 3 секунды...
тик: 1 с
тик: 2 с
цикл жил и на 3 с

Тот же принцип работает в обратную сторону — и этим пользуются watchdog-и: если пульс пропал, значит, цикл кто-то держит. Внешний монитор стучится в health-эндпоинт, который отвечает изнутри того же цикла; пока цикл жив, ответы идут мгновенно, замер — таймауты, и система принимает решение: алерт дежурному или перезапуск. Здоровье asyncio-сервиса — это здоровье его цикла, и любые проверки живости сводятся к одному вопросу: успевает ли цикл крутиться.

Блокирует не только сон

time.sleep — герой примеров, но блокирует что угодно синхронное, что держит процессор или ждёт ввод-вывод синхронно: requests.get без async-аналога, тяжёлые вычисления, чтение большого файла, hashlib на гигабайте. Тяжёлые вычисления блокируют так же честно — циклу не отдают управление даже await-ом, ведь await есть не у каждого оператора:

import asyncio

async def counter(name):
    print(name, "начала считать")
    total = sum(range(5_000_000))     # тяжёлые вычисления - тоже блокировка
    print(name, "закончила считать")

async def main():
    await asyncio.gather(counter("Альфа"), counter("Бета"))

asyncio.run(main())

Альфа начала считать
Альфа закончила считать
Бета начала считать
Бета закончила считать

Никакого sleep в коде нет — но задачи исполнились строго по очереди: до конца цикла суммирования Альфы Бета не могла даже напечатать старт. Отсюда правило по границам: asyncio создан для ввода-вывода (сеть, диск), где ожидание можно перехватывать; процессорная работа блокирует цикл и требует потоков или процессов.

К счастью, почти у каждой популярной синхронной библиотеки есть async-родственник: для sqlite3 — aiosqlite, для файловых операций — aiofiles, для сетевых клиентов — aiohttp и компания. Их устройство одинаковое: снаружи await, внутри та же работа, отданная в поток или переписанная на неблокирующие вызовы. Первое, что стоит делать, обнаружив синхронный вызов в async-коде, — искать такой async-аналог; и только если его нет, спускаться до to_thread из раздела ниже.

Правило и выход из положения

1. В async-коде только async-ожидания: asyncio.sleep вместо time.sleep, aiohttp вместо requests, aiosqlite вместо sqlite3.
2. Нужна синхронная библиотека без async-версии — уводи вызов в поток: await asyncio.to_thread(blocking_func, arg).
3. Тяжёлые вычисления — поток для одноразовой задачи, ProcessPoolExecutor для постоянной нагрузки: цикл в обоих случаях остаётся жив.
4. Диагностика подтормаживаний начинается с поиска синхронных вызовов внутри async def — это источник номер один.

import asyncio
import time

def parse_big_file():
    time.sleep(3)                     # тяжёлая синхронная работа
    return "готово"

async def main():
    result = await asyncio.to_thread(parse_big_file)
    print(result)

asyncio.run(main())

У to_thread есть предел, о котором стоит знать заранее: потоки берутся из пула, и он конечен — по умолчанию несколько десятков рабочих. Тысяча одновременных to_thread не создаст тысячу потоков: лишние вызовы встанут в очередь пула и будут ждать. Это не блокировка цикла (цикл жив и занимается другими задачами), но латентность растёт. Для тысяч одновременных тяжёлых операций пул настраивают явно или переходят на процессы — детали выходят за рамки урока, а правило «в async-коде не блокировать» остаётся прежним.

Граница правила наглядна в паре «синхронный HTTP-клиент против асинхронного». Слева — то, что ломает сервис; справа — то, что нужно (фрагменты для локального запуска, сети в песочнице нет):

# ТАК НЕЛЬЗЯ: requests внутри корутины замораживает цикл
import requests

async def get_url_bad(url):
    resp = requests.get(url)          # синхронный вызов - блокировка
    return resp.status_code

# ТАК МОЖНО: aiohttp умеет await
import aiohttp

async def get_url_good(session, url):
    async with session.get(url) as resp:
        return resp.status

Что дальше

Правило сформулировано: корутина живёт, только пока не встретила await; всё, что не await, — блокировка всего цикла. Дальше учимся строить из задач конвейеры: очередь asyncio.Queue связывает продюсера и консьюмеров так, что медленный этап не тормозит быстрый. Это готовый скелет для воркеров настоящего бота — там же пригодится всё, что вы знаете про отмену.

Последний штрих к правилу — про чужой код: библиотека не обязана быть асинхронной, даже если её вызывают из async-функции. Документация честнее любых заголовков: если в ней нет слова await у вызова, считайте вызов блокирующим и проверяйте его так же строго, как time.sleep.

У asyncio нет колдовства: пока корутина не в await, процессор делает ровно её — и весь остальной мир задач ждёт.