Posted inプログラミング

Python AsyncIOを極める：イベントループ、タスク、セマフォで構築する高パフォーマンス並行処理

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

Pythonにおける並行処理：主要アプローチの比較

Webスクレイパー、データパイプライン、マイクロサービスの処理速度が低下する最大の原因は、ほぼ常にネットワークI/O待ちにあります。アプリケーションがHTTPリクエストを送信したりPostgreSQLにクエリを発行したりすると、OSスレッドはネットワーク経由でパケットが返ってくるまでの50〜300ミリ秒間、アイドル状態のまま待機することになります。

Pythonには、この待機時間を解消するための3つの異なるパラダイムが用意されています：

  • マルチスレッド（threadingモジュール）： メモリを共有するOSスレッドを使用し、ブロッキングコールが発生した際に制御を譲渡します。しかし、CPythonのグローバルインタプリタロック（GIL）により、真の並列処理はI/O操作に限定されます。また、スレッド数が数百を超えると、コンテキストスイッチによるOSオーバーヘッドが顕著になります。
  • マルチプロセス（multiprocessingモジュール）： 独立したPythonプロセスを起動し、それぞれが独自のメモリ空間とGILを保持します。画像圧縮や行列演算のようなCPUバウンドな処理には最適です。一方で、プロセス間通信（IPC）のコストが高く、ワーカーごとに30〜50MBのベースラインメモリを消費するため、数万規模のネットワークソケットを同時に開く用途には適していません。
  • 協調的非同期I/O（asyncio）： 単一スレッド内でイベントループを使用して実行されます。コルーチンは、I/O待ちが発生するたびにawaitを使って明示的にループへ制御を戻します。

メリット・デメリット：スレッド vs プロセス vs AsyncIO

アーキテクチャの選定を誤ると、再現の難しいバグ、スレッドのスラッシング、不要なインフラ費用の増加につながります。本番環境における各アプローチの挙動は以下のとおりです：

1. マルチスレッド

  • メリット： 既存の同期ライブラリとスムーズに連携可能。共有メモリを使用するためシリアライズのオーバーヘッドがありません。
  • デメリット： レースコンディション対策として入念なミューテックスロックが必要。LinuxではOSスレッドごとにデフォルトで8MBのスタック領域が割り当てられ、同時実行スレッドが500を超えるとCPUキャッシュのミスヒットが急増しスループットが著しく低下します。

2. マルチプロセス

  • メリット： マルチCPUコアにまたがってGILを完全にバイパス可能。プロセスが分離されているため、1つのワーカーで未処理のクラッシュが発生してもシステム全体へ波及しません。
  • デメリット： インタプリタの状態が複製されるためメモリフットプリントが肥大化。プロセス境界を越えて大きなペイロードをpickle化する処理により、無視できないシリアライズ遅延が発生します。

3. Python AsyncIO

  • メリット： 極めて軽量。1つのコルーチンが消費するメモリは1KB未満であり、わずか2GB RAMのスペックでも20,000以上のオープンコネクションを処理できます。また、決定論的な実行順序により、従来のスレッドスケジューリングに起因するレースコンディションを回避できます。
  • デメリット： time.sleep()やrequests.get()など、awaitされていないブロッキングコールが1つでもあると、イベントループ全体が停止します。また、階層化されたタスクチェーンのデバッグには、コルーチンのスケジューリングに対する深い理解が求められます。

推奨されるアーキテクチャと環境構築

APIゲートウェイ、スクレイパー、イベントコンシューマーなどの用途では、モダンPython（3.11以降）の協調的マルチタスクが、RAM 1GBあたりで最も高いスループットを実現します。Python 3.11では内部ループの起動速度が約10〜15%向上したほか、バックグラウンドタスクのリークや孤立を防ぐ例外安全なコンテキストマネージャであるasyncio.TaskGroupが導入されました。

まずは依存関係をクリーンに保つため、サンプルコードを試す前に仮想環境を構築し、非同期HTTPクライアントをインストールしましょう：

# クリーンな仮想環境を作成して有効化
python3 -m venv .venv
source .venv/bin/activate

# 高パフォーマンスなHTTPクライアントをインストール
pip install aiohttp

ステップバイステップ実装ガイド

課題：制御されていない並行処理によるアップストリームサービスのクラッシュ

ヘルスチェックのために500基のマイクロサービスへpingを送信する状況を想像してみてください。順次実行（同期処理）では数分かかってしまいます。一方で、単純に500個の呼び出しをasyncio.gather()でラップすると、500本のTCPハンドシェイクが一斉に走りネットワークインターフェースがパンクします。その結果、ソケットの枯渇、接続切断、HTTP 429（レート制限）エラーが多発することになります。

根本原因：ファイルディスクリプタ上限とループの飽和

オープンされたすべてのソケットはファイルディスクリプタを1つ消費します。Linuxにおける標準的なプロセスでは、オープンディスクリプタの上限値はデフォルトで1,024（ulimit -n）に設定されています。制限を設けずにコルーチンを走らせるとすべてのソケットを同時に開こうとするため、OSの制限値に達し、相手サーバーの防衛システムをトリガーしてしまいます。そこで、同時接続数を制限する仕組み（ゲートキーパー）が必要になります。

実践的な解決策：イベントループ、タスク、セマフォの組み合わせ

耐障害性の高い非同期パイプラインは、次の3つのプリミティブで構成されます：

  1. イベントループ： 内部でepollやkqueueを使用し、ソケットにデータが到着したタイミングでコルーチンを復帰させるマルチプレクサ。
  2. タスク： 並行実行のためにコルーチンをイベントループ上にスケジューリングするラッパー。
  3. セマフォ： 共有リソースへ同時にアクセスできるコルーチンの数を制限する非同期カウンター。

以下は、制御された並行処理を実装した、本番環境でも使える完全なスクリプトです：

import asyncio
import time
import aiohttp

# 同時外部HTTP接続数の上限
CONCURRENCY_LIMIT = 5

async def fetch_url(session: aiohttp.ClientSession, url: str, semaphore: asyncio.Semaphore) -> dict:
    """並行処理数の上限を遵守しながら単一エンドポイントを取得する"""
    async with semaphore:
        # 最大でCONCURRENCY_LIMIT個のコルーチンのみがこのブロックを同時に実行可能
        start_time = time.monotonic()
        try:
            timeout = aiohttp.ClientTimeout(total=5)
            async with session.get(url, timeout=timeout) as response:
                data = await response.text()
                elapsed = time.monotonic() - start_time
                return {"url": url, "status": response.status, "duration": round(elapsed, 2), "size": len(data)}
        except Exception as exc:
            return {"url": url, "error": str(exc)}

async def main():
    urls = [
        f"https://httpbin.org/delay/{i % 2}"
        for i in range(15)
    ]

    semaphore = asyncio.Semaphore(CONCURRENCY_LIMIT)

    # プールされたTCPコネクタを使用
    connector = aiohttp.TCPConnector(limit=50)
    async with aiohttp.ClientSession(connector=connector) as session:
        # TaskGroupはタスクが未処理の例外を投げた場合に安全なキャンセルを保証する
        async with asyncio.TaskGroup() as tg:
            tasks = [
                tg.create_task(fetch_url(session, url, semaphore))
                for url in urls
            ]

        # TaskGroupを抜ける時点ですべてのコルーチンが完了している
        results = [task.result() for task in tasks]

    successful = sum(1 for r in results if "status" in r and r["status"] == 200)
    print(f"{len(results)} 件のリクエスト処理が完了しました。成功数: {successful}")
    for item in results[:3]:
        print(f" サンプル: {item}")

if __name__ == "__main__":
    # asyncio.runがループのライフサイクル、シグナルハンドラ、スレッドの後処理を一括管理
    asyncio.run(main())

AsyncIO内でのブロッキングコードの扱い方

HMACトークンの計算、巨大なJSONペイロードのパース、画像処理といったCPUバウンドな処理は、単一スレッドのイベントループをブロックしてしまいます。その計算処理が実行されている間、ネットワークパケットの受信や送信は一切行われなくなります。

レイテンシの急増を防ぐには、asyncio.to_thread()を使用して同期演算処理をスレッドプールにオフロードします：

import hashlib
import asyncio

def heavy_hashing(payload: str) -> str:
    """同期的なCPU負荷の高い計算処理"""
    result = payload
    for _ in range(500_000):
        result = hashlib.sha256(result.encode()).hexdigest()
    return result

async def process_item(item: str):
    # イベントループをブロックすることなく、PythonデフォルトのThreadPoolExecutorでheavy_hashingを実行
    digest = await asyncio.to_thread(heavy_hashing, item)
    print(f"ハッシュ値の計算完了: {digest[:10]}...")

ネットワーク操作にはasyncio.Semaphoreを適用し、負荷の高い計算処理にはasyncio.to_thread()を組み合わせることで、高トラフィック下でも安定した低レイテンシと予測可能なリソース消費を実現できます。

Share:
Tags: