More than 1 year has passed since last update.

Python高速化【並列処理】【マルチスレッド】【マルチプロセス】

Last updated at 2021-11-23Posted at 2021-11-14

はじめに

　pythonのプロダクトの一部処理（I/Oバウンドなプログラム）を高速化する必要があり、初めて並列処理を学習しました！今回は学習した内容をアウトプットしていきたいと思います。
初学者のため、理解不足な点が多々あると思います。気になった点やアドバイスなどありましたら、教えていただけますと幸いです。

前提知識

プロセス
　プログラムの実行単位、固有のメモリを持つ。
スレッド
　プロセス内の処理の実行単位、プロセス内のメモリを共有。
I/Oバウンド
- プログラムのディスクとの入出力による負担のこと。
- ファイルの読み書き、DBへの接続、ネットワーク通信で発生することが多い。
- 今回高速化したいプログラムはI/Oバウンドなプログラム
CPUバウンド
- CPUに負荷がかかること。
- 数値計算のようにCPUを使い続けるような処理とかで発生することが多い。
GIL(グローバルインタプリタロック)
- 例えば、二つのスレッドを同時に実行する場合、スレッドAが変数Cを書き換えている間にスレッドBも同じ変数Cを書き換えようとしてエラーが発生する。このようなスレッド間の競合を防ぐため、　スレッドA実行中はスレッドBを待機させる仕組みがGIL。
- GILロックを取得したスレッドのみが処理を実行する。

マルチスレッドとマルチプロセス

画像引用：https://www.youtube.com/watch?v=LQRMX-1Rzew

マルチスレッド

1つのプロセス内で複数のスレッドを実行。
スレッド同士、同じメモリ空間にあるためデータの受け渡しが容易
GILロックを取得したスレッドが処理を実行している間、他のスレッドは待機、ある期間経過したらGILロックが解放されてGILロックを取得した別のスレッドが処理を実行していく。スレッド同士でGILロックを取り合ってるイメージ。
同時にスレッドが実行されているわけではない。
高速化したい処理がI/OバウンドかCPUバウンドかで高速化できるかどうかが変わってくる。

マルチスレッドのI/OバウンドとCPUバウンド

I/Oバウンドの処理には効果的。
スレッド数がマルチプロセスより増やせるのでスケールしやすい。
- コア数により並列化できるプロセスが決まってくるから。
CPUバウンドの処理には不向き。

マルチプロセス

複数のプロセスを立ち上げて処理を実行
全プロセスが同時に走る
コアが別なので、GILロックを取得する必要なし
スレッド生成よりプロセス生成は時間とメモリ消費が大きくなる。
- プロセス同士、メモリを共有していないので必要なデータを複製する必要があるため。

マルチプロセスのI/OバウンドとCPUバウンド

GILの影響を受けない
CPUのコア数を超えて並列化できないのでスケールしない。
プロセス間通信しないとデータの受け渡しができない。

使用したライブラリ　concurrent.futures

Python3.2 から追加されたマルチスレッド、マルチプロセス用のライブラリ。
マルチスレッドは ThreadPoolExecutor()

マルチプロセスはProcessPoolExecutor()と書き換えるだけ。

サンプルコード

test1.py

# I/Oバウンドの処理をシングルスレッド、マルチスレッド、マルチプロセスでそれぞれ実行
import concurrent.futures
import os
import time


LEARGE_TEXT = 'string' * 100000000

def io_bound(file_name):
    with open(file_name, 'w+') as f:
        # ファイルにLEARGE_TEXTを書き込み
        f.write(LEARGE_TEXT)
        # ファイルの先頭に移動
        f.seek(0)
        # ファイル読み込み
        f.read()
    # ファイル削除
    os.remove(file_name)
    return 'Future is done.'

if __name__ == '__main__':
    start = time.time()
    print(io_bound('1.text'))
    print(io_bound('2.text'))
    end = time.time()
    print('シングルスレッド: TIME {:.4f}\n'.format(end - start))

    # マルチスレッド ThreadPoolExecutor
    # max_workersでスレッド数指定
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        start = time.time()
        # submitで第一引数に処理対象の関数、第二引数以降に引数指定
        future1 = executor.submit(io_bound, '1.text')
        future2 = executor.submit(io_bound, '2.text')
        print(future1.result())
        print(future2.result())
        end = time.time()
        print('マルチスレッド: TIME {:.4f}\n'.format(end - start))
    
    # マルチプロセス ProcessPoolExecutor
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        start = time.time()
        future1 = executor.submit(io_bound, '1.text')
        future2 = executor.submit(io_bound, '2.text')
        print(future1.result())
        print(future2.result())
        end = time.time()
        print('CPUの数:', os.cpu_count())
        print('マルチプロセス: TIME {:.4f}\n'.format(end - start))

ターミナル

% python test1.py
Future is done.
Future is done.
シングルスレッド: TIME 1.6771

Future is done.
Future is done.
マルチスレッド: TIME 1.2766

Future is done.
Future is done.
CPUの数: 8
マルチプロセス: TIME 1.5025

test2.py

# CPUバウンドの処理をシングルスレッド、マルチスレッド、マルチプロセスでそれぞれ実行
import concurrent.futures
import os
import time


def cpu_bound():
    i = 0
    while i < 10000000:
        i = i + 1 - 2 + 3 - 4 + 5
    return 'Future is done.'

if __name__ == '__main__':
    start = time.time()
    print(cpu_bound())
    print(cpu_bound())
    end = time.time()
    print('シングルスレッド: TIME {:.4f}\n'.format(end - start))

    # マルチスレッド ThreadPoolExecutor
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        start = time.time()
        # submitで第一引数に処理対象の関数、第二引数以降に引数指定
        future1 = executor.submit(cpu_bound)
        future2 = executor.submit(cpu_bound)
        print(future1.result())
        print(future2.result())
        end = time.time()
        print('マルチスレッド: TIME {:.4f}\n'.format(end - start))
    
    # マルチプロセス ProcessPoolExecutor
    with concurrent.futures.ProcessPoolExecutor(max_workers=2) as executor:
        start = time.time()
        future1 = executor.submit(cpu_bound)
        future2 = executor.submit(cpu_bound)
        print(future1.result())
        print(future2.result())
        end = time.time()
        print('CPUの数:', os.cpu_count())
        print('マルチプロセス: TIME {:.4f}\n'.format(end - start))