キュー（queue.Queue）とスレッドプールの基礎 — 投入順と完了順はなぜ一致しないのか

複数のサイトに対してSSH接続を並列で試みるような処理を書いていると、「同時に実行する数を絞りたい」「結果は終わった順に受け取りたい」という要求に必ず突き当たる。Pythonでこれを実現する土台になっているのが、標準ライブラリの queue モジュールと、それを内部で使っている concurrent.futures.ThreadPoolExecutor だ。今回はこの2つの関係と、「投入した順番」と「処理が終わる順番」が一致するとは限らないという、見落とされがちな性質を整理する。

queue.Queue とは何か

補足: queue.Queue は、複数のスレッドから安全に「モノを渡す・受け取る」ためのFIFO（先入れ先出し）データ構造だ。リストに append/pop するのと発想は同じだが、複数スレッドが同時に触っても壊れないようロック処理が内部に組み込まれている点が違う。

queue.Queue の基本操作は put()（キューに1件追加する）と get()（キューから1件取り出す）の2つだけだ。get() はキューが空の間、呼び出し元スレッドをブロックして待たせる（引数でタイムアウトも指定できる）。この「空なら待つ」という挙動があるおかげで、「作業を作る側（producer）」と「作業をこなす側（consumer）」を別スレッドに分けて動かす、いわゆる producer-consumer パターンが素直に書ける。producer は思いついた順に put() するだけでよく、consumer 側の実行速度を気にする必要がない。

queue.Queue にはこの他に、maxsize（キューの上限件数を決めて producer の暴走を防ぐ）や task_done()/join()（「投入した作業が全部処理し終わるまで待つ」ための同期用メソッド）がある。標準ライブラリにはこれらを持たない軽量版として queue.SimpleQueue も用意されており、後述する ThreadPoolExecutor の内部実装が実際に使っているのはこちらだ。

ThreadPoolExecutor の内部にもキューがある

concurrent.futures.ThreadPoolExecutor は「あらかじめ決めた本数のワーカースレッドを立てておき、そこに仕事を投げ込む」ための仕組みだが、その「投げ込む」部分の実体は、まさに上で説明したキューだ。CPython の実装（concurrent/futures/thread.py）を見ると、ThreadPoolExecutor.__init__ の中で

self._work_queue = queue.SimpleQueue()

という1行があり、executor.submit(fn, *args) を呼ぶたびに、実行したい関数と引数のペアがこの _work_queue に put() される。一方、起動済みの各ワーカースレッドは無限ループの中でこのキューから get() し続けており、何か取り出せたらそれを実行し、終わればまた次を get() する、を繰り返している。つまり ThreadPoolExecutor は「queue.Queue（正確には軽量版の queue.SimpleQueue）に対する producer-consumer パターンを、あらかじめ用意しておいたワーカースレッド群でラップしたもの」と説明できる。maxsize や task_done()/join() を使わない用途なら queue.Queue の代わりに SimpleQueue で十分、という判断がここに表れている。

実例: サイト横断のプラグイン更新チェック

自社の保守ツールでも、複数のWordPressサイトへ並列でSSH接続してプラグインの更新有無を調べる処理（site_manager_web.py::aggregate_pending_updates）で ThreadPoolExecutor を使っている。要点だけ抜き出すと、次のような形になる。

with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex:
    future_to_site = {
        ex.submit(_fetch_pending_plugins_for_site, s): s for s in all_sites
    }
    for fut in concurrent.futures.as_completed(future_to_site):
        r = fut.result()
        # r を集計...

max_workers は API のリクエストボディで指定でき、既定値8・上限16でクランプしている。サイトの台数がこれより多い場合、ex.submit() で作られたタスクは前述の内部キューにいったん全部積まれ、空いているワーカースレッドが順番に取り出して1サイトずつSSH接続・WP-CLI実行を行う。ここまでは投入順（＝サイトを読み込んだ順）にキューへ並ぶので、「どのサイトが何番目に着手されるか」はおおむねFIFOで決まる。

投入順と完了順は別物

ところが、着手される順番と、処理が終わる順番は話が別だ。SSHのレスポンス時間はサイトごとにサーバーの応答速度・ネットワーク経路・プラグイン数によってばらつくため、後から着手したサイトの方が先に終わることは普通に起こる。上のコードが concurrent.futures.as_completed(future_to_site) を使っているのはまさにこの点を踏まえた設計で、「終わった Future から順に」結果を回収していく。これにより、遅い1サイトの応答を待つ間に、先に終わった他サイトの結果を（このAPIの用途では最終的にまとめて返してはいるが）逐次拾い上げていくことができる。

もし代わりに ThreadPoolExecutor.map() を使っていたらどうなるか。map() は「投入した順序と同じ順序で結果を返す」ことを保証する便利なAPIだが、その保証を満たすために、内部では「先頭の要素の結果が来るまで、後続の要素がどれだけ早く終わっていても呼び出し元に結果を渡さない」という制約を伴う。1番目のサイトのSSH接続がたまたま極端に遅い場合、2番目以降がすでに終わっていても map() の反復処理はそこで足止めされる。aggregate_pending_updates のように「全サイトの応答を待って一括で結果を返す」用途では最終的な待ち時間に大差はないが、「終わったものから逐次画面に反映したい」ような用途では map() の順序保証がかえって不利に働く。submit() + as_completed() の組み合わせは、こうした場面のために「投入順は保つが、結果の受け取りは完了順でよい」という選択を明示的に行うためのAPIだと言える。

順序が重要な場面との使い分け

逆に、結果を必ず入力と同じ順序で扱いたい場面もある。例えば複数ページの内容を並列で取得して、1つの文書として順番通りに結合したいようなケースでは、map() の順序保証がそのまま役に立つ。要は「並列に実行すること」と「結果を受け取る順序」は独立した2つの設計判断であり、ThreadPoolExecutor はどちらの組み合わせも submit()+as_completed() か map() かを選ぶだけで表現できるようになっている。内部のキュー自体はどちらの場合も同じFIFOの queue.SimpleQueue であり、「順序を守るかどうか」は呼び出し側APIの選択で決まる、という点を押さえておくと、想定外の挙動に出会ったときの切り分けがしやすくなる。

まとめ

queue.Queue（および軽量版の queue.SimpleQueue）は、複数スレッド間で安全に作業を受け渡すためのFIFOデータ構造であり、ThreadPoolExecutor はこの仕組みをワーカースレッド群でラップした形で成り立っている。タスクの投入自体はキューによっておおむね順序どおりに進む一方、各タスクの処理時間はスレッドごとにばらつくため、完了する順序は投入順と一致するとは限らない。concurrent.futures.as_completed() はこの前提に立って「終わったものから順に」結果を受け取るためのAPIであり、map() の入力順保証とは異なる設計判断であることを理解しておくと、並列処理のコードを読むときにも書くときにも見通しが良くなる。