Мосты между sync и async: to_thread и экзекьюторы в одну сторону, долгоживущий цикл и run_coroutine_threadsafe в другую

Async в sync: убрать блокирующую работу с цикла

Урок 1 доказал, почему цикл не должен ждать: один блокирующий вызов морозит опрос селектора. Клапанов давления два, и выбор между ними — GIL-математика из юнита про конкурентность. await asyncio.to_thread(fn, *args) (3.9+) выполняет fn в дефолтном ThreadPoolExecutor (пул потоков из стандартной библиотеки concurrent.futures) цикла и протаскивает contextvars (контекстные переменные, аналог thread-local для корутин); loop.run_in_executor(pool, fn) — старшая форма с явным пулом. Для блокирующего I/O — синхронный драйвер БД, requests, чтение файлов — поток решает задачу целиком: он паркуется в сисколле, ничего не держа, цикл бежит свободно. Для CPU-bound работы GIL по-прежнему сериализует Python-байткод между потоками: to_thread оставляет цикл более-менее отзывчивым (GIL переходит из рук в руки каждые ~5 мс, и цикл продвигается между слайсами), но не покупает ни грамма пропускной способности и портит латентность цикла под контеншном. Настоящий CPU-параллелизм — это ProcessPoolExecutor: налог pickle в обе стороны и старт процессов, окупается от десятков миллисекунд CPU на единицу работы.

Ловушка размера — дефолтный экзекьютор: min(32, cpu_count + 4) потоков. Сорок одновременных to_thread(requests.get, ...) на 8-ядерной машине — это 12 работающих потоков и 28 вызовов, невидимо стоящих в очереди: у вашего «асинхронного» эндпоинта теперь скрытое ожидание тред-пула в p99. Та же форма, что в юните про веб-сервисы, где фреймворк выгружает синхронные эндпоинты в дефолтный пул на 40 потоков: у каждого моста есть ширина, и неизмеренная ширина — это обрыв латентности. Размеряйте явный пул под зависимость, которую он защищает (БД на 10 соединений нужны ~10 потоков, а не 32).

pool = concurrent.futures.ThreadPoolExecutor(max_workers=10, thread_name_prefix="db")

async def get_user(uid):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(pool, db.fetch_user, uid)   # ограниченный мост

Sync в async: три инструмента, три ситуации

У обратного направления — инструмент на ситуацию, и их перепутывание — инцидент из Хука. asyncio.run(coro) создаёт свежий цикл, прогоняет до конца, закрывает цикл — корректно примерно один раз за жизнь процесса: скрипты, main(), тесты. Всё, что привязано к циклу — пулы соединений, сессии, транспорты, — умирает вместе с ним; поэтому по-запросный asyncio.run пересобирал мир SDK две тысячи раз в минуту. asyncio.run_coroutine_threadsafe(coro, loop) отправляет корутину в работающий цикл из другого потока и возвращает concurrent.futures.Future (синхронный Future из стандартной библиотеки, не asyncio.Future), чей .result(timeout) блокирует вызывающий поток, а не цикл. Это серверный мост: один долгоживущий цикл в выделенном потоке, и каждый sync-вызывающий передаёт корутины через него:

class AsyncBridge:
    def __init__(self):
        self.loop = asyncio.new_event_loop()
        threading.Thread(target=self.loop.run_forever, daemon=True, name="bridge").start()

    def call(self, coro, timeout=None):
        fut = asyncio.run_coroutine_threadsafe(coro, self.loop)   # потокобезопасная передача
        return fut.result(timeout)        # блокирует ЭТОТ поток; цикл бежит свободно

bridge = AsyncBridge()                    # один цикл, один пул соединений на процесс
def view(request):                        # синхронная WSGI-вью, фреймворк не тронут
    return bridge.call(client.search(request.GET["q"]), timeout=5)

Третья ситуация — ошибка, которую вы встретите первой: вызов asyncio.run (или loop.run_until_complete) при уже работающем цикле поднимает RuntimeError: asyncio.run() cannot be called from a running event loop — цикл нереентерабелен by design; вложенный запуск должен был бы поставить итерацию на паузу посреди разгрузки. Jupyter — классическое столкновение (ноутбук сам крутит цикл — потому там и работает await на верхнем уровне). nest_asyncio, честно: он манкипатчит цикл, разрешая реентерабельный run_until_complete, — работает в ноутбуке и нарушает инварианты цикла везде остальном; в проде это диагностическая улика («кто-то воюет с архитектурой»), а не фикс.

Менеджмент заразности и правила на границе

Async заразен вверх — каждый вызывающий корутину обязан сам await-ить или мостить, — и каждое пересечение моста стоит прыжка между потоками, бухгалтерии футур и побудки целевого потока: десятки микросекунд в лучшем случае, плюс очередь на ширине пула в худшем. Архитектурное правило: одна граница. Синхронное ядро с асинхронным краем (Django-мост выше) или асинхронное ядро с синхронными листовыми вызовами через to_thread — и отказ от посыпки посреди стека, где слой три асинхронный, слой четыре синхронный, слой пять снова асинхронный, и каждый переход — мост со своей шириной, таймаутом и режимом отказа.

На самой границе потокобезопасность абсолютна: объекты цикла не потокобезопасны, и единственные законные входы из чужого потока — loop.call_soon_threadsafe(cb) и run_coroutine_threadsafe. Механизм замыкает круг урока 1: цикл может быть запаркован в selector.select(timeout) — спит в ядре. call_soon_threadsafe добавляет колбэк и пишет байт в self-pipe, за которым следит селектор, — цикл просыпается немедленно. Обычный call_soon из чужого потока пропускает побудку: колбэк лежит незамеченным до следующего естественного пробуждения — или прямо портит внутреннее состояние. Никогда не вызывайте task.cancel(), future.set_result() или любой метод цикла напрямую из другого потока; заворачивайте в call_soon_threadsafe.