Python イベントループ入門

この記事は Python の非同期処理を初めて学ぶ開発者向けです。次の点を扱います。

• Event Loop とは何か
• Coroutine、Task、Future とどう関係するか
• await でイベントループは何をするか
• ネットワーク要求の完了後、どの Task を再開するかをどう判断するか
• 連続する二つの await が逐次実行になる理由
• create_task() が並行性を生む理由
• Ready Queue、Timer Queue、I/O Selector の役割
• イベントループをブロックするコード

Python 3.14.6 の公式ドキュメントでは、Event Loop は各 asyncio アプリケーションの中核であり、非同期 Task と Callback の実行、ネットワーク I/O、サブプロセスなどを管理します。通常のアプリケーションでは、手動操作より asyncio.run() のような高水準 API を優先します。(Python documentation)

1. Event Loop とは何か

一言で言えば、次のとおりです。

Event Loop は継続的に動作するスケジューラであり、今実行する Task と、待機を終えて再開できる Task を決定します。

ビジネスロジックをすべて代行するものでも、同期ブロッキングコードを自動的に非同期化するものでもありません。

Task A: HTTP 応答を待つ
Task B: データベース問い合わせを待つ
Task C: データを処理する

イベントループは、A が I/O 待ちになれば A を中断して B や C を進め、HTTP 応答が届けば A を再開します。asyncio は 協調的スケジューリング を使います。一つのイベントループスレッドでは同時に一つの Task だけが実行され、現在の Task が待機のために await で中断したときに他の仕事が実行されます。(Python documentation)

2. Event Loop はスレッドプールではない

イベントループを「多数のバックグラウンドスレッド」と考えるのは正しくありません。典型的には、一つの Event Loop が一つの OS スレッドで、一度に一つの Task を実行します。

高い同時実行性は大量のスレッド作成ではなく、I/O 待ちの Task を中断し、その間に同じスレッドで別の Task を動かすことで得られます。一人の作業者が、外部承認待ちの書類を置いて別の書類を処理するようなものです。(Python documentation)

3. 五つの中心的な役割

async def 関数
      ↓ 呼び出し
Coroutine object
      ↓ create_task()
Task
      ↓ await に到達
Future / Future-like object を待つ
      ↓ Future 完了
Task を再開

Python は Coroutine、Task、Future を主要な awaitable と位置付けています。Task はイベントループ上でコルーチンを動かし、Future は非同期操作の将来の結果を表す低レベル awaitable です。(Python documentation)

4. Coroutine Function と Coroutine Object

async def fetch_data() -> str:
    return "done"

fetch_data はコルーチン関数です。coroutine = fetch_data() を実行するとコルーチンオブジェクトが得られます。重要なのは、コルーチン関数を呼ぶだけではイベントループに登録されず、実行も始まらないことです。

result = await fetch_data()

task = asyncio.create_task(fetch_data())
result = await task

このどちらかのように待機または Task 化します。そうしない場合、通常は RuntimeWarning: coroutine was never awaited が出ます。(Python documentation)

5. Task と Future

Task は、イベントループが管理するコルーチンの実行インスタンスです。

task = asyncio.create_task(fetch_data())

fetch_data()
→ Coroutine object を作る
→ create_task() が Task で包む
→ Event Loop が Task を実行する

Task はコルーチンを進め、await で中断し、待機対象が完了すると再開し、最終的な値、例外、キャンセル状態を保持します。Future-like なので、await、キャンセル、結果や例外の取得ができます。(Python documentation)

Future は「将来の結果の箱」です。通常は自ら業務コードを実行せず、結果、例外、またはキャンセル状態を保持します。

future = loop.create_future()
future.set_result("hello")
result = await future

Future は、低レベルの Callback 型非同期処理を高レベルの async / await コードへ接続するために主に使われます。通常のアプリケーションコードが自分で作る必要はありません。(Python documentation)

6. asyncio.run()

独立した Python プログラムの入口は、通常次のようになります。

import asyncio


async def main() -> None:
    print("start")
    await asyncio.sleep(1)
    print("end")


asyncio.run(main())

asyncio.run() はイベントループを作成し、main() を最上位の非同期 Task として動かし、完了までループを実行します。その後、非同期ジェネレーターと既定 Executor を終了し、ループを閉じます。通常は一度だけ呼びます。同じスレッドですでにループが動いている場合は呼べません。FastAPI や Jupyter ではフレームワーク側がループを開始済みなので、業務コードでは asyncio.run(...) ではなく直接 await を使います。(Python documentation)

7. Event Loop が内部で管理するもの

Event Loop
├── Ready Queue
├── Timer / Scheduled Queue
├── I/O Registration
└── Future Completion Callbacks

Ready Queue には今すぐ実行できる Callback や Task のステップが入ります。loop.call_soon(callback) は次のループ反復で Callback を実行するよう登録します。Future が完了したときに待機中 Task を再開する Callback もここに入ります。

Timer Queue は将来に実行可能になる Callback を保持します。loop.call_later(5, callback) や loop.call_at(deadline, callback) が代表例です。これらはキャンセル可能な TimerHandle を返します。asyncio.sleep() もタイマー機構を使い、常に現在の Task を中断します。(Python documentation) (Python documentation)

ネットワーク I/O では、ループは Python コードで繰り返し確認しません。ソケットが読み取り可能になったときに OS から通知を受けるよう登録します。Unix では selector を利用し、Linux の epoll、macOS/BSD の kqueue、その他の poll や select などが使われます。(Python documentation)

未完了 Future を await すると、Task は完了時に自分を再開する Callback を登録して中断します。future.add_done_callback(callback) の Callback は現在の呼び出しスタックで直ちに実行されず、loop.call_soon() を通じてスケジュールされます。(Python documentation)

8. await、再開、逐次実行

result = await operation()

では、現在の Task が operation() を呼び、awaitable を得てそれを進めます。即座に完了できれば結果を受け取り、待機が必要なら Task は中断します。ループはその間に他の仕事を実行し、待機対象の完了後に元の Task を再開します。await が必ず Task 切替を起こすわけではありませんが、await asyncio.sleep(...) は必ず現在の Task を中断します。(Python documentation)

Future が set_result、set_exception、または cancel により完了すると、完了 Callback が Ready Queue に入り、待機中 Task は後のループ反復で再開します。ネットワーク I/O では Selector または IOCP の通知が Future 完了につながります。asyncio.to_thread() ではワーカースレッドの完了がスレッドセーフにイベントループへ通知されます。(Python documentation)

result_a = await fetch_a()
result_b = await fetch_b()

これは逐次実行です。最初の行で main Task 全体が中断するため、fetch_b() はまだ呼ばれておらず、B を表す Task も存在しません。イベントループはソースの次の行を読んで新 Task に分割しません。

task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())

result_a = await task_a
result_b = await task_b

こちらは独立した二つの Task を先に作るため、各 Task が I/O を待つ間に交互に進めます。これは一つのイベントループスレッドで Python バイトコードを並列実行することではなく、並行実行です。(Python documentation)

9. Callback、Handle、ブロッキングコード

Callback は loop.call_soon() や loop.call_later() で時刻を指定して呼ばれる通常の callable です。Coroutine は async def で定義され、Task が進め、await で中断できるコードです。Handle は call_soon() で予約した Callback、TimerHandle は時刻を指定して予約した Callback を表し、実行前ならキャンセルできます。(Python documentation)

次のような処理をイベントループスレッドで実行するとループをブロックします。

time.sleep(5)
requests.get("https://example.com")
large_result = cpu_heavy_calculation()

await time.sleep(5) は解決になりません。まず同期的に五秒ブロックし、None を await しようとして失敗します。待機には await asyncio.sleep(5) を使います。避けられない同期 I/O は await asyncio.to_thread(blocking_function) でワーカースレッドへ移します。CPU 集約処理の真のマルチコア並列には、プロセスプールや別ワーカーを検討します。

10. スレッド、プラットフォーム、キャンセル

Event Loop は一つのスレッドに結び付きます。多くのアプリケーションは主スレッドに一つのループを置きます。別スレッドから通知する場合は loop.call_soon_threadsafe(callback)、別スレッドからコルーチンを投入する場合は asyncio.run_coroutine_threadsafe(coro, loop) を使います。(Python documentation)

Unix の標準ループは多くの場合 selector ベースです。Windows の ProactorEventLoop は I/O Completion Ports を利用します。通常のアプリケーションは高水準 API を使い、プラットフォーム固有のポーリング方式に依存しないようにします。(Python documentation)

task.cancel() はキャンセルを要求し、Task は適切な中断点で CancelledError を受け取ります。asyncio.timeout() や asyncio.wait_for() で外部呼び出しに期限を設定できます。キャンセルも協調的です。ブロッキングコードは制御を返すまでキャンセルに応答できません。

11. AI アプリケーションでの実例とベストプラクティス

独立したツール呼び出しは TaskGroup で並行に実行できます。

async with asyncio.TaskGroup() as group:
    weather_task = group.create_task(get_weather())
    search_task = group.create_task(search_documents())

一方、計画を作ってから実行する Agent ステップのように前の結果へ依存する処理は、直接 await で逐次実行します。LLM のストリーミングでは、非同期ジェネレーターがネットワークから次の token を待ち、待機中に他の要求を処理できます。同期ドキュメントパーサーは asyncio.to_thread() で隔離します。

実践上は、独立プログラムの入口に asyncio.run() を使い、関連する並行 Task には TaskGroup を優先し、同期ブロッキング I/O をループで実行せず、外部呼び出しにタイムアウトを設定し、下流サービスの容量に応じて並行数を制限します。開発時には asyncio の debug mode を有効にし、coroutine was never awaited や Task exception was never retrieved を確認します。(Python documentation)

まとめ

asyncio.run()
→ Event Loop を開始・管理する

Coroutine
→ 非同期コードを記述する

Task
→ Coroutine を Event Loop に渡してスケジュールする

Task が await に到達
→ Future を待つ
→ Task は中断する

Event Loop
→ 他の Ready Task を実行する
→ Timer または I/O を待つ

I/O / Timer / Thread が完了
→ Future が Done になる
→ 再開 Callback が Ready Queue に入る

次の反復
→ 元の Task が再開する

最も重要な点は、Event Loop が Task 間を任意に飛び回るのではないことです。Task が待機のために自発的に中断し、ループは Ready な別 Task を選びます。待機が終わると Future の Callback が元の Task を Ready Queue へ戻します。