Pythonでのマルチプロセッシングとマルチスレッド

Question

私は、1）ディスクから非常に大きなファイルから読み取り（〜95％の時間）、次に2）比較的小さな出力（〜5％の時間）を処理して提供するPythonプログラムを持っています。このプログラムは、TeraBytesのファイルで実行されます。

今、私はマルチプロセッシングとマルチスレッドを利用してこのプログラムを最適化しようとしています。私が実行しているプラ​​ットフォームは、仮想マシン上に4つのプロセッサを搭載した仮想マシンです。

4つのプロセス（プロセッサと同じ）を実行するスケジューラプロセスを計画しています。ほとんどの部分がI / Oであるため、各プロセスにはいくつかのスレッドが必要です。各スレッドは1つのファイルを処理し、結果をメインスレッドに報告し、メインスレッドはIPCを介してスケジューラプロセスに報告します。スケジューラはこれらをキューに入れ、最終的には順序付けられた方法でディスクに書き込むことができます

では、そのようなシナリオで作成するプロセスとスレッドの数をどのように決定するのでしょうか。最適な組み合わせを見つけるための数学的方法はありますか？

ありがとうございました

Answer 1

私はそれをあなたがしていることの逆にアレンジすると思います。つまり、結果の生成を担当する特定のサイズのスレッドプールを作成します。このプールに送信されるタスクは、CPUにバインドされた作業部分を送信するためにワーカースレッドが使用できるプロセッサプールを引数として渡されます。言い換えると、スレッドプールワーカーは主にすべてのディスク関連の操作を実行し、CPUを集中的に使用する作業をプロセッサプールに渡します。

プロセッサプールのサイズは、環境内にあるプロセッサの数だけである必要があります。スレッドプールの正確なサイズを指定することは困難です。これは、収穫逓減の法則が機能する前に処理できる同時ディスク操作の数によって異なります。また、メモリにも依存します。プールが大きいほど、特にファイル全体を処理のためにメモリに読み込む必要がある場合に、使用されるメモリリソースが大きくなります。したがって、この値を試してみる必要があるかもしれません。以下のコードは、これらのアイデアの概要を示しています。スレッドプールから得られるのは、小さなプロセッサプールを使用した場合よりも多くのI / O操作の重複です。

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from functools import partial
import os

def cpu_bound_function(arg1, arg2):
    ...
    return some_result



def io_bound_function(process_pool_executor, file_name):
    with open(file_name, 'r') as f:
        # Do disk related operations:
        . . . # code omitted
        # Now we have to do a CPU-intensive operation:
        future = process_pool_executor.submit(cpu_bound_function, arg1, arg2)
        result = future.result() # get result
        return result
    
file_list = [file_1, file_2, file_n]
N_FILES = len(file_list)
MAX_THREADS = 50 # depends on your configuration on how well the I/O can be overlapped
N_THREADS = min(N_FILES, MAX_THREADS) # no point in creating more threds than required
N_PROCESSES = os.cpu_count() # use the number of processors you have

with ThreadPoolExecutor(N_THREADS) as thread_pool_executor:
    with ProcessPoolExecutor(N_PROCESSES) as process_pool_executor:
        results = thread_pool_executor.map(partial(io_bound_function, process_pool_executor), file_list)

重要な注意：

もう1つのはるかに単純なアプローチは、サイズがCPUプロセッサの数よりも大きい単一のプロセッサプール（たとえば、25）を使用することです。ワーカープロセスは、I / O操作とCPU操作の両方を実行します。CPUよりも多くのプロセスがある場合でも、プロセスの多くはI / Oが完了するのを待機状態になり、CPUを集中的に使用する作業を実行できるようになります。

このアプローチの欠点は、N個のプロセスを作成する際のオーバーヘッドが、N個のスレッドと少数のプロセスを作成する場合のオーバーヘッドよりもはるかに大きいことです。ただし、プールに送信されたタスクの実行時間がますます大きくなると、この増加したオーバーヘッドは、合計実行時間のより小さな割合になります。したがって、タスクが簡単でない場合、これは適度にパフォーマンスの高い単純化になる可能性があります。

更新：両方のアプローチのベンチマーク

サイズが約10,000KBの24個のファイルを処理する2つのアプローチに対していくつかのベンチマークを実行しました（実際には、これらはそれぞれ8回処理された3つの異なるファイルであったため、キャッシュが行われた可能性があります）。

方法1（スレッドプール+プロセッサプール）

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from functools import partial
import os
from math import sqrt
import timeit


def cpu_bound_function(b):
    sum = 0.0
    for x in b:
        sum += sqrt(float(x))
    return sum

def io_bound_function(process_pool_executor, file_name):
    with open(file_name, 'rb') as f:
        b = f.read()
        future = process_pool_executor.submit(cpu_bound_function, b)
        result = future.result() # get result
        return result

def main():
    file_list = ['/download/httpd-2.4.16-win32-VC14.zip'] * 8 + ['/download/curlmanager-1.0.6-x64.exe'] * 8 + ['/download/Element_v2.8.0_UserManual_RevA.pdf'] * 8
    N_FILES = len(file_list)
    MAX_THREADS = 50 # depends on your configuration on how well the I/O can be overlapped
    N_THREADS = min(N_FILES, MAX_THREADS) # no point in creating more threds than required
    N_PROCESSES = os.cpu_count() # use the number of processors you have

    with ThreadPoolExecutor(N_THREADS) as thread_pool_executor:
        with ProcessPoolExecutor(N_PROCESSES) as process_pool_executor:
            results = list(thread_pool_executor.map(partial(io_bound_function, process_pool_executor), file_list))
            print(results)

if __name__ == '__main__':
    print(timeit.timeit(stmt='main()', number=1, globals=globals()))

方法2（プロセッサー・プールのみ）

from concurrent.futures import ProcessPoolExecutor
from math import sqrt
import timeit


def cpu_bound_function(b):
    sum = 0.0
    for x in b:
        sum += sqrt(float(x))
    return sum

def io_bound_function(file_name):
    with open(file_name, 'rb') as f:
        b = f.read()
        result = cpu_bound_function(b)
        return result

def main():
    file_list = ['/download/httpd-2.4.16-win32-VC14.zip'] * 8 + ['/download/curlmanager-1.0.6-x64.exe'] * 8 + ['/download/Element_v2.8.0_UserManual_RevA.pdf'] * 8
    N_FILES = len(file_list)
    MAX_PROCESSES = 50 # depends on your configuration on how well the I/O can be overlapped
    N_PROCESSES = min(N_FILES, MAX_PROCESSES) # no point in creating more threds than required

    with ProcessPoolExecutor(N_PROCESSES) as process_pool_executor:
        results = list(process_pool_executor.map(io_bound_function, file_list))
        print(results)

if __name__ == '__main__':
    print(timeit.timeit(stmt='main()', number=1, globals=globals()))

結果：

（私は8コアを持っています）

スレッドプール+プロセッサプール：13.5秒プロセッサプールのみ：13.3秒

結論：まず、すべてにプロセッサプールを使用するという単純なアプローチを試してみます。ここで注意が必要なのは、作成するプロセスの最大数を決定することです。これは元の質問の一部であり、CPUを集中的に使用する計算だけで簡単に答えることができました。読んでいるファイルの数が多すぎない場合、要点は議論の余地があります。ファイルごとに1つのプロセスを持つことができます。ただし、数百のファイルがある場合は、プールに数百のプロセスを含めたくないでしょう（作成できるプロセスの数にも上限があり、これらの厄介なメモリの制約もあります）。私があなたに正確な数を与えることができる方法は全くありません。ファイルの数が多い場合は、小さいプールサイズから始めて、それ以上のメリットが得られなくなるまで増分を続けます（もちろん、これらのテストで最大数を超えるファイルを処理したくない場合は、次のようになります。実際の実行に適したプールサイズを決定するだけで、永久に実行されます）。

Answer 2

並列処理の場合：私はこの質問を見て、受け入れられた答えを引用しました：

実際には、最適なスレッド数を見つけるのは難しい場合があり、その数でさえ、プログラムを実行するたびに変わる可能性があります。したがって、理論的には、最適なスレッド数は、マシン上にあるコアの数になります。コアが「ハイパースレッド」である場合（Intelはそれを呼んでいます）、各コアで2つのスレッドを実行できます。その場合、最適なスレッド数は、マシンのコア数の2倍になります。

マルチプロセッシングの場合：誰かがここで同様の質問をしました、そして受け入れられた答えはこれを言いました：

すべてのスレッド/プロセスが実際にCPUにバインドされている場合は、CPUがコアを報告するのと同じ数のプロセスを実行する必要があります。ハイパースレッディングにより、各物理CPUコアは複数の仮想コアを提示できる場合があります。multiprocessing.cpu_count仮想コアの数を取得するために呼び出します。

スレッドの1つのうちpのみがCPUにバインドされている場合は、pを掛けることでその数を調整できます。たとえば、プロセスの半分がCPUバウンド（p = 0.5）で、それぞれ4コアと2xハイパースレッディングを備えた2つのCPUがある場合、0.5 * 2 * 4 * 2 = 8プロセスを開始する必要があります。

ここで重要なのは、使用しているマシンを理解することです。そこから、ほぼ最適な数のスレッド/プロセスを選択して、コードの実行を分割できます。また、スクリプトを実行するたびに少し変化するため、ほぼ最適だと言いました。そのため、数学的な観点からこの最適な数を予測することは困難です。

特定の状況では、マシンに4つのコアがある場合、最大4つのスレッドのみを作成してから、それらを分割することをお勧めします。

• メインスレッドに1。
• 3ファイルの読み取りと処理。

Answer 3

複数のプロセスを使用してIOパフォーマンスを高速化することはお勧めできません。これとその下のサンプルコードをチェックして、役立つかどうかを確認してください。

Answer 4

1つのアイデアは、スレッドにファイルのみを読み取らせ（よく理解していれば、ファイルは1つだけです）、独立した部分（行など）をメッセージとともにキューにプッシュすることです。
メッセージは4つのスレッドで処理できます。このようにして、プロセッサ間の負荷を最適化できます。

Answer 5

強力なI / Oバウンドプロセス（説明しているものなど）では、必ずしもマルチスレッドやマルチプロセッシングは必要ありません。OSからより高度なI / Oプリミティブを使用することもできます。

たとえば、Linuxでは、適切なサイズの可変バッファーとともに読み取り要求をカーネルに送信し、バッファーがいっぱいになったときに通知を受けることができます。これは、純粋なpythonバインディングを記述したAIO APIを使用して実行できます：python-libaio（libaio on pypi））、またはCFFIpythonバインディングがあると思われる最新のio_uringAPI（liburing）を使用します。 on pypy）（io_uringもこのpythonバインディングも使用していません）。

これにより、レベルでの並列処理の複雑さが解消され、OS /ユーザーランドコンテキストスイッチの数が減り（CPU時間がさらに短縮され）、OSが何をしようとしているのかをより詳しく知ることができ、スケジューリングの機会が与えられます。 IOをより効率的に（仮想化環境では、自分で試したことはありませんが、データコピーの数が減っても驚かないでしょう）。

もちろん、欠点は、プログラムが実行しているOSにより緊密にバインドされ、別のOSで実行するためにより多くの労力が必要になることです。