asyncio 및 Free-Threaded Python¶

asyncio는 이벤트 루프를 스케줄러로 사용하여 작업을 전환함으로써 비차단(non-blocking) I/O 연산을 가능하게 하고 매우 효율적인 동시성을 구현합니다. 이는 I/O 바운드 사례에서 더 나은 성능을 제공합니다. 또한 CPU 바운드 작업을 스레드나 프로세스 풀로 오프로딩할 수 있지만, CPython에서는 여전히 global interpreter lock 의 제한을 받습니다.

그러나 free-threaded Python 에서는 GIL이 비활성화되어 Python이 진정한 멀티스레드 코드를 실행할 수 있습니다. 이는 asyncio가 이제 GIL에 의한 제약 없이 다중 CPU 코어의 이점을 활용할 수 있음을 의미합니다.

Python 3.14부터 asyncio는 Free-Threaded Python을 위한 일급 지원(first-class support)을 제공하며, asyncio 구현은 멀티스레드 환경에서 안전하게 사용할 수 있습니다.

단일 코어의 단일 이벤트 루프는 여러 연결을 동시에 처리할 수 있지만, 각 연결을 처리하기 위해 실행되는 Python 코드는 여전히 직렬로 실행됩니다. 요청에 상당한 양의 계산이 포함되면 해당 처리가 병목 현상이 되어 단일 코어가 더 이상 따라가지 못하게 됩니다. 이 경우 asyncio를 스레드와 결합하는 것이 매우 유용합니다. 스레드당 하나씩 이벤트 루프를 실행함으로써 여러 CPU 코어에 걸쳐 서로 다른 요청을 병렬로 처리할 수 있기 때문입니다. 또한 asyncio 애플리케이션에서 블로킹 또는 CPU 바운드 코드를 실행해야 하는 경우에도 유용합니다.

Scaling asyncio on Free-Threaded Python, a blog post by Kumar Aditya which explains the internal changes that make asyncio safe and efficient under free-threaded Python, together with benchmarks of the resulting improvements.

스레드 안전성 고려 사항¶

asyncio는 Free-Threaded Python 환경에서 스레드 안전하도록 설계되었으나, asyncio를 스레드와 함께 사용할 때는 여전히 몇 가지 고려해야 할 사항이 있습니다:

이벤트 루프: 각 스레드는 자체 이벤트 루프를 가져야 하며, 이벤트 루프는 스레드 간에 공유되어서는 안 됩니다. 이를 통해 이벤트 루프가 다른 스레드의 방해 없이 고유의 작업과 콜백을 관리할 수 있습니다.
태스크 관리: 한 스레드에서 생성된 태스크와 퓨처(future)는 다른 스레드에서 대기하거나 조작해서는 안 됩니다.
스레드 안전 API: 여러 스레드에서 asyncio와 상호작용할 때, 다른 스레드에서 이벤트 루프에 코루틴을 제출하기 위한 asyncio.run_coroutine_threadsafe() 와 같이 asyncio가 제공하는 스레드 안전 API를 사용하는 것이 중요합니다. 다른 스레드에서 콜백을 호출해야 하는 경우, loop.call_soon_threadsafe() 를 사용하여 안전하게 예약할 수 있습니다.
동기화: asyncio에서 제공하는 동기화 프리미티브(asyncio.Lock, asyncio.Event 등)는 스레드 간에 사용되도록 설계되지 않았습니다. 스레드 간의 동기화가 필요한 경우 대신 threading 모듈의 동기화 프리미티브를 사용해야 합니다.

스레드와 함께 asyncio 사용하기¶

asyncio는 스레드당 하나의 이벤트 루프 실행을 지원하며, 이를 통해 Free-Threaded Python 환경에서 다중 CPU 코어의 이점을 활용할 수 있습니다. 각 스레드는 고유의 이벤트 루프를 실행할 수 있으며, 태스크는 해당 루프에서 독립적으로 예약될 수 있습니다.

스레드와 함께 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() 을 사용하여 자체 이벤트 루프를 생성하고 해당 루프에서 코루틴을 실행합니다. 스레드들은 동시에 실행되며, free-threaded 빌드에서는 별도의 CPU 코어에서 병렬로 실행될 수 있습니다.

스레드 간의 Producer/consumer¶

일반(non-asyncio) 스레드가 다른 스레드에서 실행 중인 asyncio 이벤트 루프에 작업을 전달해야 하는 경우, 단일 이벤트 루프 내에서만 안전한 asyncio.Queue 대신 스레드 안전 프리미티브인 queue.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()

Producer는 메인 스레드에서 실행되고 consumer는 자체 스레드의 이벤트 루프 내에서 실행되지만, 둘은 queue.Queue 를 통해 안전하게 통신합니다. 큐가 비어 있으면 consumer는 잠시 대기한 후 다시 시도합니다. producer가 완료되면 shutdown() 을 호출하며, 이는 이후의 get_nowait() 호출이 queue.ShutDown 을 발생시켜 consumer가 깔끔하게 종료될 수 있도록 합니다.