到達点は、待ち時間と計算時間を分け、並行処理の方法を選ぶことです。前提は第09・13回です。TaskGroupの例にはPython 3.11以降を使います。
awaitは待ちを他の仕事へ譲る
async defで定義した関数を呼ぶとコルーチンができます。呼んだだけでは通常、処理は最後まで進みません。イベントループで実行し、awaitで他の仕事へ進む機会を渡します。独立した仕事を並行に進めるにはタスクを作ります。
import asyncio
async def read_minutes(value: int, gate: asyncio.Semaphore) -> int:
async with gate:
await asyncio.sleep(0.01)
return value
async def main():
gate = asyncio.Semaphore(2)
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(read_minutes(n, gate)) for n in [10, 20, 30]]
print(sum(task.result() for task in tasks))
asyncio.run(main())
main.pyの出力は60です。Semaphoreで同時に中へ入れる仕事を二つに制限します。例ではsleepで待ちを模擬しています。大量のタスクを一度に生成すると、同時実行を制限してもタスク自体のメモリは増えるため、大量入力ではキューと一定数のワーカーも検討します。
ブロッキング処理を混ぜない
イベントループの中でtime.sleepや長い同期I/Oを直接呼ぶと、他のコルーチンも進みにくくなります。非同期対応のAPIを使うか、同期I/Oならasyncio.to_threadなどを検討します。
CPythonの一般的なGILあり構成では、純粋なPythonのCPU計算をスレッドに分けても、複数コアの速度向上が得られるとは限りません。プロセス分割や、GILを解放するライブラリが候補です。free-threaded構成など実行環境によって条件が違うため、使う処理系を確認して測ります。
TaskGroupは子タスクの失敗時に他のタスクをキャンセルし、スコープ終了時に待ち合わせます。キャンセルを受けても後始末を行い、CancelledErrorを安易に飲み込まないようにします。タイムアウトとキャンセルは、外部サービスの更新を取り消せる保証ではありません。
練習と解答
練習：forの中で一件ずつawaitする方法と、create_taskする方法の違いを説明します。
解答：前者は各処理の終了を待ってから次へ進むため、そのループでは直列になります。後者は独立した仕事を並行に進められます。並行にしてよい順序か、相手の負荷制限を守れるかを先に判断します。
公式資料
asyncioのタスクとthreadingを参照できます。
Python全20回の目次 ｜ 前の回 ｜ 次の回