← все задачи

Async Python · задача 7 из 10

Блокирующий код внутри асинхронного

Средний 15 минут to_threadexecutorGIL

Условие

В асинхронном обработчике нужно вызвать синхронную библиотеку: запрос через requests, хеширование пароля через bcrypt и разбор большого JSON. Как сделать это, не заблокировав цикл событий? Ответьте отдельно для операции ввода-вывода и для нагрузки на процессор.

Что требуется

  • Обработчик остаётся отзывчивым для других запросов
  • Решение разное для I/O и для CPU — объясните почему
  • Объясните, что произойдёт, если ничего не делать

Пример

@app.get("/user")
async def handler():
    data = requests.get(url).json()     # блокирует весь сервис
    hashed = bcrypt.hashpw(password)    # блокирует весь сервис

Сначала уточните

Вопросы до кода — половина оценки. Молча начать печатать хуже, чем задать два вопроса.

  • Есть ли асинхронный аналог библиотеки — httpx или aiohttp вместо requests?
  • Насколько тяжёлая операция по процессору: миллисекунды или секунды?
  • Сколько таких вызовов в секунду ожидается — от этого зависит размер пула
Показать решение Скрыть решение

Решение

import asyncio
from concurrent.futures import ProcessPoolExecutor

import bcrypt
import httpx


# 1. Ввод-вывод: лучше всего — асинхронная библиотека
async def fetch(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()


# 2. Если асинхронного аналога нет — поток
async def legacy_call(argument):
    return await asyncio.to_thread(sync_library_call, argument)


# 3. Нагрузка на процессор — процессы, а не потоки
process_pool = ProcessPoolExecutor(max_workers=4)


async def hash_password(password):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        process_pool, bcrypt.hashpw, password, bcrypt.gensalt(),
    )

Почему так

Что именно происходит при блокирующем вызове

  • Цикл событий — один поток: пока в нём выполняется синхронный код, ни одна другая корутина не продвинется
  • Секундный requests.get означает секунду, в течение которой сервис не отвечает никому, а не только этому клиенту
  • Внешне это выглядит как «иногда тормозит» и почти не ловится без профилировщика — потому и спрашивают

Почему для I/O достаточно потока

  • На времени ожидания ответа GIL освобождается, поэтому поток не мешает основному
  • to_thread — это тот же run_in_executor с пулом потоков по умолчанию, только короче
  • Пул потоков не бесконечен: на тысяче одновременных вызовов это станет узким местом, и лучше искать асинхронную библиотеку

Почему для CPU нужны процессы

  • Вычисления держат GIL, и поток не даст выигрыша — блокировать будет ровно так же, просто в другом месте
  • Отдельный процесс имеет свой интерпретатор и свой GIL, поэтому работает по-настоящему параллельно
  • Цена — сериализация аргументов и результата через pickle: гонять большие объекты туда-обратно дорого

Что спросят дальше

  • Спросят, как найти такие места в проде: asyncio debug mode логирует колбэки дольше 100 мс
  • Спросят про ORM: синхронный Django ORM в async-вью — это sync_to_async, и заодно вопрос про пул соединений
  • Спросят, что нельзя отменить операцию в потоке — задача отменится, поток продолжит работать

Следующая задача

Фоновая задача по расписанию — Периодическое обновление кэша: важно, чтобы одна ошибка не убила задачу навсегда.

Перейти