Pythonでのプロセス同期とプーリング徹底解説【multiprocessing活用ガイド】

Pythonのmultiprocessingモジュールを使うと、複数のプロセスを活用した並行・並列処理を簡単に実装できます。本記事では、プロセス間の排他制御に欠かせない同期プリミティブ(Lock)と、複数のタスクを効率的に分散処理するPool(プーリング)について、サンプルコード付きでわかりやすく解説します。

プロセス間の同期

multiprocessingは、APIを通じてプロセスを生成・管理するためのパッケージです。ローカル環境とリモート環境の両方での並行処理をサポートしており、このモジュールを利用することで、プログラマは1台のマシン上の複数のCPUコアを最大限に活用できます。WindowsおよびUNIX系OSの両方で動作します。

また、このパッケージには、スレッド処理と同等の同期プリミティブ(ロック、セマフォなど)がすべて用意されています。

同期のサンプルコード

from multiprocessing import Process, Lock

def my_function(x, y):
    x.acquire()
    print('hello world', y)
    x.release()

if __name__ == '__main__':
    lock = Lock()
    for num in range(10):
        Process(target=my_function, args=(lock, num)).start()

この例では、Lockインスタンスを使用することで、一度に標準出力へ書き込めるプロセスを1つだけに制限しています。acquire()でロックを取得し、処理後にrelease()で解放することで、複数プロセスからの出力が混在するのを防ぎます。

プーリング(Pool)

プーリングにはPoolクラスを使用します。Poolクラスを使うと、投入されたすべてのタスクを処理するワーカープロセスのプールを作成できます。

class multiprocessing.Pool([processes[, initializer[, initargs[, maxtasksperchild]]]])

Poolオブジェクトはワーカーのプールを管理し、どのジョブを投入するかを制御します。さらに、タイムアウトやコールバックを備えた非同期結果の取得や、並列map実装もサポートしています。

引数processesがNoneの場合はcpu_count()の値が使用され、initializerがNoneでない場合はinitializer(*initargs)が呼び出されます。

Poolクラスの主なメソッド

apply(func[, args[, kwds]])

組み込み関数apply()と同じ動作をします。結果が準備できるまでブロックするため、並列に実行したい場合はapply_async()メソッドの方が適しています。

apply_async(func[, args[, kwds[, callback]]])

非同期実行を行い、AsyncResultオブジェクトを返します。

map(func, iterable[, chunksize])

組み込み関数map()と同様ですが、扱えるイテラブル引数は1つだけです。結果が準備できるまでブロックします。

このメソッドでは、イテラブルが複数の小さなチャンクに分割され、それぞれが独立したタスクとしてプロセスプールに投入されます。

map_async(func, iterable[, chunksize[, callback]])

map()の非同期版で、AsyncResultオブジェクトを返します。

imap(func, iterable[, chunksize])

itertools.imap()と同等の動作をします。引数のchunksizeはmap()で使われるものと同じ意味を持ちます。

imap_unordered(func, iterable[, chunksize])

imap()とほぼ同じですが、返されるイテレータの結果順序は順不同(任意の順序)である点が異なります。完了したものから順に結果を受け取れます。

close()

これ以上タスクを投入しないことを示します。すべてのタスクが完了すると、ワーカープロセスは終了します。

terminate()

タスクの完了を待たずに、ワーカープロセスを即座に停止させたい場合に使用します。

join()

ワーカープロセスの終了を待ちます。join()を呼び出す前に、必ずclose()またはterminate()を実行しておく必要があります。

AsyncResultクラス(multiprocessing.pool.AsyncResult)

Pool.apply_async()やPool.map_async()によって返されるオブジェクトのクラスです。

get([timeout])

結果が到着した時点でその値を返します。timeoutを指定した場合、指定秒数以内に結果が得られないとTimeoutErrorが発生します。

wait([timeout])

結果が利用可能になるまで、またはtimeout秒が経過するまで待機します。

ready()

呼び出しが完了しているかどうかを真偽値で返します。

successful()

呼び出しがエラーなしで完了した場合にTrueを返します。

プーリングのサンプルコード

# -*- coding: utf-8 -*-
"""
Created on Sun Sep 30 12:17:58 2018
@author: Tutorials Point
"""
from multiprocessing import Pool
import time

def myfunction(m):
    return m * m

if __name__ == '__main__':
    my_pool = Pool(processes=4)  # ワーカープロセスを4つ起動
    result = my_pool.apply_async(myfunction, (10,))  # 単一プロセス内で f(10) を非同期に評価
    print(result.get(timeout=1))
    print(my_pool.map(myfunction, range(10)))  # [0, 1, 4, ..., 81] を出力
    my_it = my_pool.imap(myfunction, range(10))
    print(my_it.next())            # 0 を出力
    print(my_it.next())            # 1 を出力
    print(my_it.next(timeout=1))   # 4 を出力(非常に遅いマシンでない限り)
    result = my_pool.apply_async(time.sleep, (10,))
    print(result.get(timeout=1))   # multiprocessing.TimeoutError を発生させる

サンプルコードのポイント

• Pool(processes=4):4つのワーカープロセスを持つプールを生成します。
• apply_async():f(10)の計算を非同期で実行し、get(timeout=1)で結果を取得します。
• map():0〜9の各値を2乗し、[0, 1, 4, ..., 81]というリストをまとめて返します。
• imap():結果をイテレータとして順次受け取れます。必要な分だけ取り出せるのが特徴です。
• タイムアウトの挙動:最後の例では10秒かかるtime.sleepに対して1秒でタイムアウトを設定しているため、multiprocessing.TimeoutErrorが発生します。

このように、multiprocessingのPoolクラスを活用すれば、CPU負荷の高い処理を複数のコアに分散させ、プログラム全体の実行時間を大幅に短縮できます。同期プリミティブと組み合わせれば、安全かつ効率的な並列処理が実現できます。