0.2秒のI/O待ちを20回、逐次で回すと5.34秒、ThreadPoolExecutorのスレッド20本なら0.27秒。ただ、同じプールにfib(32)を8個入れても逐次と同じ0.89秒のまま縮みません。この差が生まれる理由と、Python 3.14でmapに入ったbuffersize引数まで、実測で整理します。
スレッドで速くなるのはI/O待ちだけ
実測環境はPython 3.14.6、18コアのApple Silicon Macです。ThreadPoolExecutorは標準ライブラリのconcurrent.futures(公式ドキュメントの章題は “Launching parallel tasks”)に入っているので、インストールは不要です。
sleep(0.2)を20回、スレッド数を変えて測る
HTTPリクエストやDB問い合わせの代わりに、0.2秒眠るだけの関数で測ります。
import time
from concurrent.futures import ThreadPoolExecutor
def fake_io(i):
time.sleep(0.2) # I/O待ちの疑似
return i
t0 = time.perf_counter()
for i in range(20):
fake_io(i)
print(f"sequential: {time.perf_counter() - t0:.2f}s")
for w in (4, 8, 20):
t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=w) as ex:
list(ex.map(fake_io, range(20)))
print(f"max_workers={w}: {time.perf_counter() - t0:.2f}s")sequential: 5.34s
max_workers=4: 1.33s
max_workers=8: 0.78s
max_workers=20: 0.27s20タスクをスレッド20本で受ければ全員が同時に眠るので、理論値の0.2秒+プール起動のコストに収束します。逐次比で約20倍です。
fib(32)はスレッド8本でも0.89秒のまま
同じ書き方のまま、処理だけCPUを使うものに替えます。
def fib(n):
return n if n < 2 else fib(n - 1) + fib(n - 2)
t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=8) as ex:
list(ex.map(fib, [32] * 8))
print(f"threads: {time.perf_counter() - t0:.2f}s")sequential fib(32) x 8: 0.89s
ThreadPoolExecutor: 0.89s小数第2位まで同じでした。CPythonの通常ビルドにはGIL(Global Interpreter Lock)があり、Pythonバイトコードを実行できるスレッドは常に1本だけです。fib(32)には待ち時間がないので、スレッドを8本並べてもGILを順番に取り合うだけで、並列には走りません。
time.sleep()やソケットの読み書きの間、スレッドはGILを手放します。だからI/O待ちは重なり、計算は重ならない。冒頭の5.34秒と0.89秒の差はここから来ています。
どちらのバウンドか迷ったらプロファイラで測る
「この処理はI/OバウンドかCPUバウンドか」を推測で決めると、スレッド化しても1秒も縮まないコードを書くことになります。稼働中のプロセスに後付けできるサンプリングプロファイラをPython 3.15のprofiling.sampling—稼働中プロセスを実測するで扱いました。sleepやrecvのフレームが支配的ならスレッド化が効きます。
submitとmapの基本の使い方
submitはFutureを即座に返す
with ThreadPoolExecutor() as ex:
future = ex.submit(fake_io, 1) # ブロックしない
print(future.done()) # False
print(future.result()) # 完了まで待って 1submit()は関数を1個キューに積み、Futureオブジェクトを即座に返します。結果はfuture.result()で受け取り、未完了のうちに呼べばその行でブロックします。タスクごとに引数や関数を変えて投げられるのが利点で、as_completed()で完了順に拾う書き方(後述)もsubmit()が前提です。
mapは投入順に結果を返す
with ThreadPoolExecutor(max_workers=8) as ex:
results = ex.map(fake_io, range(20)) # イテレータが返る
print(list(results)) # [0, 1, 2, ..., 19]組み込みのmap()と同じ形で書けて、返るのはイテレータ。順序は完了順ではなく投入順です。3番目のタスクが先に終わっても、1番目の結果を取り出すまで3番目は出てきません。入力と出力の対応が保たれるので、リストの一括変換はmap()が素直です。
max_workersのデフォルトは3.13で変わった
import os
print(os.process_cpu_count()) # 18
with ThreadPoolExecutor() as ex:
print(ex._max_workers) # 22引数を省略したときの本数はmin(32, os.process_cpu_count() + 4)で、18コアの手元では22本でした。Python 3.12まではos.cpu_count()基準でしたが、Python 3.13からcgroupsなどでプロセスに割り当てられたコア数を見るos.process_cpu_count()に変わっています。公式ドキュメントはこの式の意図を “This default value preserves at least 5 workers for I/O bound tasks” と説明しています。1コアのコンテナでもI/O用に5本は確保する設計です。
終了待ちの道具はmap・as_completed・wait
違いは結果を受け取る順序に出ます。
| 道具 | 返る順序・形 | 使いどころ |
|---|---|---|
Executor.map() | 投入順のイテレータ | 入力と出力の対応を保ちたい |
as_completed() | 完了順のイテレータ | 終わったものから後続処理へ流したい |
wait() | (done, not_done)の2タプル | FIRST_EXCEPTIONなど条件付きで待ちたい |
as_completed()の完了順は実測で確認できます。0.3秒・0.2秒・0.1秒かかるタスクを0, 1, 2の順にsubmitすると、出てくる順は[2, 1, 0]でした。wait()はreturn_when=FIRST_EXCEPTIONを渡すと、どれかが例外を出した時点で制御が戻ります。
submit投げっぱなしは例外を握りつぶす
アンチパターン: result()を呼ばない
# NG: 投げっぱなし
def boom(i):
raise ValueError(f"boom {i}")
with ThreadPoolExecutor(max_workers=2) as ex:
ex.submit(boom, 1)
print("finished")finishedトレースバックはどこにも出ません。ワーカー内の例外はFutureの中に格納され、result()を呼ぶまで再送出されないためです。実測でもf.done()はTrueを返し、result()を呼んだ行で初めてValueErrorが上がりました。ログ上は正常終了に見えるので、バッチ処理でこれをやると欠損に気づけません。
# OK: 完了順に回して必ずresult()を呼ぶ
from concurrent.futures import as_completed
with ThreadPoolExecutor(max_workers=2) as ex:
futures = [ex.submit(boom, i) for i in range(3)]
for f in as_completed(futures):
try:
f.result()
except ValueError as e:
print(f"task failed: {e}")map()の場合は結果イテレータを回した時点で例外が再送出されるので、この事故は起きにくいです。submit()を使うなら、Futureを放置しない導線をセットで書いてください。
cancel_futuresが取り消すのは未開始のタスクだけ
途中で残りを打ち切る手段はshutdown()のcancel_futures引数です。スレッド2本のプールに0.5秒のタスクを6個積み、0.1秒後に打ち切ります。
ex = ThreadPoolExecutor(max_workers=2)
futs = [ex.submit(slow, i) for i in range(6)]
time.sleep(0.1)
ex.shutdown(wait=True, cancel_futures=True)
print([f.cancelled() for f in futs])[False, False, True, True, True, True]実行中だった2個は取り消されず、未開始の4個だけがキャンセルされます。公式ドキュメントの記述どおり “Any futures that are completed or running won’t be cancelled” です。走り出したスレッドを外から止めるAPIはありません。with文を抜けるときも暗黙にshutdown(wait=True)が走るので、submitしただけのつもりでも全タスクの完了を待ちます。
mapはiterableを先に全部消費する – 3.14のbuffersize
ex.map()にジェネレータを渡しても、遅延評価にはなりません。mapを呼んだ瞬間に全要素が消費され、全件がタスクとしてキューに積まれます。公式ドキュメントにも “The iterables are collected immediately rather than lazily” と明記されていますが、組み込みmap()の感覚でいると見落とします。
yieldログで即時消費を確認する
yieldのたびにログを出すジェネレータを渡すと、消費のタイミングが見えます。workは0.1秒かかる関数です。
def gen():
for i in range(10):
print(f"yield {i}")
yield i
with ThreadPoolExecutor(max_workers=2) as ex:
it = ex.map(work, gen())
print("(map returned)")
print("first:", next(it))yield 0
yield 1
（中略: yield 2〜8）
yield 9
(map returned)
first: 0最初の結果を取り出す前に10件すべてのyieldが走っています。ワーカーは2本なのに、です。入力10件なら実害はありませんが、数百万行のファイルやDBカーソルを渡すと全行ぶんのFutureがメモリに乗ります。itertools.count()のような無限イテレータに至っては、mapが永遠に返りません。手元でサブプロセスにtimeout=3秒を掛けて確認したところ、本当に返ってきませんでした。
buffersizeで投入数を絞る – Python 3.14で追加
Python 3.14でExecutor.map()にbuffersize引数が入りました。公式の説明は “unless a buffersize is specified to limit the number of submitted tasks whose results have not yet been yielded”。結果をまだ取り出していないタスク数を、この値で頭打ちにします。
with ThreadPoolExecutor(max_workers=2) as ex:
it = ex.map(work, gen(), buffersize=2)
print("(map returned)")
print("first:", next(it))yield 0
yield 1
(map returned)
yield 2
first: 0消費されたのは2件だけ。next()で1件取り出すと、空いた枠にyield 2が補充されます。バッファが埋まっている間、入力側の消費は止まったまま。無限イテレータも同じ理屈でストリーム処理できます。
import itertools
with ThreadPoolExecutor(max_workers=4) as ex:
it = ex.map(work, itertools.count(), buffersize=4)
print([next(it) for _ in range(5)]) # [0, 2, 4, 6, 8]3.13以前の回避策
3.13以前にこの引数はありません。同じことをしたければ、itertools.isliceで入力をチャンクに切って回すか、セマフォでsubmit数を自前で絞るしかありません。ストリーム処理が主目的なら、そもそもasyncioに寄せる判断もあります。buffersize=Nの1引数で済むのは3.14からです。
CPUバウンドはProcessPoolExecutorへ
fib(32)が0.89秒から0.22秒になる
from concurrent.futures import ProcessPoolExecutor
if __name__ == "__main__":
t0 = time.perf_counter()
with ProcessPoolExecutor(max_workers=8) as ex:
list(ex.map(fib, [32] * 8))
print(f"processes: {time.perf_counter() - t0:.2f}s")sequential fib(32) x 8: 0.89s
ThreadPoolExecutor: 0.89s
ProcessPoolExecutor: 0.22sプロセスを分ければGILもプロセスごとに分かれるので、0.89秒が0.22秒(約4倍)になりました。8プロセスで4倍に留まるのは、プロセス起動と引数・結果のpickle転送に固定費が掛かるためです。数ミリ秒で終わる細かいタスクを大量に投げると、この固定費が支配して逐次より遅くなります。
spawn前提で書く
手元のmacOSでは子プロセスの起動方式(multiprocessing.get_start_method())はspawnでした。子プロセスがモジュールを再importするため、実行部をif __name__ == "__main__":で囲わないと子が孫を起動し続けます。渡す関数と引数はpicklableである必要があり、ラムダやローカル関数はそのまま渡せません。Python 3.14からは、macOSとWindows以外のプラットフォームでもデフォルトがforkからforkserverに変わり、fork固有のデッドロックを踏みにくくなっています。
3.14世代の新しい選択肢
同じconcurrent.futuresに、3.14でInterpreterPoolExecutorが追加されました。ThreadPoolExecutorのサブクラスで、公式いわく “each worker has its own interpreter, and runs each task using that interpreter”。スレッドごとに独立したGILを持つ形です。仕組みと実測はPython 3.14のサブインタプリタ入門—GILを外さずCPU処理を並列化するで扱いました。GILそのものを外すビルドの実測はPython 3.14のfree-threadingでGILを外す—効果と使いどころにあります。どれを選んでもExecutorのインターフェースは共通なので、submit・map・as_completedの書き方は変わりません。
まとめ
- ThreadPoolExecutorが効くのはI/O待ちだけ。sleep(0.2)×20は5.34秒→0.27秒、fib(32)×8は0.89秒のまま
- submitしたFutureを放置すると例外は沈黙する。as_completed()で回してresult()を呼ぶ導線をセットにする
- map()はiterableを即時に全消費する。Python 3.14ならbuffersizeで未消化タスク数を絞れる
- CPUバウンドはProcessPoolExecutorで0.89秒→0.22秒。数ミリ秒のタスクはpickle転送の固定費で逆転する
