Python マルチプロセッシング ロギング

1. マルチプロセッシング
2. Python でのマルチプロセッシング
3. Python でのマルチプロセッシング ロギング

この記事では、マルチプロセッシングの概念について説明します。 この後、Python でのマルチプロセッシングと、Python コードを使用したマルチプロセッシングのログ処理について説明します。

マルチプロセッシング

マルチプロセッシングは、複数のプロセッサが同じプログラムの異なる部分を同時に処理するコンピューティング パラダイムです。

マルチプロセッシング オペレーティング システムで実行されるアプリケーションは、パフォーマンスを向上させるために異なるプロセッサに割り当てることによって独立して実行される小さなサブルーチンに分割されます。

マルチプロセッシングには 2つの異なるタイプがあります。

1. 対称型マルチプロセッシング: このマルチプロセッシング手法では、単一のオペレーティング システムが共有メイン メモリを使用して複数のコンピューター プロセッサを管理します。 一般に、マルチプロセッシング能力を有するコンピュータシステムは対称型マルチプロセッシングをサポートする。
2. 非対称マルチプロセッシング: このマルチプロセッシングでは、コンピューター プロセッサがオペレーティング システム関連のタスクを処理するために割り当てられ、別のコンピューター プロセッサがアプリケーション関連のタスクを割り当てられます。 非対称マルチプロセッシングでは、同じ時間間隔で 1つのプロセッサがアイドル状態になり、別のプロセッサが完全にビジー状態になる可能性があるため、対称型マルチプロセッシングと比較して非効率的であると見なされます。

Python でのマルチプロセッシング

Python では、マルチプロセッシング タスクに multiprocessing ライブラリが使用されます。

次の例を検討してください。

import multiprocessing


def func1(arg):
    print("func1: with parameter", arg)


def func2(arg):
    print("func2: with parameter", arg)


if __name__ == "__main__":
    process1 = multiprocessing.Process(target=func1, args=(1,))
    process2 = multiprocessing.Process(target=func2, args=(2,))

    process1.start()
    process2.start()

    process1.join()
    process2.join()
    print("Processes Ended")

上記のコードでは、multiprocessing モジュールをインクルードするために import multiprocessing を使用しています。 Process クラス オブジェクトは、プロセスの作成に使用されます。

Processクラスのオブジェクトはtarget(プロセス内で実行する関数)をパラメータとして受け取り、target関数の引数としてargsを受け取ります。

Process クラスの start メソッドを使用して、プロセスを開始します。 上記の例では、2つのプロセスを開始しました。

process1 と process2 が終了しないまで、join メソッドを使用して現在のプログラムを実行し続けます。

process1 と process2 がタスクを完了すると、現在のプログラムは print("Processes Ended") ステートメントを実行します。

上記のプログラムの出力は次のとおりです。

func1: with parameter 1
func2: with parameter 2
Processes Ended

マルチプロセッシング中の Python のプロセス ID

2つの異なる方法を使用して、プロセス ID (PID) を出力することもできます。

1. os.getpid()
2. Process クラスのオブジェクトメンバ変数 pid

次のコードを検討してください。

import multiprocessing
import os


def func1(arg):
    print("func1: with parameter ", arg)
    print("ID of func1 process:", os.getpid())


def func2(arg):
    print("func2: with parameter ", arg)
    print("ID of func2 process:", os.getpid())


if __name__ == "__main__":
    process1 = multiprocessing.Process(target=func1, args=(1,))
    process2 = multiprocessing.Process(target=func2, args=(2,))

    process1.start()
    process2.start()

    print("Process 1 / function 1 PID: ", process1.pid)
    print("Process 2 / function 2 PID: ", process2.pid)
    process1.join()
    process2.join()
    print("Processes Ended")

上記の例では、os.getpid() と process1.pid がプロセス ID を示しています。 次のコードの出力は次のとおりです。

Process 1 / function 1 PID:  11368
Process 2 / function 2 PID:  14876
func1: with parameter  1
ID of func1 process: 11368
func2: with parameter  2
ID of func2 process: 14876
Processes Ended

os.getpid() と process1.pid は同じプロセス ID を示します。

Pool クラスによるマルチプロセッシング

Python multiprocessing モジュールの Pool クラスを使用して、同じ関数を異なる入力値で並列に実行します。 たとえば、次のコードを考えてみましょう。

import multiprocessing


def func1(arg):
    print("func1: with parameter ", arg)


if __name__ == "__main__":
    process_pool = multiprocessing.Pool(3)
    process_pool.map(func1, [1, 2, 3])
    process_pool.close()
    process_pool.join()

上記のコードでは、multiprocessing.Pool は、異なる引数で func1 を呼び出す 3つのプロセスを作成します。 次のコードの出力は次のとおりです。

func1: with parameter  1
func1: with parameter  2
func1: with parameter  3

Python でのマルチプロセッシング ロギング

Python の multiprocessing ライブラリを使用して、複数のプロセスからの情報をログに記録できます。 マルチプロセッシングのロギングにはさまざまな方法があります。

ロギングのために Python で logging.handlers、QueueHandler、および QueueListener クラスを使用できます。

次のコードを検討してください。

import multiprocessing
import logging
import os
from logging.handlers import QueueHandler, QueueListener


def func(arg):
    logging.info(
        "Process/function with argument {} and PID {}".format(arg, os.getpid())
    )


def Process_init(q):
    queue_handler = QueueHandler(q)
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    logger.addHandler(queue_handler)


if __name__ == "__main__":
    print("Main Started")
    mp_queue = multiprocessing.Queue()
    lg_handler = logging.StreamHandler()

    lg_handler.setFormatter(
        logging.Formatter("%(levelname)s: %(asctime)s - %(process)s - %(message)s")
    )

    queue_listener = QueueListener(mp_queue, lg_handler)
    queue_listener.start()

    process_pool = multiprocessing.Pool(2, Process_init, [mp_queue])
    process_pool.map(func, [1, 2])
    process_pool.close()
    process_pool.join()

    queue_listener.stop()
    print("Main Ended")

上記のコードは、1つの引数を受け取る関数 func を定義しています。

func メソッドは、引数値とプロセス ID を含む文字列をログに記録します。 os.getpid() を使用してプロセス ID を取得します。

main メソッドでは、QueueListener の multiprocessing.Queue() および logging.StreamHandler() オブジェクトを作成します。 handler.setFormatter は、ログ形式を設定するために使用されます。

multiprocessing.Pool オブジェクトを使用して 2つのプロセスを作成し、両方のプロセスで 1 と 2 を引数値として func メソッドを呼び出します。

Process_init メソッドは、queue_handler を初期化するために使用されます。 上記のコードの出力は次のとおりです。

上記のコードは、プロセス ID 7024 と 15680 を持つ 2つのプロセスのログ情報を示しています。

lg_handler = logging.StreamHandler() ステートメントを lg_handler = logging.FileHandler('info.log') に置き換えることで、情報を log ファイルに記録することもできます。

logging.FileHandler はファイル info.log を作成し、ログを info.log ファイルに保存します。