Python講座 / スレッド・プロセス・asyncioを選ぶ

スレッド、プロセス、asyncioは複数の処理を進める異なる方法です。CPU計算か外部の待ち時間か、対応ライブラリは何かを見て選び、並行化で速くなると決めつけず実際の処理を測定します。

並行と同時実行の条件を整理する

通常のGIL付きCPythonでは、スレッドを増やすだけで任意の純PythonのCPU計算を並列化できません。GILを解放する拡張やfree-threadedビルドでは条件が変わるため、インタープリターのビルドとライブラリを合わせて考えます。

待機処理の制限と取得失敗は、HTTP APIとJSON応答・通信エラーを扱うで扱う規則とつながります。

同じ全体例で呼び出し方を比較する

concurrency.pyとして保存し、python concurrency.pyを実行します。ノートブックや対話環境でもプロセスが同じように動くとは考えず、.pyファイルを使います。

import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def square(value):
    return value * value

async def async_square(value):
    await asyncio.sleep(0.01)
    return square(value)

async def gather_values():
    return await asyncio.gather(*(async_square(x) for x in [1, 2, 3]))

def main():
    with ThreadPoolExecutor(max_workers=2) as pool:
        print(list(pool.map(square, [1, 2, 3])))
    with ProcessPoolExecutor(max_workers=2) as pool:
        print(list(pool.map(square, [1, 2, 3])))
    print(asyncio.run(gather_values()))

if __name__ == "__main__":
    main()

予想結果です。

[1, 4, 9]
[1, 4, 9]
[1, 4, 9]

起動の入口と協調的な待ちを守る

Windowsなどのプロセス起動ではメインモジュールが読み込まれます。mainガードは再帰的な起動を避けます。squareをモジュール直下で定義すると、子プロセスから参照できます。

async関数の中でも、長いCPU計算や同期通信を譲らず実行するとイベントループを止めます。asyncio.sleepは協調的な待ちの例で、実際のHTTP取得はしていません。既に動作しているイベントループ内へasyncio.runを入れ子にできません。

実際の負荷と失敗処理を設計する

小さな二乗の例はAPIの比較で、速度のベンチマークではありません。プロセスの起動が計算より高くつくこともあります。代表的なデータで環境、ワーカー数、メモリーも記録して測定します。

mapの結果は入力順ですが、完了順は別の場合があります。同時数を制限し、共有状態の同期、例外、取り消し、終了を設計します。無制限にタスクを作ったり、同期なしの共有リスト変更へ依存したりしないでください。API収集実習の通信と保存の段階を理解してから、並行リクエストの必要性を判断します。

GIL（Global Interpreter Lock）は、通常のGIL有効CPythonで複数スレッドによるPythonバイトコードの同時実行を制限するロックです。ブロッキングI/Oはスレッド、純粋なPythonのCPU処理はプロセス、非同期対応I/Oはasyncioを検討し、実際の負荷で測ります。free-threadedビルドやGILを解放する拡張には別の条件があります。二乗の例はAPIの練習で速度比較ではありません。タスク数・待ち時間を制限し、Future.resultの例外も確認します。下の最小例はfinallyの片付け後に呼び出し元がキャンセルを確認します。

全体を cancel.py に保存し、同じ練習フォルダーで python cancel.py を実行します。

import asyncio
async def worker():
    try:
        await asyncio.sleep(60)
    finally:
        print("worker cleanup")
async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(0)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancel confirmed")
if __name__ == "__main__":
    asyncio.run(main())

予想結果：

worker cleanup
cancel confirmed

次は投入したタスクの失敗をresultで確認する例です。max_workersは同時実行数で、待ち行列全体の上限ではありません。timeoutは待ち時間を制限するだけで実行中の処理を強制停止しません。大量の入力は小さなバッチに分け、asyncioの同時I/OにはSemaphoreなどで別の上限を設けます。

全体を future_failure.py に保存し、同じ練習フォルダーで python future_failure.py を実行します。

from concurrent.futures import ThreadPoolExecutor
def calculate(value):
    if value < 0:
        raise ValueError("negative input")
    return value * value
with ThreadPoolExecutor(max_workers=2) as pool:
    futures = [pool.submit(calculate, value) for value in [2, -1]]
    for future in futures:
        try:
            print(future.result(timeout=5))
        except ValueError as error:
            print("failed:", error)

予想結果：

4
failed: negative input

動作と適用条件の公式説明：Python: asyncio task cancellation · Python: concurrent.futures

関連する公式リファレンス

• Python: concurrent.futures