asyncio и Python со свободной многопоточностью

Особенности потокобезопасности

Хотя asyncio спроектирован с учетом потокобезопасности в среде Python со свободной многопоточностью, при использовании asyncio с потоками следует помнить о некоторых особенностях:

1. Цикл событий: у каждого потока должен быть собственный цикл событий, который нельзя использовать совместно с другими потоками. Это позволяет циклу событий управлять собственными задачами и обратными вызовами без вмешательства других потоков.
2. Управление задачами: задачи и объекты future, созданные в одном потоке, нельзя ожидать или изменять из другого потока.
3. Потокобезопасные API: при взаимодействии с asyncio из нескольких потоков важно использовать предоставляемые asyncio потокобезопасные API, например asyncio.run_coroutine_threadsafe() для передачи сопрограмм циклу событий из другого потока. Если нужно вызвать обратный вызов из другого потока, можно безопасно запланировать его с помощью loop.call_soon_threadsafe().
4. Синхронизация: примитивы синхронизации asyncio (такие как asyncio.Lock и asyncio.Event) не предназначены для использования между потоками. Для синхронизации потоков следует вместо них использовать примитивы синхронизации из модуля threading.

Использование asyncio с потоками

asyncio поддерживает запуск одного цикла событий в каждом потоке, что позволяет использовать несколько ядер процессора в среде Python со свободной многопоточностью. Каждый поток может выполнять собственный цикл событий, а задачи можно независимо планировать в этих циклах.

Пример использования asyncio с потоками:

import asyncio
import threading

async def worker(name: str) -> None:
    print(f"Worker {name} starting")
    await asyncio.sleep(1)
    print(f"Worker {name} done")

def run_loop(name: str) -> None:
    asyncio.run(worker(name))

threads = [
    threading.Thread(target=run_loop, args=(f"T{i}",))
    for i in range(4)
]
for t in threads:
    t.start()
for t in threads:
    t.join()

В этом примере каждый поток создает собственный цикл событий с помощью asyncio.run() и выполняет в нем сопрограмму. Потоки выполняются одновременно, а в сборке со свободной многопоточностью могут параллельно работать на разных ядрах процессора.

Производитель и потребитель в разных потоках

Если обычному потоку (не asyncio) нужно передать работу циклу событий asyncio, выполняющемуся в другом потоке, используйте потокобезопасный примитив, например queue.Queue, а не asyncio.Queue, который безопасно использовать только в пределах одного цикла событий:

import asyncio
import queue
import threading

def producer(q: queue.Queue[int]) -> None:
    for i in range(5):
        print(f"Producing {i}")
        q.put(i)
    q.shutdown()

async def consumer(q: queue.Queue[int]) -> None:
    while True:
        try:
            item = q.get_nowait()
        except queue.Empty:
            await asyncio.sleep(0.1)
            continue
        except queue.ShutDown:
            break
        print(f"Consumed {item}")
        await asyncio.sleep(item)

q: queue.Queue[int] = queue.Queue()
consumer_thread = threading.Thread(
    target=lambda: asyncio.run(consumer(q))
)
consumer_thread.start()
producer(q)
consumer_thread.join()

Производитель работает в главном потоке, а потребитель — внутри цикла событий в отдельном потоке, но они безопасно обмениваются данными через queue.Queue. Когда очередь пуста, потребитель ненадолго засыпает и пробует снова. После завершения работы производитель вызывает shutdown(). В результате последующие вызовы get_nowait() вызывают исключение queue.ShutDown, что позволяет потребителю корректно завершить работу.