ThreadPoolExecutor(max_workers=4)なら、メモリ上にも4件前後しか仕事を抱えない。そう考えると、巨大なgeneratorをExecutor.map()へ渡したときの挙動は少し意外です。
max_workersが制限するのは、同時に実行するworker数です。入力iterableからどこまで先に仕事を投入するかは別の話で、Python 3.14ではその制御のためにbuffersizeが追加されました。
つまり、worker数を決めることと、未回収の仕事を何件まで先行させるかは、別々に設計する必要があります。
workerを4にしてもgeneratorは止まらない
まず、入力側がどこまで読まれるかだけを観察します。
実行環境はPython 3.14です。外部パッケージは使いません。
from concurrent.futures import ThreadPoolExecutor
import time
def source(n):
for i in range(n):
print(f"generate: {i}")
yield i
def work(x):
time.sleep(0.1)
return x * 2
with ThreadPoolExecutor(max_workers=4) as executor:
results = executor.map(work, source(20))
print("map returned")
for result in results:
print(f"consume: {result}")
time.sleep(0.2)
ここで見たいのは処理速度ではなく、consumeが始まる前にgenerateがどこまで進むかです。
Python 3.14の公式ドキュメントには、Executor.map()へ渡したiterableは、buffersizeを指定しない限り即座に収集されると明記されています。
concurrent.futures、Launching parallel tasks
そのため、max_workers=4だからといってgeneratorの生成も4件程度で止まるわけではありません。
20件程度なら問題になりませんが、入力が大きなbytesを生成する場合は話が変わります。generatorそのものがlazyでも、その先にいるmap()が先へ読み進めれば、期待したメモリ特性にはなりません。
generatorであることと、パイプライン全体にbackpressureがあることは別です。
max_workersはキューの深さを制限しない
ここを混同すると原因を追いにくくなります。
ThreadPoolExecutorのmax_workersは、同時に仕事を実行するthread数の上限です。
producer -> submitted tasks -> workers -> map iterator -> consumer
^
max_workers
一方、今回問題にしているのはproducerがどこまで先行できるかです。
workerが4つしかなくても、その手前に多数の仕事を投入できるなら、入力オブジェクトやFutureなどを保持する余地があります。
さらにExecutor.map()の返すiteratorは入力順に結果を返します。consumer側が遅ければ、「workerは処理できているが、結果をまだconsumerへ渡していない」という状態も考える必要があります。
そこでPython 3.14のbuffersizeが効きます。公式ドキュメントでは、まだyieldされていない投入済みタスク数を制限し、bufferが満杯になると、結果がyieldされるまで入力iterableのiterationを停止すると説明されています。
with ThreadPoolExecutor(max_workers=4) as executor:
results = executor.map(
work,
source(20),
buffersize=8,
)
for result in results:
print(f"consume: {result}")
time.sleep(0.2)
先ほどのコードにbuffersize=8を足しただけです。
これでproducerは20件すべてを一気に生成するのではなく、consumerによる回収に応じて先へ進むようになります。
ここでのポイントは、buffersize=8を「queueの最大サイズ8」と読み替えないことです。公式の定義は、submittedされ、まだ結果がyieldされていないタスク数に対する制限です。
また、ThreadPoolExecutorではchunksizeを変更してもこの問題の解決にはなりません。公式ドキュメント上、chunksizeはThreadPoolExecutorとInterpreterPoolExecutorでは効果がなく、ProcessPoolExecutorで長いiterableを分割するときに使われます。
メモリを使う入力で差を観察する
次は、各入力に大きめのbytesを持たせて、tracemallocでPythonが追跡するメモリの推移を表示します。
数値そのものは実行環境に依存するので、ここでは計測コードだけを置きます。
from concurrent.futures import ThreadPoolExecutor
import time
import tracemalloc
ITEMS = 200
BYTES_PER_ITEM = 512 * 1024
def source():
for i in range(ITEMS):
data = bytes(BYTES_PER_ITEM)
current, peak = tracemalloc.get_traced_memory()
print(
f"generate={i:3d} "
f"current={current / 1024 / 1024:6.1f}MiB "
f"peak={peak / 1024 / 1024:6.1f}MiB"
)
yield i, data
def work(item):
i, data = item
time.sleep(0.02)
return i, len(data)
def run(buffersize):
tracemalloc.start()
with ThreadPoolExecutor(max_workers=4) as executor:
results = executor.map(
work,
source(),
buffersize=buffersize,
)
for i, size in results:
time.sleep(0.05)
_, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
print(f"peak={peak / 1024 / 1024:.1f}MiB")
run(buffersize=8)
比較するときは最後をrun(buffersize=None)に変えます。Noneはデフォルトの挙動です。
このコードではworker側よりconsumer側を意図的に遅くしています。見たいのは「4並列が速いか」ではなく、consumerが追いつかない状況でproducerがどこまで先行するかだからです。
tracemallocが測るのはPythonによって割り当てられたメモリです。OSから見たプロセス全体のRSSと同じ数字になるわけではありません。ネイティブライブラリが確保する領域まで含めて評価したい場合は、別の計測手段も必要です。
それでも、この例ならgeneratorのgenerateログとメモリ推移を同時に見られるため、buffersizeによって入力側が止まる位置を確認しやすくなります。
worker数とbuffersizeは別々に決める
max_workersとbuffersizeを同じ基準で決める必要はありません。
max_workersで考えるのは、主に実行側です。I/O待ちの割合、接続先に許容される並行数、threadを増やしたときの負荷などを見ます。
buffersizeで考えるのは、パイプライン上に何件の仕事を先行して保持してよいかです。1件あたりの入力が大きい、入力生成自体に副作用がある、consumerが遅い、といった条件ではこちらが効いてきます。
たとえば1件の入力が大きな画像やファイル内容を保持する処理なら、workerが少ないから安全とは限りません。先行投入数が大きければ、それだけ入力を長く保持する可能性があります。
逆にbuffersizeを小さくしすぎると、producerやconsumerのタイミング次第ではworkerへ十分な仕事を供給できず、待ち時間が増える可能性があります。
そのため私は、次の二つを分けて考えるのが分かりやすいと思っています。
-
max_workers: 同時に何件を実行してよいか -
buffersize: 未回収の仕事を何件まで先行させてよいか
並列度とbackpressureは別のパラメータです。
なお、Python 3.13以前でもsubmit()する件数を自前で管理すれば同じ種類の制御はできます。3.14の変化は、Executor.map()の順序を保つインターフェースを使ったまま、その制御をbuffersizeとして指定できるようになった点です。
無限iterableや非常に大きな入力をmap()へ渡す処理では、Python 3.14だから自動的に安全になるわけではありません。buffersize=Noneなら従来どおり先読みされるため、入力1件の保持コストとconsumerの速度を測ったうえで、worker数とは独立してbufferの深さを決める必要があります。