【Python】ThreadPoolExecutorとProcessPoolExecutorの比較

# はじめに

この記事では `concurrent.futures` の `ThreadPoolExecutor` と `ProcessPoolExecutor` を使用して並列で四則演算を行い、処理時間を比較します。

これらは以下の用途によって使う場面が分かれます。

`ThreadPoolExecutor`: I/O が多い処理向き
`ProcessPoolExecutor`: CPU を多く消費する処理向き

https://docs.python.org/ja/3.13/library/concurrent.futures.html

行う処理は I/O ではなく CPU を多く消費する処理のため、`ProcessPoolExecutor` のほうが高速になる想定です。

環境：

```
Python 3.10.12
```

# 比較

プロセスごとに CPU 使用率をモニタリングします。

```python
import concurrent.futures
import time
import threading
import psutil

def check_cpu():
    """論理コア数、物理コア数を確認"""
    print(f"Logical CPU cores: {psutil.cpu_count(logical=True)}")
    print(f"Physical CPU cores: {psutil.cpu_count(logical=False)}")

def compute(n):
    """疑似的な計算処理"""
    total = 0
    for i in range(n):
        for j in range(100):
            total += (i * j) % 7
    return total

def monitor_cpu(stop_event, label=""):
    """各コアのCPU使用率をモニタリング"""
    while not stop_event.is_set():
        # 0.5秒ごとにCPU使用率を出力
        usage_per_core = psutil.cpu_percent(interval=0.5, percpu=True)
        usage_str = "\n".join(f"Core{i+1}: {u:.1f}%" for i, u in enumerate(usage_per_core))
        print(f"[{label}]")
        print(f"{usage_str}")

def start_monitoring(label=""):
    """CPUモニタリング開始"""
    stop_event = threading.Event()
    monitor_thread = threading.Thread(target=monitor_cpu, args=(stop_event, label))
    monitor_thread.start()

    return stop_event, monitor_thread

def stop_monitoring(stop_event, monitor_thread):
    """モニタリング終了"""
    stop_event.set()
    monitor_thread.join()

def run_normal(sample_date):
    """逐次処理"""
    start = time.time()
    for data in sample_date:
        compute(data)
    print("Normal:", time.time() - start)

def run_ThreadPoolExecutor(sample_data):
    """ThreadPoolExecutorを使用した並列処理を実行"""
    start = time.time()
    with concurrent.futures.ThreadPoolExecutor() as executor:
        list(executor.map(compute, sample_data))
    print("ThreadPoolExecutor:", time.time() - start)

def run_ProcessPoolExecutor(sample_data):
    """ProcessPoolExecutorを使用した並列処理を実行"""
    start = time.time()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        list(executor.map(compute, sample_data))
    print("ProcessPoolExecutor:", time.time() - start)

def main():
    # 100個の疑似的な計算を行う
    n_tasks = 1000
    # サンプル入力データ
    sample_data = [10_000] * n_tasks

    # 逐次処理
    stop_event, monitor_thread = start_monitoring("Normal")
    run_normal(sample_data)
    stop_monitoring(stop_event, monitor_thread)

    # スレッドによる並列処理
    stop_event, monitor_thread = start_monitoring("ThreadPool")
    run_ThreadPoolExecutor(sample_data)
    stop_monitoring(stop_event, monitor_thread)

    # プロセスによる並列処理
    stop_event, monitor_thread = start_monitoring("ProcessPool")
    run_ProcessPoolExecutor(sample_data)
    stop_monitoring(stop_event, monitor_thread)

if __name__ == '__main__':
    check_cpu()
    main()
```

コア数の確認：

```
Logical CPU cores: 20
Physical CPU cores: 10
```

## 並列処理なし

CPU 使用率（ある1つの出力）：

```
[Normal]
Core1: 0.0%
Core2: 7.4%
Core3: 0.0%
Core4: 0.0%
Core5: 0.0%
Core6: 0.0%
Core7: 0.0%
Core8: 0.0%
Core9: 2.0%
Core10: 0.0%
Core11: 0.0%
Core12: 0.0%
Core13: 100.0%
Core14: 0.0%
Core15: 0.0%
Core16: 0.0%
Core17: 0.0%
Core18: 0.0%
Core19: 0.0%
Core20: 0.0%
```

結果：

```
Normal: 31.72087550163269
```


## スレッドによる並列処理

CPU 使用率（ある1つの出力）：

```
[ThreadPool]
Core1: 10.2%
Core2: 1.8%
Core3: 1.8%
Core4: 5.3%
Core5: 5.4%
Core6: 8.9%
Core7: 1.9%
Core8: 3.5%
Core9: 6.8%
Core10: 8.6%
Core11: 15.3%
Core12: 3.5%
Core13: 3.4%
Core14: 5.1%
Core15: 8.8%
Core16: 3.8%
Core17: 8.8%
Core18: 1.8%
Core19: 3.6%
Core20: 5.2%
```

処理時間：

```
ThreadPoolExecutor: 32.59864163398743
```

## プロセスによる並列処理

CPU 使用率（ある1つの出力）：

```
[ProcessPool]
Core1: 100.0%
Core2: 100.0%
Core3: 100.0%
Core4: 100.0%
Core5: 100.0%
Core6: 100.0%
Core7: 100.0%
Core8: 100.0%
Core9: 100.0%
Core10: 100.0%
Core11: 100.0%
Core12: 100.0%
Core13: 100.0%
Core14: 100.0%
Core15: 100.0%
Core16: 100.0%
Core17: 100.0%
Core18: 100.0%
Core19: 100.0%
Core20: 100.0%
```

処理時間：

```
ProcessPoolExecutor: 6.43767237663269
```

# 結果

```
Normal: 31.72087550163269
ThreadPoolExecutor: 32.59864163398743
ProcessPoolExecutor: 6.43767237663269
```

`ProcessPoolExecutor` が高速でした。`ThreadPoolExecutor` は逐次処理とほぼ変わりませんでした。

`ThreadPoolExecutor` は GIL（Global Interpreter Lock）により、CPU を多く消費するタスクでは複数スレッドでも実質的に並列に動作できず、効率が上がりにくいという制限があるらしいです。

参考：

https://logmi.jp/main/technology/329470

# 用語

## コア

1つの命令を実行するコンピュータのコンポーネントを指します。コア数は以下のコマンドで確認できます。

```bash
lscpu
```

WSL での実行結果：

```
Architecture:             x86_64
  CPU op-mode(s):         32-bit, 64-bit
  Address sizes:          46 bits physical, 48 bits virtual
  Byte Order:             Little Endian
CPU(s):                   20
  On-line CPU(s) list:    0-19
Vendor ID:                GenuineIntel
  Model name:             13th Gen Intel(R) Core(TM) i9-13900H
    CPU family:           6
    Model:                186
    Thread(s) per core:   2
```

## プロセス

実行するプログラムの単位のことです。専用のメモリ空間を持つため、他のプロセスとメモリを共有しません。python script.py のようにスクリプトを1回実行すると、1つのプロセスが作られます。

## スレッド

ハードウェアの意味でのスレッド（ハードウェアスレッド）とソフトウェアの意味でのスレッド（ソフトウェアスレッド）があります。ハードウェアスレッドは、ハイパースレッディング（1つのコアに複数のスレッドを持つ）などの文脈で使われます。ソフトウェアスレッドは、プロセス内部での実行単位のことを指します。この記事ではスレッドはソフトウェアスレッドのことを指しています。

1つのプロセスで実行される複数のスレッドは、メモリを共有するため他のスレッド内のデータに簡単にアクセスできます。しかしこれがデータの競合を引き起こすことがあります。なお、データの競合が発生しないことを保証することをスレッドセーフといいます。

# 参考

https://docs.python.org/ja/3.13/library/concurrent.futures.html

https://www.isus.jp/hpc/adv-computer-concepts-terminology-1/

https://www.isus.jp/hpc/adv-computer-concepts-terminology-2/

https://logmi.jp/main/technology/329470

https://gihyo.jp/admin/serial/01/rdbms/0007