multiprocessing --- プロセスベースの並列処理¶
ソースコード: Lib/multiprocessing/
利用可能な環境: not Android, not iOS, not WASI.
このモジュールは モバイルプラットフォーム と WebAssemblyプラットフォーム をサポートしません。
はじめに¶
multiprocessing is a package that supports spawning processes using an
API similar to the threading module. The multiprocessing package
offers both local and remote concurrency, effectively side-stepping the
Global Interpreter Lock by using
subprocesses instead of threads. Due
to this, the multiprocessing module allows the programmer to fully
leverage multiple processors on a given machine. It runs on both POSIX and
Windows.
The multiprocessing module also introduces the
Pool object which offers a convenient means of
parallelizing the execution of a function across multiple input values,
distributing the input data across processes (data parallelism). The following
example demonstrates the common practice of defining such functions in a module
so that child processes can successfully import that module. This basic example
of data parallelism using 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]
The multiprocessing module also introduces APIs which do not have
analogs in the threading module, like the ability to terminate, interrupt or kill a running process.
参考
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__' という記述が必要かは プログラミングガイドライン を参照してください。
The arguments to Process usually need to be picklable so they can be
passed to the child process. If you tried typing the above example directly
into a REPL it could lead to an AttributeError in the child process
trying to locate the f function in the __main__ module.
コンテキストと開始方式¶
プラットフォームにもよりますが、multiprocessing はプロセスを開始するために 3 つの方法をサポートしています。それら 開始方式 は以下のとおりです
- spawn
親プロセスは新たに Python インタープリタープロセスを開始します。子プロセスはプロセスオブジェクトの
run()メソッドの実行に必要なリソースのみ継承します。特に、親プロセスからの不要なファイル記述子とハンドルは継承されません。この方式を使用したプロセスの開始は fork や forkserver に比べ遅くなります。POSIX と Windows で利用できます。 Windows と macOS ではデフォルトの方式です。
- fork
親プロセスは
os.fork()を使用して Python インタープリターをフォークします。子プロセスはそれが開始されるとき、事実上親プロセスと同一になります。親プロセスのリソースはすべて子プロセスに継承されます。マルチスレッドプロセスのフォークは安全性に問題があることに注意してください。Available on POSIX systems. Currently the default on POSIX except macOS.
注釈
The default start method will change away from fork in Python 3.14. Code that requires fork should explicitly specify that via
get_context()orset_start_method().バージョン 3.12 で変更: プロセスが複数のスレッドを持つと Python が検出できる場合、この開始方式が内部で呼び出す
os.fork()関数はDeprecationWarningを出すようになりました。 別の開始方式を使ってください。 詳しくはos.fork()のドキュメントを参照してください。
- forkserver
プログラムの開始時に forkserver 開始方式を選ぶと、サーバープロセスを生成します。 それ以降、新しいプロセスが必要になるたびに、親プロセスがサーバーに接続し、新しいプロセスをフォークするよう要求します。 システムライブラリや事前のインポートによる副作用でスレッドが生成されない限り、フォークサーバーは単一スレッドなので、通常は安全に
os.fork()を使えます。 不要なリソースは引き継ぎません。Available on POSIX platforms which support passing file descriptors over Unix pipes such as Linux.
バージョン 3.4 で変更: すべての POSIX プラットフォームで spawn を、一部の POSIX プラットフォームで forkserver を追加しました。 Windows では、親の継承可能なハンドルを子プロセスがすべて引き継ぐことはなくなりました。
バージョン 3.8 で変更: macOS では spawn がデフォルトの開始方式になりました。 macOS のシステムライブラリがスレッドを開始し、サブプロセスのクラッシュにつながる場合があるため、fork 開始方式は安全ではないと考えてください。 bpo-33725 を参照してください。
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 を使用して開始されたプロセスに渡すことはできません。
Libraries using multiprocessing or
ProcessPoolExecutor should be designed to allow
their users to provide their own multiprocessing context. Using a specific
context of your own within a library can lead to incompatibilities with the
rest of the library user's application. Always document if your library
requires a specific start method.
警告
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()Queues are thread and process safe. Any object put into a
multiprocessingqueue will be serialized.
パイプ (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()
ロックを使用しないで標準出力に書き込んだ場合は、各プロセスからの出力がごちゃまぜになってしまいます。
ワーカープロセスのプールを使用¶
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 とほとんど同じです。
Global start method¶
Python supports several ways to create and initialize a process. The global start method sets the default mechanism for creating a process.
Several multiprocessing functions and methods that may also instantiate certain objects will implicitly set the global start method to the system's default, if it hasn’t been set already. The global start method can only be set once. If you need to change the start method from the system default, you must proactively set the global start method before calling functions or methods, or creating these objects.
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 に渡す引数のリストまたはタプルを指定するために使用されます。If a subclass overrides the constructor, it must make sure it invokes the base class constructor (
super().__init__()) before doing anything else to the process.注釈
In general, all arguments to
Processmust be picklable. This is frequently observed when trying to create aProcessor use aconcurrent.futures.ProcessPoolExecutorfrom a REPL with a locally defined target function.Passing a callable object defined in the current REPL session causes the child process to die via an uncaught
AttributeErrorexception when starting as target must have been defined within an importable module in order to be loaded during unpickling.Example of this uncatchable error from the child:
>>> 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>
See 開始方式が spawn および forkserver の場合. While this restriction is not true if using the
"fork"start method, as of Python3.14that is no longer the default on any platform. See コンテキストと開始方式. See also 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-N1:N2:...:Nk' 形式の名前が構築されます。ここでそれぞれの Nk はその親のN番目の子供です。
- 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.
- 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() を呼び出さなければ なりません 。さもないと、いつか完了していないタスクを数えるためのセマフォがオーバーフローし、例外を発生させるでしょう。
One difference from other Python queue implementations, is that multiprocessing
queues serializes all objects that are put into them using pickle.
The object return by the get method is a re-created object that does not share memory
with the original object.
管理オブジェクトを使用することで共有キューを作成できることも覚えておいてください。詳細は マネージャー を参照してください。
注釈
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])¶
パイプの両端を表す
Connectionオブジェクトのペア(conn1, conn2)を返します。duplex が
True(デフォルト) ならパイプは双方向性です。duplex がFalseならパイプは一方向性で、conn1はメッセージの受信専用、conn2はメッセージの送信専用になります。send()メソッドはpickleでオブジェクトをシリアライズし、recv()メソッドはオブジェクトを再構築します。
- class multiprocessing.Queue([maxsize])¶
パイプといくつかのロックやセマフォで実装された、プロセス間で共有するキューを返します。 あるプロセスが初めてキューに要素を入れると、バッファーからパイプへオブジェクトを転送する供給スレッドが開始されます。
Instantiating this class may set the global start method. See Global start method for more details.
標準ライブラリの
queueモジュールの通常のqueue.Emptyやqueue.Full例外がタイムアウトを伝えるために送出されます。Queueimplements all the methods ofqueue.Queueexcept fortask_done(),join(), andshutdown().- 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()¶
キューを閉じ、内部リソースを解放します。
A queue must not be used anymore after it is closed. For example,
get(),put()andempty()methods must no longer be called.The background thread will quit once it has flushed all buffered data to the pipe. This is called automatically when the queue is garbage collected.
- 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と非常に似ています。Instantiating this class may set the global start method. See Global start method for more details.
- close()¶
キューを閉じ、内部リソースを解放します。
閉じたキューをそれ以降使ってはいけません。 例えば、
get()、put()、empty()メソッドを呼び出してはいけません。Added in version 3.9.
- get()¶
キューから要素を削除して返します。
- put(item)¶
item をキューに追加します。
- class multiprocessing.JoinableQueue([maxsize])¶
JoinableQueueはQueueのサブクラスであり、task_done()やjoin()メソッドが追加されているキューです。Instantiating this class may set the global start method. See Global start method for more details.
- 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を送出します。バージョン 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()¶
Add support for when a program which uses
multiprocessinghas been frozen to produce an executable. (Has been tested with py2exe, PyInstaller and 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を発生させます。Calling
freeze_support()has no effect when the start method is not spawn. In addition, if the module is being run normally by the Python interpreter (the program has not been frozen), thenfreeze_support()has no effect.
- multiprocessing.get_all_start_methods()¶
サポートされている開始方式のリストを返します。 最初の要素がデフォルトの方式です。 開始方式には
'fork'、'spawn'、'forkserver'があります。 すべてのプラットフォームがすべての方式をサポートしているわけではありません。 コンテキストと開始方式 を参照してください。Added in version 3.4.
- multiprocessing.get_context(method=None)¶
multiprocessingモジュールと同じ属性を持つコンテキストオブジェクトを返します。If method is
Nonethen the default context is returned. Note that if the global start method has not been set, this will set it to the system default See Global start method for more details. Otherwise method should be'fork','spawn','forkserver'.ValueErroris raised if the specified start method is not available. See コンテキストと開始方式.Added in version 3.4.
- multiprocessing.get_start_method(allow_none=False)¶
開始するプロセスで使用する開始方式名を返します。
If the global start method is not set and allow_none is
False, the global start method is set to the default, and its name is returned. See Global start method for more details.戻り値は
'fork'、'spawn'、'forkserver'、Noneのいずれかです。 コンテキストと開始方式 を参照してください。Added in version 3.4.
バージョン 3.8 で変更: macOS では、 spawn 開始方式がデフォルトになりました。 fork 開始方法は、サブプロセスのクラッシュを引き起こす可能性があるため、安全ではありません。 bpo-33725 を参照。
- multiprocessing.set_executable(executable)¶
Set the path of the Python interpreter to use when starting a child process. (By default
sys.executableis used). Embedders will probably need to do something likeset_executable(os.path.join(sys.exec_prefix, 'pythonw.exe'))
子プロセスを作成する前に行ってください。
バージョン 3.4 で変更:
'spawn'開始方式を使う場合、POSIX でもサポートするようになりました。バージョン 3.11 で変更: path-like object を受け入れるようになりました。
- multiprocessing.set_forkserver_preload(module_names)¶
Set a list of module names for the forkserver main process to attempt to import so that their already imported state is inherited by forked processes. Any
ImportErrorwhen doing so is silently ignored. This can be used as a performance enhancement to avoid repeated work in every process.これを有効にするには、フォークサーバープロセスが起動する前 (
Poolを作成したりProcessを開始したりする前) に呼び出す必要があります。'forkserver'開始方式を使う場合だけ意味を持ちます。 コンテキストと開始方式 を参照してください。Added in version 3.4.
- 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 から完全なメッセージとしてバイトデータを送ります。
If offset is given then data is read from that position in buf. If size is given then that many bytes will be read from buf. Very large buffers (approximately 32 MiB+, though it depends on the OS) may raise a
ValueErrorexception
- recv_bytes([maxlength])¶
接続の相手側から送られたバイトデータの完全なメッセージを、バイト列として返します。 受信するものがあるまでブロックします。 受信するデータが残っておらず、相手側が接続を閉じた場合は
EOFErrorを送出します。maxlength を指定していて、かつメッセージが maxlength より長い場合、
OSErrorが発生してコネクションからそれ以上読めなくなります。
- recv_bytes_into(buf[, offset])¶
Read into buf a complete message of byte data sent from the other end of the connection and return the number of bytes in the message. Blocks until there is something to receive. Raises
EOFErrorif there is nothing left to receive and the other end was closed.buf must be a writable bytes-like object. If offset is given then the message will be written into the buffer from that position. Offset must be a non-negative integer less than the length of buf (in bytes).
バッファーがあまりに小さいと
BufferTooShort例外が発生します。eが例外インスタンスとすると完全なメッセージはe.args[0]で確認できます。
バージョン 3.3 で変更:
Connection.send()とConnection.recv()を使用して Connection オブジェクト自体をプロセス間で転送できるようになりました。Connection オブジェクトがコンテキスト管理プロトコルもサポートするようになりました (Context Manager Types を参照)。
__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のクローンです。Instantiating this class may set the global start method. See Global start method for more details.
Added in version 3.3.
- class multiprocessing.BoundedSemaphore([value])¶
有限セマフォオブジェクト:
threading.BoundedSemaphoreの類似物です。Instantiating this class may set the global start method. See Global start method for more details.
よく似た
threading.BoundedSemaphoreとは、次の一点だけ異なります。acquireメソッドの第一引数名は block で、Lock.acquire()と一致しています。注釈
macOS では
sem_getvalue()が実装されていないのでSemaphoreと区別がつきません。
- class multiprocessing.Condition([lock])¶
条件変数:
threading.Conditionの別名です。lock を指定するなら
multiprocessingのLockかRLockオブジェクトにすべきです。Instantiating this class may set the global start method. See Global start method for more details.
バージョン 3.3 で変更:
wait_for()メソッドが追加されました。
- class multiprocessing.Event¶
threading.Eventのクローンです。Instantiating this class may set the global start method. See Global start method for more details.
- class multiprocessing.Lock¶
再帰しないロックオブジェクトで、
threading.Lock相当のものです。 プロセスやスレッドがいったんロックを獲得 (acquire) すると、その後は、どのプロセスやスレッドから獲得しようとしても、ロックが解放 (release) されるまでブロックします。 解放はどのプロセスやスレッドからも行えます。 特に明記した点を除き、threading.Lockがスレッドに適用する概念と動作を、multiprocessing.Lockはプロセスとスレッドの両方に適用します。Lockは実際にはファクトリ関数で、デフォルトコンテキストで初期化されたmultiprocessing.synchronize.Lockのインスタンスを返すことに注意してください。Instantiating this class may set the global start method. See Global start method for more details.
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となる点だけが違います。
- class multiprocessing.RLock¶
再帰ロックオブジェクトで、
threading.RLock相当のものです。再帰ロックオブジェクトはそれを獲得 (acquire) したプロセスやスレッドが解放 (release) しなければなりません。プロセスやスレッドがロックをいったん獲得すると、同じプロセスやスレッドはブロックされずに再度獲得出来ます。そのプロセスやスレッドは獲得した回数ぶん解放しなければなりません。RLockは実際にはファクトリ関数で、デフォルトコンテキストで初期化されたmultiprocessing.synchronize.RLockのインスタンスを返すことに注意してください。Instantiating this class may set the global start method. See Global start method for more details.
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()実装とは例外の型が異なるので注意してください。
- class multiprocessing.Semaphore([value])¶
セマフォオブジェクト:
threading.Semaphoreのクローンです。Instantiating this class may set the global start method. See Global start method for more details.
よく似た
threading.BoundedSemaphoreとは、次の一点だけ異なります。acquireメソッドの第一引数名は block で、Lock.acquire()と一致しています。
注釈
macOS では sem_timedwait がサポートされていないので、acquire() にタイムアウトを与えて呼ぶと、ループ内でスリープすることでこの関数がエミュレートされます。
注釈
このパッケージに含まれる機能には、ホストとなるオペレーティングシステム上で動作している共有セマフォを使用しているものがあります。これが使用できない場合には、multiprocessing.synchronize モジュールが無効になり、このモジュールのインポート時に ImportError が発生します。詳細は bpo-3770 を参照してください。
マネージャー¶
マネージャーは異なるプロセス間で共有されるデータの作成方法を提供します。これには別のマシン上で走るプロセス間のネットワーク越しの共有も含まれます。マネージャーオブジェクトは 共有オブジェクト を管理するサーバープロセスを制御します。他のプロセスはプロキシ経由で共有オブジェクトへアクセスすることができます。
- 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 is a context object, or
None(use the current context). IfNone, calling this may set the global start method. See Global start method for more details.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()
- 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 で変更: マネージャーオブジェクトはコンテキストマネージメント・プロトコルをサポートします -- Context Manager Types を参照してください。
__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オブジェクトを作成して、そのプロキシを返します。
- Queue([maxsize])¶
共有
queue.Queueオブジェクトを作成して、そのプロキシを返します。
- RLock()¶
共有
threading.RLockオブジェクトを作成して、そのプロキシを返します。
- Semaphore([value])¶
共有
threading.Semaphoreオブジェクトを作成して、そのプロキシを返します。
- Array(typecode, sequence)¶
配列を作成して、そのプロキシを返します。
- Value(typecode, value)¶
書き込み可能な
value属性を持つオブジェクトを作成し、そのプロキシを返します。
バージョン 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 can be used to specify the context used for starting the worker processes. Usually a pool is created using the function
multiprocessing.Pool()or thePool()method of a context object. In both cases context is set appropriately. IfNone, calling this function will have the side effect of setting the current global start method if it has not been set already. See theget_context()function.プールオブジェクトのメソッドは、そのプールを作成したプロセスのみが呼び出すべきです。
警告
multiprocessing.poolオブジェクトは内部リソースを持ちます。 他のリソースと同様、プールをコンテキストマネージャとして使うか、close()とterminate()を手動で呼び出し、適切に管理する必要があります。 これを怠ると、終了処理でプロセスがハングする可能性があります。CPython はプールのファイナライザーが呼ばれることを保証しないため、プールの破棄をガベージコレクターに任せるのは 正しくありません。 詳しくは
object.__del__()を参照してください。バージョン 3.2 で変更: maxtasksperchild 引数を追加しました。
バージョン 3.4 で変更: context パラメータを追加しました。
バージョン 3.13 で変更: processes のデフォルト値に、
os.cpu_count()ではなくos.process_cpu_count()を使うようになりました。注釈
Worker processes within a
Pooltypically live for the complete duration of the Pool's work queue. A frequent pattern found in other systems (such as Apache, mod_wsgi, etc) to free resources held by workers is to allow a worker within a pool to complete only a set amount of work before exiting, being cleaned up and a new process spawned to replace the old one. The maxtasksperchild argument to thePoolexposes this ability to the end user.- 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])¶
An iterator-based version of
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 オブジェクトがコンテキストマネージメント・プロトコルをサポートするようになりました。 -- Context Manager Types を参照してください。
__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 オブジェクトがコンテキストマネージメント・プロトコルをサポートするようになりました。 -- Context Manager Types を参照してください。
__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オブジェクト; または
読み取ることのできるデータがある場合、あるいは相手側の端が閉じられている場合、コネクションまたはソケットオブジェクトは準備ができています。
POSIX:
wait(object_list, timeout)almost equivalentselect.select(object_list, [], [], timeout). The difference is that, ifselect.select()is interrupted by a signal, it can raiseOSErrorwith an error number ofEINTR, whereaswait()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() を使用して生成することもできます。
This authentication protects Listener and Client() connections,
which are reachable by address. It is not applied to the anonymous pipes
created by Pipe() or used internally by
Queue.
multiprocessing treats all local processes running as the same user as
trusted; on most operating systems such processes can access each other's pipe
file descriptors regardless. Applications that require isolation between
processes of the same user must arrange it at the operating-system level --
for example, by running workers under a different user account or in a sandbox.
ログ記録¶
ロギングをサポートする機能もいくつかあります。
ただし、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 モジュールのラッパーでしかありません。
In particular, the Pool function provided by multiprocessing.dummy
returns an instance of ThreadPool, which is a subclass of
Pool that supports all the same method calls but uses a pool of
worker threads rather than worker processes.
- 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
開始方式が spawn および forkserver の場合¶
fork 開始方式には当てはまらない、いくつかの追加の制限があります。
さらなる pickle 化の可能性
Ensure that all arguments to
Processare picklable. Also, if you subclassProcess.__init__, you must make sure that instances will be picklable when theProcess.startmethod is called.
グローバル変数
子プロセスで実行されるコードがグローバル変数にアクセスしようとする場合、子プロセスが見るその値は
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()