PythonのmultiprocessingでCPU集約的処理を実行する

1. 概要

Pythonコードが大量の計算が必要な場合（例：画像処理、チェックサム計算、数値シミュレーションなど）、スレッドを使ってもほとんど高速化しないことに気づくでしょう。その原因はGIL（Global Interpreter Lock）です。この記事ではこの問題を簡潔に説明し、その後、作業を複数のプロセスに分割するためにmultiprocessingモジュールとProcessPoolExecutorを使う方法をガイドします。

読み終わると、CPU-boundなタスクを並列化する方法、それをテストする方法、よくあるエラーを避ける方法がわかります。

すべての例はPython 3.12以上を使用し、標準ライブラリのみが必要で、追加のインストールは不要です。

2. なぜthreadingが役に立たないのか

2.1. CPU集約的なタスク

純粋なPythonで書かれた素数を数える関数を使います。これは典型的なCPU集約的な作業です：

# cpu_task.py
import math

def count_primes(start: int, end: int) -> int:
    count = 0
    for n in range(max(start, 2), end):
        if all(n % d for d in range(2, math.isqrt(n) + 1)):
            count += 1
    return count

この関数は[start, end)内の各数を走査し、2から平方根までの数で割り切れるかをテストします。この関数はファイルを読み込まず、ネットワークを呼び出さず、単に計算するだけです。

2.2. タイミング測定：順序実行とスレッド

# bench_threads.py
import time
from concurrent.futures import ThreadPoolExecutor
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

def run_sequential() -> int:
    return sum(count_primes(a, b) for a, b in RANGES)

def run_threads() -> int:
    starts, ends = zip(*RANGES)
    with ThreadPoolExecutor(max_workers=8) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    for fn in (run_sequential, run_threads):
        t0 = time.perf_counter()
        result = fn()
        print(f"{fn.__name__}: {result} primes, {time.perf_counter() - t0:.2f}s")

RANGESは[0, 1_000_000)を8つの部分に分割します。run_threadsは各部分を1つのスレッドに割り当てます。8コアマシンでの結果（お使いのマシンでは異なる可能性があります）：

run_sequential: 78498 primes, 6.41s
run_threads: 78498 primes, 6.58s

スレッドを使用した版はさらに若干遅いです。標準的なCPythonでは、GILを保持する必要があるため、同時に複数のスレッドがPythonバイトコードを実行することはできません。8つのスレッドは単一のコアを交互に使用するだけであり、さらにスレッド間で切り替えるためのオーバーヘッドが追加されます。

注記： Python 3.13以降では、GILを削除したfree-threadedビルド（python3.13t）があります。このビルドはまだ実験段階であり、多くのC拡張ライブラリはまだサポートされていません。ほとんどの現在のプロジェクトでは、プロセスを使用することが最も確実な方法です。

3. multiprocessingの基本

各プロセスには独自のインタプリタとGILがあります。したがって、N個のプロセスはN個のコア上で実際に並列に実行できます。

3.1. Processを手動で作成する

# basic_process.py
from multiprocessing import Process, Queue
from cpu_task import count_primes

def worker(start: int, end: int, results: Queue) -> None:
    results.put(count_primes(start, end))

if __name__ == "__main__":
    results = Queue()
    ranges = [(0, 500_000), (500_000, 1_000_000)]
    processes = [Process(target=worker, args=(a, b, results)) for a, b in ranges]

    for p in processes:
        p.start()
    total = sum(results.get() for _ in processes)
    for p in processes:
        p.join()

    print("Total primes:", total)

各部分を説明します：

• Process(target=..., args=...)はプロセスの子を記述します。start()を呼ぶまでは実行されません。
• 子プロセスは親プロセスとメモリを共有しません。結果を返すには、通信チャネルを使用する必要があります。ここではQueueです。
• プログラムはjoin()の前にresults.get()を呼び出します。キューにまだ読み込まれていない大量のデータがある間にjoin()を実行すると、プログラムがデッドロックする可能性があります。

Total primes: 78498

このアプローチは柔軟ですが、冗長です。実際には、Processを手動で管理することはほとんどありません。

3.2. Poolを使用する

Poolは事前に作成されたワーカーのグループを作成し、タスクを自動的に分配します：

# pool_example.py
from multiprocessing import Pool
from cpu_task import count_primes

RANGES = [(i, i + 125_000) for i in range(0, 1_000_000, 125_000)]

if __name__ == "__main__":
    with Pool(processes=8) as pool:
        counts = pool.starmap(count_primes, RANGES)
    print(counts)
    print("Total:", sum(counts))

starmapはRANGES内の各タプルをstart, endパラメータに展開します。結果は入力の正確な順序で返されます：

[10746, 9895, 9622, 9480, 9637, 9554, 9454, 9610]
Total: 78498

processesを指定しない場合、Poolはos.cpu_count()ワーカーを使用します。withブロックは終了時にワーカーをクリーンアップします。

4. ProcessPoolExecutor：より現代的なAPI

concurrent.futures.ProcessPoolExecutorはThreadPoolExecutorと同じインターフェースを持っています。これにより、クラス名を変更するだけで、ほぼスレッドからプロセスに切り替えることができます：

# parallel.py
import os
from concurrent.futures import ProcessPoolExecutor
from cpu_task import count_primes

def split_range(start: int, end: int, parts: int) -> list[tuple[int, int]]:
    step = (end - start + parts - 1) // parts
    return [(s, min(s + step, end)) for s in range(start, end, step)]

def parallel_count_primes(start: int, end: int, workers: int | None = None) -> int:
    workers = workers or os.cpu_count() or 1
    chunks = split_range(start, end, workers)
    starts, ends = zip(*chunks)
    with ProcessPoolExecutor(max_workers=workers) as executor:
        return sum(executor.map(count_primes, starts, ends))

if __name__ == "__main__":
    import time
    t0 = time.perf_counter()
    print(parallel_count_primes(0, 1_000_000, workers=8))
    print(f"{time.perf_counter() - t0:.2f}s")

78498
1.27s

時間は約6.4秒から1.3秒に減少しました。速度は正確に8倍に増加しません。これには2つの理由があります。第1に、ワークロードがセグメント間で均等でありません：大きな数のセグメントはより長い検査が必要です。第2に、プロセスの起動とデータ転送にも時間がかかります。

完了したら各結果を即座に処理したい場合は、submitとas_completedを組み合わせます：

from concurrent.futures import ProcessPoolExecutor, as_completed
from cpu_task import count_primes

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        futures = {executor.submit(count_primes, a, a + 250_000): a
                   for a in range(0, 1_000_000, 250_000)}
        for future in as_completed(futures):
            print(f"Range starting {futures[future]}: {future.result()} primes")

ワーカーが例外をスローした場合、future.result()は親プロセスでその例外を再スローします。したがって、通常と同じようにtry/exceptで例外をキャッチできます。

5. 複数の小さなタスク：chunksizeとimap_unordered

ワーカーにタスクを送信するたびに、Pythonはパラメータをpickle化し、パイプ経由で送信し、結果をunpickle化する必要があります。100,000個の非常に小さなタスクがある場合、このコストは実際の計算よりも大きくなる可能性があります。chunksizeでタスクをバッチにグループ化することで対処できます：

# chunks.py
from multiprocessing import Pool

def is_prime(n: int) -> bool:
    if n < 2:
        return False
    return all(n % d for d in range(2, int(n ** 0.5) + 1))

if __name__ == "__main__":
    numbers = range(1_000_000)
    with Pool() as pool:
        total = sum(pool.imap_unordered(is_prime, numbers, chunksize=10_000))
    print(total)

• chunksize=10_000は、各送信に1つの数ではなく10,000個の数を含めます。プロセス間の通信回数が大幅に削減されます。
• imap_unorderedは遅延的に結果を返し、順序を保たないイテレータです。これは集約計算（例えば合算）を行うだけで、メモリ使用量を節約したい場合に適しています。
• ProcessPoolExecutor.mapもchunksizeパラメータを受け入れます。デフォルトは1です。小さなタスクが多い場合は、必ず調整してください。

6. 各ワーカーごとにリソースを一度初期化する

各タスクがビルドに時間がかかるルックアップテーブルを必要とする場合を想像してください。関数の呼び出しのたびにテーブルを再度作成すると、非常に無駄になります。initializerパラメータを使用すると、ワーカーの起動時に正確に1回関数を実行できます：

# initializer_example.py
from concurrent.futures import ProcessPoolExecutor

_price_table: dict[str, float] = {}

def load_prices(currency: str) -> None:
    global _price_table
    rate = 25_000 if currency == "VND" else 1
    _price_table = {f"SKU-{i}": i * 1.5 * rate for i in range(100_000)}

def order_total(skus: list[str]) -> float:
    return sum(_price_table[s] for s in skus)

if __name__ == "__main__":
    orders = [["SKU-1", "SKU-2"], ["SKU-10", "SKU-20", "SKU-30"]]
    with ProcessPoolExecutor(max_workers=2, initializer=load_prices,
                             initargs=("VND",)) as executor:
        print(list(executor.map(order_total, orders)))

[112500.0, 2250000.0]

変数_price_tableは各ワーカー内でのグローバル変数です。親プロセスはそれらの変更を見ることはできません。これは、各ワーカーがプライベートなリソースを持つ安全な方法であり、タスクごとにルックアップテーブルを送信する必要がありません。

7. 並列コードのテストを書く

通常のコードと同じようにpytestを使用してプロセスを使用するコードをテストできます。条件は、ワーカー関数がモジュールレベルで定義されていることです：

# test_parallel.py
from cpu_task import count_primes
from parallel import parallel_count_primes, split_range

def test_count_primes_known_value():
    assert count_primes(0, 100) == 25

def test_split_range_covers_whole_interval():
    chunks = split_range(0, 10, 3)
    assert chunks == [(0, 4), (4, 8), (8, 10)]

def test_parallel_matches_sequential():
    expected = count_primes(0, 50_000)
    assert parallel_count_primes(0, 50_000, workers=4) == expected

pip install pytest
pytest -q test_parallel.py

...                                                    [100%]
3 passed in 0.62s

最も重要なテストはtest_parallel_matches_sequentialです。セグメント分割がどの要素も漏らしたり、重複してカウントしたりしないことを保証します。これは独自のsplit_rangeを書くときに非常に一般的なエラーです。

8. よくあるエラーと使用しないべき場合

if __name__ == "__main__":を忘れる。 start methodがspawn（Windows と macOS のデフォルト）またはforkserver（Python 3.14以降のLinuxのデフォルト）の場合、子プロセスはメインモジュールを再度インポートします。プールを作成するコードがモジュールレベルにある場合、各子プロセスが新しいプールを作成します。この場合、RuntimeErrorが発生するか、プロセスが継続的に生成されます。

lambda または入れ子関数を渡す。 ワーカーに送信されるすべてはpickle化可能である必要があります：

from concurrent.futures import ProcessPoolExecutor

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(lambda x: x * x, range(5))))

_pickle.PicklingError: Can't pickle <function <lambda> at 0x...>: attribute lookup <lambda> on __main__ failed

モジュールレベルでdefを使って関数を定義してください。いくつかのパラメータを固定する必要がある場合は、functools.partialを使用します。

大量のデータを渡す。 各タスクに渡される数百万要素のリストは何度もpickle化されます。コピー時間は計算自体を上回る可能性があります。いくつかの回避策があります：インデックスまたはファイルパスのみを送信する、initializerを使用する、または配列形式のデータでmultiprocessing.shared_memoryを使用する。

forkの動作に依存する。 Linux でfork（子プロセスがグローバル変数を継承する）のおかげで機能するコードは、macOS、Windows、または Python 3.14 で失敗する可能性があります。一貫した動作を実現するには、start methodを明示的に選択します：

import multiprocessing as mp

if __name__ == "__main__":
    mp.set_start_method("spawn")

注記： HTTP呼び出しやデータベースクエリなどのI/O-boundなタスクにはmultiprocessingを使用しないでください。I/Oを待機する際、スレッドはGILを解放するため、ThreadPoolExecutorまたはasyncioがはるかに軽量です。また、各タスクが数マイクロ秒しかかからない場合（バッチ処理なし）、または計算が既にGILを解放するCライブラリ（NumPy、大規模データのハッシング関数など）にある場合はmultiprocessingを使用しないでください。これらの場合、スレッドまたはベクトル化が通常は十分です。

ワーカーを作成しすぎる。 CPUコア数よりも多くのワーカーを設定しても、CPU-boundコードは高速化しません。RAMを消費するだけです。各プロセスは完全なインタプリタだからです。os.cpu_count()から始めるのが理にかなっています。

9. 結論

GILにより、スレッドはPythonの純粋なCPU集約的コードを高速化することはできません。各プロセスは独自のGILを持つため、作業を複数のプロセスに分割することは、すべてのCPUコアを活用するシンプルな方法です。ほとんどの場合、ProcessPoolExecutorが最も簡潔な選択肢です。imap_unorderedまたはstarmapが必要な場合はPoolが便利です。手動Processは特殊な状況用です。

multiprocessingを使用する場合は、以下の点に注意してください：

• 常にif __name__ == "__main__":を使用してください。
• Pickle化可能な関数とデータのみを渡します。
• 小さなタスクが多い場合はchunksizeを増やします。
• 重いリソースにはinitializerを使用します。
• セグメント分割が正しいことを確認するために、並列結果と順序実行結果を比較するテストを書きます。