Краткий ответ

Работу из другого потока следует передавать в event loop через потокобезопасные механизмы: asyncio.run_coroutine_threadsafe для корутины или loop.call_soon_threadsafe для обычного callback. Нельзя напрямую создавать в чужом потоке задачу, менять состояние loop или обращаться к непотокобезопасным объектам asyncio.

Исторический контекст

asyncio строится вокруг однопоточного event loop и кооперативного выполнения корутин. Такой подход упрощает синхронизацию операций внутри loop, но создаёт границу между потоком loop и внешними потоками.

Потоки часто нужны для запуска блокирующего кода, интеграции со старой библиотекой или получения событий от API, которое вызывает обработчики вне event loop. Для пересечения этой границы asyncio предоставляет специальные потокобезопасные точки входа.

Постановка проблемы

Объекты вроде задач, futures и очередей asyncio обычно связаны с конкретным event loop и не предназначены для произвольного доступа из других потоков. Прямое обращение к ним может привести к исключению, гонке или потере уведомления.

Особенно опасна попытка вызвать обычный метод планирования задачи из рабочего потока. Даже если это иногда работает в тестах, такой код опирается на незащищённое внутреннее состояние loop и может сломаться при нагрузке или смене версии Python.

Подробное решение

Для передачи корутины используется asyncio.run_coroutine_threadsafe. Она планирует корутину в указанном loop и возвращает объект concurrent.futures.Future, через который внешний поток может дождаться результата или получить исключение.

Для передачи обычной функции или callback используется loop.call_soon_threadsafe. Он добавляет callback в очередь loop и будит его, если тот сейчас ожидает события. Сам callback затем выполнится уже в потоке event loop.

import asyncio
import threading

async def work():
    await asyncio.sleep(0.1)
    return 42

def worker(loop):
    future = asyncio.run_coroutine_threadsafe(work(), loop)
    print(future.result())

async def main():
    thread = threading.Thread(target=worker, args=(asyncio.get_running_loop(),))
    thread.start()
    await asyncio.sleep(0.2)
    thread.join()

asyncio.run(main())

В примере другой поток не запускает отдельный loop и не создаёт задачу напрямую. Он передаёт корутину уже работающему loop, а future.result() блокирует только рабочий поток, не event loop.

Важно не вызывать future.result() из самого event loop: это заблокирует его ожиданием собственного результата и может привести к взаимной блокировке. Также нужно учитывать жизненный цикл loop: после его остановки новые работы в него передавать нельзя.

Потокобезопасность метода передачи не делает автоматически потокобезопасными данные приложения. Если callback и внешний поток совместно меняют обычное изменяемое состояние, для этого состояния всё равно может понадобиться синхронизация или передача сообщений через очередь.

Ситуация из практики

Сервис получает уведомления от клиентской библиотеки в отдельном потоке, а обработка уведомлений должна выполняться в основном event loop. Прямой вызов создания задачи из callback библиотеки иногда приводит к ошибке о другом потоке или к нестабильному поведению.

Рассматривались два варианта. Запустить отдельный event loop в потоке библиотеки можно, но тогда усложняются обмен данными, отмена задач и завершение приложения. Передавать каждое событие через общий потокобезопасный контейнер проще концептуально, но появляется отдельный механизм пробуждения loop и контроля очереди.

Выбран вызов loop.call_soon_threadsafe, который помещает обработку события в loop. Для операций, где нужен результат корутины, используется run_coroutine_threadsafe; результат читается в рабочем потоке с тайм-аутом. Это сохраняет единственного владельца состояния внутри loop и явно ограничивает блокировку внешнего потока.

Что кандидаты часто упускают

Можно ли вызвать asyncio.create_task из рабочего потока, если передать ему корутину?

Нет, этого делать не следует. create_task рассчитан на вызов в контексте текущего работающего event loop и не является универсальным межпоточным API. Из другого потока нужно передать корутину через asyncio.run_coroutine_threadsafe, указав целевой loop.

Чем отличается run_coroutine_threadsafe от asyncio.to_thread?

run_coroutine_threadsafe направляет корутину из внешнего потока в уже работающий event loop. asyncio.to_thread, наоборот, позволяет корутине передать синхронную функцию на выполнение в рабочий поток, чтобы не блокировать event loop. Эти операции имеют противоположное направление пересечения границы.

Что произойдёт, если event loop остановится до выполнения переданной корутины?

Переданная работа может не завершиться, а возвращённый внешний future не даст обычного успешного результата. При завершении приложения нужно явно организовать остановку: прекратить приём новых работ, дождаться или отменить уже переданные задачи и только затем закрыть loop. Иначе возможны зависшие ожидания, предупреждения о незавершённых корутинах и потеря результатов.