【ITニュース解説】Queues and Thread Pools — Why Submission Order and Completion Order Aren't the Same

システムエンジニアを目指す上で、複数の処理を効率良く、同時に実行する方法を学ぶことは非常に重要だ。その中でも、Pythonの標準ライブラリにあるqueueモジュールと、それを利用したconcurrent.futures.ThreadPoolExecutorは、並行処理を理解する上で欠かせない要素となる。特に、処理を投入した順番と、それが完了する順番が必ずしも同じではない、という点は、並行処理を扱う上で理解しておくべき重要な概念である。

まず、queue.Queueとは何かについて説明する。これは、複数の処理（スレッド）間で安全にデータをやり取りするための「箱」のようなものだと考えると良い。この箱は「先入れ先出し（FIFO: First-In, First-Out）」というルールで動く。つまり、最初に入れたものが最初に、次に入れたものが次に、という具合に取り出される。リストにデータを追加したり取り出したりするのと概念的には似ているが、queue.Queueの最大の特徴は、複数のスレッドが同時に触っても問題が起きないように、内部でロック（鍵をかけるような仕組み）が施されている点だ。データを箱に入れる操作をput()、箱からデータを取り出す操作をget()と呼ぶ。もし箱が空の状態でget()を呼ぶと、データが来るまでその処理は一時停止（ブロック）する。この「データが来るまで待つ」という挙動があるため、データを生み出す側（プロデューサー）と、それを使う側（コンシューマー）がそれぞれのペースで動いても、安全に連携できる「プロデューサー・コンシューマーパターン」を簡単に実装できる。queue.Queueには、箱の最大サイズを指定できるmaxsizeや、箱に入れたすべてのデータが処理されるまで待つためのtask_done()やjoin()といった便利な機能も備わっている。一方、queue.SimpleQueueは、これらの追加機能がない、より軽量なキューとして提供されている。

このqueue.SimpleQueueは、concurrent.futures.ThreadPoolExecutorの内部で重要な役割を担っている。ThreadPoolExecutorは、決められた数の「ワーカー（作業者）」スレッドを起動し、それらのワーカーに仕事を割り振る仕組みだ。この仕事の割り振りメカニズムこそが、まさにSimpleQueueなのである。ThreadPoolExecutorの内部実装を見ると、self._work_queue = queue.SimpleQueue()という行があり、これが作業キューとして使われていることがわかる。開発者がexecutor.submit(fn, *args)という形で実行したい関数と引数を渡すと、そのタスクのペアがこの_work_queueに追加される。一方、起動されたワーカー・スレッドたちは、この_work_queueに対して永遠にget()を呼び出し続けている。キューに新しいタスクが入ると、ワーカーはそれを取り出し、実行し、終わったらまた次のタスクを求めてget()を呼び出す、というループを繰り返す。つまり、ThreadPoolExecutorは、SimpleQueueの上に構築されたプロデューサー・コンシューマーパターンであり、固定数のワーカー・スレッドを効率的に管理するための仕組みなのである。

具体的な例として、複数のWebサイトにSSH接続して、それぞれでプラグインの更新状況を確認するような状況を考えてみよう。この種の並行処理にはThreadPoolExecutorが非常に有効だ。例えば、最大で8つの接続を同時に実行するように設定し、処理したいサイトの数が多い場合は、executor.submit()を使って次々とタスクをキューに投入していく。この段階では、タスクは投入された順に内部のキューに格納され、ワーカー・スレッドが空き次第、ほぼ投入順にタスクがキューから取り出され、SSH接続とプラグインチェックが実行される。ここまでは投入順が守られているように見える。

しかし、タスクがワーカー・スレッドによって実行され始めた後の挙動は全く異なる。SSH接続にかかる時間や、各Webサイトでのプラグインチェックにかかる時間は、サーバーの負荷、ネットワークの状態、インストールされているプラグインの数などによって様々だ。このため、後から実行を開始したタスクが、先に開始したタスクよりも早く完了することは日常茶飯事である。つまり、タスクの「投入順（キューに入れた順番）」と「完了順（実際に処理が終わった順番）」は、多くの場合、一致しないのだ。

この「投入順と完了順のずれ」という現実に対応するために、concurrent.futures.as_completed()という便利な関数が用意されている。これは、投入したタスクの中から、完了したものから順に結果を返してくれる。例えば、10個のサイトのチェックを依頼し、そのうちの1つが非常に遅延しているとする。as_completed()を使えば、その遅い1つを待つことなく、先に完了した9つのサイトの結果から順次処理を進めることができる。これは、遅延しているタスクが全体の処理をブロックしてしまうのを防ぎ、効率的なデータ処理を可能にする。

これに対して、ThreadPoolExecutor.map()という別の機能もある。map()は、結果を「投入順」に返すことを保証する。例えば、タスクA、B、Cをこの順で投入したとする。もしBとCがAよりも早く完了したとしても、map()はAの結果が返ってくるまで、BとCの結果を保持し続ける。これは、結果を投入順に処理する必要がある場合に便利だが、もし最初のタスクAが極端に遅い場合、残りのタスクが全て完了していても、Aが終わるまで全体の処理が停止してしまうというデメリットがある。

しかし、結果の順序が投入順と一致することがどうしても必要なケースも存在する。例えば、Webサイトの複数ページを並行して取得し、それらを元のページの順序で一つのドキュメントに結合するような場合だ。このようなケースでは、map()が提供する「投入順に結果を返す」という保証がまさに求められるものとなる。ThreadPoolExecutorは、submit()とas_completed()の組み合わせを選ぶか、map()を選ぶかによって、このような異なる要件に対応できる柔軟性を持っている。内部で使用されるキューはどちらの場合も同じFIFOのqueue.SimpleQueueだが、その上で提供されるAPIによって、結果の順序がどう扱われるかが変わってくるのである。

まとめると、queue.Queue（およびSimpleQueue）は、スレッド間で安全にタスクをやり取りするための先入れ先出しの仕組みであり、ThreadPoolExecutorはこの仕組みを基盤としてワーカー・スレッドのプールを管理する。タスクはほぼ投入順にキューから取り出され実行されるが、個々のタスクにかかる時間が異なるため、完了する順番は投入順とは一致しないことがほとんどだ。concurrent.futures.as_completed()は、この現実を受け入れ、完了したタスクから順に結果を返すことで、全体の処理を効率化する。一方、ThreadPoolExecutor.map()は、結果を投入順に返すことを保証する。並行処理のコードを設計する際には、これらの違いを理解し、どちらの順序で結果を処理すべきかという要件に基づいて適切なAPIを選択することが、意図しない挙動や性能の低下を防ぐ上で非常に重要となる。