mpi4pyで関数呼び出しを並列化する

Question

mpi4pyを使用して最適化問題を並列化したいと考えています。私はmpi4pyを使用して、私の機能を並列化したい場合は、私の機能を最小限に抑えるために、私は、scipyのダウンロード

今

from scipy.optimize import minimize 

def f(x, data) : 
    #returns f(x) 
x = minimize(f, x0, args=(data)) 

から最小化ルーチンを使用します。最小化アルゴリズムの実装は順次であり、1つのプロセスでしか実行できないので、関数呼び出しはほとんどの時間を要するため、問題はない。しかし、この問題を実装する方法を、パラレルとシーケンシャルの部分で把握することはできません。ここで

は私の試みです：

from scipy.optimize import minimize 
from mpi4py import MPI 

comm = MPI.COMM_WORLD 
size = comm.Get_size() 
rank = comm.Get_rank() 

N = 100 # for testing 
step = N//size # say that N is divisible by size 
def mpi_f(x, data) : 
    x0 = x[rank*step:(rank+1)*step] 
    res = f(x0, data) 
    res = comm.gather(res, root=0) 
    if rank == 0 : 
     return res 

if rank == 0 : 
    x = np.zeros(N) 
    xs = minimize(mpi_f, x, args=(data)) 

mpi_fだけだから私はどのように私が進むべきお願いいたしプロセス0上で実行されますので、これは明らかに動作していませんか？

ありがとうございました。

Answer 1

コードでは、ルートプロセスがパラレル化されたコスト関数を呼び出す唯一のプロセスであるため、comm.gather()を呼び出す唯一のルートプロセスです。したがって、プログラムはデッドロックに直面しています。あなたはこの問題をよく承知しています。

このデッドロックを克服するには、他のプロセスがminimizeに必要な回数だけコスト関数を呼び出す必要があります。このコール数は事前には分かっていないので、whileループがこれらのプロセスに適しているようです。

whileループの停止条件を定義します。このフラグは、ルートプロセスがminimize()関数が終了したという事実を認識している唯一のプロセスであるため、ルートプロセスからすべてのプロセスにbrodcastされます。全てのプロセスが各反復において最小化関数の終わりをテストしなければならないので、ブロードキャストはコスト関数で実行されなければならない。 minimizeは関数の戻り値を使用するため、フラグはpassed by reference via a mutable type

です。最後に、ここで問題を解決する方法を示します。それはmpirun -np 4 python main.pyによって実行されます。私はminimize()の代わりにfmin()を使用しました。私のscipyのバージョンは古くなっているからです。

#from scipy.optimize import minimize 
from scipy.optimize import fmin 
from mpi4py import MPI 
import numpy as np 

comm = MPI.COMM_WORLD 
size = comm.Get_size() 
rank = comm.Get_rank() 

N = 100 # for testing 
step = N//size # say that N is divisible by size 

def parallel_function_caller(x,stopp): 
    stopp[0]=comm.bcast(stopp[0], root=0) 
    summ=0 
    if stopp[0]==0: 
     #your function here in parallel 
     x=comm.bcast(x, root=0) 
     array= np.arange(x[0]-N/2.+rank*step-42,x[0]-N/2.+(rank+1)*step-42,1.) 
     summl=np.sum(np.square(array)) 
     summ=comm.reduce(summl,op=MPI.SUM, root=0) 
     if rank==0: 
      print "value is "+str(summ) 
    return summ 

if rank == 0 : 
    stop=[0] 
    x = np.zeros(1) 
    x[0]=20 
    #xs = minimize(parallel_function_caller, x, args=(stop)) 
    xs = fmin(parallel_function_caller,x0= x, args=(stop,)) 
    print "the argmin is "+str(xs) 
    stop=[1] 
    parallel_function_caller(x,stop) 

else : 
    stop=[0] 
    x=np.zeros(1) 
    while stop[0]==0: 
     parallel_function_caller(x,stop)