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

Python の multiprocessing モジュールを使うと、処理を別々のプロセスで実行できます。個々のプロセスを開始して終了を待つ方法、Pool を使って複数の入力に同じ関数を適用する方法、ワーカープロセスのログレコードを親プロセスのリスナーへ送る方法を紹介します。以下の例では、それぞれの設定方法と、メッセージを出力したプロセスの識別方法を説明します。

マルチプロセッシング

マルチプロセッシングでは、メモリ空間がそれぞれ独立した複数のプロセスで処理を実行します。独立したタスクに分割できるCPU負荷の高い処理に適しています。プロセスはオペレーティングシステムによってスケジュールされるため、出力される順序は実行ごとに変わることがあります。

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

プロセスを作成するには、multiprocessing モジュールをインポートします。Process には、呼び出し可能なオブジェクトを target として、引数のタプルを args として渡します。start() でワーカーを開始し、join() で終了を待ちます。メインガードを使うと、メインモジュールのインポートによってワーカーを開始するプラットフォームも含め、子プロセスがプログラムのトップレベルにあるプロセス作成コードを再実行するのを防げます。

次の例を実行します。

import multiprocessing


def report(value):
    print(f"worker received: {value}", flush=True)


if __name__ == "__main__":
    process = multiprocessing.Process(target=report, args=(5,))
    process.start()
    process.join()
    print("parent: process finished")

worker received: 5
parent: process finished

join() が戻る前にワーカーが出力するため、親プロセスのメッセージはその後に表示されます。複数のワーカーを使うと、メッセージが入り交じったり、別の順序で表示されたりすることがあります。出力を一元的に処理する必要がある場合は、キューを使います。

プロセス ID

ワーカー内で os.getpid() を使うと、実行中のプロセスを識別できます。start() の呼び出し後は、親プロセスから Process オブジェクトの pid 属性を参照して同じプロセス ID を取得できます。PID はオペレーティングシステムが割り当て、実行ごとに異なるため、固定の出力値ではなく実行時の値として扱ってください。

import multiprocessing
import os


def report_pid():
    print(f"worker PID: {os.getpid()}", flush=True)


if __name__ == "__main__":
    process = multiprocessing.Process(target=report_pid)
    process.start()
    print(f"Process.pid: {process.pid}")
    process.join()

Pool で処理を割り当てる

Pool はワーカープロセスのグループを管理します。map() メソッドは、反復可能オブジェクトの各要素に同じ関数を適用し、その反復順に結果を返します。プールのサイズはワーカー数の上限であり、入力ごとに1つのプロセスが使われるとは限りません。コンテキストマネージャーを抜けると、ブロックの終了時にプールは強制終了されます。正常に完了させることが重要な場合は、代わりにプールを明示的に閉じて新しい処理を受け付けないようにし、join() でワーカーの終了を待ってください。

import multiprocessing


def square(value):
    return value * value


if __name__ == "__main__":
    with multiprocessing.Pool(processes=2) as pool:
        results = pool.map(square, [2, 3, 4])
    print(results)

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

ワーカープロセスから1つのキューへログレコードを送り、親プロセスの QueueListener から出力用ハンドラーへ渡します。ファイルやコンソールへの出力を1つのプロセスに集約できるため、複数のワーカーが同じハンドラーへ個別に書き込むのを避けられます。各ワーカーのロガーにはキューハンドラーを設定し、リスナーにはキューと1つ以上のハンドラーを渡します。

次の完全な例では、プールのワーカーに QueueHandler を設定します。親プロセスはプールを作る前にリスナーを開始します。マッピングが終わったら、新しい処理を受け付けないようプールを閉じ、ワーカーの終了を待ってからリスナーを停止します。ワーカーの終了時にはキューのフィーダーにバッファーされたレコードがフラッシュされるため、リスナーはそれらを読み取れます。マッピングで例外が発生した場合も同じ終了処理が行われ、完了済みのワーカーのレコードを読み取ってから例外が伝播します。フォーマッターは各メッセージにログレベルを付け、親プロセスはレコードを標準出力に書き込みます。ワーカーが1つの場合、項目は順番に処理されます。ワーカーをプロセス ID で識別する必要がある場合は、先ほどの PID の例を使ってください。

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


def configure_worker(log_queue):
    logger = logging.getLogger()
    logger.handlers.clear()
    logger.setLevel(logging.INFO)
    logger.addHandler(QueueHandler(log_queue))


def report(value):
    logger = logging.getLogger(__name__)
    logger.info("item=%s", value)
    return value


if __name__ == "__main__":
    log_queue = multiprocessing.Queue()
    output_handler = logging.StreamHandler(sys.stdout)
    output_handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
    listener = QueueListener(log_queue, output_handler)
    listener.start()

    try:
        pool = multiprocessing.Pool(
            processes=1,
            initializer=configure_worker,
            initargs=(log_queue,),
        )
        try:
            results = pool.map(report, [1, 2, 3])
        finally:
            pool.close()
            pool.join()
    finally:
        listener.stop()
        log_queue.close()
        log_queue.join_thread()

INFO item=1
INFO item=2
INFO item=3

この例は終了コード 0 で完了し、マッピングされた各項目について INFO レコードを1件ずつ出力しました。ワーカーが1つの場合、レコードはマッピングの順に表示されました。出力例のログレベル表示 INFO は、コードと出力の一致を保つため英語のままです。

コンソールではなくファイルにレコードを書き込むには、logging.StreamHandler(sys.stdout) を logging.FileHandler("multiprocessing.log", encoding="utf-8") に置き換えます。そのハンドラーは親プロセスが所有して閉じます。インタープリター終了時の処理に任せず明示的に管理する場合は、listener.stop() の後に閉じてください。ログのローテーションや複数の出力先が必要な場合は、追加のハンドラーを QueueListener に渡し、それぞれに必要なフォーマッターを設定します。