asyncio とフリースレッド Python
*******************************

asyncio uses an event loop as a scheduler to enable highly efficient
concurrency by switching between tasks to allow non-blocking I/O
operations. This results in better performance for I/O-bound use
cases. It also allows off-loading CPU-bound work to a thread or
process pool, but that is still limited by the *global interpreter
lock* in CPython.

一方、フリースレッド Python では GIL が無効になり、Python コードを複数
のスレッドで真に並列実行できます。 これにより、asyncio でも GIL の制限
を受けずに複数の CPU コアを活用できます。

Python 3.14 以降、asyncio はフリースレッド Python を正式にサポートして
おり、マルチスレッド環境でも安全に使える実装になっています。

1 つのコア上のイベントループでも多数の接続を並行に処理できますが、各接
続を処理する Python コードは依然として逐次実行されます。 リクエストご
とに相応の計算が必要になると、その処理がボトルネックとなり、1 つのコア
では追いつかなくなります。 asyncio とスレッドの組み合わせが特に役立つ
のはこのような場合です。 スレッドごとにイベントループを実行すると、異
なるリクエストの処理を複数の CPU コアで並列に実行できます。 asyncio ア
プリケーションからブロックするコードや CPU バウンドのコードを実行する
必要がある場合にも役立ちます。

参考:

  Kumar Aditya によるブログ記事 Scaling asyncio on Free-Threaded
  Python では、フリースレッド Python 上で asyncio を安全かつ効率的に動
  作させるための内部的な変更と、改善の効果を示すベンチマークを紹介して
  います。


スレッド安全性に関する注意点
============================

asyncio はフリースレッド Python 環境で安全に動作するように設計されてい
ますが、スレッドと組み合わせて使う際には、次の点に注意する必要がありま
す。

1. **イベントループ**: 各スレッドが専用のイベントループを持ち、スレッ
   ド間で共有しないようにしてください。 これにより、イベントループは他
   のスレッドから干渉されずに自身のタスクやコールバックを管理できます
   。

2. **タスクの管理**: あるスレッドで作成した Task や Future を、別のス
   レッドから await したり操作したりしないでください。

3. **スレッドセーフな API**: 複数のスレッドから asyncio を使う場合は、
   asyncio が提供するスレッドセーフな API を使うことが重要です。 例え
   ば、別のスレッドからイベントループにコルーチンを送るには
   "asyncio.run_coroutine_threadsafe()" を使います。 別のスレッドから
   コールバックを呼び出す必要がある場合は、
   "loop.call_soon_threadsafe()" で安全にスケジュールできます。

4. **同期**: asyncio が提供する同期プリミティブ ("asyncio.Lock" や
   "asyncio.Event" など) は、スレッドをまたいで使うようには設計されて
   いません。 スレッド間で同期する必要がある場合は、代わりに
   "threading" モジュールの同期プリミティブを使ってください。


asyncio とスレッドの併用
========================

asyncio はスレッドごとに 1 つのイベントループを実行できるため、フリー
スレッド 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()" で専用のイベントループを作成
し、その上でコルーチンを実行します。 スレッドは並行に実行され、フリー
スレッドビルドでは別々の CPU コアで並列に実行できます。


スレッドをまたぐ生産者と消費者
==============================

通常の (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()

The producer runs on the main thread while the consumer runs inside an
event loop on its own thread, yet they communicate safely through
"queue.Queue". When the queue is empty the consumer sleeps briefly and
tries again. When the producer is done it calls "shutdown()", which
causes subsequent "get_nowait()" calls to raise "queue.ShutDown" so
the consumer can exit cleanly.
