【Pythonでパフォーマンスを向上させる: マルチスレッドとマルチプロセスの基本と選択法】

はじめに

マルチスレッドとマルチプロセスは、プログラムの実行速度を向上させたり、リソースをより効率的に利用するための手法です。
本記事では、Python を勉強中の筆者がそれらの基本概念や実装方法、どのように使い分けを行うのかをまとめてみました。

マルチスレッドとマルチプロセスの基本概念

マルチスレッド

マルチスレッドは、1 つのプロセス内で複数のスレッドを実行します。各スレッドは独立した実行コンテキストを持ち、タスクを並行して実行します。

マルチプロセス

マルチプロセスは、複数のプロセスを同時に実行することで、複数のタスクを同時に処理します。各プロセスは独立したメモリ空間を持ち、互いに影響を与えずにタスクを実行します。

Python でのマルチスレッドとマルチプロセスの実装方法

マルチスレッド

以下は concurrent.futures モジュールを使用して、マルチスレッドで処理するコードです。
sleep処理を入れることでスレッドがスリープ状態にある間、他のスレッドがアクティブになります。
そのため、処理結果を見れば分かる通りスレッドが切り替わりながら処理が行われているのが分かると思います。

import concurrent.futures
import time

# 1 2 3 4 5を出力する関数
def show_sequence(identifier):
    for i in range(1, 6):
        print(f"{identifier}: {i}", flush=True)
        time.sleep(0.1)

# 6 7 8 9 10を出力する関数
def show_sequence2(identifier):
    for i in range(6, 11):
        print(f"{identifier}: {i}", flush=True)
        time.sleep(0.2)
        


# ThreadPoolExecutorを使用してマルチスレッドを実装
with concurrent.futures.ThreadPoolExecutor() as executor:
    # スレッドでshow_sequenceを実行
    executor.submit(show_sequence, "A")

    # スレッドでshow_sequence2を実行
    executor.submit(show_sequence2, "B")

#結果
#A: 1
#B: 6
#A: 2
#A: 3
#B: 7
#A: 4
#A: 5
#B: 8
#B: 9
#B: 10

マルチプロセス

以下は concurrent.futures モジュールを使用して、マルチプロセスで処理するコードです。
※ここでスリープ処理を入れるのはマルチスレッドとは別の意味があります。sleep処理を削除して実行してもらえると分かるのですが出力結果はシーケンシャルな並びになります。
これは各プロセスが高速に実行され、標準出力にアクセスするタイミングが非常に近いためです。
そのため、わざと処理速度を遅くすることで各プロセスが独立して実行されていることを視覚的に確認しやすくしています。
つまり、ここでのsleep処理というのはあくまで並列で処理が動いているということを感じていただくためのデモンストレーション目的であることに留意してください。

import concurrent.futures
import time

# 1 2 3 4 5を出力する関数
def show_sequence(identifier):
    for i in range(1, 6):
        print(f"{identifier}: {i}", flush=True)
        time.sleep(0.1)


# 6 7 8 9 10を出力する関数
def show_sequence2(identifier):
    for i in range(6, 11):
        print(f"{identifier}: {i}", flush=True)
        time.sleep(0.2)

# ProcessPoolExecutorを使用してマルチプロセスを実装
with concurrent.futures.ProcessPoolExecutor() as executor:
    time.sleep(1)
    # プロセスでshow_sequenceを実行
    executor.submit(show_sequence, "A")

    # プロセスでshow_sequence2を実行
    executor.submit(show_sequence2, "B")

#結果
#A: 1
#B: 6
#A: 2
#A: 3
#B: 7
#A: 4
#A: 5
#B: 8
#B: 9
#B: 10

スレッド間、プロセス間でのデータを共有する

スレッドは、プロセス内のメモリを共有するため、グローバル変数を使ってデータをやり取りすることもできます。
ただし、同時アクセスによる競合を防ぐために、ロックを使用することが推奨されます。
以下はスレッド間でデータを共有する処理のコードです

import concurrent.futures
import threading

# スレッド間で共有するデータ
shared_data = []

# 排他制御用のロックを生成
lock = threading.Lock()

# 1 2 3 4 5を出力する関数
def output_number(number):
    with lock:
        shared_data.append(number)

# 2 4 6 8 10を出力する関数
def output_double(number):
    with lock:
        double_number = number * 2
        shared_data.append(double_number)


numbers = [1,2,3,4,5]

# ThreadPoolExecutorを使用してマルチスレッドを実装
with concurrent.futures.ThreadPoolExecutor() as executor:
    # スレッドで実行
    results1 = executor.map(output_number, numbers)

    # スレッドで実行
    results2 = executor.map(output_double, numbers)

print("Shared data:", shared_data)

#出力結果
#Shared data: [1, 2, 3, 4, 5, 2, 4, 6, 8, 10]

プロセス間でもデータの共有は可能です。
multiprocessing.Manager モジュールを使用します。
以下はプロセス間でデータを共有する処理のコードです

import concurrent.futures
import multiprocessing

# プロセス間で共有するデータ
manager = multiprocessing.Manager()
shared_data = manager.list()

# 排他制御用のロックを生成
lock = manager.Lock()

# 1 2 3 4 5を出力する関数
def output_number(number):
    with lock:
        shared_data.append(number)

# 2 4 6 8 10を出力する関数
def output_double(number):
    with lock:
        double_number = number * 2
        shared_data.append(double_number)

numbers = [1, 2, 3, 4, 5]

# ProcessPoolExecutorを使用してマルチプロセスを実装
with concurrent.futures.ProcessPoolExecutor() as executor:
    # プロセスで実行
    results1 = executor.map(output_number, numbers)

    # プロセスで実行
    results2 = executor.map(output_double, numbers)

print("Shared data:", shared_data)

#出力結果
#Shared data: [1, 2, 5, 3, 4, 6, 4, 2, 8, 10]

マルチプロセスは初期化時にオーバーヘッドが発生する

実はマルチプロセスは初期化時にオーバーヘッドが発生します。
以下のようなコードで検証してみます。
マルチスレッドとマルチプロセスの初期化のみの実行時間を計測しているコードです。

import concurrent.futures
import time
from functools import wraps

# 実行時間を計測するデコレータ
def timer_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        elapsed_time = end_time - start_time
        print(f"{func.__name__} took {elapsed_time:.5f} seconds to execute.")
        return result
    return wrapper

# マルチスレッド
@timer_decorator
def multithread():
    with concurrent.futures.ThreadPoolExecutor() as executor:
        pass

multithread()

# マルチプロセス
@timer_decorator
def multiprocess():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        pass

multiprocess()

5 回の試行の結果

マルチスレッド	マルチプロセス
0.00059	0.03601
0.00061	0.03764
0.00081	0.03904
0.00075	0.03980
0.00131	0.04182

5 回程度の試行結果ですが明らかにマルチプロセスはマルチスレッドに比べ初期化に時間がかかっています。
理由はマルチプロセスでは各プロセスが独立したメモリ空間を確保し初期化するという処理が行われるためです。
そのため短時間の処理を繰り返しマルチプロセスで実行するとパフォーマンスに影響する場合があります。

マルチスレッド、マルチプロセスをどう使い分けるのか

どう使い分けるのかについてですが、
マルチスレッドは I/O バウンドな処理に適しています。
I/O バウンドな処理というのは、CPU 処理速度よりも外部リソースへのアクセス速度がパフォーマンスの制限要因になるような処理のことを指します。
例えば、データベースへのクエリやファイルの読み書きなどがそうです。
こういった処理はデータの読み書きに時間がかかるため、処理の終了を待つ間に CPU に遊びの時間ができてしまいます。マルチスレッドを使うことで、1 つのスレッドが 処理の終了を待っている間に、他のスレッドが CPU を利用して処理を行うことができます。

マルチプロセスは CPU バウンドな処理に適しています。
CPU バウンドな処理というのは CPU 処理速度がパフォーマンスの制限要因になるような処理のことを指します。
例えば、暗号解読などの計算量の多いアルゴリズムを用いる処理などがそうです。
こういった処理は CPU の計算能力がボトルネックとなるため、マルチコアの CPU をフル活用することで処理速度を向上させることができます。

まとめ

マルチスレッドとマルチプロセスについてひととおりまとめてみました。
実務でこれが必要な場面に遭遇したことはないですがいつか役に立てばいいなと思います。