### ナビゲーション

- 索引
- モジュール |
- 次へ |
- 前へ |
- Python v2.6.2 documentation »
- Python 標準ライブラリ »
- 17. オプションのオペレーティングシステムサービス »

# 17.6. `multiprocessing` — プロセスベースの “並列処理” インタフェース

バージョン 2.6 で追加.

## 17.6.1. はじめに

`multiprocessing` はPythonの標準ライブラリのパッケージで `threading` とよく似た API を使ってプロセスを生成することができます。 `multiprocessing` パッケージを使用すると、 ローカルとリモート両方の並列制御を行うことができます。 また、このパッケージはスレッドの代わりにサブプロセスを使用することにより、 グローバルインタプリタロック ( Global Interpreter Lock ) の問題を避ける工夫が行われています。 このような特徴があるため `multiprocessing` モジュールを使うことで、 マルチプロセッサマシンの性能を最大限に活用することができるでしょう。 なお、このモジュールは Unix と Windows で動作します。

警告

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

ノート

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

    >>> from multiprocessing import Pool
    >>> p = Pool(5)
    >>> def f(x):
    ...     return x*x
    ...
    >>> p.map(f, [1,2,3])
    Process PoolWorker-1:
    Process PoolWorker-2:
    Traceback (most recent call last):
    AttributeError: 'module' object has no attribute 'f'
    AttributeError: 'module' object has no attribute 'f'
    AttributeError: 'module' object has no attribute 'f'

### 17.6.1.1. `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()

(Windows 環境で) `if __name__ == '__main__'` という文が必要な理由については、 プログラミングガイドライン を参照してください。

### 17.6.1.2. プロセス間でのオブジェクト交換

`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()

> キューはスレッドセーフであり、プロセスセーフです。

パイプ (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つのプロセス (もしくはスレッド) が同時に 同じ パイプに対して、 読み書きをおこなった場合に起こります。 もちろん、異なったパイプを使用していれば、同時に読み書きをおこなっても データが破壊されてしまう危険性はありません。

### 17.6.1.3. プロセス間の同期

`multiprocessing` は `threading` モジュールと同じプロセス間同期の仕組みを 備えています。以下の例では、ロックを使用して、一度に1つのプロセスしか 標準出力に書き込まないようにしています。

    from multiprocessing import Process, Lock

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

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

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

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

### 17.6.1.4. プロセス間での状態の共有

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

しかし、どうしてもプロセス間のデータ共有が必要な場合のために `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` 、 `Namespace` 、 `Lock` 、 `RLock` 、 `Semaphore` 、 `BoundedSemaphore` 、 `Condition` 、 `Event` 、 `Queue` 、 `Value` 、 `Array` をサポートします。 以下にサンプルコードを示します。

>     from multiprocessing import Process, Manager

>     def f(d, l):
>         d[1] = '1'
>         d['2'] = 2
>         d[0.25] = None
>         l.reverse()

>     if __name__ == '__main__':
>         manager = Manager()

>         d = manager.dict()
>         l = manager.list(range(10))

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

>         print d
>         print l

> このサンプルコードを実行すると以下のように表示されます。

>     {0.25: None, 1: '1', '2': 2}
>     [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

> サーバプロセスのマネージャオブジェクトは共有メモリのオブジェクトよりも 柔軟であるといえます。それは、どのような型のオブジェクトでも使えるからです。 また、1つのマネージャオブジェクトはネットワーク経由で他のコンピュータ上の プロセスによって共有することもできます。 しかし、共有メモリより動作が遅いという欠点があります。

### 17.6.1.5. ワーカープロセスのプールを使用

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

以下に例を示します。

    from multiprocessing import Pool

    def f(x):
        return x*x

    if __name__ == '__main__':
        pool = Pool(processes=4)              # 4つのワーカープロセスで開始
        result = pool.apply_async(f, [10])    # 非同期で "f(10)" を評価
        print result.get(timeout=1)           # あなたのコンピュータが *かなり* 遅くない限りは "100" を表示
        print pool.map(f, range(10))          # "[0, 1, 4,..., 81]" を表示

## 17.6.2. リファレンス

`multiprocessing` パッケージは `threading` モジュールの API とほとんど同じです。

### 17.6.2.1. `Process` クラスと例外

class `multiprocessing.``Process`([group[, target[, name[, args[, kwargs]]]]])

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

  コンストラクタは必ずキーワード引数で呼び出すべきです。 引数 group には必ず `None` を渡してください。 この引数は `threading.Thread` クラスとの互換性のためだけに残されています。 引数 target には、呼び出し可能オブジェクト (Collable Object) を渡します。 このオブジェクトは `run()` メソッドから呼び出されます。 この引数はデフォルトで `None` となっており、何も呼び出されません。 引数 name にはプロセス名を渡します。 デフォルトでは、自動でユニークな名前が割り当てられます。 命名規則は、 ‘Process-N1:N2:...:Nk‘ となります。 ここで N1,N2,...,Nk は整数の数列で、 作成した プロセス数に対応します。 引数 args は target で指定された呼び出し可能オブジェクトへの引数を渡します。 同じく、引数 kwargs はキーワード引数を渡します。 デフォルトでは、 target には引数が渡されないようになっています。

  サブクラスがコンストラクタをオーバーライドする場合は、 そのプロセスに対する処理を行う前に基底クラスのコンストラクタ (`Process.__init__()`) を実行しなければなりません。

  `run`()

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

    このメソッドはサブクラスでオーバーライドすることができます。 標準の `run()` メソッドは呼び出し可能オブジェクトを呼び出します。 この呼び出されるオブジェクトはコンストラクタの target 引数として渡されます。 もしコンストラクタに args もしくは kwargs 引数が渡されていれば、 呼び出すオブジェクトにこれらの引数を渡します。

  `start`()

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

    このメソッドをプロセスごとに呼び出す必要があります。 各プロセスの `run()` メソッドを呼び出す準備が完了します。

  `join`([timeout])

    `join()` されたプロセスが terminate を呼び出すまで、もしくは オプションで指定したタイムアウトが発生するまで呼び出し側のスレッドを ブロックします。

    timeout が `None` ならタイムアウトは設定されません。

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

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

  `name`

    プロセス名です。

    この名前は文字列で、プロセスの識別にのみ使用されます。特別な命名規則はありません。 複数のプロセスが同じ名前を持つ場合もあります。 また、この名前はコンストラクタにより初期化されます。

  `is_alive`()

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

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

  `daemon`

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

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

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

    デーモンプロセスは子プロセスを作成できないことに注意してください。 もしそうでなければ、そのデーモンの親プロセスが終了したときに子プロセスが孤児になってしまう場合があるからです。

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

  `pid`

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

  `exitcode`

    子プロセスの終了コードです。 子プロセスがまだ終了していない場合は `None` が返されます。 負の値 -N は子プロセスがシグナル N で終了したことを表します。

  `authkey`

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

    `multiprocessing` モジュールがメインプロセスにより初期化される場合には、 `os.random()` 関数を使用してランダムな値が設定されます。

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

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

  `terminate`()

    プロセスを終了します。Unix 環境では `SIGTERM` シグナルを、 Windows 環境では `TerminateProcess()` を使用して終了させます。 終了ハンドラや finally 節などは、実行されないことに注意してください。

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

    警告

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

  プロセスオブジェクトが作成したプロセスのみが `start()`, `join()`, `is_alive()` と `exit_code` のメソッドを呼び出すべきです。

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

      >>> import multiprocessing, time, signal
      >>> p = multiprocessing.Process(target=time.sleep, args=(1000,))
      >>> print p, p.is_alive()
      <Process(Process-1, initial)> False
      >>> p.start()
      >>> print p, p.is_alive()
      <Process(Process-1, started)> True
      >>> p.terminate()
      >>> print p, p.is_alive()
      <Process(Process-1, stopped[SIGTERM])> False
      >>> p.exitcode == -signal.SIGTERM
      True

exception `multiprocessing.``BufferTooShort`

  この例外は `Connection.recv_bytes_into()` によって発生し、 バッファオブジェクトが小さすぎてメッセージが読み込めないことを示します。

  `e` が `BufferTooShort` のインスタンスとすると、 `e.args[0]` はバイト文字列でそのメッセージを取得できます。

### 17.6.2.2. パイプ (Pipe) とキュー (Queue)

マルチプロセス環境では、 一般的にメッセージパッシングをプロセス間通信のために使用し、 ロックのような同期プリミティブの使用しないようにします。

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

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

もし `JoinableQueue` を使用するなら、キューから削除される各タスクのために `JoinableQueue.task_done()` を呼び出さなければ なりません 。 もしくは、ある例外を発生させてオーバーフローする可能性がある未終了タスクを 数えるためにセマフォが使用されます。

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

ノート

`multiprocessing` は通常の `Queue.Empty` と、 タイムアウトのシグナルを送るために `Queue.Full` 例外を使用します。 それらは `Queue` からインポートする必要があるので `multiprocessing` の名前空間では利用できません。

警告

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

警告

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

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

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

プロセス間通信におけるキューの使用方法は 例 を参照してください。

`multiprocessing.``Pipe`([duplex])

  パイプの終端を表す `Connection` オブジェクトのタプル `(conn1, conn2)` を返します。

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

class `multiprocessing.``Queue`([maxsize])

  パイプや2～3個のロック/セマフォを使用して実装されたプロセス共有キューを返します。 あるプロセスが最初に要素をキューへ追加するとき、 バッファからパイプの中へオブジェクトを転送する供給スレッドが開始されます。

  標準ライブラリの `Queue` モジュールからの通常の `Queue.Empty` や `Queue.Full` 例外はタイムアウトのシグナルを送るために発生します。

  `Queue` は `task_done()` や `join()` を除く `Queue.Queue` の全てのメソッドを実装します。

  `qsize`()

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

    これは `sem_getvalue()` が実装されていない Mac OS X のような Unix プラットホーム上で `NotImplementedError` を発生させる可能性があることを覚えておいてください。

  `empty`()

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

  `full`()

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

  `put`(item[, block[, timeout]])

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

  `put_nowait`(item)

    `put(item, False)` と等価です。

  `get`([block[, timeout]])

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

  `get_nowait`()
  `get_no_wait`()

    `get(False)` と等価です。

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

  `close`()

    カレントプロセスからこのキューへそれ以上データが追加されないことを表します。 バックグラウンドスレッドはパイプへバッファされた全てのデータをフラッシュするとすぐに終了します。 これはキューがガベージコレクトされるときに自動的に呼び出されます。

  `join_thread`()

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

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

  `cancel_join_thread`()

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

class `multiprocessing.``JoinableQueue`([maxsize])

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

  `task_done`()

    以前にキューへ追加されたタスクが完了したことを表します。キュー消費スレッドによって使用されます。 タスクをフェッチするために使用されるそれぞれの `get()` では、 次に `task_done()` を呼び出してタスクの処理が完了したことをキューへ伝えます。

    もし `join()` がブロッキング状態なら、 全ての要素が処理されたときに復帰します( `task_done()` 呼び出しが 全ての要素からキュー内へ `put()` されたと受け取ったことを意味します)。

    キューにある要素より多く呼び出された場合 `ValueError` が発生します。

  `join`()

    キューにある全ての要素が取り出されて処理されるまでブロッキングします。

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

### 17.6.2.3. その他の関数(Miscellaneous)

`multiprocessing.``active_children`()

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

  これを呼び出すと “join” して既に終了しているプロセスには副作用があります。

`multiprocessing.``cpu_count`()

  システムの CPU 数を返します。 もしかしたら `NotImplementedError` が発生するかもしれません。

`multiprocessing.``current_process`()

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

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

`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` を発生させます。

  そのモジュールが Python インタプリタによって普通に実行されるなら `freeze_support()` は何の影響もありません。

`multiprocessing.``set_executable`()

  子プロセスを開始するときに使用する Python インタプリタのパスを設定します。 (デフォルトでは `sys.executable` が使用されます)。 コードに組み込むときは、おそらく次のようにする必要があります。

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

      子プロセスを生成する前に行います。 (Windows 専用)

ノート

`multiprocessing` には `threading.active_count()`, `threading.enumerate()`, `threading.settrace()`, `threading.setprofile()`, `threading.Timer` や `threading.local` のような関数はありません。

### 17.6.2.4. Connection Objects

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

Connection オブジェクトは通常は `Pipe()` を使用して作成されます。 詳細は Listeners and Clients も参照してください。

class `multiprocessing.``Connection`
  `send`(obj)

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

    オブジェクトは pickle でシリアライズ可能でなければなりません。

  `recv`()

    コネクションの向こう側から `send()` を使用して送られたオブジェクトを返します。 何も受け取らずにコネクションの向こう側でクローズされた場合 `EOFError` が発生します。

  `fileno`()

    コネクションが使用するハンドラか、ファイルディスクリプタを返します。

  `close`()

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

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

  `poll`([timeout])

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

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

  `send_bytes`(buffer[, offset[, size]])

    バッファインタフェースをサポートするオブジェクトから完全なメッセージとしてバイトデータを送ります。

    offset が指定されると buffer のその位置からデータが読み込まれます。 size が指定されると大量データがバッファから読み込まれます。

  `recv_bytes`([maxlength])

    文字列のようにコネクションの向こう側から送られたバイトデータの完全なメッセージを返します。 何も受け取らずにコネクションの向こう側でクローズされた場合 `EOFError` が発生します。

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

  `recv_bytes_into`(buffer[, offset])

    コネクションの向こう側から送られたバイトデータを buffer に読み込み、メッセージのバイト数を返します。 何も受け取らずにコネクションの向こう側でクローズされた場合 `EOFError` が発生します。

    buffer は書き込み可能なバッファインタフェースを備えたオブジェクトでなければなりません。 offset が与えられたら、その位置からバッファへメッセージが書き込まれます。 オフセットは buffer バイトよりも小さい正の数でなければなりません。

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

例:

    >>> from multiprocessing import Pipe
    >>> a, b = Pipe()
    >>> a.send([1, 'hello', None])
    >>> b.recv()
    [1, 'hello', None]
    >>> b.send_bytes('thank you')
    >>> a.recv_bytes()
    '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 されると、 メッセージの境界が正しいかどうか分からないので、 そのパイプのデータは破壊されたようになります。

### 17.6.2.5. 同期プリミティブ

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

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

class `multiprocessing.``BoundedSemaphore`([value])

  束縛されたセマフォオブジェクト: `threading.BoundedSemaphore` のクローンです。

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

class `multiprocessing.``Condition`([lock])

  状態変数: `threading.Condition` のクローンです。

  lock を指定するなら `multiprocessing` の `Lock` か `RLock` オブジェクトにすべきです。

class `multiprocessing.``Event`

  `threading.Event` のクローンです。

class `multiprocessing.``Lock`

  非再帰的なロックオブジェクト: `threading.Lock` のクローンです。

class `multiprocessing.``RLock`

  再帰的なロックオブジェクト: `threading.RLock` のクローンです。

class `multiprocessing.``Semaphore`([value])

  束縛されたセマフォオブジェクト: `threading.Semaphore` のクローンです。

ノート

`BoundedSemaphore`, `Lock`, `RLock` と `Semaphore` の `acquire()` メソッドは `threading` ではサポートされていない タイムアウトパラメータを取ります。その引数はキーワード引数で受け取れる `acquire(block=True, timeout=None)` です。 block が `True` 且つ timeout が `None` ではないなら、 タイムアウトが秒単位で設定されます。 block が `False` なら timeout は無視されます。

Mac OS X では `sem_timedwait` がサポートされていないので、 タイムアウトの引数は無視されることに注意してください。

ノート

メインスレッドが `BoundedSemaphore.acquire()`, `Lock.acquire()`, `RLock.acquire()`, `Semaphore.acquire()`, `Condition.acquire()` 又は `Condition.wait()` を呼び出してブロッキング状態のときに Ctrl-C で 生成される SIGINT シグナルを受け取ると、その呼び出しはすぐに中断されて `KeyboardInterrupt` が発生します。

これは同等のブロッキング呼び出しが実行中のときに SIGINT が無視される `threading` の振る舞いとは違っています。

### 17.6.2.6. 共有 `ctypes` オブジェクト

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

`multiprocessing.``Value`(typecode_or_type, *args[, lock])

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

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

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

  lock はキーワード引数でのみ指定することに注意してください。

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

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

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

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

  lock はキーワード引数でのみ指定することに注意してください。

  `ctypes.c_char` の配列は文字列を格納して取り出せる value と raw 属性を持っていることを覚えておいてください。

#### 17.6.2.6.1. `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, *args[, lock])

  lock の値に依存する点を除けば `RawArray()` と同様です。 プロセスセーフな同期ラッパーが raw ctypes 配列の代わりに返されるでしょう。

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

  lock はキーワード引数でのみ指定することに注意してください。

`multiprocessing.sharedctypes.``Value`(typecode_or_type, *args[, lock])

  lock の値に依存する点を除けば `RawValue()` と同様です。 プロセスセーフな同期ラッパーが ctypes オブジェクトの代わりに返されるでしょう。

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

  lock はキーワード引数でのみ指定することに注意してください。

`multiprocessing.sharedctypes.``copy`(obj)

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

`multiprocessing.sharedctypes.``synchronized`(obj[, lock])

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

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

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

次の表は通常の 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(ctypes.c_double, 1.0/3.0, lock=False)
        s = Array('c', '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)]

### 17.6.2.7. Managers

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

`multiprocessing.``Manager`()

  プロセス間で共有オブジェクトのために使用される `SyncManager` オブジェクトを返します。 返されたマネージャオブジェクトは生成される子プロセスに対応して、 共有オブジェクトを作成するメソッドを持ち、応答プロキシを返します。

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

class `multiprocessing.managers.``BaseManager`([address[, authkey]])

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

  作成後、マネージャオブジェクトが開始されたマネージャプロセスの参照を保証するために `start()` か `serve_forever()` を呼び出します。

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

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

  `start`()

    マネージャを開始するためにサブプロセスを開始します。

  `serve_forever`()

    カレントプロセスでサーバを実行します。

  `from_address`(address, authkey)

    引数として渡されたアドレスと認証キーを使用して既存のサーバプロセスを参照する マネージャオブジェクトを作成するクラスメソッドです。

  `get_server`()

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

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

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

  `connect`()

    ローカルからリモートのマネージャオブジェクトへ接続します。

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

  `shutdown`()

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

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

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

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

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

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

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

    exposed は `BaseProxy._callMethod()` を使用してアクセスされる この typeid のプロキシになるメソッド名のシーケンスを指定するために使用されます。 ( exposed が `None` の場合 `proxytype._exposed_` が存在すれば、 それが代わりに使用されます。) exposed リストが指定されない場合は、 共有オブジェクトの全ての “パブリックメソッド” にアクセスされます。 (ここで言う “パブリックメソッド” は `__call__()` メソッドを持ち、 `'_'` で始まらない名前の属性を意味します。)

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

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

  `BaseManager` インスタンスも読み取り専用属性を1つ持っています。

  `address`

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

class `multiprocessing.managers.``SyncManager`

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

  また共有リストやディクショナリの作成もサポートします。

  `BoundedSemaphore`([value])

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

  `Condition`([lock])

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

    lock が提供される場合 `threading.Lock` か `threading.RLock` オブジェクトのためのプロキシになります。

  `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` オブジェクトを作成して、そのプロキシを返します。

#### 17.6.2.7.1. Namespace オブジェクト

Namespace オブジェクトはプライベートなメソッドを持っていますが、書き込み属性を持ちます。 そのオブジェクト表現はその属性の値を表示します。

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

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

#### 17.6.2.7.2. カスタマイズされたマネージャ

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

    from multiprocessing.managers import BaseManager

    class MathsClass(object):
        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__':
        manager = MyManager()
        manager.start()
        maths = manager.Maths()
        print maths.add(4, 3)         # 7 を表示
        print maths.mul(7, 8)         # 56 を表示

#### 17.6.2.7.3. リモートマネージャを使用する

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

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

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

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

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

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

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

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

    >>> from multiprocessing import Process, Queue
    >>> from multiprocessing.managers import BaseManager
    >>> class Worker(Process):
    ...     def __init__(self, q):
    ...         self.q = q
    ...         super(Worker, self).__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='abracadabra')
    >>> s = m.get_server()
    >>> s.serve_forever()

### 17.6.2.8. Proxy オブジェクト

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

プロキシオブジェクトはその参照対象が持つ対応メソッドを実行するメソッドを持ちます。 (そうは言っても、参照対象の全てのメソッドが必ずしもプロキシ経由で利用可能ではありません) プロキシは通常その参照対象ができることと同じ方法で使用されます。

    >>> from multiprocessing import Manager
    >>> manager = 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 0xb799974c>
    >>> l[4]
    16
    >>> l[2:5]
    [4, 9, 16]

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

プロキシオブジェクトの重要な機能はプロセス間で受け渡し可能な pickle 化ができることです。 しかし、プロキシが対応するマネージャプロセスに対して送信される場合、 そのプロキシを unpickle するとその参照対象を生成することを覚えておいてください。 例えば、これはある共有オブジェクトに別の共有オブジェクトが含められることを意味します。

    >>> a = manager.list()
    >>> b = manager.list()
    >>> a.append(b)         # a の参照対象に b の参照対象を含める
    >>> print a, b
    [[]] []
    >>> b.append('hello')
    >>> print a, b
    [['hello']] ['hello']

ノート

`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('__getslice__', (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`()

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

    参照対象が unpickle 化できるなら例外を発生します。

  `__repr__`()

    プロキシオブジェクトのオブジェクト表現を返します。

  `__str__`()

    参照対象のオブジェクト表現を返します。

#### 17.6.2.8.1. クリーンアップ

プロキシオブジェクトは弱参照(weakref)コールバックを使用します。 プロキシオブジェクトがガベージコレクトされるときに その参照対象が所有するマネージャからその登録を取り消せるようにするためです。

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

### 17.6.2.9. プロセスプール

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

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

  プロセスプールオブジェクトはジョブが実行されるようにワーカープロセスのプールを制御します。 タイムアウトやコールバックで非同期の実行をサポートして、並列 map 実装を持ちます。

  processes は使用するワーカープロセスの数です。 processes が `None` の場合 `cpu_count()` が返す数を使用します。 initializer が `None` の場合、 各ワーカープロセスが開始時に `initializer(*initargs)` を呼び出します。

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

    `apply()` 組み込み関数と同じです。その結果を返せるようになるまでブロックします。

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

    `apply()` メソッドの一種で結果オブジェクトを返します。

    callback が指定された場合、1つの引数を受け取って呼び出されます。 その結果を返せるようになったときに callback が結果オブジェクトに対して (その呼び出しが失敗しない限り)適用されます。 その結果を扱う別スレッドはブロックされるので callback はすぐに終了します。

  `map`(func, iterable[, chunksize])

    並列な `map()` 組み込み関数と同じです( iterable な引数を1つだけサポートします)。 その結果を返せるようになるまでブロックします。

    このメソッドは独立したタスクのようにプロセスプールに対して実行するチャンク数に分割します。 チャンク(概算)サイズは chunksize に正の整数を指定することで行います。

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

    `map()` メソッドの一種で結果オブジェクトを返します。

    callback が指定された場合、1つの引数を受け取って呼び出されます。 その結果を返せるようになったときに callback が結果オブジェクトに対して (その呼び出しが失敗しない限り)適用されます。 その結果を扱う別スレッドはブロックされるので callback はすぐに終了します。

  `imap`(func, iterable[, chunksize])

    `itertools.imap()` と同じです。

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

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

  `imap_unordered`(func, iterable[, chunksize])

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

  `close`()

    これ以上プールでタスクが実行されないようにします。 全てのタスクが完了した後でワーカープロセスが終了します。

  `terminate`()

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

  `join`()

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

class `multiprocessing.pool.``AsyncResult`

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

  `get`([timeout])

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

  `wait`([timeout])

    その結果が有効になるか timeout 秒経つまで待ちます。

  `ready`()

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

  `successful`()

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

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

    from multiprocessing import Pool

    def f(x):
        return x*x

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

        result = pool.apply_async(f, (10,))   # 非同期で "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 it.next()                       # "0" を表示
        print it.next()                       # "1" を表示
        print it.next(timeout=1)              # あなたのコンピュータが *かなり* 遅くない限りは "4" を表示

        import time
        result = pool.apply_async(time.sleep, (10,))
        print result.get(timeout=1)           # TimeoutError を発生

### 17.6.2.10. Listeners and Clients

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

しかし `multiprocessing.connection` モジュールはさらに柔軟な仕組みがあります。 基本的にはソケットもしくは Windows の名前付きパイプを扱う 高レベルのメッセージ指向 API を提供して `hmac` モジュールを 使用して ダイジェスト認証 もサポートします。

`multiprocessing.connection.``deliver_challenge`(connection, authkey)

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

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

`multiprocessing.connection.``answerChallenge`(connection, authkey)

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

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

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

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

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

  authentication が `True` か authkey が文字列の場合、 ダイジェスト認証が使用されます。認証に使用されるキーは authkey 、 又は authkey が `None` の場合は `current_process().authkey` のどちらかです。 認証が失敗した場合 `AuthenticationError` が発生します。 認証キー を参照してください。

class `multiprocessing.connection.``Listener`([address[, family[, backlog[, authenticate[, 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()` メソッドに対して渡されます。

  authentication が `True` (デフォルトでは `False` ) か authkey が `None` ではない場合、ダイジェスト認証が使用されます。

  authkey が文字列の場合、認証キーとして使用されます。そうでない場合は None でなければいけません。

  authkey が `None` 且つ authenticate が `True` の場合 `current_process().authkey` が認証キーとして使用されます。 authkey が `None` 且つ authentication が `False` の場合、認証は行われません。 もし認証が失敗した場合 `AuthenticationError` が発生します。 詳細 認証キー を参照してください。

  `accept`()

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

  `close`()

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

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

  `address`

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

  `last_accepted`

    最後にコネクションを受け付けたアドレスです。 有効なアドレスがない場合は `None` になります。

このモジュールは2つの例外を定義します。

exception `multiprocessing.connection.``AuthenticationError`

  認証エラーが起こったときに例外が発生します。

例

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

    from multiprocessing.connection import Listener
    from array import array

    address = ('localhost', 6000)     # family is deduced to be 'AF_INET'
    listener = Listener(address, authkey='secret password')

    conn = listener.accept()
    print 'connection accepted from', listener.last_accepted

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

    conn.send_bytes('hello')

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

    conn.close()
    listener.close()

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

    from multiprocessing.connection import Client
    from array import array

    address = ('localhost', 6000)
    conn = Client(address, authkey='secret password')

    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])

    conn.close()

#### 17.6.2.10.1. アドレスフォーマット

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

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

### 17.6.2.11. 認証キー

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

認証キーはパスワードとして見なされる文字列です。 コネクションが確立すると、双方の終点で正しい接続先であることを証明するために 知っているお互いの認証キーを要求します。 (双方の終点が同じキーを使用して通信しようとしても、 コネクション上でそのキーを送信することは できません 。)

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

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

### 17.6.2.12. ロギング

ロギングのために幾つかの機能が利用可能です。しかし `logging` パッケージは、 (ハンドラ種別に依存して)違うプロセスからのメッセージがごちゃ混ぜになるので、 プロセスの共有ロックを使用しないことに注意してください。

`multiprocessing.``get_logger`()

  `multiprocessing` が使用するロガーを返します。必要に応じて新たなロガーを作成します。

  最初に作成するとき、ロガーはレベルに `logging.NOTSET` が設定されていて デフォルトハンドラがありません。 このロガーへ送られるメッセージはデフォルトではルートロガーへ伝播されません。

  Windows 上では子プロセスが親プロセスのロガーレベルを継承しないことに注意してください。 さらにその他のロガーのカスタマイズ内容も全て継承されません。

`multiprocessing.``log_to_stderr`()

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

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

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

これらの2つのロギング関数があることに加えて、 multiprocessing モジュールも2つの追加ロギングレベル属性を提供します。 それは `SUBWARNING` と `SUBDEBUG` です。 次の表は通常のレベル階層にうまく適合していることを表します。

Level | Numeric value
--- | ---
`SUBWARNING` | 25
`SUBDEBUG` | 5

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

こういった追加のロギングレベルは主に multiprocessing モジュールの 信頼できるデバッグメッセージのために使用されます。 以下に上述の例に `SUBDEBUG` を有効にしたものを紹介します。

    >>> import multiprocessing, logging
    >>> logger = multiprocessing.log_to_stderr()
    >>> logger.setLevel(multiprocessing.SUBDEBUG)
    >>> logger.warning('doomed')
    [WARNING/MainProcess] doomed
    >>> m = multiprocessing.Manager()
    [INFO/SyncManager-1] child process calling self.run()
    [INFO/SyncManager-1] created temp directory /.../pymp-djGBXN
    [INFO/SyncManager-1] manager serving at '/.../pymp-djGBXN/listener-knBYGe'
    >>> del m
    [SUBDEBUG/MainProcess] finalizer calling ...
    [INFO/MainProcess] sending shutdown message to manager
    [DEBUG/SyncManager-1] manager received shutdown message
    [SUBDEBUG/SyncManager-1] calling <Finalize object, callback=unlink, ...
    [SUBDEBUG/SyncManager-1] finalizer calling <built-in function unlink> ...
    [SUBDEBUG/SyncManager-1] calling <Finalize object, dead>
    [SUBDEBUG/SyncManager-1] finalizer calling <function rmtree at 0x5aa730> ...
    [INFO/SyncManager-1] manager exiting with exitcode 0

### 17.6.2.13. `multiprocessing.dummy` モジュール

`multiprocessing.dummy` は `multiprocessing` の API を複製しますが `threading` モジュールのラッパーでしかありません。

## 17.6.3. プログラミングガイドライン

`multiprocessing` を使用するときに守るべき確かなガイドラインとイディオムです。

### 17.6.3.1. 全てのプラットホーム

共有状態を避ける

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

> `threading` モジュールのプリミティブな低レベルの同期を使用するよりも、 キューかパイプをプロセス間通信に使用することがおそらく最善の方法です。

pickle 機能

> プロキシのメソッドへの引数は pickle 化できることを保証します。

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

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

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

ゾンビプロセスを join する

> Unix 上ではプロセスが終了したときに join しないと、そのプロセスはゾンビになります。 新たなプロセスが開始する(又は `active_children()` が呼ばれる)ときに、 join されていない全ての完了プロセスが join されるので、 あまり多くにはならないでしょう。また、終了したプロセスの `Process.is_alive()` はそのプロセスを join します。 そうは言っても、自分で開始した全てのプロセスを明示的に join することは おそらく良いプラクティスです。

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

> Windows 上では `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()` の行を削除します)。

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

> Unix 上では子プロセスはグローバルなリソースを使用する親プロセスが作成した 共有リソースを使用することができます。しかし、引数としてそのオブジェクトを 子プロセスのコンストラクタへ渡す方が良いです。

> (潜在的に) 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()

### 17.6.3.2. Windows

Windows では `os.fork()` がないので幾つか追加制限があります。

さらなる pickle 機能

> `Process.__init__()` へ渡す全ての引数は pickle 化できることを保証します。 これは特に束縛、又は非束縛メソッドが Windows 上の `target` 引数として 直接的に使用できないことを意味します。その代わり、まさに関数を定義してください。

> また `Process` をサブクラス化する場合、そのインスタンスが `Process.start()` メソッドが呼ばれたときに pickle 化できることを保証します。

グローバル変数

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

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

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

> 新たに Python インタプリタによって、意図しない副作用(新たなプロセスを開始する等) を起こさずにメインモジュールを安全にインポートできることを保証します。

> 例えば Windows で次のモジュールを実行しようとすると `RuntimeError` で失敗します。

>     from multiprocessing import Process

>     def foo():
>         print 'hello'

>     p = Process(target=foo)
>     p.start()

> 代わりに、次のように `if __name__ == '__main__':` を使用して プログラムの “エントリポイント” を保護すべきです。

>     from multiprocessing import Process, freeze_support

>     def foo():
>         print 'hello'

>     if __name__ == '__main__':
>         freeze_support()
>         p = Process(target=foo)
>         p.start()

> ( `freeze_support()` 行はプログラムが固まらずに実行されるなら通常は取り除かれます。)

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

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

## 17.6.4. 例

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

    #
    # This module shows how to use arbitrary callables with a subclass of
    # `BaseManager`.
    #

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

    ##

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

    # A simple generator function
    def baz():
        for i in xrange(10):
            yield i*i

    # Proxy type for generator objects
    class GeneratorProxy(BaseProxy):
        _exposed_ = ('next', '__next__')
        def __iter__(self):
            return self
        def next(self):
            return self._callmethod('next')
        def __next__(self):
            return self._callmethod('__next__')

    # Function to return the operator module
    def get_operator_module():
        return operator

    ##

    class MyManager(BaseManager):
        pass

    # register the Foo class; make `f()` and `g()` accessible via proxy
    MyManager.register('Foo1', Foo)

    # register the Foo class; make `g()` and `_h()` accessible via proxy
    MyManager.register('Foo2', Foo, exposed=('g', '_h'))

    # register the generator function baz; use `GeneratorProxy` to make proxies
    MyManager.register('baz', baz, proxytype=GeneratorProxy)

    # register get_operator_module(); make public functions accessible via proxy
    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,
        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.getslice(range(10), 2, 6) =', op.getslice(range(10), 2, 6)
        print 'op.repeat(range(5), 3) =', op.repeat(range(5), 3)
        print 'op._exposed_ =', op._exposed_

    ##

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

`Pool` の使用例を紹介します。

    #
    # A test of `multiprocessing.Pool` class
    #

    import multiprocessing
    import time
    import random
    import sys

    #
    # Functions used by test code
    #

    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

    #
    # Test code
    #

    def test():
        print 'cpu_count() = %d\n' % multiprocessing.cpu_count()

        #
        # Create pool
        #

        PROCESSES = 4
        print 'Creating pool with %d processes\n' % PROCESSES
        pool = multiprocessing.Pool(PROCESSES)
        print 'pool = %s' % pool
        print

        #
        # Tests
        #

        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

        #
        # Simple benchmarks
        #

        N = 100000
        print 'def pow3(x): return x**3'

        t = time.time()
        A = map(pow3, xrange(N))
        print '\tmap(pow3, xrange(%d)):\n\t\t%s seconds' % \
              (N, time.time() - t)

        t = time.time()
        B = pool.map(pow3, xrange(N))
        print '\tpool.map(pow3, xrange(%d)):\n\t\t%s seconds' % \
              (N, time.time() - t)

        t = time.time()
        C = list(pool.imap(pow3, xrange(N), chunksize=N//8))
        print '\tlist(pool.imap(pow3, xrange(%d), chunksize=%d)):\n\t\t%s' \
              ' seconds' % (N, N//8, time.time() - t)

        assert A == B == C, (len(A), len(B), len(C))
        print

        L = [None] * 1000000
        print 'def noop(x): pass'
        print 'L = [None] * 1000000'

        t = time.time()
        A = map(noop, L)
        print '\tmap(noop, L):\n\t\t%s seconds' % \
              (time.time() - t)

        t = time.time()
        B = pool.map(noop, L)
        print '\tpool.map(noop, L):\n\t\t%s seconds' % \
              (time.time() - t)

        t = time.time()
        C = list(pool.imap(noop, L, chunksize=len(L)//8))
        print '\tlist(pool.imap(noop, L, chunksize=%d)):\n\t\t%s seconds' % \
              (len(L)//8, time.time() - t)

        assert A == B == C, (len(A), len(B), len(C))
        print

        del A, B, C, L

        #
        # Test error handling
        #

        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, range(10))
        except ZeroDivisionError:
            print '\tGot ZeroDivisionError as expected from pool.map()'
        else:
            raise AssertionError, 'expected ZeroDivisionError'

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

        it = pool.imap(f, range(10))
        for i in range(10):
            try:
                x = it.next()
            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

        #
        # Testing timeouts
        #

        print 'Testing ApplyResult.get() with timeout:',
        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:',
        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

        #
        # Testing callback
        #

        print 'Testing callback:'

        A = []
        B = [56, 0, 1, 8, 27, 64, 125, 216, 343, 512, 729]

        r = pool.apply_async(mul, (7, 8), callback=A.append)
        r.wait()

        r = pool.map_async(pow3, range(10), callback=A.extend)
        r.wait()

        if A == B:
            print '\tcallbacks succeeded\n'
        else:
            print '\t*** callbacks failed\n\t\t%s != %s\n' % (A, B)

        #
        # Check there are no outstanding tasks
        #

        assert not pool._cache, 'cache = %r' % pool._cache

        #
        # Check close() methods
        #

        print 'Testing close():'

        for worker in pool._pool:
            assert worker.is_alive()

        result = pool.apply_async(time.sleep, [0.5])
        pool.close()
        pool.join()

        assert result.get() is None

        for worker in pool._pool:
            assert not worker.is_alive()

        print '\tclose() succeeded\n'

        #
        # Check terminate() method
        #

        print 'Testing terminate():'

        pool = multiprocessing.Pool(2)
        DELTA = 0.1
        ignore = pool.apply(pow3, [2])
        results = [pool.apply_async(time.sleep, [DELTA]) for i in range(100)]
        pool.terminate()
        pool.join()

        for worker in pool._pool:
            assert not worker.is_alive()

        print '\tterminate() succeeded\n'

        #
        # Check garbage collection
        #

        print 'Testing garbage collection:'

        pool = multiprocessing.Pool(2)
        DELTA = 0.1
        processes = pool._pool
        ignore = pool.apply(pow3, [2])
        results = [pool.apply_async(time.sleep, [DELTA]) for i in range(100)]

        results = pool = None

        time.sleep(DELTA * 2)

        for worker in processes:
            assert not worker.is_alive()

        print '\tgarbage collection succeeded\n'

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

        assert len(sys.argv) in (1, 2)

        if len(sys.argv) == 1 or sys.argv[1] == 'processes':
            print ' Using processes '.center(79, '-')
        elif sys.argv[1] == 'threads':
            print ' Using threads '.center(79, '-')
            import multiprocessing.dummy as multiprocessing
        else:
            print 'Usage:\n\t%s [processes | threads]' % sys.argv[0]
            raise SystemExit(2)

        test()

ロック、コンディションやキューのような同期の例を紹介します

    #
    # A test file for the `multiprocessing` package
    #

    import time, sys, random
    from Queue import Empty

    import multiprocessing               # may get overwritten

    #### TEST_VALUE

    def value_func(running, mutex):
        random.seed()
        time.sleep(random.random()*4)

        mutex.acquire()
        print '\n\t\t\t' + str(multiprocessing.current_process()) + ' has finished'
        running.value -= 1
        mutex.release()

    def test_value():
        TASKS = 10
        running = multiprocessing.Value('i', TASKS)
        mutex = multiprocessing.Lock()

        for i in range(TASKS):
            p = multiprocessing.Process(target=value_func, args=(running, mutex))
            p.start()

        while running.value > 0:
            time.sleep(0.08)
            mutex.acquire()
            print running.value,
            sys.stdout.flush()
            mutex.release()

        print
        print 'No more running processes'

    #### TEST_QUEUE

    def queue_func(queue):
        for i in range(30):
            time.sleep(0.5 * random.random())
            queue.put(i*i)
        queue.put('STOP')

    def test_queue():
        q = multiprocessing.Queue()

        p = multiprocessing.Process(target=queue_func, args=(q,))
        p.start()

        o = None
        while o != 'STOP':
            try:
                o = q.get(timeout=0.3)
                print o,
                sys.stdout.flush()
            except Empty:
                print 'TIMEOUT'

        print

    #### TEST_CONDITION

    def condition_func(cond):
        cond.acquire()
        print '\t' + str(cond)
        time.sleep(2)
        print '\tchild is notifying'
        print '\t' + str(cond)
        cond.notify()
        cond.release()

    def test_condition():
        cond = multiprocessing.Condition()

        p = multiprocessing.Process(target=condition_func, args=(cond,))
        print cond

        cond.acquire()
        print cond
        cond.acquire()
        print cond

        p.start()

        print 'main is waiting'
        cond.wait()
        print 'main has woken up'

        print cond
        cond.release()
        print cond
        cond.release()

        p.join()
        print cond

    #### TEST_SEMAPHORE

    def semaphore_func(sema, mutex, running):
        sema.acquire()

        mutex.acquire()
        running.value += 1
        print running.value, 'tasks are running'
        mutex.release()

        random.seed()
        time.sleep(random.random()*2)

        mutex.acquire()
        running.value -= 1
        print '%s has finished' % multiprocessing.current_process()
        mutex.release()

        sema.release()

    def test_semaphore():
        sema = multiprocessing.Semaphore(3)
        mutex = multiprocessing.RLock()
        running = multiprocessing.Value('i', 0)

        processes = [
            multiprocessing.Process(target=semaphore_func,
                                    args=(sema, mutex, running))
            for i in range(10)
            ]

        for p in processes:
            p.start()

        for p in processes:
            p.join()

    #### TEST_JOIN_TIMEOUT

    def join_timeout_func():
        print '\tchild sleeping'
        time.sleep(5.5)
        print '\n\tchild terminating'

    def test_join_timeout():
        p = multiprocessing.Process(target=join_timeout_func)
        p.start()

        print 'waiting for process to finish'

        while 1:
            p.join(timeout=1)
            if not p.is_alive():
                break
            print '.',
            sys.stdout.flush()

    #### TEST_EVENT

    def event_func(event):
        print '\t%r is waiting' % multiprocessing.current_process()
        event.wait()
        print '\t%r has woken up' % multiprocessing.current_process()

    def test_event():
        event = multiprocessing.Event()

        processes = [multiprocessing.Process(target=event_func, args=(event,))
                     for i in range(5)]

        for p in processes:
            p.start()

        print 'main is sleeping'
        time.sleep(2)

        print 'main is setting event'
        event.set()

        for p in processes:
            p.join()

    #### TEST_SHAREDVALUES

    def sharedvalues_func(values, arrays, shared_values, shared_arrays):
        for i in range(len(values)):
            v = values[i][1]
            sv = shared_values[i].value
            assert v == sv

        for i in range(len(values)):
            a = arrays[i][1]
            sa = list(shared_arrays[i][:])
            assert a == sa

        print 'Tests passed'

    def test_sharedvalues():
        values = [
            ('i', 10),
            ('h', -2),
            ('d', 1.25)
            ]
        arrays = [
            ('i', range(100)),
            ('d', [0.25 * i for i in range(100)]),
            ('H', range(1000))
            ]

        shared_values = [multiprocessing.Value(id, v) for id, v in values]
        shared_arrays = [multiprocessing.Array(id, a) for id, a in arrays]

        p = multiprocessing.Process(
            target=sharedvalues_func,
            args=(values, arrays, shared_values, shared_arrays)
            )
        p.start()
        p.join()

        assert p.exitcode == 0

    ####

    def test(namespace=multiprocessing):
        global multiprocessing

        multiprocessing = namespace

        for func in [ test_value, test_queue, test_condition,
                      test_semaphore, test_join_timeout, test_event,
                      test_sharedvalues ]:

            print '\n\t######## %s\n' % func.__name__
            func()

        ignore = multiprocessing.active_children()      # cleanup any old processes
        if hasattr(multiprocessing, '_debug_info'):
            info = multiprocessing._debug_info()
            if info:
                print info
                raise ValueError, 'there should be no positive refcounts left'

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

        assert len(sys.argv) in (1, 2)

        if len(sys.argv) == 1 or sys.argv[1] == 'processes':
            print ' Using processes '.center(79, '-')
            namespace = multiprocessing
        elif sys.argv[1] == 'manager':
            print ' Using processes and a manager '.center(79, '-')
            namespace = multiprocessing.Manager()
            namespace.Process = multiprocessing.Process
            namespace.current_process = multiprocessing.current_process
            namespace.active_children = multiprocessing.active_children
        elif sys.argv[1] == 'threads':
            print ' Using threads '.center(79, '-')
            import multiprocessing.dummy as namespace
        else:
            print 'Usage:\n\t%s [processes | manager | threads]' % sys.argv[0]
            raise SystemExit, 2

        test(namespace)

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

    #
    # Simple example which uses a pool of workers to carry out some tasks.
    #
    # Notice that the results will probably not come out of the output
    # queue in the same in the same order as the corresponding tasks were
    # put on the input queue.  If it is important to get the results back
    # in the original order then consider using `Pool.map()` or
    # `Pool.imap()` (which will save on the amount of code needed anyway).
    #

    import time
    import random

    from multiprocessing import Process, Queue, current_process, freeze_support

    #
    # Function run by worker processes
    #

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

    #
    # Function used to calculate result
    #

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

    #
    # Functions referenced by tasks
    #

    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)]

        # Create queues
        task_queue = Queue()
        done_queue = Queue()

        # Submit tasks
        for task in TASKS1:
            task_queue.put(task)

        # Start worker processes
        for i in range(NUMBER_OF_PROCESSES):
            Process(target=worker, args=(task_queue, done_queue)).start()

        # Get and print results
        print 'Unordered results:'
        for i in range(len(TASKS1)):
            print '\t', done_queue.get()

        # Add more tasks using `put()`
        for task in TASKS2:
            task_queue.put(task)

        # Get and print some more results
        for i in range(len(TASKS2)):
            print '\t', done_queue.get()

        # Tell child processes to stop
        for i in range(NUMBER_OF_PROCESSES):
            task_queue.put('STOP')

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

ワーカープロセスのプールが1つのソケットを共有して それぞれの `SimpleHTTPServer.HttpServer` インスタンスを 実行する方法の例を紹介します。

    #
    # Example where a pool of http servers share a single listening socket
    #
    # On Windows this module depends on the ability to pickle a socket
    # object so that the worker processes can inherit a copy of the server
    # object.  (We import `multiprocessing.reduction` to enable this pickling.)
    #
    # Not sure if we should synchronize access to `socket.accept()` method by
    # using a process-shared lock -- does not seem to be necessary.
    #

    import os
    import sys

    from multiprocessing import Process, current_process, freeze_support
    from BaseHTTPServer import HTTPServer
    from SimpleHTTPServer import SimpleHTTPRequestHandler

    if sys.platform == 'win32':
        import multiprocessing.reduction    # make sockets pickable/inheritable

    def note(format, *args):
        sys.stderr.write('[%s]\t%s\n' % (current_process().name, format%args))

    class RequestHandler(SimpleHTTPRequestHandler):
        # we override log_message() to show which process is handling the request
        def log_message(self, format, *args):
            note(format, *args)

    def serve_forever(server):
        note('starting server')
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            pass

    def runpool(address, number_of_processes):
        # create a single server object -- children will each inherit a copy
        server = HTTPServer(address, RequestHandler)

        # create child processes to act as workers
        for i in range(number_of_processes-1):
            Process(target=serve_forever, args=(server,)).start()

        # main process also acts as a worker
        serve_forever(server)

    def test():
        DIR = os.path.join(os.path.dirname(__file__), '..')
        ADDRESS = ('localhost', 8000)
        NUMBER_OF_PROCESSES = 4

        print 'Serving at http://%s:%d using %d worker processes' % \
              (ADDRESS[0], ADDRESS[1], NUMBER_OF_PROCESSES)
        print 'To exit press Ctrl-' + ['C', 'Break'][sys.platform=='win32']

        os.chdir(DIR)
        runpool(ADDRESS, NUMBER_OF_PROCESSES)

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

`multiprocessing` と `threading` を比較した簡単なベンチマークです。

    #
    # Simple benchmarks for the multiprocessing package
    #

    import time, sys, multiprocessing, threading, Queue, gc

    if sys.platform == 'win32':
        _timer = time.clock
    else:
        _timer = time.time

    delta = 1

    #### TEST_QUEUESPEED

    def queuespeed_func(q, c, iterations):
        a = '0' * 256
        c.acquire()
        c.notify()
        c.release()

        for i in xrange(iterations):
            q.put(a)

        q.put('STOP')

    def test_queuespeed(Process, q, c):
        elapsed = 0
        iterations = 1

        while elapsed < delta:
            iterations *= 2

            p = Process(target=queuespeed_func, args=(q, c, iterations))
            c.acquire()
            p.start()
            c.wait()
            c.release()

            result = None
            t = _timer()

            while result != 'STOP':
                result = q.get()

            elapsed = _timer() - t

            p.join()

        print iterations, 'objects passed through the queue in', elapsed, 'seconds'
        print 'average number/sec:', iterations/elapsed

    #### TEST_PIPESPEED

    def pipe_func(c, cond, iterations):
        a = '0' * 256
        cond.acquire()
        cond.notify()
        cond.release()

        for i in xrange(iterations):
            c.send(a)

        c.send('STOP')

    def test_pipespeed():
        c, d = multiprocessing.Pipe()
        cond = multiprocessing.Condition()
        elapsed = 0
        iterations = 1

        while elapsed < delta:
            iterations *= 2

            p = multiprocessing.Process(target=pipe_func,
                                        args=(d, cond, iterations))
            cond.acquire()
            p.start()
            cond.wait()
            cond.release()

            result = None
            t = _timer()

            while result != 'STOP':
                result = c.recv()

            elapsed = _timer() - t
            p.join()

        print iterations, 'objects passed through connection in',elapsed,'seconds'
        print 'average number/sec:', iterations/elapsed

    #### TEST_SEQSPEED

    def test_seqspeed(seq):
        elapsed = 0
        iterations = 1

        while elapsed < delta:
            iterations *= 2

            t = _timer()

            for i in xrange(iterations):
                a = seq[5]

            elapsed = _timer()-t

        print iterations, 'iterations in', elapsed, 'seconds'
        print 'average number/sec:', iterations/elapsed

    #### TEST_LOCK

    def test_lockspeed(l):
        elapsed = 0
        iterations = 1

        while elapsed < delta:
            iterations *= 2

            t = _timer()

            for i in xrange(iterations):
                l.acquire()
                l.release()

            elapsed = _timer()-t

        print iterations, 'iterations in', elapsed, 'seconds'
        print 'average number/sec:', iterations/elapsed

    #### TEST_CONDITION

    def conditionspeed_func(c, N):
        c.acquire()
        c.notify()

        for i in xrange(N):
            c.wait()
            c.notify()

        c.release()

    def test_conditionspeed(Process, c):
        elapsed = 0
        iterations = 1

        while elapsed < delta:
            iterations *= 2

            c.acquire()
            p = Process(target=conditionspeed_func, args=(c, iterations))
            p.start()

            c.wait()

            t = _timer()

            for i in xrange(iterations):
                c.notify()
                c.wait()

            elapsed = _timer()-t

            c.release()
            p.join()

        print iterations * 2, 'waits in', elapsed, 'seconds'
        print 'average number/sec:', iterations * 2 / elapsed

    ####

    def test():
        manager = multiprocessing.Manager()

        gc.disable()

        print '\n\t######## testing Queue.Queue\n'
        test_queuespeed(threading.Thread, Queue.Queue(),
                        threading.Condition())
        print '\n\t######## testing multiprocessing.Queue\n'
        test_queuespeed(multiprocessing.Process, multiprocessing.Queue(),
                        multiprocessing.Condition())
        print '\n\t######## testing Queue managed by server process\n'
        test_queuespeed(multiprocessing.Process, manager.Queue(),
                        manager.Condition())
        print '\n\t######## testing multiprocessing.Pipe\n'
        test_pipespeed()

        print

        print '\n\t######## testing list\n'
        test_seqspeed(range(10))
        print '\n\t######## testing list managed by server process\n'
        test_seqspeed(manager.list(range(10)))
        print '\n\t######## testing Array("i", ..., lock=False)\n'
        test_seqspeed(multiprocessing.Array('i', range(10), lock=False))
        print '\n\t######## testing Array("i", ..., lock=True)\n'
        test_seqspeed(multiprocessing.Array('i', range(10), lock=True))

        print

        print '\n\t######## testing threading.Lock\n'
        test_lockspeed(threading.Lock())
        print '\n\t######## testing threading.RLock\n'
        test_lockspeed(threading.RLock())
        print '\n\t######## testing multiprocessing.Lock\n'
        test_lockspeed(multiprocessing.Lock())
        print '\n\t######## testing multiprocessing.RLock\n'
        test_lockspeed(multiprocessing.RLock())
        print '\n\t######## testing lock managed by server process\n'
        test_lockspeed(manager.Lock())
        print '\n\t######## testing rlock managed by server process\n'
        test_lockspeed(manager.RLock())

        print

        print '\n\t######## testing threading.Condition\n'
        test_conditionspeed(threading.Thread, threading.Condition())
        print '\n\t######## testing multiprocessing.Condition\n'
        test_conditionspeed(multiprocessing.Process, multiprocessing.Condition())
        print '\n\t######## testing condition managed by a server process\n'
        test_conditionspeed(multiprocessing.Process, manager.Condition())

        gc.enable()

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

プロセスを分散して、SSH 経由でアクセスできるネットワーク上のマシンの “クラスタ” に 対する分散キューを経由するシステム上で構築された `managers.SyncManager`, `Process` やその他の使用方法の例/デモです。 この処理を実行するために全てのホストで秘密鍵認証を行う必要があります。

    #
    # Module to allow spawning of processes on foreign host
    #
    # Depends on `multiprocessing` package -- tested with `processing-0.60`
    #

    __all__ = ['Cluster', 'Host', 'get_logger', 'current_process']

    #
    # Imports
    #

    import sys
    import os
    import tarfile
    import shutil
    import subprocess
    import logging
    import itertools
    import Queue

    try:
        import cPickle as pickle
    except ImportError:
        import pickle

    from multiprocessing import Process, current_process, cpu_count
    from multiprocessing import util, managers, connection, forking, pool

    #
    # Logging
    #

    def get_logger():
        return _logger

    _logger = logging.getLogger('distributing')
    _logger.propogate = 0

    util.fix_up_logger(_logger)
    _formatter = logging.Formatter(util.DEFAULT_LOGGING_FORMAT)
    _handler = logging.StreamHandler()
    _handler.setFormatter(_formatter)
    _logger.addHandler(_handler)

    info = _logger.info
    debug = _logger.debug

    #
    # Get number of cpus
    #

    try:
        slot_count = cpu_count()
    except NotImplemented:
        slot_count = 1

    #
    # Manager type which spawns subprocesses
    #

    class HostManager(managers.SyncManager):
        '''
        Manager type used for spawning processes on a (presumably) foreign host
        '''
        def __init__(self, address, authkey):
            managers.SyncManager.__init__(self, address, authkey)
            self._name = 'Host-unknown'

        def Process(self, group=None, target=None, name=None, args=(), kwargs={}):
            if hasattr(sys.modules['__main__'], '__file__'):
                main_path = os.path.basename(sys.modules['__main__'].__file__)
            else:
                main_path = None
            data = pickle.dumps((target, args, kwargs))
            p = self._RemoteProcess(data, main_path)
            if name is None:
                temp = self._name.split('Host-')[-1] + '/Process-%s'
                name = temp % ':'.join(map(str, p.get_identity()))
            p.set_name(name)
            return p

        @classmethod
        def from_address(cls, address, authkey):
            manager = cls(address, authkey)
            managers.transact(address, authkey, 'dummy')
            manager._state.value = managers.State.STARTED
            manager._name = 'Host-%s:%s' % manager.address
            manager.shutdown = util.Finalize(
                manager, HostManager._finalize_host,
                args=(manager._address, manager._authkey, manager._name),
                exitpriority=-10
                )
            return manager

        @staticmethod
        def _finalize_host(address, authkey, name):
            managers.transact(address, authkey, 'shutdown')

        def __repr__(self):
            return '<Host(%s)>' % self._name

    #
    # Process subclass representing a process on (possibly) a remote machine
    #

    class RemoteProcess(Process):
        '''
        Represents a process started on a remote host
        '''
        def __init__(self, data, main_path):
            assert not main_path or os.path.basename(main_path) == main_path
            Process.__init__(self)
            self._data = data
            self._main_path = main_path

        def _bootstrap(self):
            forking.prepare({'main_path': self._main_path})
            self._target, self._args, self._kwargs = pickle.loads(self._data)
            return Process._bootstrap(self)

        def get_identity(self):
            return self._identity

    HostManager.register('_RemoteProcess', RemoteProcess)

    #
    # A Pool class that uses a cluster
    #

    class DistributedPool(pool.Pool):

        def __init__(self, cluster, processes=None, initializer=None, initargs=()):
            self._cluster = cluster
            self.Process = cluster.Process
            pool.Pool.__init__(self, processes or len(cluster),
                               initializer, initargs)

        def _setup_queues(self):
            self._inqueue = self._cluster._SettableQueue()
            self._outqueue = self._cluster._SettableQueue()
            self._quick_put = self._inqueue.put
            self._quick_get = self._outqueue.get

        @staticmethod
        def _help_stuff_finish(inqueue, task_handler, size):
            inqueue.set_contents([None] * size)

    #
    # Manager type which starts host managers on other machines
    #

    def LocalProcess(**kwds):
        p = Process(**kwds)
        p.set_name('localhost/' + p.name)
        return p

    class Cluster(managers.SyncManager):
        '''
        Represents collection of slots running on various hosts.

        `Cluster` is a subclass of `SyncManager` so it allows creation of
        various types of shared objects.
        '''
        def __init__(self, hostlist, modules):
            managers.SyncManager.__init__(self, address=('localhost', 0))
            self._hostlist = hostlist
            self._modules = modules
            if __name__ not in modules:
                modules.append(__name__)
            files = [sys.modules[name].__file__ for name in modules]
            for i, file in enumerate(files):
                if file.endswith('.pyc') or file.endswith('.pyo'):
                    files[i] = file[:-4] + '.py'
            self._files = [os.path.abspath(file) for file in files]

        def start(self):
            managers.SyncManager.start(self)

            l = connection.Listener(family='AF_INET', authkey=self._authkey)

            for i, host in enumerate(self._hostlist):
                host._start_manager(i, self._authkey, l.address, self._files)

            for host in self._hostlist:
                if host.hostname != 'localhost':
                    conn = l.accept()
                    i, address, cpus = conn.recv()
                    conn.close()
                    other_host = self._hostlist[i]
                    other_host.manager = HostManager.from_address(address,
                                                                  self._authkey)
                    other_host.slots = other_host.slots or cpus
                    other_host.Process = other_host.manager.Process
                else:
                    host.slots = host.slots or slot_count
                    host.Process = LocalProcess

            self._slotlist = [
                Slot(host) for host in self._hostlist for i in range(host.slots)
                ]
            self._slot_iterator = itertools.cycle(self._slotlist)
            self._base_shutdown = self.shutdown
            del self.shutdown

        def shutdown(self):
            for host in self._hostlist:
                if host.hostname != 'localhost':
                    host.manager.shutdown()
            self._base_shutdown()

        def Process(self, group=None, target=None, name=None, args=(), kwargs={}):
            slot = self._slot_iterator.next()
            return slot.Process(
                group=group, target=target, name=name, args=args, kwargs=kwargs
                )

        def Pool(self, processes=None, initializer=None, initargs=()):
            return DistributedPool(self, processes, initializer, initargs)

        def __getitem__(self, i):
            return self._slotlist[i]

        def __len__(self):
            return len(self._slotlist)

        def __iter__(self):
            return iter(self._slotlist)

    #
    # Queue subclass used by distributed pool
    #

    class SettableQueue(Queue.Queue):
        def empty(self):
            return not self.queue
        def full(self):
            return self.maxsize > 0 and len(self.queue) == self.maxsize
        def set_contents(self, contents):
            # length of contents must be at least as large as the number of
            # threads which have potentially called get()
            self.not_empty.acquire()
            try:
                self.queue.clear()
                self.queue.extend(contents)
                self.not_empty.notifyAll()
            finally:
                self.not_empty.release()

    Cluster.register('_SettableQueue', SettableQueue)

    #
    # Class representing a notional cpu in the cluster
    #

    class Slot(object):
        def __init__(self, host):
            self.host = host
            self.Process = host.Process

    #
    # Host
    #

    class Host(object):
        '''
        Represents a host to use as a node in a cluster.

        `hostname` gives the name of the host.  If hostname is not
        "localhost" then ssh is used to log in to the host.  To log in as
        a different user use a host name of the form
        "username@somewhere.org"

        `slots` is used to specify the number of slots for processes on
        the host.  This affects how often processes will be allocated to
        this host.  Normally this should be equal to the number of cpus on
        that host.
        '''
        def __init__(self, hostname, slots=None):
            self.hostname = hostname
            self.slots = slots

        def _start_manager(self, index, authkey, address, files):
            if self.hostname != 'localhost':
                tempdir = copy_to_remote_temporary_directory(self.hostname, files)
                debug('startup files copied to %s:%s', self.hostname, tempdir)
                p = subprocess.Popen(
                    ['ssh', self.hostname, 'python', '-c',
                     '"import os; os.chdir(%r); '
                     'from distributing import main; main()"' % tempdir],
                    stdin=subprocess.PIPE
                    )
                data = dict(
                    name='BoostrappingHost', index=index,
                    dist_log_level=_logger.getEffectiveLevel(),
                    dir=tempdir, authkey=str(authkey), parent_address=address
                    )
                pickle.dump(data, p.stdin, pickle.HIGHEST_PROTOCOL)
                p.stdin.close()

    #
    # Copy files to remote directory, returning name of directory
    #

    unzip_code = '''"
    import tempfile, os, sys, tarfile
    tempdir = tempfile.mkdtemp(prefix='distrib-')
    os.chdir(tempdir)
    tf = tarfile.open(fileobj=sys.stdin, mode='r|gz')
    for ti in tf:
        tf.extract(ti)
    print tempdir
    "'''

    def copy_to_remote_temporary_directory(host, files):
        p = subprocess.Popen(
            ['ssh', host, 'python', '-c', unzip_code],
            stdout=subprocess.PIPE, stdin=subprocess.PIPE
            )
        tf = tarfile.open(fileobj=p.stdin, mode='w|gz')
        for name in files:
            tf.add(name, os.path.basename(name))
        tf.close()
        p.stdin.close()
        return p.stdout.read().rstrip()

    #
    # Code which runs a host manager
    #

    def main():
        # get data from parent over stdin
        data = pickle.load(sys.stdin)
        sys.stdin.close()

        # set some stuff
        _logger.setLevel(data['dist_log_level'])
        forking.prepare(data)

        # create server for a `HostManager` object
        server = managers.Server(HostManager._registry, ('', 0), data['authkey'])
        current_process()._server = server

        # report server address and number of cpus back to parent
        conn = connection.Client(data['parent_address'], authkey=data['authkey'])
        conn.send((data['index'], server.address, slot_count))
        conn.close()

        # set name etc
        current_process().set_name('Host-%s:%s' % server.address)
        util._run_after_forkers()

        # register a cleanup function
        def cleanup(directory):
            debug('removing directory %s', directory)
            shutil.rmtree(directory)
            debug('shutting down host manager')
        util.Finalize(None, cleanup, args=[data['dir']], exitpriority=0)

        # start host manager
        debug('remote host manager starting in %s', data['dir'])
        server.serve_forever()

### ナビゲーション

- 索引
- モジュール |
- 次へ |
- 前へ |
- Python v2.6.2 documentation »
- Python 標準ライブラリ »
- 17. オプションのオペレーティングシステムサービス »