← назад к разделу

Самая частая причина «сервис на asyncio тормозит под нагрузкой» это не asyncio, а один синхронный вызов внутри async def: клиент к старой системе без асинхронной версии, библиотека шифрования, requests вместо httpx, чтение файла в 200 мегабайт. Разберём, как такие места находить, как выносить их в потоки, не сломав остальное, и что делать, когда потоков не хватает.

Обязательно

Как выглядит блокировка

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

БлокируетЧто вместо
requests.get(...)httpx.AsyncClient
psycopg2, pymysql, синхронный Session SQLAlchemyasyncpg, psycopg async, AsyncSession
time.sleep(1)await asyncio.sleep(1)
open(path).read() на большом файлеasyncio.to_thread(read_file, path) или aiofiles
boto3aiobotocore, или to_thread
json.loads на 50 МБ, hashlib на большом буфере, bcryptto_thread, для тяжёлого счёта процессы
subprocess.runasyncio.create_subprocess_exec

Проверить догадку можно режимом отладки: python -X dev или PYTHONASYNCIODEBUG=1 включают замер каждого шага цикла, и шаг дольше 0,1 секунды попадает в лог как Executing <Task ...> took 0.200 seconds. Проверено на Python 3.14: time.sleep(0.2) внутри корутины под -X dev даёт именно такое предупреждение. Подробнее в статье про отладку.

to_thread: вынести вызов в поток

asyncio.to_thread(func, *args) выполняет синхронную функцию в пуле потоков и возвращает результат как корутину, которую можно await. Цикл событий в это время свободен и обслуживает другие задачи.

import asyncio, bcrypt

async def verify_password(password: str, stored_hash: bytes) -> bool:
    return await asyncio.to_thread(bcrypt.checkpw, password.encode(), stored_hash)

Два факта, которые стоит знать. to_thread переносит в поток контекст contextvars: идентификатор запроса из middleware виден в логах, которые пишет синхронная функция. Проверено: ContextVar, установленная в задаче, читается внутри to_thread. И to_thread это тот же loop.run_in_executor(None, ...), только с контекстом и именованными аргументами; старая форма нужна, когда хотите передать свой пул.

Поток не отменяется. Если задачу, которая ждёт to_thread, отменить, await поднимет CancelledError, но функция в потоке доработает до конца и займёт поток всё это время. Для долгих синхронных операций это означает, что отмена запроса клиентом не освобождает ресурсы, и единственная защита это таймаут внутри самой синхронной функции (например, таймаут сокета у драйвера).

Пул потоков и его исчерпание

Пул по умолчанию это ThreadPoolExecutor с min(32, число_ядер + 4) потоками; на машине с 11 ядрами это 15. Пока все потоки заняты, следующий to_thread ждёт свободного, и это ожидание не видно в коде: обработчик просто становится медленным. Под нагрузкой, когда сотня запросов ждёт синхронный драйвер, пул из 15 потоков превращается в узкое место.

Варианты: увеличить пул через loop.set_default_executor(ThreadPoolExecutor(max_workers=64)) в lifespan; завести отдельный пул для тяжёлых операций, чтобы они не вытесняли лёгкие, и передавать его в run_in_executor; или убрать причину, заменив синхронную библиотеку асинхронной. Третий вариант единственный, который масштабируется: поток стоит мегабайты памяти и переключения, а задача asyncio килобайты.

from concurrent.futures import ThreadPoolExecutor

heavy_pool = ThreadPoolExecutor(max_workers=4, thread_name_prefix="heavy")

async def render_pdf(report: Report) -> bytes:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(heavy_pool, build_pdf, report)

Число потоков в пуле это одновременно и лимит параллелизма на внешнюю систему, что иногда удобно: четыре потока на генерацию PDF означают не больше четырёх одновременных генераций.

sync def во FastAPI не бесплатен

Обработчик, объявленный как def, а не async def, FastAPI выполняет в пуле потоков, чтобы не блокировать цикл. Это удобно для синхронного кода, но пул общий на всё приложение и ограничен: Starlette использует ограничитель anyio на 40 потоков. Сорок одновременных синхронных обработчиков, и сорок первый ждёт. Проверено на Starlette 1.7: current_default_thread_limiter().total_tokens равен 40.

То же касается синхронных зависимостей Depends и фоновых задач с def. Правило: async def для всего, что ждёт сеть через асинхронные клиенты; def только для короткого синхронного кода без ввода-вывода или с осознанно принятым лимитом пула. Поднять лимит можно через anyio.to_thread.current_default_thread_limiter().total_tokens = 100 в lifespan. Как FastAPI выбирает способ выполнения, подробно в статье про async во FastAPI.

Из потока обратно в цикл

Поток не может напрямую вызывать корутины или трогать объекты цикла: asyncio.Lock, Queue и задачи не потокобезопасны. Чтобы передать результат из потока в цикл, есть два инструмента. loop.call_soon_threadsafe(callback, *args) ставит обычную функцию в очередь цикла из любого потока; это единственный потокобезопасный метод цикла. asyncio.run_coroutine_threadsafe(coro, loop) запускает корутину в цикле из потока и возвращает concurrent.futures.Future, у которого можно ждать результат синхронно.

def on_message_from_legacy_sdk(payload: bytes, loop: asyncio.AbstractEventLoop, queue: asyncio.Queue):
    loop.call_soon_threadsafe(queue.put_nowait, payload)   # безопасно из чужого потока

Так подключают библиотеки с колбэками в своих потоках (SDK брокеров, драйверы устройств): колбэк только перекладывает данные в очередь цикла, а вся обработка идёт в задачах. Обратный путь, из цикла в поток, это to_thread; ловушка в том, что синхронная функция внутри потока не должна вызывать asyncio.run (упадёт, если цикл есть в этом потоке, или создаст второй цикл), ей дают run_coroutine_threadsafe с ссылкой на основной цикл.

Потоки и общие данные

Как только появились потоки, однопоточные гарантии asyncio заканчиваются. Словарь, который меняют и задача, и функция в to_thread, нужно защищать threading.Lock, причём захватывать его в задаче без await внутри, иначе цикл встанет на блокировке. Проще не делить состояние: функция в потоке получает копию входа и возвращает результат, а в структуры цикла пишет только задача после await to_thread. Глобальная блокировка интерпретатора (GIL) защищает отдельные операции над встроенными типами от повреждения, но не защищает последовательности операций: d[k] += 1 из двух потоков теряет инкременты. Что меняется в сборке без GIL, рассказывает статья про свободные потоки и процессы.

Дополнительно: при первом чтении можно пропустить

Глубже: что делает поток с циклом, а цикл с потокомрасширенное

Пока функция выполняется в потоке пула, цикл событий продолжает крутиться, но оба делят один GIL: поток, который считает хеш на Python, держит GIL и отдаёт его каждые 5 миллисекунд (sys.getswitchinterval()), так что цикл получает процессор небольшими квантами. Для функций, которые ждут ввода-вывода (драйвер базы, HTTP-клиент), GIL отпускается на время ожидания, и цикл не страдает. Для чисто вычислительных функций на Python to_thread разгружает цикл лишь частично: он перестаёт стоять намертво, но замедляется; библиотеки на C (hashlib, bcrypt, NumPy) отпускают GIL сами, и с ними to_thread работает хорошо. Отсюда правило выбора: ввод-вывод и C-библиотеки в потоки, вычисления на чистом Python в процессы или подинтерпретаторы.

Второй эффект: контекст. to_thread копирует contextvars на момент вызова, но изменения, сделанные в потоке, в задачу не возвращаются. Если синхронная функция выставляет ContextVar (например, устанавливает идентификатор трассы), в вызвавшей задаче этого не будет видно.

Коротко

  • Блокировка цикла выглядит как рост задержки всех запросов при свободном процессоре; подозреваемые: requests, синхронные драйверы, time.sleep, большие файлы, тяжёлый счёт.
  • python -X dev пишет Executing <Task> took N seconds для шагов дольше 0,1 секунды и находит виновника.
  • asyncio.to_thread выносит синхронный вызов в пул потоков с переносом contextvars; отмена задачи поток не прерывает.
  • Пул по умолчанию min(32, ядра + 4) потоков; исчерпание выглядит как тихое замедление; отдельный пул для тяжёлых операций или замена библиотеки на асинхронную.
  • def-обработчики и зависимости FastAPI идут в общий пул на 40 потоков; async def для всего, что ждёт сеть асинхронно.
  • Из потока в цикл только call_soon_threadsafe и run_coroutine_threadsafe; примитивы asyncio не потокобезопасны.
  • Общие данные между потоком и задачей защищают threading.Lock или не делят вовсе; GIL не спасает от потерянных инкрементов.
  • Поток с чистым Python делит GIL с циклом; C-библиотеки и ввод-вывод отпускают его, вычисления на Python выносят в процессы.

Что почитать дальше