`multiprocessing` --- プロセスベースの並列処理¶

利用可能な環境: not Android, not iOS, not WASI.

このモジュールはモバイルプラットフォームと WebAssemblyプラットフォームをサポートしません。

はじめに¶

multiprocessing は、threading モジュールに似た API でプロセスを生成できるパッケージです。 multiprocessing パッケージは、ローカルとリモートの両方で並行処理を提供し、スレッドの代わりにサブプロセスを使うことでグローバルインタープリタロックを実質的に回避します。そのため、multiprocessing モジュールを使うと、マシンに搭載された複数のプロセッサーを十分に活用できます。 POSIX と Windows の両方で動作します。

multiprocessing モジュールは Pool オブジェクトも提供します。これは、入力データをプロセス間に分配し、複数の入力値に対する関数の実行を並列化する (データ並列処理) ための便利な手段です。次の例では、子プロセスがモジュールをインポートできるように、その中に関数を定義するという一般的な方法を示します。 Pool を使ったデータ並列処理の基本的な例です。

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))

標準出力に以下が出力されます:

[1, 4, 9]

multiprocessing モジュールには、実行中のプロセスの terminate、interrupt、kill など、threading モジュールには対応するものがない API もあります。

参考

concurrent.futures.ProcessPoolExecutor は、呼び出し元プロセスの実行をブロックせずに、バックグラウンドのプロセスへタスクを渡す高水準のインターフェースを提供します。 Pool インターフェースを直接使う場合に比べ、concurrent.futures API では、内部のプロセスプールへの処理の投入と結果の待機を分離しやすくなります。

`Process` クラス¶

multiprocessing では、Process オブジェクトを作成して start() メソッドを呼び出すことで、プロセスを生成します。 Process は threading.Thread の API に従っています。以下は、複数のプロセスを使うプログラムの簡単な例です。

from multiprocessing import Process

def f(name):
    print('hello', name)

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

実行された個々のプロセス ID を表示するために拡張したサンプルコードを以下に示します:

from multiprocessing import Process
import os

def info(title):
    print(title)
    print('module name:', __name__)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())

def f(name):
    info('function f')
    print('hello', name)

if __name__ == '__main__':
    info('main line')
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()

なぜ if __name__ == '__main__' という記述が必要かはプログラミングガイドラインを参照してください。

Process の引数は、子プロセスに渡せるよう、通常は pickle 化可能でなければなりません。上の例を REPL に直接入力すると、子プロセスが __main__ モジュール内の f 関数を見つけようとして AttributeError になる場合があります。

コンテキストと開始方式¶

プラットフォームに応じて、multiprocessing はプロセスを開始する 3 つの方式をサポートします。これらの 開始方式 は次のとおりです。

spawn
親プロセスは新たに Python インタープリタープロセスを開始します。子プロセスはプロセスオブジェクトの run() メソッドの実行に必要なリソースのみ継承します。特に、親プロセスからの不要なファイル記述子とハンドルは継承されません。この方式を使用したプロセスの開始は fork や forkserver に比べ遅くなります。

POSIX と Windows で利用できます。 Windows と macOS ではデフォルトの方式です。

fork
親プロセスは os.fork() を使用して Python インタープリターをフォークします。子プロセスはそれが開始されるとき、事実上親プロセスと同一になります。親プロセスのリソースはすべて子プロセスに継承されます。マルチスレッドプロセスのフォークは安全性に問題があることに注意してください。

POSIX システムで利用できます。

バージョン 3.14 で変更: どのプラットフォームでも、デフォルトの開始方式ではなくなりました。 fork を必要とするコードでは、get_context() または set_start_method() で明示的に指定しなければなりません。

バージョン 3.12 で変更: プロセスが複数のスレッドを持つと Python が検出できる場合、この開始方式が内部で呼び出す os.fork() 関数は DeprecationWarning を出すようになりました。別の開始方式を使ってください。詳しくは os.fork() のドキュメントを参照してください。

forkserver
プログラムの開始時に forkserver 開始方式を選ぶと、サーバープロセスを生成します。それ以降、新しいプロセスが必要になるたびに、親プロセスがサーバーに接続し、新しいプロセスをフォークするよう要求します。システムライブラリや事前のインポートによる副作用でスレッドが生成されない限り、フォークサーバーは単一スレッドなので、通常は安全に os.fork() を使えます。不要なリソースは引き継ぎません。

Linux など、Unix パイプでのファイル記述子の受け渡しをサポートする POSIX プラットフォームで利用できます。これらの環境ではデフォルトの方式です。ただし、フォークサーバーが必要とする書き込み可能な一時ディレクトリがない場合 (tempfile.gettempdir() を参照) は、spawn がデフォルトになります。

バージョン 3.14 で変更: POSIX プラットフォームのデフォルトの開始方式になりました。

バージョン 3.15.1 で変更: 読み取り専用ファイルシステムなど、書き込み可能な一時ディレクトリがない場合のデフォルトの開始方式が spawn になりました。以前は、この場合も forkserver がデフォルトのままで、プロセスの開始が OSError で失敗していました。デフォルトの方式は、multiprocessing のインポート時ではなく、最初に必要になった時点で決まります。 gh-155717 を参照してください。

バージョン 3.4 で変更: すべての POSIX プラットフォームで spawn を、一部の POSIX プラットフォームで forkserver を追加しました。 Windows では、親の継承可能なハンドルを子プロセスがすべて引き継ぐことはなくなりました。

バージョン 3.8 で変更: macOS では spawn がデフォルトの開始方式になりました。 macOS のシステムライブラリがスレッドを開始し、サブプロセスのクラッシュにつながる場合があるため、fork 開始方式は安全ではないと考えてください。 bpo-33725 を参照してください。

バージョン 3.14 で変更: POSIX プラットフォームのデフォルトの開始方式を fork から forkserver に変更しました。これは、性能を維持しつつ、マルチスレッドプロセスでよく起こる互換性の問題を避けるためです。 gh-84559 を参照してください。

On POSIX using the spawn or forkserver start methods will also start a resource tracker process which tracks the unlinked named system resources (such as named semaphores or SharedMemory objects) created by processes of the program. When all processes have exited the resource tracker unlinks any remaining tracked object. Usually there should be none, but if a process was killed by a signal there may be some "leaked" resources. (Neither leaked semaphores nor shared memory segments will be automatically unlinked until the next reboot. This is problematic for both objects because the system allows only a limited number of named semaphores, and shared memory segments occupy some space in the main memory.)

開始方式はメインモジュールの if __name__ == '__main__' 節内で、関数 set_start_method() によって指定します。以下に例を示します:

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    mp.set_start_method('spawn')
    q = mp.Queue()
    p = mp.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

関数 set_start_method() はプログラム内で複数回使用してはいけません。

もうひとつの方法として、get_context() を使用してコンテキストオブジェクトを取得することができます。コンテキストオブジェクトは multiprocessing モジュールと同じ API を持ち、同じプログラム内で複数の開始方式を使用できます。

import multiprocessing as mp

def foo(q):
    q.put('hello')

if __name__ == '__main__':
    ctx = mp.get_context('spawn')
    q = ctx.Queue()
    p = ctx.Process(target=foo, args=(q,))
    p.start()
    print(q.get())
    p.join()

あるコンテキストに関連したオブジェクトは、異なるコンテキストのプロセスとは互換性がない場合があることに注意してください。特に、fork コンテキストを使用して作成されたロックは、spawn あるいは forkserver を使用して開始されたプロセスに渡すことはできません。

multiprocessing や ProcessPoolExecutor を使うライブラリは、利用者が独自の multiprocessing コンテキストを渡せるように設計してください。ライブラリ内で特定のコンテキストを選んで使うと、利用者のアプリケーションの他の部分と互換性の問題が生じる場合があります。ライブラリが特定の開始方式を必要とする場合は、必ずドキュメントに記載してください。

警告

POSIX システムでは、通常、'spawn' と 'forkserver' 開始方式を「凍結された」実行ファイル (PyInstaller や cx_Freeze などのパッケージで生成したバイナリ) とともに使うことはできません。コードがスレッドを使わなければ、'fork' 開始方式は動作する可能性があります。

プロセス間でのオブジェクト交換¶

multiprocessing は、プロセス間で 2 種類の通信路をサポートします。

キュー (Queue)

Queue クラスは queue.Queue クラスとほとんど同じように使うことができます。以下に例を示します:
from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())    # "[42, None, 'hello']" を表示
    p.join()
キューはスレッドセーフかつプロセスセーフです。 multiprocessing のキューに入れたオブジェクトはすべてシリアライズされます。

パイプ (Pipe)

Pipe() 関数はパイプで繋がれたコネクションオブジェクトのペアを返します。デフォルトでは双方向性パイプを返します。以下に例を示します:
from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())   # "[42, None, 'hello']" を表示
    p.join()
パイプのそれぞれの端を表す2つのコネクションオブジェクトが Pipe() 関数から返されます。各コネクションオブジェクトには、 send()、 recv()、その他のメソッドがあります。2つのプロセス (またはスレッド) がパイプの同じ端で同時に読み込みや書き込みを行うと、パイプ内のデータが破損してしまうかもしれないことに注意してください。もちろん、各プロセスがパイプの別々の端を同時に使用するならば、データが破壊される危険性はありません。

send() メソッドがオブジェクトをシリアライズし、recv() メソッドがオブジェクトを再構築します。

プロセス間の同期¶

multiprocessing には、threading のすべての同期プリミティブに相当するものがあります。例えばロックを使って、一度に標準出力へ表示するプロセスを 1 つだけにできます。

from multiprocessing import Process, Lock

def f(l, i):
    l.acquire()
    try:
        print('hello world', i)
    finally:
        l.release()

if __name__ == '__main__':
    lock = Lock()

    for num in range(10):
        Process(target=f, args=(lock, num)).start()

ロックを使用しないで標準出力に書き込んだ場合は、各プロセスからの出力がごちゃまぜになってしまいます。

プロセス間での状態の共有¶

これまでの話の流れで触れたとおり、並行プログラミングを行うときには、できるかぎり状態を共有しないのが定石です。複数のプロセスを使用するときは特にそうでしょう。

それでも共有データを使う必要がある場合、multiprocessing はいくつかの方法を提供しています。

共有メモリ (Shared memory)

データを共有メモリ上に保持するために Value クラス、もしくは Array クラスを使用することができます。以下のサンプルコードを使って、この機能についてみていきましょう
from multiprocessing import Process, Value, Array

def f(n, a):
    n.value = 3.1415927
    for i in range(len(a)):
        a[i] = -a[i]

if __name__ == '__main__':
    num = Value('d', 0.0)
    arr = Array('i', range(10))

    p = Process(target=f, args=(num, arr))
    p.start()
    p.join()

    print(num.value)
    print(arr[:])
このサンプルコードを実行すると以下のように表示されます
3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]
num と arr を生成するときに使用されている、引数 'd' と 'i' は array モジュールにより使用される種別の型コードです。ここで使用されている 'd' は倍精度浮動小数、 'i' は符号付整数を表します。これらの共有オブジェクトは、プロセスセーフでありスレッドセーフです。

共有メモリを使用して、さらに柔軟なプログラミングを行うには multiprocessing.sharedctypes モジュールを使用します。このモジュールは共有メモリから割り当てられた任意の ctypes オブジェクトの生成をサポートします。

サーバープロセス (Server process)

Manager() が返すマネージャーオブジェクトは、Python オブジェクトを保持するサーバープロセスを管理します。他のプロセスはプロキシを使って、その Python オブジェクトを操作できます。

Manager() が返すマネージャーは、list、dict、set、Namespace、 Lock、RLock、Semaphore、BoundedSemaphore、Condition、Event、Barrier、 Queue、Value、Array の各型をサポートします。以下に例を示します。
from multiprocessing import Process, Manager

def f(d, l, s):
    d[1] = '1'
    d['2'] = 2
    d[0.25] = None
    l.reverse()
    s.add('a')
    s.add('b')

if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()
        l = manager.list(range(10))
        s = manager.set()

        p = Process(target=f, args=(d, l, s))
        p.start()
        p.join()

        print(d)
        print(l)
        print(s)
このサンプルコードを実行すると以下のように表示されます
{0.25: None, 1: '1', '2': 2}
[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]
{'a', 'b'}
サーバープロセスのマネージャーオブジェクトは共有メモリのオブジェクトよりも柔軟であるといえます。それは、どのような型のオブジェクトでも使えるからです。また、1つのマネージャーオブジェクトはネットワーク経由で他のコンピューター上のプロセスによって共有することもできます。しかし、共有メモリより動作が遅いという欠点があります。

ワーカープロセスのプールを使用¶

Pool クラスは、ワーカープロセスをプールする機能を備えています。このクラスには、異なる方法でワーカープロセスへタスクを割り当てるいくつかのメソッドがあります。

例えば:

from multiprocessing import Pool, TimeoutError
import time
import os

def f(x):
    return x*x

if __name__ == '__main__':
    # 4 つのワーカープロセスを開始
    with Pool(processes=4) as pool:

        # "[0, 1, 4,..., 81]" を表示
        print(pool.map(f, range(10)))

        # 同じ数値を任意の順序で表示
        for i in pool.imap_unordered(f, range(10)):
            print(i)

        # "f(20)" を非同期で評価
        res = pool.apply_async(f, (20,))      # 1 つのプロセスだけで実行
        print(res.get(timeout=1))             # "400" を表示

        # "os.getpid()" を非同期で評価
        res = pool.apply_async(os.getpid, ()) # 1 つのプロセスだけで実行
        print(res.get(timeout=1))             # そのプロセスの PID を表示

        # 複数の評価を非同期に開始すると、より多くのプロセスを使う場合があります
        multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
        print([res.get(timeout=1) for res in multiple_results])

        # 1 つのワーカーを 10 秒間スリープさせる
        res = pool.apply_async(time.sleep, (10,))
        try:
            print(res.get(timeout=1))
        except TimeoutError:
            print("We lacked patience and got a multiprocessing.TimeoutError")

        print("For the moment, the pool remains available for more work")

    # with ブロックを抜けたため、プールは停止しています
    print("Now the pool is closed and no longer available")

プールオブジェクトのメソッドは、そのプールを作成したプロセスのみが呼び出すべきです。

注釈

このパッケージに含まれる機能を使用するためには、子プロセスから __main__ モジュールをインポートできる必要があります。このことについてはプログラミングガイドラインで触れていますが、ここであらためて強調しておきます。なぜかというと、いくつかのサンプルコード、例えば multiprocessing.pool.Pool のサンプルはインタラクティブシェル上では動作しないからです。以下に例を示します:

>>> from multiprocessing import Pool
>>> p = Pool(5)
>>> def f(x):
...     return x*x
...
>>> with p:
...     p.map(f, [1,2,3])
Process PoolWorker-1:
Process PoolWorker-2:
Process PoolWorker-3:
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>
AttributeError: Can't get attribute 'f' on <module '__main__' (<class '_frozen_importlib.BuiltinImporter'>)>

(もしこのコードを試すなら、実際には3つの完全なトレースバックがばらばらの順番で出力されますし、親プロセスを何らかの方法で止める必要があります。)

リファレンス¶

multiprocessing パッケージは、おおむね threading モジュールと同じ API を提供します。

グローバルな開始方式¶

Python は、プロセスを作成して初期化する方法をいくつかサポートしています。グローバルな開始方式は、プロセスを作成するデフォルトの仕組みを設定します。

multiprocessing の一部の関数やメソッド、および特定のオブジェクトの作成では、グローバルな開始方式がまだ設定されていなければ、システムのデフォルトに暗黙的に設定します。グローバルな開始方式は 1 回しか設定できません。システムのデフォルトから変更する必要がある場合は、これらの関数やメソッドの呼び出し、オブジェクトの作成より前に、明示的に設定してください。

`Process` クラスと例外¶

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)¶

Process オブジェクトは各プロセスの処理を表します。 Process クラスは threading.Thread クラスのすべてのメソッドと同じインターフェースを提供します。

コンストラクタは常にキーワード引数で呼び出してください。 group は常に None にしてください。これは threading.Thread との互換性のためだけに存在します。 target は run() メソッドが呼び出す呼び出し可能オブジェクトです。デフォルトの None は、何も呼び出さないことを意味します。 name はプロセス名です (詳しくは name を参照)。 args は target の呼び出しに渡す引数のタプルです。 kwargs は target の呼び出しに渡すキーワード引数の辞書です。キーワード専用引数 daemon を指定すると、プロセスの daemon フラグを True または False に設定します。 None (デフォルト) の場合は、作成元のプロセスからこのフラグを継承します。

デフォルトでは、target に何の引数も与えられません。(実) 引数 args (デフォルトは ())は、target に渡す引数のリストまたはタプルを指定するために使用されます。

サブクラスでコンストラクタをオーバーライドする場合は、プロセスに対して他の処理をする前に、必ず基底クラスのコンストラクタ (super().__init__()) を呼び出してください。

注釈

一般に、Process に渡す引数はすべて pickle 化可能でなければなりません。これは、REPL 内でローカルに定義した target 関数を使って Process を作成したり、concurrent.futures.ProcessPoolExecutor を使ったりするときによく問題になります。

現在の REPL セッションで定義した呼び出し可能オブジェクトを渡すと、子プロセスは起動時に未捕捉の AttributeError で終了します。 pickle から復元するときに読み込めるよう、target はインポート可能なモジュール内で定義されていなければならないためです。

子プロセスで起きる、この捕捉できないエラーの例です。

>>> import multiprocessing as mp
>>> def knigit():
...     print("Ni!")
...
>>> process = mp.Process(target=knigit)
>>> process.start()
>>> Traceback (most recent call last):
  File ".../multiprocessing/spawn.py", line ..., in spawn_main
  File ".../multiprocessing/spawn.py", line ..., in _main
AttributeError: module '__main__' has no attribute 'knigit'
>>> process
<SpawnProcess name='SpawnProcess-1' pid=379473 parent=378707 stopped exitcode=1>

開始方式が spawn および forkserver の場合を参照してください。 "fork" 開始方式を使う場合はこの制限はありませんが、Python 3.14 以降、どのプラットフォームでもデフォルトではなくなっています。コンテキストと開始方式を参照してください。 gh-132898 も参照してください。

バージョン 3.3 で変更: daemon 引数を追加しました。

run()¶

プロセスが実行する処理を表すメソッドです。

このメソッドはサブクラスでオーバーライドできます。標準の run() メソッドは、コンストラクターの target 引数に呼び出し可能オブジェクトが渡されていれば、それを呼び出します。その際、args の要素を位置引数として、kwargs の内容をキーワード引数として渡します。

Process に渡される args の引数にリストやタプルを使っても、同じ効果が得られます。

以下はプログラム例です:

>>> from multiprocessing import Process
>>> p = Process(target=print, args=[1])
>>> p.run()
1
>>> p = Process(target=print, args=(1,))
>>> p.run()
1

start()¶

プロセスの処理を開始するためのメソッドです。

各 Process オブジェクトに対し、このメソッドが2回以上呼び出されてはいけません。各プロセスでオブジェクトの run() メソッドを呼び出す準備を行います。

join([timeout])¶

オプションの引数 timeout が None (デフォルト) の場合、このメソッドは join() の対象プロセスが終了するまでブロックします。 timeout が正の数である場合、最大 timeout 秒ブロックします。対象プロセスが終了した場合も、メソッドがタイムアウトした場合も、戻り値は None であることに注意してください。プロセスの exitcode を確認し、終了したかどうかを判断してください。

1つのプロセスは何回も join されることができます。

プロセスは自分自身を join することはできません。それはデッドロックを引き起こすことがあるからです。プロセスが start される前に join しようとするとエラーが発生します。

name¶

プロセスの名前。名前は識別のためだけに使用される文字列です。それ自体には特別な意味はありません。複数のプロセスに同じ名前が与えられても構いません。

最初の名前はコンストラクターによってセットされます。コンストラクターに明示的な名前が渡されない場合、 'Process-N₁:N₂:...:N_k' 形式の名前が構築されます。ここでそれぞれの N_k はその親のN番目の子供です。

is_alive()¶

プロセスが実行中かを判別します。

おおまかに言って、プロセスオブジェクトは start() メソッドから戻ってから子プロセス終了までの期間が実行中となります。

daemon¶

デーモンプロセスであるかのフラグであり、ブール値です。この属性は start() が呼び出される前に設定されている必要があります。

初期値は作成するプロセスから継承します。

あるプロセスが終了するとき、そのプロセスはその子プロセスであるデーモンプロセスすべてを終了させようとします。

デーモンプロセスは子プロセスを作成できないことに注意してください。もし作成できてしまうと、そのデーモンプロセスの親プロセスが終了したときにデーモンプロセスの子プロセスが孤児になってしまう場合があるからです。さらに言えば、デーモンプロセスはUnix デーモンやサービスではなく通常のプロセスであり、非デーモンプロセスが終了すると終了されます (そして join されません)。

threading.Thread クラスの API に加えて Process クラスのオブジェクトには以下の属性およびメソッドがあります:

pid¶: プロセスIDを返します。プロセスの生成前は None が設定されています。

exitcode¶

子プロセスの終了コードです。プロセスがまだ終了していない場合は None です。

子プロセスの run() メソッドが正常に戻った場合、終了コードは 0 です。整数の引数 N で sys.exit() を呼び出して終了した場合、終了コードは N です。

run() 内で捕捉されなかった例外によって子プロセスが終了した場合、終了コードは 1 です。シグナル N によって終了した場合は、負の値 -N になります。

authkey¶

プロセスの認証キーです (バイト文字列です)。

multiprocessing の初期化時に、os.urandom() を使ってメインプロセスにランダムなバイト列を割り当てます。

Process クラスのオブジェクトの作成時にその親プロセスから認証キーを継承します。もしくは authkey に別のバイト文字列を設定することもできます。

詳細は認証キーを参照してください。

sentinel¶

プロセスが終了するときに "ready" となるシステムオブジェクトの数値ハンドル。

multiprocessing.connection.wait() で複数のイベントを同時に待ちたい場合は、この値を使えます。それ以外では join() を呼び出す方が簡単です。

Windows では、WaitForSingleObject および WaitForMultipleObjects 系の API 呼び出しに使える OS ハンドルです。 POSIX では、select モジュールのプリミティブで使えるファイル記述子です。

Added in version 3.3.

interrupt()¶

プロセスを終了させます。 POSIX では SIGINT シグナルを使います。 Windows での動作は未定義です。

デフォルトでは、KeyboardInterrupt を送出して子プロセスを終了させます。子プロセスで signal.signal() を使って SIGINT のシグナルハンドラーを設定すると、この動作を変更できます。

注意: 子プロセスが KeyboardInterrupt を捕捉して無視する場合、プロセスは終了しません。

注意: デフォルトの動作では、子プロセスで未捕捉の例外が送出された場合と同じように、exitcode が 1 になります。別の exitcode にするには、KeyboardInterrupt を捕捉して exit(your_code) を呼び出せばよいでしょう。

Added in version 3.14.

terminate()¶

プロセスを終了させます。 POSIX では SIGTERM シグナルを、Windows では TerminateProcess() を使います。終了ハンドラーや finally 節などは実行されません。

このメソッドにより終了するプロセスの子孫プロセスは、終了 しません 。そういった子孫プロセスは単純に孤児になります。

警告

このメソッドの使用時に、関連付けられたプロセスがパイプやキューを使用している場合には、使用中のパイプやキューが破損して他のプロセスから使用できなくなる可能性があります。同様に、プロセスがロックやセマフォなどを取得している場合には、このプロセスが終了してしまうと他のプロセスのデッドロックの原因になるでしょう。

kill()¶: terminate() と同じですが、POSIX では SIGKILL シグナルを使います。

Added in version 3.7.

close()¶: Process オブジェクトを閉じ、関連付けられていたすべてのリソースを開放します。中のプロセスが実行中であった場合、ValueError を送出します。close() が成功した場合、Process オブジェクトの他のメソッドや属性は、ほとんどが ValueError を送出します。

Added in version 3.7.

start()、join()、is_alive()、terminate() の呼び出しと exitcode の参照は、そのプロセスオブジェクトを作成したプロセスだけが行うべきです。

以下の例では Process のメソッドの使い方を示しています:

>>> import multiprocessing, time, signal
>>> mp_context = multiprocessing.get_context('spawn')
>>> p = mp_context.Process(target=time.sleep, args=(1000,))
>>> print(p, p.is_alive())
<...Process ... initial> False
>>> p.start()
>>> print(p, p.is_alive())
<...Process ... started> True
>>> p.terminate()
>>> time.sleep(0.1)
>>> print(p, p.is_alive())
<...Process ... stopped exitcode=-SIGTERM> False
>>> p.exitcode == -signal.SIGTERM
True

exception multiprocessing.ProcessError¶: multiprocessing のすべての例外の基底クラスです。

exception multiprocessing.BufferTooShort¶

読み込んだメッセージに対して、渡されたバッファーオブジェクトが小さすぎる場合に Connection.recv_bytes_into() が送出する例外です。

e が BufferTooShort のインスタンスであるとすると、 e.args[0] はそのメッセージをバイト文字列で与えるものです。

exception multiprocessing.AuthenticationError¶: 認証エラーがあった場合に送出されます。

exception multiprocessing.TimeoutError¶: タイムアウトをサポートするメソッドでタイムアウトが過ぎたときに送出されます。

パイプ (Pipe) とキュー (Queue)¶

複数のプロセスを使う場合、一般的にはメッセージパッシングをプロセス間通信に使用し、ロックのような同期プリミティブを使用しないようにします。

メッセージのやりとりのために Pipe() (2つのプロセス間の通信用)、もしくはキュー (複数のメッセージ生成プロセス (producer)、消費プロセス (consumer) の実現用) を使うことができます。

Queue, SimpleQueue と JoinableQueue 型は複数プロセスから生成/消費を行う FIFO キューです。これらのキューは標準ライブラリの queue.Queue を模倣しています。 Queue には Python 2.5 の queue.Queue クラスで導入された task_done() と join() メソッドがないことが違う点です。

もし JoinableQueue を使用するなら、キューから削除される各タスクのために JoinableQueue.task_done() を呼び出さなければ なりません 。さもないと、いつか完了していないタスクを数えるためのセマフォがオーバーフローし、例外を発生させるでしょう。

Python の他のキュー実装と異なり、multiprocessing のキューは、追加されたすべてのオブジェクトを pickle でシリアライズします。 get() メソッドが返すのは再構築されたオブジェクトであり、元のオブジェクトとメモリを共有しません。

管理オブジェクトを使用することで共有キューを作成できることも覚えておいてください。詳細はマネージャーを参照してください。

注釈

multiprocessing は、通常の queue.Empty 例外と queue.Full 例外でタイムアウトを通知します。これらは multiprocessing 名前空間では利用できないため、queue からインポートする必要があります。

注釈

オブジェクトをキューに追加すると、そのオブジェクトは pickle 化されます。その後、バックグラウンドのスレッドが pickle 化されたデータを下層のパイプにフラッシュします。このため、少し意外な動作をすることがありますが、通常は実用上の問題にはなりません。問題になる場合は、代わりにマネージャーで作成したキューを使えます。

空のキューにオブジェクトを追加してから、キューの empty() メソッドが False を返し、get_nowait() が queue.Empty を送出せずにオブジェクトを返せるようになるまで、ごくわずかな遅延が生じることがあります。
複数のプロセスがオブジェクトをキューに詰めている場合、キューの反対側ではオブジェクトが詰められたのとは違う順序で取得される可能性があります。ただし、同一のプロセスから詰め込まれたオブジェクトは、それらのオブジェクト間では、必ず期待どおりの順序になります。

警告

Queue を利用しようとしている最中にプロセスを Process.terminate() や os.kill() で終了させる場合、キューにあるデータは破損し易くなります。終了した後で他のプロセスがキューを利用しようとすると、例外を発生させる可能性があります。

警告

上述したように、もし子プロセスがキューへ要素を追加するなら (かつ JoinableQueue.cancel_join_thread を使用しないなら) そのプロセスはバッファーされたすべての要素がパイプへフラッシュされるまで終了しません。

これは、そのプロセスを join しようとする場合、キューに追加されたすべての要素が消費されたことが確実でないかぎり、デッドロックを発生させる可能性があることを意味します。似たような現象で、子プロセスが非デーモンプロセスの場合、親プロセスは終了時に非デーモンのすべての子プロセスを join しようとしてハングアップする可能性があります。

マネージャーを使用して作成されたキューではこの問題はありません。詳細はプログラミングガイドラインを参照してください。

プロセス間通信におけるキューの使用例を知りたいなら使用例を参照してください。

multiprocessing.Pipe(duplex=True)¶

パイプの両端を表す Connection オブジェクトのペア (conn1, conn2) を返します。

duplex が True (デフォルト) ならパイプは双方向性です。duplex が False ならパイプは一方向性で、conn1 はメッセージの受信専用、conn2 はメッセージの送信専用になります。

send() メソッドは pickle でオブジェクトをシリアライズし、recv() メソッドはオブジェクトを再構築します。

class multiprocessing.Queue([maxsize])¶

パイプといくつかのロックやセマフォで実装された、プロセス間で共有するキューを返します。あるプロセスが初めてキューに要素を入れると、バッファーからパイプへオブジェクトを転送する供給スレッドが開始されます。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

標準ライブラリの queue モジュールの通常の queue.Empty や queue.Full 例外がタイムアウトを伝えるために送出されます。

Queue は、task_done()、join()、shutdown() を除く queue.Queue のすべてのメソッドを実装しています。

qsize()¶

おおよそのキューのサイズを返します。マルチスレッディング/マルチプロセスの特性上、この数値は信用できません。

macOS など sem_getvalue() が実装されていないプラットフォームでは、NotImplementedError を送出する場合があります。

empty()¶

キューが空っぽなら True を、そうでなければ False を返します。マルチスレッディング/マルチプロセシングの特性上、これは信用できません。

閉じたキューでは OSError を送出する場合があります (保証はされません)。

full()¶: キューがいっぱいなら True を、そうでなければ False を返します。マルチスレッディング/マルチプロセシングの特性上、これは信用できません。

put(obj[, block[, timeout]])¶: キューの中へ obj を追加します。オプションの引数 block が True (デフォルト) 且つ timeout が None (デフォルト) なら、空きスロットが利用可能になるまで必要であればブロックします。 timeout が正の数なら、最大 timeout 秒ブロックして、その時間内に空きスロットが利用できなかったら queue.Full 例外を発生させます。それ以外 (block が False) で、空きスロットがすぐに利用可能な場合はキューに要素を追加します。そうでなければ queue.Full 例外が発生します(その場合 timeout は無視されます)。

バージョン 3.8 で変更: キューが閉じられている場合、AssertionError の代わりに ValueError を送出するようになりました。

put_nowait(obj)¶: put(obj, False) と等価です。

get([block[, timeout]])¶: キューから要素を取り出して削除します。オプションの引数 block が True (デフォルト) 且つ timeout が None (デフォルト) なら、要素が取り出せるまで必要であればブロックします。 timeout が正の数なら、最大 timeout 秒ブロックして、その時間内に要素が取り出せなかったら queue.Empty 例外を発生させます。それ以外 (block が False) で、要素がすぐに取り出せる場合は要素を返します。そうでなければ queue.Empty 例外が発生します(その場合 timeout は無視されます)。

バージョン 3.8 で変更: キューが閉じられている場合、OSError の代わりに ValueError を送出するようになりました。

get_nowait()¶: get(False) と等価です。

multiprocessing.Queue は queue.Queue にはない追加メソッドがあります。これらのメソッドは通常、ほとんどのコードに必要ありません:

close()¶

キューを閉じ、内部リソースを解放します。

閉じたキューをそれ以降使ってはいけません。例えば、get()、put()、empty() メソッドを呼び出してはいけません。

バックグラウンドスレッドは、バッファー内のデータをすべてパイプに書き出すと終了します。このメソッドは、キューがガベージコレクションされるときに自動で呼び出されます。

join_thread()¶

バックグラウンドスレッドを join します。このメソッドは close() が呼び出された後でのみ使用されます。バッファーされたすべてのデータがパイプへフラッシュされるのを保証するため、バックグラウンドスレッドが終了するまでブロックします。

デフォルトでは、あるプロセスがキューを作成していない場合、終了時にキューのバックグラウンドスレッドを join しようとします。そのプロセスは join_thread() が何もしないように cancel_join_thread() を呼び出すことができます。

cancel_join_thread()¶

join_thread() がブロッキングするのを防ぎます。特にこれはバックグラウンドスレッドがそのプロセスの終了時に自動的に join されるのを防ぎます。詳細は join_thread() を参照してください。

このメソッドは allow_exit_without_flush() という名前のほうがよかったかもしれません。キューに追加されたデータが失われてしまいがちなため、このメソッドを使う必要はほぼ確実にないでしょう。本当にこれが必要になるのは、キューに追加されたデータを下位層のパイプにフラッシュすることなくカレントプロセスを直ちに終了する必要があり、かつ失われるデータに関心がない場合です。

注釈

このクラスに含まれる機能には、ホストとなるオペレーティングシステム上で動作している共有セマフォ (shared semaphore) を使用しているものがあります。これが使用できない場合には、このクラスが無効になり、 Queue をインスタンス化する時に ImportError が発生します。詳細は bpo-3770 を参照してください。同様のことが、以下に列挙されている特殊なキューでも成り立ちます。

class multiprocessing.SimpleQueue¶

単純化された Queue 型です。ロックされた Pipe と非常に似ています。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

close()¶

キューを閉じ、内部リソースを解放します。

閉じたキューをそれ以降使ってはいけません。例えば、get()、put()、empty() メソッドを呼び出してはいけません。

Added in version 3.9.

empty()¶

キューが空ならば True を、そうでなければ False を返します。

SimpleQueue が閉じられている場合は、常に OSError を送出します。

get()¶: キューから要素を削除して返します。

put(item)¶: item をキューに追加します。

class multiprocessing.JoinableQueue([maxsize])¶

JoinableQueue は Queue のサブクラスであり、 task_done() や join() メソッドが追加されているキューです。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

task_done()¶

以前にキューへ追加されたタスクが完了したことを表します。キューのコンシューマによって使用されます。タスクをフェッチするために使用されるそれぞれの get() に対して、後続の task_done() 呼び出しはタスクの処理が完了したことをキューへ伝えます。

join() が現在ブロックしている場合、すべての要素が処理されると復帰します。これは、キューに put() されたすべての要素について task_done() の呼び出しを受け取ったことを意味します。

キューに追加された要素の総数を超えて呼び出すと ValueError を送出します。

join()¶

キューにあるすべてのアイテムが取り出されて処理されるまでブロックします。

キューに要素が追加されると未完了タスク数が増えます。コンシューマがキューの要素が取り出されてすべての処理が完了したことを表す task_done() を呼び出すと数が減ります。未完了タスク数がゼロになると join() はブロッキングを解除します。

その他¶

multiprocessing.active_children()¶

カレントプロセスのすべてのアクティブな子プロセスのリストを返します。

この関数を呼び出すと、副作用として終了済みのプロセスがすべて "join" されます。

multiprocessing.cpu_count()¶

システムの CPU 数を返します。

この数は、現在のプロセスが利用できる CPU 数とは異なります。利用可能な CPU 数は、os.process_cpu_count() (または len(os.sched_getaffinity(0))) で取得できます。

CPU 数を調べられない場合は NotImplementedError を送出します。

参考

os.cpu_count() os.process_cpu_count()

バージョン 3.13 で変更: これは os の CPU 数取得 API のラッパーにすぎないため、-X cpu_count オプションまたは PYTHON_CPU_COUNT で戻り値を上書きすることもできます。

multiprocessing.current_process()¶

カレントプロセスに対応する Process オブジェクトを返します。

threading.current_thread() とよく似た関数です。

multiprocessing.parent_process()¶: current_process() が返すプロセスの親に対応する Process オブジェクトを返します。メインプロセスでは None を返します。

Added in version 3.8.

multiprocessing.freeze_support()¶

multiprocessing を使うプログラムを凍結して実行ファイルにした場合のサポートを追加します。 (py2exe、PyInstaller、cx_Freeze でテストされています。)

メインモジュールの if __name__ == '__main__' の直後にこの関数を呼び出す必要があります。以下に例を示します:

from multiprocessing import Process, freeze_support

def f():
    print('hello world!')

if __name__ == '__main__':
    freeze_support()
    Process(target=f).start()

もし freeze_support() の行がない場合、フリーズされた実行可能形式を実行しようとすると RuntimeError を発生させます。

開始方式が spawn でない場合、freeze_support() の呼び出しは何もしません。また、モジュールを Python インタープリタで通常どおり実行している場合 (プログラムが凍結されていない場合) も、freeze_support() は何もしません。

multiprocessing.get_all_start_methods()¶: サポートされている開始方式のリストを返します。最初の要素がデフォルトの方式です。開始方式には 'fork'、'spawn'、'forkserver' があります。すべてのプラットフォームがすべての方式をサポートしているわけではありません。コンテキストと開始方式を参照してください。

Added in version 3.4.

multiprocessing.get_context(method=None)¶

multiprocessing モジュールと同じ属性を持つコンテキストオブジェクトを返します。

method が None の場合は、デフォルトのコンテキストを返します。グローバルな開始方式がまだ設定されていなければ、システムのデフォルトに設定します。詳しくはグローバルな開始方式を参照してください。それ以外の場合、method は 'fork'、'spawn'、'forkserver' のいずれかでなければなりません。指定された開始方式が利用できない場合は ValueError を送出します。コンテキストと開始方式を参照してください。

Added in version 3.4.

multiprocessing.get_start_method(allow_none=False)¶

開始するプロセスで使用する開始方式名を返します。

グローバルな開始方式が未設定で、allow_none が False の場合は、開始方式をデフォルトに設定してその名前を返します。詳しくはグローバルな開始方式を参照してください。

戻り値は 'fork'、'spawn'、'forkserver'、None のいずれかです。コンテキストと開始方式を参照してください。

Added in version 3.4.

バージョン 3.8 で変更: macOS では、 spawn 開始方式がデフォルトになりました。 fork 開始方法は、サブプロセスのクラッシュを引き起こす可能性があるため、安全ではありません。 bpo-33725 を参照。

multiprocessing.set_executable(executable)¶

子プロセスの開始に使う Python インタープリタのパスを設定します (デフォルトでは sys.executable を使います)。 Python を組み込む場合は、次のような設定が必要になるでしょう。

set_executable(os.path.join(sys.exec_prefix, 'pythonw.exe'))

子プロセスを作成する前に行ってください。

バージョン 3.4 で変更: 'spawn' 開始方式を使う場合、POSIX でもサポートするようになりました。

バージョン 3.11 で変更: path-like object を受け入れるようになりました。

multiprocessing.set_forkserver_preload(module_names, *, on_error='ignore')¶

フォークサーバーのメインプロセスがインポートを試みるモジュール名のリストを設定し、インポート済みの状態をフォークしたプロセスに引き継がせます。各プロセスで同じ処理を繰り返さずに済むため、性能の改善に利用できます。

これを有効にするには、フォークサーバープロセスが起動する前 (Pool を作成したり Process を開始したりする前) に呼び出す必要があります。

on_error 引数は、モジュールの事前読み込み中に起きた ImportError の扱いを指定します。 "ignore" (デフォルト) は失敗を黙って無視します。 "warn" では、フォークサーバーのサブプロセスが ImportWarning を stderr に出力します。 "fail" では、フォークサーバーのサブプロセスが例外のトレースバックを stderr に出力して終了し、以降のプロセス作成が EOFError または ConnectionError で失敗します。

'forkserver' 開始方式を使う場合だけ意味を持ちます。コンテキストと開始方式を参照してください。

Added in version 3.4.

バージョン 3.15 で変更: on_error 引数を追加しました。

multiprocessing.set_start_method(method, force=False)¶

子プロセスの開始に使う方式を設定します。 method 引数には 'fork'、'spawn'、'forkserver' を指定できます。開始方式がすでに設定されていて、force が True でなければ RuntimeError を送出します。 method が None、force が True の場合は、開始方式を None に設定します。 method が None、force が False の場合は、デフォルトのコンテキストに設定します。

これは一度しか呼び出すことができず、その場所もメインモジュールの if __name__ == '__main__' 節内で保護された状態でなければなりません。

コンテキストと開始方式を参照してください。

Added in version 3.4.

注釈

multiprocessing には、threading.active_count()、threading.enumerate()、threading.settrace()、threading.setprofile()、threading.Timer、threading.local に相当するものはありません。

Connection オブジェクト¶

Connection オブジェクトは pickle でシリアライズ可能なオブジェクトか文字列を送ったり、受け取ったりします。そういったオブジェクトはメッセージ指向の接続ソケットと考えられます。

Connection オブジェクトは通常は Pipe を使用して作成されます。詳細はリスナーとクライアントも参照してください。

class multiprocessing.connection.Connection¶

send(obj)¶

コネクションの相手側へ recv() を使用して読み込むオブジェクトを送ります。

オブジェクトは pickle でシリアライズ可能でなければなりません。 pickle が極端に大きすぎる (OS にも依りますが、およそ 32 MiB+) と、 ValueError 例外が送出されることがあります。

recv()¶: コネクションの相手側から send() を使用して送られたオブジェクトを返します。何か受け取るまでブロックします。受信するデータが残っておらず、コネクションの相手側がクローズされている場合 EOFError が発生します。

fileno()¶: コネクションが使用するファイル記述子またはハンドルを返します。

close()¶

コネクションをクローズします。

コネクションがガベージコレクトされるときに自動的に呼び出されます。

poll([timeout])¶

読み込み可能なデータがあるかどうかを返します。

timeout が指定されていなければすぐに返します。 timeout に数値を指定すると、最大指定した秒数をブロッキングします。 timeout に None を指定するとタイムアウトせずにずっとブロッキングします。

multiprocessing.connection.wait() を使って複数のコネクションオブジェクトを同時にポーリングできることに注意してください。

send_bytes(buf[, offset[, size]])¶

bytes-like object から完全なメッセージとしてバイトデータを送ります。

offset を指定すると、buf のその位置からデータを読み込みます。 size を指定すると、buf からそのバイト数だけ読み込みます。非常に大きいバッファー (OS にもよりますが、およそ 32 MiB 以上) では ValueError を送出する場合があります。

recv_bytes([maxlength])¶

接続の相手側から送られたバイトデータの完全なメッセージを、バイト列として返します。受信するものがあるまでブロックします。受信するデータが残っておらず、相手側が接続を閉じた場合は EOFError を送出します。

maxlength を指定していて、かつメッセージが maxlength より長い場合、 OSError が発生してコネクションからそれ以上読めなくなります。

バージョン 3.3 で変更: この関数は以前は IOError を送出していました。今では OSError の別名です。

recv_bytes_into(buf[, offset])¶

接続の相手側から送られたバイトデータの完全なメッセージを buf に読み込み、メッセージのバイト数を返します。受信するものがあるまでブロックします。受信するデータが残っておらず、相手側が接続を閉じた場合は EOFError を送出します。

buf は書き込み可能な bytes-like object でなければなりません。 offset を指定すると、バッファーのその位置からメッセージを書き込みます。 offset は、buf のバイト単位の長さより小さい、非負の整数でなければなりません。

バッファーがあまりに小さいと BufferTooShort 例外が発生します。 e が例外インスタンスとすると完全なメッセージは e.args[0] で確認できます。

バージョン 3.3 で変更: Connection.send() と Connection.recv() を使用して Connection オブジェクト自体をプロセス間で転送できるようになりました。

Connection オブジェクトがコンテキスト管理プロトコルもサポートするようになりました (コンテキストマネージャ型を参照)。 __enter__() は接続オブジェクトを返し、__exit__() は close() を呼び出します。

例えば:

>>> from multiprocessing import Pipe
>>> a, b = Pipe()
>>> a.send([1, 'hello', None])
>>> b.recv()
[1, 'hello', None]
>>> b.send_bytes(b'thank you')
>>> a.recv_bytes()
b'thank you'
>>> import array
>>> arr1 = array.array('i', range(5))
>>> arr2 = array.array('i', [0] * 10)
>>> a.send_bytes(arr1)
>>> count = b.recv_bytes_into(arr2)
>>> assert count == len(arr1) * arr1.itemsize
>>> arr2
array('i', [0, 1, 2, 3, 4, 0, 0, 0, 0, 0])

警告

Connection.recv() メソッドは受信したデータを自動的に unpickle 化します。それはメッセージを送ったプロセスが信頼できる場合を除いてセキュリティリスクになります。

そのため、Pipe() で生成したコネクションオブジェクトを除き、recv() と send() は何らかの認証を済ませてから使うべきです。認証キーを参照してください。

警告

もしプロセスがパイプの読み込みまたは書き込み中に kill されると、メッセージの境界がどこなのか分からなくなってしまうので、そのパイプ内のデータは破損してしまいがちです。

同期プリミティブ¶

一般的にマルチプロセスプログラムは、マルチスレッドプログラムほどは同期プリミティブを必要としません。詳細は threading モジュールのドキュメントを参照してください。

マネージャーオブジェクトを使用して同期プリミティブを作成できることも覚えておいてください。詳細はマネージャーを参照してください。

class multiprocessing.Barrier(parties[, action[, timeout]])¶

バリアーオブジェクト: threading.Barrier のクローンです。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

Added in version 3.3.

class multiprocessing.BoundedSemaphore([value])¶

有限セマフォオブジェクト: threading.BoundedSemaphore の類似物です。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

よく似た threading.BoundedSemaphore とは、次の一点だけ異なります。 acquire メソッドの第一引数名は block で、Lock.acquire() と一致しています。

locked()¶: このオブジェクトが現在ロックされているかどうかを真偽値で返します。

Added in version 3.14.

注釈

macOS では sem_getvalue() が実装されていないので Semaphore と区別がつきません。

class multiprocessing.Condition([lock])¶

条件変数: threading.Condition の別名です。

lock を指定する場合は、multiprocessing の Lock オブジェクトまたは RLock オブジェクトでなければなりません。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

バージョン 3.3 で変更: wait_for() メソッドが追加されました。

class multiprocessing.Event¶

threading.Event のクローンです。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

class multiprocessing.Lock¶

再帰しないロックオブジェクトで、 threading.Lock 相当のものです。プロセスやスレッドがいったんロックを獲得 (acquire) すると、その後は、どのプロセスやスレッドから獲得しようとしても、ロックが解放 (release) されるまでブロックします。解放はどのプロセスやスレッドからも行えます。特に明記した点を除き、threading.Lock がスレッドに適用する概念と動作を、multiprocessing.Lock はプロセスとスレッドの両方に適用します。

Lock は実際にはファクトリ関数で、デフォルトコンテキストで初期化された multiprocessing.synchronize.Lock のインスタンスを返すことに注意してください。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

Lock は context manager プロトコルをサポートしています。つまり with 文で使うことができます。

acquire(block=True, timeout=None)¶

ブロックあり、またはブロックなしでロックを獲得します。

引数 block を True (デフォルト) に設定して呼び出した場合、ロックがアンロック状態になるまでブロックします。ブロックから抜けるとそれをロック状態にしてから True を返します。 threading.Lock.acquire() の最初の引数とは名前が違っているので注意してください。

引数 block の値を False にして呼び出すとブロックしません。現在ロック状態であれば、直ちに False を返します。それ以外の場合には、ロックをロック状態にして True を返します。

timeout として正の浮動小数点数を与えて呼び出すと、ロックが獲得できない限り、指定された秒数だけブロックします。 timeout 値に負数を与えると、ゼロを与えた場合と同じになります。 timeout 値の None (デフォルト) を与えると、無限にブロックします。 timeout 引数の負数と None の扱いは、 threading.Lock.acquire() に実装された動作と異なるので注意してください。 block が False の場合、 timeout は実際的な意味を持たなくなるので無視されます。ロックを獲得した場合は True 、タイムアウトした場合は False で戻ります。

release()¶

ロックを解放します。これはロックを獲得したプロセスやスレッドだけでなく、任意のプロセスやスレッドから呼ぶことができます。

threading.Lock.release() と同じように振舞いますが、ロックされていない場合に呼び出すと ValueError となる点だけが違います。

locked()¶: このオブジェクトが現在ロックされているかどうかを真偽値で返します。

Added in version 3.14.

class multiprocessing.RLock¶

再帰ロックオブジェクトで、 threading.RLock 相当のものです。再帰ロックオブジェクトはそれを獲得 (acquire) したプロセスやスレッドが解放 (release) しなければなりません。プロセスやスレッドがロックをいったん獲得すると、同じプロセスやスレッドはブロックされずに再度獲得出来ます。そのプロセスやスレッドは獲得した回数ぶん解放しなければなりません。

RLock は実際にはファクトリ関数で、デフォルトコンテキストで初期化された multiprocessing.synchronize.RLock のインスタンスを返すことに注意してください。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

RLock は context manager プロトコルをサポートしています。つまり with 文で使うことができます。

acquire(block=True, timeout=None)¶

ブロックあり、またはブロックなしでロックを獲得します。

block 引数を True にして呼び出した場合、現在のプロセスまたはスレッドが既にロックを所有している場合を除き、アンロック状態 (どのプロセス、スレッドも所有していない状態) になるまでブロックします。ブロックから抜けるとカレントプロセスもしくはカレントスレッドが (既に持っていなければ) 所有権を得て、再帰レベルをインクリメントし、 True で戻ります。 threading.RLock.acquire() の実装とはこの最初の引数の振る舞いが、その名前自身を始めとしていくつか違うので注意してください。

block 引数を False にして呼び出した場合、ブロックしません。ロックが他のプロセスもしくはスレッドにより獲得済み (つまり所有されている) であれば、カレントプロセスまたはカレントスレッドは所有権を得ず、再帰レベルも変更せずに、 False で戻ります。ロックがアンロック状態の場合、カレントプロセスもしくはカレントスレッドは所有権を得て再帰レベルがインクリメントされ、 True で戻ります。(---訳注: block の True/False 関係なくここでの説明では「所有権を持っている場合の2度目以降の aquire」の説明が欠けています。2度目以降の acquire では再帰レベルがインクリメントされて即座に返ります。全体読めばわかるとは思いますが一応。---)

timeout 引数の使い方と振る舞いは Lock.acquire() と同じです。 timeout 引数の振る舞いがいくつかの点で threading.RLock.acquire() と異なるので注意してください。

release()¶

再帰レベルをデクリメントしてロックを解放します。デクリメント後に再帰レベルがゼロになった場合、ロックの状態をアンロック (いかなるプロセス、いかなるスレッドにも所有されていない状態) にリセットし、ロックの状態がアンロックになるのを待ってブロックしているプロセスもしくはスレッドがある場合にはその中のただ一つだけが処理を進行できるようにします。デクリメント後も再帰レベルがゼロでない場合、ロックの状態はロックのままで、呼び出し側のプロセスもしくはスレッドに所有されたままになります。

このメソッドは呼び出しプロセスあるいはスレッドがロックを所有している場合に限り呼び出してください。所有者でないプロセスもしくはスレッドによって呼ばれるか、あるいはアンロック (未所有) 状態で呼ばれた場合、 AssertionError が送出されます。同じ状況での threading.RLock.release() 実装とは例外の型が異なるので注意してください。

locked()¶: このオブジェクトが現在ロックされているかどうかを真偽値で返します。

Added in version 3.14.

class multiprocessing.Semaphore([value])¶

セマフォオブジェクト: threading.Semaphore のクローンです。

このクラスのインスタンス化によって、グローバルな開始方式が設定される場合があります。詳しくはグローバルな開始方式を参照してください。

よく似た threading.BoundedSemaphore とは、次の一点だけ異なります。 acquire メソッドの第一引数名は block で、Lock.acquire() と一致しています。

get_value()¶

セマフォの現在の値を返します。

macOS など sem_getvalue() が実装されていないプラットフォームでは、NotImplementedError を送出する場合があります。

locked()¶: このオブジェクトが現在ロックされているかどうかを真偽値で返します。

Added in version 3.14.

注釈

macOS では sem_timedwait がサポートされていないので、acquire() にタイムアウトを与えて呼ぶと、ループ内でスリープすることでこの関数がエミュレートされます。

注釈

このパッケージに含まれる機能には、ホストとなるオペレーティングシステム上で動作している共有セマフォを使用しているものがあります。これが使用できない場合には、multiprocessing.synchronize モジュールが無効になり、このモジュールのインポート時に ImportError が発生します。詳細は bpo-3770 を参照してください。

共有 `ctypes` オブジェクト¶

子プロセスにより継承される共有メモリを使用する共有オブジェクトを作成することができます。

multiprocessing.Value(typecode_or_type, *args, lock=True)¶

共有メモリから割り当てられた ctypes オブジェクトを返します。デフォルトでは、返り値は実際のオブジェクトの同期ラッパーです。オブジェクトそれ自身は、 Value の value 属性によってアクセスできます。

typecode_or_type は返されるオブジェクトの型を決めます。それは ctypes の型か array モジュールで使用されるような1文字の型コードかのどちらか一方です。 *args は型のコンストラクターへ渡されます。

lock が True (デフォルト) なら、値へ同期アクセスするために新たに再帰的なロックオブジェクトが作成されます。 lock が Lock か RLock なら値への同期アクセスに使用されます。 lock が False なら、返されたオブジェクトへのアクセスはロックにより自動的に保護されません。そのため、必ずしも "プロセスセーフ" ではありません。

+= のような演算は、読み込みと書き込みを含むためアトミックでありません。このため、たとえばアトミックに共有の値を増加させたい場合、以下のようにするのでは不十分です

counter.value += 1

関連するロックが再帰的 (それがデフォルトです) なら、かわりに次のようにします

with counter.get_lock():
    counter.value += 1

lock はキーワード専用引数であることに注意してください。

multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)¶

共有メモリから割り当てられた ctypes 配列を返します。デフォルトでは、返り値は実際の配列の同期ラッパーです。

typecode_or_type は、返す配列の要素の型を指定します。 ctypes の型、または array モジュールで使うような 1 文字の型コードを指定しますが、'w' はサポートしません。さらに、型コード 'c' は ctypes.c_char の別名です。 size_or_initializer が整数の場合は配列の長さを指定し、配列をゼロで初期化します。それ以外の場合、size_or_initializer は配列の初期化に使うシーケンスで、その長さが配列の長さになります。

lock が True (デフォルト) なら、値へ同期アクセスするために新たなロックオブジェクトが作成されます。 lock が Lock か RLock なら値への同期アクセスに使用されます。 lock が False なら、返されたオブジェクトへのアクセスはロックにより自動的に保護されません。そのため、必ずしも "プロセスセーフ" ではありません。

lock はキーワード引数としてのみ利用可能なことに注意してください。

ctypes.c_char の配列には value 属性と raw 属性があり、どちらもバイト列の格納と取得に使えます。 raw では配列全体のサイズの bytes オブジェクトを扱えますが、value の読み出しは、多くのプログラミング言語の文字列処理と同様に、ヌルバイトに達すると終了します。

`multiprocessing.sharedctypes` モジュール¶

multiprocessing.sharedctypes モジュールは、子プロセスに引き継げる共有メモリに ctypes オブジェクトを割り当てる関数を提供します。

注釈

共有メモリにポインターを格納することは可能ではありますが、特定プロセスのアドレス空間の位置を参照するということを覚えておいてください。しかし、そのポインターは別のプロセスのコンテキストにおいて無効になる確率が高いです。そして、別のプロセスからそのポインターを逆参照しようとするとクラッシュを引き起こす可能性があります。

multiprocessing.sharedctypes.RawArray(typecode_or_type, size_or_initializer)¶

共有メモリから割り当てられた ctypes 配列を返します。

typecode_or_type は返される配列の要素の型を決めます。それは ctypes の型か array モジュールで使用されるような1文字の型コードのどちらか一方です。 size_or_initializer が整数なら、それが配列の長さになり、その配列はゼロで初期化されます。別の使用方法として size_or_initializer には配列の初期化に使用されるシーケンスを設定することもでき、その場合はシーケンスの長さが配列の長さになります。

要素を取得したり設定したりすることは潜在的に非アトミックであることに注意してください。ロックを使用して自動的に同期化されたアクセスを保証するには Array() を使用してください。

multiprocessing.sharedctypes.RawValue(typecode_or_type, *args)¶

共有メモリから割り当てられた ctypes オブジェクトを返します。

typecode_or_type は返されるオブジェクトの型を決めます。それは ctypes の型か array モジュールで使用されるような1文字の型コードかのどちらか一方です。 *args は型のコンストラクターへ渡されます。

値を取得したり設定したりすることは潜在的に非アトミックであることに注意してください。ロックを使用して自動的に同期化されたアクセスを保証するには Value() を使用してください。

ctypes.c_char の配列には、バイト列の格納と取得に使える value 属性と raw 属性があります。 ctypes のドキュメントを参照してください。

multiprocessing.sharedctypes.Array(typecode_or_type, size_or_initializer, *, lock=True, ctx=None)¶

RawArray() と同様ですが、 lock の値によっては ctypes 配列をそのまま返す代わりに、プロセスセーフな同期ラッパーが返されます。

lock が True (デフォルト) なら、値へ同期アクセスするために新たなロックオブジェクトが作成されます。 lock が Lock か RLock なら値への同期アクセスに使用されます。 lock が False なら、返されたオブジェクトへのアクセスはロックにより自動的に保護されません。そのため、必ずしも "プロセスセーフ" ではありません。

ctx はコンテキストオブジェクト、または None (現在のコンテキストを使う) です。 None の場合、この呼び出しによってグローバルな開始方式が設定されることがあります。詳しくはグローバルな開始方式を参照してください。

lock と ctx はキーワード専用引数です。

multiprocessing.sharedctypes.Value(typecode_or_type, *args, lock=True, ctx=None)¶

RawValue() と同様ですが、 lock の値によっては ctypes オブジェクトをそのまま返す代わりに、プロセスセーフな同期ラッパーが返されます。

lock が True (デフォルト) なら、値へ同期アクセスするために新たなロックオブジェクトが作成されます。 lock が Lock か RLock なら値への同期アクセスに使用されます。 lock が False なら、返されたオブジェクトへのアクセスはロックにより自動的に保護されません。そのため、必ずしも "プロセスセーフ" ではありません。

ctx はコンテキストオブジェクト、または None (現在のコンテキストを使う) です。 None の場合、この呼び出しによってグローバルな開始方式が設定されることがあります。詳しくはグローバルな開始方式を参照してください。

lock と ctx はキーワード専用引数です。

multiprocessing.sharedctypes.copy(obj)¶: ctypes オブジェクト obj のコピーを共有メモリに割り当てて返します。

multiprocessing.sharedctypes.synchronized(obj, lock=None, ctx=None)¶

同期アクセスに lock を使用する ctypes オブジェクトのためにプロセスセーフなラッパーオブジェクトを返します。 lock が None (デフォルト) なら、 multiprocessing.RLock オブジェクトが自動的に作成されます。

ctx はコンテキストオブジェクト、または None (現在のコンテキストを使う) です。 None の場合、この呼び出しによってグローバルな開始方式が設定されることがあります。詳しくはグローバルな開始方式を参照してください。

同期ラッパーがラップするオブジェクトに加えて2つのメソッドがあります。 get_obj() はラップされたオブジェクトを返します。 get_lock() は同期のために使用されるロックオブジェクトを返します。

ラッパー経由で ctypes オブジェクトにアクセスすることは raw ctypes オブジェクトへアクセスするよりずっと遅くなることに注意してください。

バージョン 3.5 で変更: synchronized オブジェクトはコンテキストマネージャプロトコルをサポートしています。

次の表は、共有メモリ上に共有 ctypes オブジェクトを作成する構文と、通常の ctypes の構文を比較したものです。表中の MyStruct は ctypes.Structure のサブクラスです。

ctypes	type を使用する sharedctypes	typecode を使用する sharedctypes
c_double(2.4)	RawValue(c_double, 2.4)	RawValue('d', 2.4)
MyStruct(4, 6)	RawValue(MyStruct, 4, 6)
(c_short * 7)()	RawArray(c_short, 7)	RawArray('h', 7)
(c_int * 3)(9, 2, 8)	RawArray(c_int, (9, 2, 8))	RawArray('i', (9, 2, 8))

以下に子プロセスが多くの ctypes オブジェクトを変更する例を紹介します:

from multiprocessing import Process, Lock
from multiprocessing.sharedctypes import Value, Array
from ctypes import Structure, c_double

class Point(Structure):
    _fields_ = [('x', c_double), ('y', c_double)]

def modify(n, x, s, A):
    n.value **= 2
    x.value **= 2
    s.value = s.value.upper()
    for a in A:
        a.x **= 2
        a.y **= 2

if __name__ == '__main__':
    lock = Lock()

    n = Value('i', 7)
    x = Value(c_double, 1.0/3.0, lock=False)
    s = Array('c', b'hello world', lock=lock)
    A = Array(Point, [(1.875,-6.25), (-5.75,2.0), (2.375,9.5)], lock=lock)

    p = Process(target=modify, args=(n, x, s, A))
    p.start()
    p.join()

    print(n.value)
    print(x.value)
    print(s.value)
    print([(a.x, a.y) for a in A])

結果は以下のように表示されます

49
0.1111111111111111
HELLO WORLD
[(3.515625, 39.0625), (33.0625, 4.0), (5.640625, 90.25)]

マネージャー¶

マネージャーは異なるプロセス間で共有されるデータの作成方法を提供します。これには別のマシン上で走るプロセス間のネットワーク越しの共有も含まれます。マネージャーオブジェクトは 共有オブジェクト を管理するサーバープロセスを制御します。他のプロセスはプロキシ経由で共有オブジェクトへアクセスすることができます。

multiprocessing.Manager()¶: プロセス間でオブジェクトを共有するための、起動済みの SyncManager オブジェクトを返します。返されるマネージャーオブジェクトは起動された子プロセスに対応し、共有オブジェクトを作成してそのプロキシを返すメソッドを持ちます。

マネージャープロセスは親プロセスが終了するか、ガベージコレクトされると停止します。マネージャークラスは multiprocessing.managers モジュールで定義されています:

class multiprocessing.managers.BaseManager(address=None, authkey=None, serializer='pickle', ctx=None, *, shutdown_timeout=1.0)¶

BaseManager オブジェクトを作成します。

作成後、start() または get_server().serve_forever() を呼び出して、マネージャーオブジェクトが、開始されたマネージャープロセスを確実に参照するようにしてください。

address はマネージャープロセスが新たなコネクションを待ち受けるアドレスです。address が None の場合、任意のアドレスが設定されます。

authkey はサーバープロセスへ接続しようとするコネクションの正当性を検証するために使用される認証キーです。authkey が None の場合 current_process().authkey が使用されます。authkey を使用する場合はバイト文字列でなければなりません。

serializer は、'pickle' (pickle でシリアライズ) または 'xmlrpclib' (xmlrpc.client でシリアライズ) でなければなりません。

ctx はコンテキストオブジェクト、または None (現在のコンテキストを使う) です。 None の場合、この呼び出しによってグローバルな開始方式が設定されることがあります。詳しくはグローバルな開始方式を参照してください。

shutdown_timeout は、shutdown() メソッドで、マネージャーが使うプロセスの終了を待つタイムアウトを秒単位で指定します。シャットダウンがタイムアウトすると、プロセスを terminate します。その処理もタイムアウトすると、プロセスを kill します。

バージョン 3.11 で変更: shutdown_timeout パラメータを追加しました。

start([initializer[, initargs]])¶: マネージャーを開始するためにサブプロセスを開始します。initializer が None でなければ、サブプロセスは開始時に initializer(*initargs) を呼び出します。

get_server()¶

マネージャーの制御下にある実際のサーバーを表す Server オブジェクトを返します。 Server オブジェクトは serve_forever() メソッドをサポートします:

>>> from multiprocessing.managers import BaseManager
>>> manager = BaseManager(address=('', 50000), authkey=b'abc')
>>> server = manager.get_server()
>>> server.serve_forever()

Server はさらに address 属性も持っています。

connect()¶

ローカルのマネージャーオブジェクトをリモートのマネージャープロセスに接続します。

>>> from multiprocessing.managers import BaseManager
>>> m = BaseManager(address=('127.0.0.1', 50000), authkey=b'abc')
>>> m.connect()

shutdown()¶

マネージャーが使用するプロセスを停止します。これはサーバープロセスを開始するために start() が使用された場合のみ有効です。

これは複数回呼び出すことができます。

register(typeid[, callable[, proxytype[, exposed[, method_to_typeid[, create_method]]]]])¶

マネージャークラスで呼び出し可能オブジェクト(callable)や型を登録するために使用されるクラスメソッドです。

typeid は特に共有オブジェクトの型を識別するために使用される "型識別子" です。これは文字列でなければなりません。

callable はこの型識別子のオブジェクトを作成するために使用される呼び出し可能オブジェクトです。マネージャーインスタンスが connect() メソッドを使ってサーバーに接続されているか、 create_method 引数が False の場合は、 None でも構いません。

proxytype はこの typeid で共有オブジェクトのプロキシを作成するために使用される BaseProxy のサブクラスです。 None の場合、プロキシクラスは自動的に作成されます。

exposed には、この typeid のプロキシが BaseProxy._callmethod() を使ってアクセスできるメソッド名のシーケンスを指定します。 exposed が None の場合は、proxytype._exposed_ があれば代わりに使います。公開するメソッドのリストを指定しなければ、共有オブジェクトのすべての「パブリックメソッド」にアクセスできます。ここでいう「パブリックメソッド」は、__call__() メソッドを持ち、かつ名前が '_' で始まらない属性を指します。

method_to_typeid は公開されるメソッドのうち、プロキシを返すメソッドの返り値の型を指定するために使用されるマッピングで、メソッド名を typeid 文字列にマップします。 (method_to_typeid が None の場合 proxytype._method_to_typeid_ が存在すれば、それが代わりに使用されます。 ) メソッド名がこのマッピングのキーではないか、マッピングが None の場合、そのメソッドによって返されるオブジェクトが値として (by value) コピーされます。

create_method は、共有オブジェクトを作成し、それに対するプロキシを返すようサーバープロセスに伝える、名前 typeid のメソッドを作成するかを決定します。デフォルトでは True です。

BaseManager インスタンスも読み出し専用属性を1つ持っています:

address¶: マネージャーが使用するアドレスです。

バージョン 3.3 で変更: マネージャーオブジェクトはコンテキストマネージメント・プロトコルをサポートします -- コンテキストマネージャ型を参照してください。 __enter__() は (まだ開始していない場合) サーバープロセスを開始してから、マネージャーオブジェクトを返します。 __exit__() は shutdown() を呼び出します。

旧バージョンでは、 __enter__() はマネージャーのサーバープロセスがまだ開始していなかった場合でもプロセスを開始しませんでした。

class multiprocessing.managers.SyncManager¶

プロセス間の同期のために使用される BaseManager のサブクラスです。 multiprocessing.Manager() はこの型のオブジェクトを返します。

そのメソッドは、プロセス間で同期するためによく使う各種データ型の Proxy オブジェクトを作成して返します。特に、共有のリストや辞書が含まれます。

Barrier(parties[, action[, timeout]])¶: 共有 threading.Barrier オブジェクトを作成して、そのプロキシを返します。

Added in version 3.3.

BoundedSemaphore([value])¶: 共有 threading.BoundedSemaphore オブジェクトを作成して、そのプロキシを返します。

Condition([lock])¶

共有 threading.Condition オブジェクトを作成して、そのプロキシを返します。

lock を指定する場合は、threading.Lock または threading.RLock オブジェクトのプロキシを渡してください。

バージョン 3.3 で変更: wait_for() メソッドが追加されました。

Event()¶: 共有 threading.Event オブジェクトを作成して、そのプロキシを返します。

Lock()¶: 共有 threading.Lock オブジェクトを作成して、そのプロキシを返します。

Namespace()¶: 共有 Namespace オブジェクトを作成して、そのプロキシを返します。

Queue([maxsize])¶: 共有 queue.Queue オブジェクトを作成して、そのプロキシを返します。

RLock()¶: 共有 threading.RLock オブジェクトを作成して、そのプロキシを返します。

Semaphore([value])¶: 共有 threading.Semaphore オブジェクトを作成して、そのプロキシを返します。

Array(typecode, sequence)¶: 配列を作成して、そのプロキシを返します。

Value(typecode, value)¶: 書き込み可能な value 属性を持つオブジェクトを作成し、そのプロキシを返します。

dict()¶
dict(mapping)
dict(sequence): 共有 dict オブジェクトを作成して、そのプロキシを返します。

list()¶
list(sequence): 共有 list オブジェクトを作成して、そのプロキシを返します。

set()¶
set(sequence)
set(mapping): 共有 set オブジェクトを作成し、そのプロキシを返します。

Added in version 3.14: set のサポートを追加しました。

バージョン 3.6 で変更: 共有オブジェクトは入れ子もできます。例えば、共有リストのような共有コンテナオブジェクトは、 SyncManager がまとめて管理し同期を取っている他の共有オブジェクトを保持できます。

class multiprocessing.managers.Namespace¶

SyncManager に登録することのできる型です。

Namespace オブジェクトにはパブリックなメソッドはありませんが、書き込み可能な属性を持ちます。そのオブジェクト表現はその属性の値を表示します。

しかし、Namespace オブジェクトのためにプロキシを使用するとき '_' が先頭に付く属性はプロキシの属性になり、参照対象の属性にはなりません:

>>> mp_context = multiprocessing.get_context('spawn')
>>> manager = mp_context.Manager()
>>> Global = manager.Namespace()
>>> Global.x = 10
>>> Global.y = 'hello'
>>> Global._z = 12.3    # これはプロキシの属性
>>> print(Global)
Namespace(x=10, y='hello')

カスタマイズされたマネージャー¶

独自のマネージャーを作成するには、BaseManager のサブクラスを作成して、マネージャークラスで呼び出し可能なオブジェクトか新たな型を登録するために register() クラスメソッドを使用します。例えば:

from multiprocessing.managers import BaseManager

class MathsClass:
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # 7 を表示
        print(maths.mul(7, 8))         # 56 を表示

リモートマネージャーを使用する¶

あるマシン上でマネージャーサーバーを実行して、他のマシンからそのサーバーを使用するクライアントを持つことができます(ファイアウォールを通過できることが前提)。

次のコマンドを実行することでリモートクライアントからアクセスを受け付ける1つの共有キューのためにサーバーを作成します:

>>> from multiprocessing.managers import BaseManager
>>> from queue import Queue
>>> queue = Queue()
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue', callable=lambda:queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

あるクライアントからサーバーへのアクセスは次のようになります:

>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.put('hello')

別のクライアントもそれを使用することができます:

>>> from multiprocessing.managers import BaseManager
>>> class QueueManager(BaseManager): pass
>>> QueueManager.register('get_queue')
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey=b'abracadabra')
>>> m.connect()
>>> queue = m.get_queue()
>>> queue.get()
'hello'

ローカルプロセスもそのキューへアクセスすることができます。クライアント上で上述のコードを使用してアクセスします:

>>> from multiprocessing import Process, Queue
>>> from multiprocessing.managers import BaseManager
>>> class Worker(Process):
...     def __init__(self, q):
...         self.q = q
...         super().__init__()
...     def run(self):
...         self.q.put('local hello')
...
>>> queue = Queue()
>>> w = Worker(queue)
>>> w.start()
>>> class QueueManager(BaseManager): pass
...
>>> QueueManager.register('get_queue', callable=lambda: queue)
>>> m = QueueManager(address=('', 50000), authkey=b'abracadabra')
>>> s = m.get_server()
>>> s.serve_forever()

Proxy オブジェクト¶

プロキシは別のプロセスで(おそらく)有効な共有オブジェクトを 参照する オブジェクトです。共有オブジェクトはプロキシの 参照対象 になるということができます。複数のプロキシオブジェクトが同じ参照対象を持つ可能性もあります。

プロキシオブジェクトはその参照対象の対応するメソッドを呼び出すメソッドを持ちます (そうは言っても、参照対象のすべてのメソッドが必ずしもプロキシ経由で利用可能なわけではありません)。この方法で、プロキシオブジェクトはまるでその参照先と同じように使えます:

>>> mp_context = multiprocessing.get_context('spawn')
>>> manager = mp_context.Manager()
>>> l = manager.list([i*i for i in range(10)])
>>> print(l)
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
>>> print(repr(l))
<ListProxy object, typeid 'list' at 0x...>
>>> l[4]
16
>>> l[2:5]
[4, 9, 16]

プロキシに str() を適用すると参照対象のオブジェクト表現を返すのに対して、 repr() を適用するとプロキシのオブジェクト表現を返すことに注意してください。

プロキシオブジェクトの重要な機能は pickle 化ができることで、これによりプロセス間での受け渡しができます。そのため、参照対象が Proxy オブジェクトを持てます。これによって管理されたリスト、辞書、その他 Proxy オブジェクトをネストできます:

>>> a = manager.list()
>>> b = manager.list()
>>> a.append(b)         # referent of a now contains referent of b
>>> print(a, b)
[<ListProxy object, typeid 'list' at ...>] []
>>> b.append('hello')
>>> print(a[0], b)
['hello'] ['hello']

同様に、辞書とリストのプロキシも他のプロキシの内部に入れてネストできます:

>>> l_outer = manager.list([ manager.dict() for i in range(2) ])
>>> d_first_inner = l_outer[0]
>>> d_first_inner['a'] = 1
>>> d_first_inner['b'] = 2
>>> l_outer[1]['c'] = 3
>>> l_outer[1]['z'] = 26
>>> print(l_outer[0])
{'a': 1, 'b': 2}
>>> print(l_outer[1])
{'c': 3, 'z': 26}

(プロキシでない) 標準の list オブジェクトや dict オブジェクトが参照対象に含まれていた場合、それらの可変な値の変更はマネージャーからは伝搬されません。というのも、プロキシには参照対象の中に含まれる値がいつ変更されたかを知る術が無いのです。しかし、コンテナプロキシに値を保存する (これはプロキシオブジェクトの __setitem__ を起動します) 場合はマネージャーを通して変更が伝搬され、その要素の変更を反映するには、変更後の値をコンテナプロキシに再代入できます。

# リストのプロキシを作り、ミュータブルなオブジェクト (辞書) を追加
lproxy = manager.list()
lproxy.append({})
# 辞書を変更
d = lproxy[0]
d['a'] = 1
d['b'] = 2
# この時点では d の変更はまだ同期されていませんが、
# 辞書を代入し直すことで、プロキシに変更が通知されます
lproxy[0] = d

この方法は、多くの用途では Proxy オブジェクトを入れ子にするより不便かもしれませんが、同期をある程度制御できることも示しています。

注釈

multiprocessing のプロキシ型は、値による比較をサポートしていません。例えば、次のようになります。

>>> manager.list([1,2,3]) == [1,2,3]
False

比較を行いたいときは参照対象のコピーを使用してください。

class multiprocessing.managers.BaseProxy¶

プロキシオブジェクトは BaseProxy のサブクラスのインスタンスです。

_callmethod(methodname[, args[, kwds]])¶

プロキシの参照対象のメソッドの実行結果を返します。

proxy がプロキシで、プロキシ内の参照対象が obj ならこの式

proxy._callmethod(methodname, args, kwds)

はこの式を評価します

getattr(obj, methodname)(*args, **kwds)

(マネージャープロセス内の)。

返される値はその呼び出し結果のコピーか、新たな共有オブジェクトに対するプロキシになります。詳細は BaseManager.register() の method_to_typeid 引数のドキュメントを参照してください。

その呼び出しによって例外が発生した場合、_callmethod() によってその例外は再送出されます。他の例外がマネージャープロセスで発生したなら、RemoteError 例外に変換されたものが _callmethod() によって送出されます。

特に methodname が公開されていない場合は例外が発生することに注意してください。

_callmethod() の使用例になります:

>>> l = manager.list(range(10))
>>> l._callmethod('__len__')
10
>>> l._callmethod('__getitem__', (slice(2, 7),)) # l[2:7] と同等
[2, 3, 4, 5, 6]
>>> l._callmethod('__getitem__', (20,))          # l[20] と同等
Traceback (most recent call last):
...
IndexError: list index out of range

_getvalue()¶

参照対象のコピーを返します。

参照対象を pickle 化できなければ例外を送出します。

__repr__()¶: プロキシオブジェクトのオブジェクト表現を返します。

__str__()¶: 参照対象のオブジェクト表現を返します。

クリーンアップ¶

プロキシオブジェクトは弱参照のコールバックを使います。これにより、プロキシがガベージコレクトされると、その参照対象を所有するマネージャーからプロキシの登録が解除されます。

共有オブジェクトはプロキシが参照しなくなったときにマネージャープロセスから削除されます。

プロセスプール¶

Pool クラスでタスクを実行するプロセスのプールを作成することができます。

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

プロセスプールオブジェクトは、ジョブを送り込めるワーカープロセスのプールを制御します。タイムアウトやコールバックのある非同期の実行をサポートし、並列 map 実装を持ちます。

processes は、使用するワーカープロセスの数です。 processes が None の場合は、os.process_cpu_count() が返す数を使います。

initializer が None ではない場合、各ワーカープロセスは開始時に initializer(*initargs) を呼び出します。

maxtasksperchild は、ワーカープロセスが exit して新たなワーカープロセスと置き替えられるまでの間に、ワーカープロセスが完了することのできるタスクの数です。この設定により未利用のリソースが解放されるようなります。デフォルトの maxtasksperchild は None で、これはワーカープロセスがプールと同じ期間だけ生き続けるということを意味します。

context で、ワーカープロセスの開始に使うコンテキストを指定できます。通常、プールは multiprocessing.Pool() 関数か、コンテキストオブジェクトの Pool() メソッドで作成します。どちらの場合も context は適切に設定されます。 None の場合、この関数を呼び出すと、グローバルな開始方式が未設定ならそれを設定する副作用があります。 get_context() 関数を参照してください。

プールオブジェクトのメソッドは、そのプールを作成したプロセスのみが呼び出すべきです。

警告

multiprocessing.pool オブジェクトは内部リソースを持ちます。他のリソースと同様、プールをコンテキストマネージャとして使うか、close() と terminate() を手動で呼び出し、適切に管理する必要があります。これを怠ると、終了処理でプロセスがハングする可能性があります。

CPython はプールのファイナライザーが呼ばれることを保証しないため、プールの破棄をガベージコレクターに任せるのは 正しくありません。詳しくは object.__del__() を参照してください。

バージョン 3.2 で変更: maxtasksperchild 引数を追加しました。

バージョン 3.4 で変更: context パラメータを追加しました。

バージョン 3.13 で変更: processes のデフォルト値に、os.cpu_count() ではなく os.process_cpu_count() を使うようになりました。

注釈

Pool 内のワーカープロセスは通常、プールの作業キューが存続する間ずっと生存します。 Apache や mod_wsgi など他のシステムでは、ワーカーが保持するリソースを解放するために、一定量の作業を終えたワーカーを終了して後始末し、代わりに新しいプロセスを生成する方法がよく使われます。 Pool の maxtasksperchild 引数により、この機能を利用できます。

apply(func[, args[, kwds]])¶: 引数 args とキーワード引数 kwds を伴って func を呼びます。結果が準備できるまでブロックします。このブロックがあるため、 apply_async() の方が並行作業により適しています。加えて、 func は、プール内の1つのワーカーだけで実行されます。

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

apply() メソッドの派生版で AsyncResult オブジェクトを返します。

callback が指定された場合、それは単一の引数を受け取る呼び出し可能オブジェクトでなければなりません。結果を返せるようになったときに callback が結果オブジェクトに対して適用されます。ただし呼び出しが失敗した場合は、代わりに error_callback が適用されます。

error_callback が指定された場合、それは単一の引数を受け取る呼び出し可能オブジェクトでなければなりません。対象の関数が失敗した場合、例外インスタンスを伴って error_callback が呼ばれます。

コールバックは直ちに完了すべきです。なぜなら、そうしなければ、結果を扱うスレッドがブロックするからです。

map(func, iterable[, chunksize])¶

map() 組み込み関数の並列版です (iterable な引数を1つだけサポートするという違いはあります。もしも複数のイテラブルを使いたいのならば starmap() を参照)。結果が出るまでブロックします。

このメソッドはイテラブルをいくつものチャンクに分割し、プロセスプールにそれぞれ独立したタスクとして送ります。(概算の) チャンクサイズは chunksize を正の整数に設定することで指定できます。

非常に長いイテラブルでは、大量のメモリを使う可能性があります。効率を上げるために、chunksize を明示的に指定して imap() または imap_unordered() を使うことを検討してください。

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

map() メソッドの派生版で AsyncResult オブジェクトを返します。

callback が指定された場合、それは単一の引数を受け取る呼び出し可能オブジェクトでなければなりません。結果を返せるようになったときに callback が結果オブジェクトに対して適用されます。ただし呼び出しが失敗した場合は、代わりに error_callback が適用されます。

error_callback が指定された場合、それは単一の引数を受け取る呼び出し可能オブジェクトでなければなりません。対象の関数が失敗した場合、例外インスタンスを伴って error_callback が呼ばれます。

コールバックは直ちに完了すべきです。なぜなら、そうしなければ、結果を扱うスレッドがブロックするからです。

imap(func, iterable[, chunksize])¶

map() のイテレータ版です。

chunksize 引数は map() メソッドで使用されるものと同じです。引数 iterable がとても長いなら chunksize に大きな値を指定して使用する方がデフォルト値の 1 を使用するよりもジョブの完了が かなり 速くなります。

また chunksize が 1 の場合 imap() メソッドが返すイテレーターの next() メソッドはオプションで timeout パラメーターを持ちます。 next(timeout) は、その結果が timeout 秒以内に返されないときに multiprocessing.TimeoutError を発生させます。

imap_unordered(func, iterable[, chunksize])¶: イテレーターが返す結果の順番が任意の順番で良いと見なされることを除けば imap() と同じです。 (ワーカープロセスが1つしかない場合のみ "正しい" 順番になることが保証されます。)

starmap(func, iterable[, chunksize])¶

iterable の要素が、引数として unpack されるイテレート可能オブジェクトであると期待される以外は、 map() と似ています。

そのため、iterable が [(1,2), (3, 4)] なら、結果は [func(1,2), func(3,4)] になります。

Added in version 3.3.

starmap_async(func, iterable[, chunksize[, callback[, error_callback]]])¶: starmap() と map_async() の組み合わせです。イテレート可能オブジェクトの iterable をイテレートして、 unpack したイテレート可能オブジェクトを伴って func を呼び出します。結果オブジェクトを返します。

Added in version 3.3.

close()¶: これ以上プールにタスクが投入されないようにします。すべてのタスクが完了すると、ワーカープロセスは終了します。

terminate()¶: 実行中の処理を完了させずにワーカープロセスをすぐに停止します。プールオブジェクトがガベージコレクトされるときに terminate() が呼び出されます。

join()¶: ワーカープロセスが終了するのを待ちます。 join() を使用する前に close() か terminate() を呼び出さなければなりません。

バージョン 3.3 で変更: Pool オブジェクトがコンテキストマネージメント・プロトコルをサポートするようになりました。 -- コンテキストマネージャ型を参照してください。 __enter__() は Pool オブジェクトを返します。また __exit__() は terminate() を呼び出します。

class multiprocessing.pool.AsyncResult¶

Pool.apply_async() や Pool.map_async() で返される結果のクラスです。

get([timeout])¶: 結果を受け取ったときに返します。 timeout が None ではなくて、その結果が timeout 秒以内に受け取れない場合 multiprocessing.TimeoutError が発生します。リモートの呼び出しが例外を発生させる場合、その例外は get() が再発生させます。

wait([timeout])¶: その結果が有効になるか timeout 秒経つまで待ちます。

ready()¶: その呼び出しが完了しているかどうかを返します。

successful()¶: その呼び出しが例外を発生させることなく完了したかどうかを返します。その結果が返せる状態でない場合 ValueError が発生します。

バージョン 3.7 で変更: 結果がまだ準備できていない場合、AssertionError の代わりに ValueError を送出するようになりました。

次の例はプールの使用例を紹介します:

from multiprocessing import Pool
import time

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(processes=4) as pool:         # 4 つのワーカープロセスを開始
        result = pool.apply_async(f, (10,)) # 1 つのプロセスで "f(10)" を非同期に評価
        print(result.get(timeout=1))        # コンピューターが非常に遅くなければ "100" を表示

        print(pool.map(f, range(10)))       # "[0, 1, 4,..., 81]" を表示

        it = pool.imap(f, range(10))
        print(next(it))                     # "0" を表示
        print(next(it))                     # "1" を表示
        print(it.next(timeout=1))           # コンピューターが非常に遅くなければ "4" を表示

        result = pool.apply_async(time.sleep, (10,))
        print(result.get(timeout=1))        # multiprocessing.TimeoutError を送出

リスナーとクライアント¶

通常、プロセス間でメッセージを渡すにはキューを使用するか Pipe() が返す Connection オブジェクトを使用します。

multiprocessing.connection モジュールを使うと、より柔軟に通信できます。基本的には、ソケットや Windows の名前付きパイプを扱う高水準のメッセージ指向 API を提供します。 hmac モジュールを使う ダイジェスト認証 や、複数の接続の同時ポーリングもサポートします。

multiprocessing.connection.deliver_challenge(connection, authkey)¶

ランダム生成したメッセージをコネクションの相手側へ送信して応答を待ちます。

その応答がキーとして authkey を使用するメッセージのダイジェストと一致する場合、コネクションの相手側へ歓迎メッセージを送信します。そうでなければ AuthenticationError を発生させます。

multiprocessing.connection.answer_challenge(connection, authkey)¶

メッセージを受信して、そのキーとして authkey を使用するメッセージのダイジェストを計算し、ダイジェストを送り返します。

歓迎メッセージを受け取れない場合 AuthenticationError が発生します。

multiprocessing.connection.Client(address[, family[, authkey]])¶

address で渡したアドレスを使用するリスナーに対してコネクションを確立しようとして Connection を返します。

コネクション種別は family 引数で決定しますが、一般的には address のフォーマットから推測できるので、これは指定されません。 (アドレスフォーマットを参照してください)

authkey を指定し、その値が None でない場合は、バイト列でなければなりません。この値を HMAC に基づくチャレンジ認証の秘密鍵として使います。 authkey が None の場合は認証しません。認証に失敗すると AuthenticationError を送出します。認証キーを参照してください。

class multiprocessing.connection.Listener([address[, family[, backlog[, authkey]]]])¶

コネクションを '待ち受ける' 束縛されたソケットか Windows の名前付きパイプのラッパーです。

address はリスナーオブジェクトの束縛されたソケットか名前付きパイプが使用するアドレスです。

注釈

'0.0.0.0' のアドレスを使用する場合、Windows 上の終点へ接続することができません。終点へ接続したい場合は '127.0.0.1' を使用すべきです。

family は使用するソケット(名前付きパイプ)の種別です。これは 'AF_INET' (TCP ソケット), 'AF_UNIX' (Unix ドメインソケット) または 'AF_PIPE' (Windows 名前付きパイプ) という文字列のどれか1つになります。これらのうち 'AF_INET' のみが利用可能であることが保証されています。 family が None の場合 address のフォーマットから推測されたものが使用されます。 address も None の場合はデフォルトが選択されます。詳細はアドレスフォーマットを参照してください。 family が 'AF_UNIX' で address が None の場合 tempfile.mkstemp() を使用して作成されたプライベートな一時ディレクトリにソケットが作成されます。

リスナーオブジェクトがソケットを使用する場合、ソケットに束縛されるときに backlog (デフォルトでは1つ) がソケットの listen() メソッドに対して渡されます。

authkey を指定し、その値が None でない場合は、バイト列でなければなりません。この値を HMAC に基づくチャレンジ認証の秘密鍵として使います。 authkey が None の場合は認証しません。認証に失敗すると AuthenticationError を送出します。認証キーを参照してください。

accept()¶: リスナーオブジェクトの名前付きパイプか束縛されたソケット上でコネクションを受け付けて Connection オブジェクトを返します。認証が失敗した場合 AuthenticationError が発生します。

close()¶: リスナーオブジェクトの名前付きパイプか束縛されたソケットをクローズします。これはリスナーがガベージコレクトされるときに自動的に呼ばれます。そうは言っても、明示的に close() を呼び出す方が望ましいです。

リスナーオブジェクトは次の読み出し専用属性を持っています:

address¶: リスナーオブジェクトが使用中のアドレスです。

last_accepted¶: 最後に受け付けた接続の接続元アドレスです。取得できない場合は None になります。

バージョン 3.3 で変更: Listener オブジェクトがコンテキストマネージメント・プロトコルをサポートするようになりました。 -- コンテキストマネージャ型を参照してください。 __enter__() はリスナーオブジェクトを返します。また __exit__() は close() を呼び出します。

multiprocessing.connection.wait(object_list, timeout=None)¶

object_list 中のオブジェクトが準備ができるまで待機します。準備ができた object_list 中のオブジェクトのリストを返します。timeout が浮動小数点なら、最大でその秒数だけ呼び出しがブロックします。timeout が None の場合、無制限の期間ブロックします。負のタイムアウトは0と等価です。

POSIX と Windows のどちらでも、object_list に含められるオブジェクトは次のとおりです。

読み取り可能な Connection オブジェクト;
接続された読み取り可能な socket.socket オブジェクト; または
Process オブジェクトの sentinel 属性。

読み取ることのできるデータがある場合、あるいは相手側の端が閉じられている場合、コネクションまたはソケットオブジェクトは準備ができています。

POSIX: wait(object_list, timeout) almost equivalent select.select(object_list, [], [], timeout). The difference is that, if select.select() is interrupted by a signal, it can raise OSError with an error number of EINTR, whereas wait() will not.

Windows: object_list の要素は、待機可能な整数ハンドル (Win32 関数 WaitForMultipleObjects() のドキュメントで定義されるもの)、またはソケットハンドルやパイプハンドルを返す fileno() メソッドを持つオブジェクトでなければなりません。 (パイプハンドルとソケットハンドル自体は、待機可能なハンドルでは ありません。)

Added in version 3.3.

例

次のサーバーコードは認証キーとして 'secret password' を使用するリスナーを作成します。このサーバーはコネクションを待ってクライアントへデータを送信します:

from multiprocessing.connection import Listener
from array import array

address = ('localhost', 6000)     # family は 'AF_INET' と推測されます

with Listener(address, authkey=b'secret password') as listener:
    with listener.accept() as conn:
        print('connection accepted from', listener.last_accepted)

        conn.send([2.25, None, 'junk', float])

        conn.send_bytes(b'hello')

        conn.send_bytes(array('i', [42, 1729]))

次のコードはサーバーへ接続して、サーバーからデータを受信します:

from multiprocessing.connection import Client
from array import array

address = ('localhost', 6000)

with Client(address, authkey=b'secret password') as conn:
    print(conn.recv())                  # => [2.25, None, 'junk', float]

    print(conn.recv_bytes())            # => 'hello'

    arr = array('i', [0, 0, 0, 0, 0])
    print(conn.recv_bytes_into(arr))    # => 8
    print(arr)                          # => array('i', [42, 1729, 0, 0, 0])

次のコードは wait() を使って複数のプロセスからのメッセージを同時に待ちます:

from multiprocessing import Process, Pipe, current_process
from multiprocessing.connection import wait

def foo(w):
    for i in range(10):
        w.send((i, current_process().name))
    w.close()

if __name__ == '__main__':
    readers = []

    for i in range(4):
        r, w = Pipe(duplex=False)
        readers.append(r)
        p = Process(target=foo, args=(w,))
        p.start()
        # ここでパイプの書き込み側を閉じ、そのハンドルを持つプロセスを
        # p だけにします。
        # これにより、p が書き込み側のハンドルを閉じると、
        # wait() が読み込み側の準備完了をすぐに通知します。
        w.close()

    while readers:
        for r in wait(readers):
            try:
                msg = r.recv()
            except EOFError:
                readers.remove(r)
            else:
                print(msg)

アドレスフォーマット¶

'AF_INET' アドレスは (hostname, port) のタプルになります。 hostname は文字列で port は整数です。
'AF_UNIX' アドレスはファイルシステム上のファイル名の文字列です。
'AF_PIPE' アドレスは r'\\.\pipe\PipeName' という形式の文字列です。 ServerName という名前のリモートコンピューター上の名前付きパイプに接続するために Client() を使用するには、代わりに r'\\ServerName\pipe\PipeName' 形式のアドレスを使用する必要があります。

デフォルトでは、2つのバックスラッシュで始まる文字列は 'AF_UNIX' よりも 'AF_PIPE' として推測されることに注意してください。

認証キー¶

Connection.recv を使用するとき、データは自動的に unpickle されて受信します。信頼できない接続元からのデータを unpickle することはセキュリティリスクがあります。そのため Listener や Client() はダイジェスト認証を提供するために hmac モジュールを使用します。

認証キーはパスワードと考えられるバイト文字列です。接続が確立すると、双方が相手に対して認証キーを知っている証拠を要求します。双方が同じキーを使っていることを証明する際に、接続を通じてキー自体を送ることは ありません。

認証が要求されているにもかかわらず認証キーが指定されていない場合 current_process().authkey の返す値が使用されます。 (詳細は Process を参照してください。 ) この値はカレントプロセスが作成する Process オブジェクトによって自動的に継承されます。これは(デフォルトでは)複数プロセスのプログラムの全プロセスが相互にコネクションを確立するときに使用される1つの認証キーを共有することを意味します。

適当な認証キーを os.urandom() を使用して生成することもできます。

この認証は、アドレスを指定して接続できる Listener と Client() の接続を保護します。 Pipe() が作成する匿名パイプや、Queue が内部で使う匿名パイプには適用しません。 multiprocessing は、同じユーザーとして動作するすべてのローカルプロセスを信頼できるものとして扱います。ほとんどの OS では、いずれにせよ同じユーザーのプロセス同士は互いのパイプのファイル記述子にアクセスできます。同じユーザーのプロセス間で隔離が必要なアプリケーションは、例えば別のユーザーアカウントやサンドボックスでワーカーを実行するなど、OS レベルで隔離を設ける必要があります。

ログ記録¶

ロギングをサポートする機能もいくつかあります。ただし、logging パッケージはプロセス間で共有するロックを使わないため、ハンドラーの種類によっては異なるプロセスからのメッセージが混ざる可能性があります。

multiprocessing.get_logger()¶

multiprocessing が使うロガーを返します。必要なら新しく作成します。

作成直後のロガーのレベルは logging.NOTSET で、デフォルトのハンドラーはありません。このロガーに送ったメッセージは、デフォルトではルートロガーに伝搬しません。

Windows では、子プロセスは親プロセスのロガーのレベルだけを継承します。それ以外のロガーのカスタマイズは継承しません。

multiprocessing.log_to_stderr(level=None)¶: この関数は get_logger() に対する呼び出しを実行しますが、 get_logger によって作成されるロガーを返すことに加えて、 '[%(levelname)s/%(processName)s] %(message)s' のフォーマットを使用して sys.stderr へ出力を送るハンドラーを追加します。level 引数を渡すことによってロガーの levelname を変更できます。

以下にロギングを有効にした例を紹介します:

>>> import multiprocessing, logging
>>> logger = multiprocessing.log_to_stderr()
>>> logger.setLevel(logging.INFO)
>>> logger.warning('doomed')
[WARNING/MainProcess] doomed
>>> m = multiprocessing.Manager()
[INFO/SyncManager-...] child process calling self.run()
[INFO/SyncManager-...] created temp directory /.../pymp-...
[INFO/SyncManager-...] manager serving at '/.../listener-...'
>>> del m
[INFO/MainProcess] sending shutdown message to manager
[INFO/SyncManager-...] manager exiting with exitcode 0

完全なロギングレベルの表については logging モジュールを参照してください。

`multiprocessing.dummy` モジュール¶

multiprocessing.dummy は multiprocessing と同じ API を提供しますが、threading モジュールのラッパーにすぎません。

特に、multiprocessing.dummy の Pool 関数は ThreadPool のインスタンスを返します。これは Pool のサブクラスで、同じメソッド呼び出しをすべてサポートしますが、ワーカープロセスの代わりにワーカースレッドのプールを使います。

class multiprocessing.pool.ThreadPool([processes[, initializer[, initargs]]])¶

ジョブを投入できるワーカースレッドのプールを管理する、スレッドプールオブジェクトです。 ThreadPool インスタンスのインターフェースは Pool インスタンスと完全に互換性があります。プールをコンテキストマネージャとして使うか、close() と terminate() を手動で呼び出し、リソースも適切に管理する必要があります。

processes は、使用するワーカースレッドの数です。 processes が None の場合は、os.process_cpu_count() が返す数を使います。

initializer が None ではない場合、各ワーカープロセスは開始時に initializer(*initargs) を呼び出します。

Pool と異なり、maxtasksperchild と context は指定できません。

注釈

ThreadPool は、プロセスのプールを中心に設計され、concurrent.futures モジュールの導入より前に作られた Pool と同じインターフェースを持ちます。そのため、スレッドを使うプールには意味のない操作も引き継いでいます。また、非同期ジョブの状態を表す独自の型 AsyncResult を使い、この型は他のライブラリでは扱えません。

一般には、concurrent.futures.ThreadPoolExecutor を使う方がよいでしょう。これは最初からスレッド向けに設計されたシンプルなインターフェースを持ち、asyncio を含む多くの他のライブラリと互換性のある concurrent.futures.Future インスタンスを返します。

プログラミングガイドライン¶

multiprocessing を使うときには、従うべきガイドラインや定石があります。

すべての開始方式について¶

以下はすべての開始方式に当てはまります。

共有状態を避ける

できるだけプロセス間で巨大なデータを移動することは避けるようにすべきです。

プロセス間の通信には、低レベルな同期プリミティブを使うのではなく、キューやパイプを使うのが良いでしょう。

pickle 化の可能性

プロキシのメソッドへの引数は、 pickle 化できるものにしてください。

プロキシのスレッドセーフ性

1 つのプロキシオブジェクトは、ロックで保護しないかぎり、2 つ以上のスレッドから使用してはいけません。

(異なるプロセスで同じプロキシを使用することは問題ではありません。)

ゾンビプロセスを join する

POSIX では、終了していても join されていないプロセスはゾンビになります。新しいプロセスを開始するたび (または active_children() を呼び出すたび) に、終了済みで未 join のプロセスはすべて join されるため、大量に残ることはないはずです。終了したプロセスの Process.is_alive を呼び出しても、そのプロセスを join します。それでも、開始したプロセスはすべて明示的に join するのがよいでしょう。

pickle/unpickle より継承する方が良い

spawn または forkserver 開始方式を使う場合、子プロセスが利用できるよう、multiprocessing の多くの型は pickle 化可能である必要があります。ただし、共有オブジェクトをパイプやキューで他のプロセスに送ることは、一般に避けるべきです。代わりに、別の場所で作られた共有リソースにアクセスする必要があるプロセスが、祖先のプロセスからそのリソースを引き継げるようにプログラムを構成してください。

プロセスの強制終了を避ける

あるプロセスを停止するために Process.terminate メソッドを使用すると、そのプロセスが現在使用されている (ロック、セマフォ、パイプやキューのような) 共有リソースを破壊したり他のプロセスから利用できない状態を引き起こし易いです。

そのため、共有リソースを使用しないプロセスでのみ Process.terminate を使用することを考慮することがおそらく最善の方法です。

キューを使用するプロセスを join する

キューに要素を追加するプロセスは、すべてのバッファーされた要素が "feeder" スレッドによって下位層のパイプに対してフィードされるまで終了を待つということを覚えておいてください。 (子プロセスはこの動作を避けるためにキューの Queue.cancel_join_thread メソッドを呼ぶことができます。)

これはキューを使用するときに、キューに追加されたすべての要素が最終的にそのプロセスが join される前に削除されていることを確認する必要があることを意味します。そうしないと、そのキューに要素が追加したプロセスの終了を保証できません。デーモンではないプロセスは自動的に join されることも覚えておいてください。

次の例はデッドロックを引き起こします:
from multiprocessing import Process, Queue

def f(q):
    q.put('X' * 1000000)

if __name__ == '__main__':
    queue = Queue()
    p = Process(target=f, args=(queue,))
    p.start()
    p.join()                    # ここでデッドロックします
    obj = queue.get()
修正するには最後の2行を入れ替えます(または単純に p.join() の行を削除します)。

明示的に子プロセスへリソースを渡す

POSIX で fork 開始方式を使う場合、子プロセスはグローバルなリソースを通じて、親プロセスが作成した共有リソースを利用できます。ただし、子プロセスのコンストラクタに引数としてオブジェクトを渡す方がよいでしょう。

Windows や他の開始方式と (将来的にでも) 互換性を持つコードにできることに加えて、これは子プロセスが実行中である限りは親プロセス内でオブジェクトがガベージコレクトされないことも保証します。これは親プロセス内でオブジェクトがガベージコレクトされたときに一部のリソースが開放されてしまう場合に重要かもしれません。

そのため、例えば
from multiprocessing import Process, Lock

def f():
    ... do something using "lock" ...

if __name__ == '__main__':
    lock = Lock()
    for i in range(10):
        Process(target=f).start()
は、次のように書き直すべきです
from multiprocessing import Process, Lock

def f(l):
    ... do something using "l" ...

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

sys.stdin を file-like オブジェクトに置き換えることに注意する

multiprocessing は、もともと無条件に次の呼び出しを行っていました。
os.close(sys.stdin.fileno())
を multiprocessing.Process._bootstrap() メソッドの中で呼び出していました --- これはプロセス内プロセス (processes-in-processes) で問題が起こしてしまいます。そこで、これは以下のように変更されました:
sys.stdin.close()
sys.stdin = open(os.open(os.devnull, os.O_RDONLY), closefd=False)
これにより、プロセス同士の衝突で不正なファイル記述子のエラーが起きるという根本的な問題は解決しましたが、 sys.stdin() を出力バッファー付きの「ファイル類似オブジェクト」に置き換えるアプリケーションでは、別の危険が生じます。複数のプロセスがこのオブジェクトの close() を呼び出すと、同じデータを何度も書き出して破損させる可能性があります。

もし file-like オブジェクトを書いて独自のキャッシュを実装するなら、キャッシュするときに常に pid を記録しておき、pid が変わったらキュッシュを捨てることで、フォークセーフにできます。例:
@property
def cache(self):
    pid = os.getpid()
    if pid != self._pid:
        self._pid = pid
        self._cache = []
    return self._cache
より詳しい情報は bpo-5155 、 bpo-5313 、 bpo-5331 を見てください

開始方式が spawn および forkserver の場合¶

fork 開始方式には当てはまらない、いくつかの追加の制限があります。

さらなる pickle 化の可能性

Process に渡す引数はすべて pickle 化可能にしてください。また、サブクラスで Process.__init__ をオーバーライドする場合は、Process.start を呼び出す時点でインスタンスを pickle 化できるようにしてください。

グローバル変数

子プロセスで実行されるコードがグローバル変数にアクセスしようとする場合、子プロセスが見るその値は Process.start が呼ばれたときの親プロセスの値と同じではない可能性があります。

しかし、単にモジュールレベルの定数であるグローバル変数なら問題にはなりません。

メインモジュールの安全なインポート

新しい Python インタープリタが、意図しない副作用 (新しいプロセスの開始など) を起こさずにメインモジュールを安全にインポートできるようにしてください。

例えば、開始方式に spawn あるいは forkserver を使用した場合に以下のモジュールを実行すると RuntimeError で失敗します:
from multiprocessing import Process

def foo():
    print('hello')

p = Process(target=foo)
p.start()
代わりに、次のように if __name__ == '__main__': を使用してプログラムの "エントリポイント" を保護すべきです:
from multiprocessing import Process, freeze_support, set_start_method

def foo():
    print('hello')

if __name__ == '__main__':
    freeze_support()
    set_start_method('spawn')
    p = Process(target=foo)
    p.start()
(プログラムをフリーズせずに通常通り実行するなら freeze_support() 行は取り除けます。)

これは新たに生成された Python インタープリターがそのモジュールを安全にインポートして、モジュールの foo() 関数を実行します。

プールまたはマネージャーがメインモジュールで作成される場合に似たような制限が適用されます。

使用例¶

カスタマイズされたマネージャーやプロキシの作成方法と使用方法を紹介します:

from multiprocessing import freeze_support
from multiprocessing.managers import BaseManager, BaseProxy
import operator

##

class Foo:
    def f(self):
        print('you called Foo.f()')
    def g(self):
        print('you called Foo.g()')
    def _h(self):
        print('you called Foo._h()')

# 簡単なジェネレータ関数
def baz():
    for i in range(10):
        yield i*i

# ジェネレータオブジェクトのプロキシ型
class GeneratorProxy(BaseProxy):
    _exposed_ = ['__next__']
    def __iter__(self):
        return self
    def __next__(self):
        return self._callmethod('__next__')

# operator モジュールを返す関数
def get_operator_module():
    return operator

##

class MyManager(BaseManager):
    pass

# Foo クラスを登録し、プロキシ経由で f() と g() にアクセスできるようにする
MyManager.register('Foo1', Foo)

# Foo クラスを登録し、プロキシ経由で g() と _h() にアクセスできるようにする
MyManager.register('Foo2', Foo, exposed=('g', '_h'))

# ジェネレータ関数 baz を登録し、GeneratorProxy でプロキシを作成する
MyManager.register('baz', baz, proxytype=GeneratorProxy)

# get_operator_module() を登録し、プロキシ経由で公開関数にアクセスできるようにする
MyManager.register('operator', get_operator_module)

##

def test():
    manager = MyManager()
    manager.start()

    print('-' * 20)

    f1 = manager.Foo1()
    f1.f()
    f1.g()
    assert not hasattr(f1, '_h')
    assert sorted(f1._exposed_) == sorted(['f', 'g'])

    print('-' * 20)

    f2 = manager.Foo2()
    f2.g()
    f2._h()
    assert not hasattr(f2, 'f')
    assert sorted(f2._exposed_) == sorted(['g', '_h'])

    print('-' * 20)

    it = manager.baz()
    for i in it:
        print('<%d>' % i, end=' ')
    print()

    print('-' * 20)

    op = manager.operator()
    print('op.add(23, 45) =', op.add(23, 45))
    print('op.pow(2, 94) =', op.pow(2, 94))
    print('op._exposed_ =', op._exposed_)

##

if __name__ == '__main__':
    freeze_support()
    test()

Pool を使用する例です:

import multiprocessing
import time
import random
import sys

#
# テストコードで使う関数
#

def calculate(func, args):
    result = func(*args)
    return '%s says that %s%s = %s' % (
        multiprocessing.current_process().name,
        func.__name__, args, result
        )

def calculatestar(args):
    return calculate(*args)

def mul(a, b):
    time.sleep(0.5 * random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5 * random.random())
    return a + b

def f(x):
    return 1.0 / (x - 5.0)

def pow3(x):
    return x ** 3

def noop(x):
    pass

#
# テストコード
#

def test():
    PROCESSES = 4
    print('Creating pool with %d processes\n' % PROCESSES)

    with multiprocessing.Pool(PROCESSES) as pool:
        #
        # テスト
        #

        TASKS = [(mul, (i, 7)) for i in range(10)] + \
                [(plus, (i, 8)) for i in range(10)]

        results = [pool.apply_async(calculate, t) for t in TASKS]
        imap_it = pool.imap(calculatestar, TASKS)
        imap_unordered_it = pool.imap_unordered(calculatestar, TASKS)

        print('Ordered results using pool.apply_async():')
        for r in results:
            print('\t', r.get())
        print()

        print('Ordered results using pool.imap():')
        for x in imap_it:
            print('\t', x)
        print()

        print('Unordered results using pool.imap_unordered():')
        for x in imap_unordered_it:
            print('\t', x)
        print()

        print('Ordered results using pool.map() --- will block till complete:')
        for x in pool.map(calculatestar, TASKS):
            print('\t', x)
        print()

        #
        # エラー処理のテスト
        #

        print('Testing error handling:')

        try:
            print(pool.apply(f, (5,)))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from pool.apply()')
        else:
            raise AssertionError('expected ZeroDivisionError')

        try:
            print(pool.map(f, list(range(10))))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from pool.map()')
        else:
            raise AssertionError('expected ZeroDivisionError')

        try:
            print(list(pool.imap(f, list(range(10)))))
        except ZeroDivisionError:
            print('\tGot ZeroDivisionError as expected from list(pool.imap())')
        else:
            raise AssertionError('expected ZeroDivisionError')

        it = pool.imap(f, list(range(10)))
        for i in range(10):
            try:
                x = next(it)
            except ZeroDivisionError:
                if i == 5:
                    pass
            except StopIteration:
                break
            else:
                if i == 5:
                    raise AssertionError('expected ZeroDivisionError')

        assert i == 9
        print('\tGot ZeroDivisionError as expected from IMapIterator.next()')
        print()

        #
        # タイムアウトのテスト
        #

        print('Testing ApplyResult.get() with timeout:', end=' ')
        res = pool.apply_async(calculate, TASKS[0])
        while 1:
            sys.stdout.flush()
            try:
                sys.stdout.write('\n\t%s' % res.get(0.02))
                break
            except multiprocessing.TimeoutError:
                sys.stdout.write('.')
        print()
        print()

        print('Testing IMapIterator.next() with timeout:', end=' ')
        it = pool.imap(calculatestar, TASKS)
        while 1:
            sys.stdout.flush()
            try:
                sys.stdout.write('\n\t%s' % it.next(0.02))
            except StopIteration:
                break
            except multiprocessing.TimeoutError:
                sys.stdout.write('.')
        print()
        print()


if __name__ == '__main__':
    multiprocessing.freeze_support()
    test()

ワーカープロセスのコレクションに対してタスクをフィードしてその結果をまとめるキューの使い方の例を紹介します:

import time
import random

from multiprocessing import Process, Queue, current_process, freeze_support

#
# ワーカープロセスが実行する関数
#

def worker(input, output):
    for func, args in iter(input.get, 'STOP'):
        result = calculate(func, args)
        output.put(result)

#
# 結果の計算に使う関数
#

def calculate(func, args):
    result = func(*args)
    return '%s says that %s%s = %s' % \
        (current_process().name, func.__name__, args, result)

#
# タスクが参照する関数
#

def mul(a, b):
    time.sleep(0.5*random.random())
    return a * b

def plus(a, b):
    time.sleep(0.5*random.random())
    return a + b

#
#
#

def test():
    NUMBER_OF_PROCESSES = 4
    TASKS1 = [(mul, (i, 7)) for i in range(20)]
    TASKS2 = [(plus, (i, 8)) for i in range(10)]

    # キューを作成
    task_queue = Queue()
    done_queue = Queue()

    # タスクを投入
    for task in TASKS1:
        task_queue.put(task)

    # ワーカープロセスを開始
    for i in range(NUMBER_OF_PROCESSES):
        Process(target=worker, args=(task_queue, done_queue)).start()

    # 結果を取得して表示
    print('Unordered results:')
    for i in range(len(TASKS1)):
        print('\t', done_queue.get())

    # put() でタスクを追加
    for task in TASKS2:
        task_queue.put(task)

    # 追加の結果を取得して表示
    for i in range(len(TASKS2)):
        print('\t', done_queue.get())

    # 子プロセスに停止を指示
    for i in range(NUMBER_OF_PROCESSES):
        task_queue.put('STOP')


if __name__ == '__main__':
    freeze_support()
    test()

`multiprocessing` --- プロセスベースの並列処理¶

はじめに¶

`Process` クラス¶

コンテキストと開始方式¶

プロセス間でのオブジェクト交換¶

プロセス間の同期¶

ワーカープロセスのプールを使用¶

リファレンス¶

グローバルな開始方式¶

`Process` クラスと例外¶

パイプ (Pipe) とキュー (Queue)¶

その他¶

Connection オブジェクト¶

同期プリミティブ¶

共有 `ctypes` オブジェクト¶

`multiprocessing.sharedctypes` モジュール¶

マネージャー¶

カスタマイズされたマネージャー¶

リモートマネージャーを使用する¶

Proxy オブジェクト¶

クリーンアップ¶

プロセスプール¶

リスナーとクライアント¶

アドレスフォーマット¶

認証キー¶

ログ記録¶

`multiprocessing.dummy` モジュール¶

プログラミングガイドライン¶

すべての開始方式について¶

開始方式が spawn および forkserver の場合¶

使用例¶

目次

前のトピックへ

次のトピックへ

このページ

multiprocessing --- プロセスベースの並列処理¶

はじめに¶

Process クラス¶

コンテキストと開始方式¶

プロセス間でのオブジェクト交換¶

プロセス間の同期¶

プロセス間での状態の共有¶

ワーカープロセスのプールを使用¶

リファレンス¶

グローバルな開始方式¶

Process クラスと例外¶

パイプ (Pipe) とキュー (Queue)¶

その他¶

Connection オブジェクト¶

同期プリミティブ¶

共有 ctypes オブジェクト¶

multiprocessing.sharedctypes モジュール¶

マネージャー¶

カスタマイズされたマネージャー¶

リモートマネージャーを使用する¶

Proxy オブジェクト¶

クリーンアップ¶

プロセスプール¶

リスナーとクライアント¶

アドレスフォーマット¶

認証キー¶

ログ記録¶

multiprocessing.dummy モジュール¶

プログラミングガイドライン¶

すべての開始方式について¶

開始方式が spawn および forkserver の場合¶

使用例¶

`multiprocessing` --- プロセスベースの並列処理¶

`Process` クラス¶

`Process` クラスと例外¶

共有 `ctypes` オブジェクト¶

`multiprocessing.sharedctypes` モジュール¶

`multiprocessing.dummy` モジュール¶