Pythonでのプロセスの同期とプーリング

プロセス間の同期

マルチプロセッシングは、APIを使用したスポーンプロセスをサポートするパッケージです。このパッケージは、ローカルとリモートの両方の同時実行に使用されます。このモジュールを使用すると、プログラマーは特定のマシンで複数のプロセッサーを使用できます。 WindowsおよびUNIXOSで動作します。

このパッケージには、同等の同期プリミティブがすべて含まれています。

サンプルコード

from multiprocessing import Process, Lock
   def my_function(x, y):
      x.acquire()
      print ('hello world', y)
      x.release()
      if __name__ == '__main__':
      lock = Lock()
   for num in range(10):
Process(target= my_function, args=(lock, num)).start()

ここでは、1つのインスタンスをロックして、一度に1つのプロセスのみが標準出力を表示できるようにすることができます。

プーリング

プーリングには、Poolクラスを使用します。送信されたタスク全体を実行するプロセスのプールを作成できる場合。

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

プールオブジェクトは、送信できるジョブを選択するためにワーカーのプールを制御し、タイムアウト、コールバック、および並列マップの実装を含む非同期結果をサポートします。

cpu_count（）は、プロセスがnoneの場合に使用され、initializer（* initargs）は、initializerがnoneでない場合にこの関数が呼び出します。

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

これは、apply（）組み込み関数と同じです。これは、結果の準備ができるまでブロックされます。並列で実行する場合は、apply_async（）メソッドの方が適しています。

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

結果オブジェクトを返します。

map（func、iterable [、chunksize]）

map（）は組み込み関数であり、反復可能な引数を1つだけサポートします。結果の準備ができるまでブロックします。

この方法では、反復可能オブジェクトがいくつかの小さなチャンクに分割され、これらの小さな部分が個別のタスクとしてプロセスプールに送信されます。

map_async（func、iterable [、chunksize [、callback]]）

結果オブジェクトを返します。

imap（func、iterable [、chunksize]）

itertools.imap（）と同じです。

引数のサイズは、map（）で使用されているものと同じです。

imap_unordered（func、iterable [、chunksize]）

これは、再調整イテレータを注文する必要があることを除いて、imap（）と同じです。

close（）

ワーカーがすべてのタスクを完了すると、ワーカーはプロセスを終了します。

terminate（）

タスクを完了せずにワーカープロセスをすぐに停止する場合は、この方法が使用されます。

join（）

join（）メソッドを使用する前に、close（）関数とterminate（）関数を使用する必要があります。

クラスmultiprocessing.pool.AsyncResult

Pool.apply_async（）およびPool.map_async（）によって返されます。

get（[timeout]）

この関数は、到着時に結果を返します。

待機（[タイムアウト]）

この待機機能を使用して、結果が利用可能になるまで、またはタイムアウト秒が経過するまで待機します。

ready（）

この関数は、呼び出しが完了したかどうかを返します。

success（）

この関数は、呼び出しがエラーなしで完了したときに戻ります。

サンプルコード

# -*- coding: utf-8 -*-
"""
Created on Sun Sep 30 12:17:58 2018
@author: Tutorials Point
"""
from multiprocessing import Pool
import time
def myfunction(m):
return m*m
if __name__ == '__main__':
my_pool = Pool(processes=4) # start 4 worker processes
result = my_pool.apply_async(myfunction, (10,)) # evaluate "f(10)" asynchronously in a single process
print (result.get(timeout=1))
print (my_pool.map(myfunction, range(10))) # prints "[0, 1, 4,..., 81]"
my_it = my_pool.imap(myfunction, range(10))
print (my_it.next() ) # prints "0"
print (my_it.next() ) # prints "1"
print (my_it.next(timeout=1) ) # prints "4" unless your computer is *very* slow
result = my_pool.apply_async(time.sleep, (10,))
print (result.get(timeout=1) ) # raises multiprocessing.TimeoutError