python - 进程在 MPIPoolExecutor 初始化之前生成。此外，我可以将 HPC 与 ProccesPoolExecutor 一起使用吗？

Question

我正在使用 mpi4py 库进行并行编程。但似乎进程是在我创建 MPIPoolExecutor 之前生成的，因为我在 Executor 初始化之前收到了 4 次相同的警告。这是正常现象吗？我怎样才能防止这种情况发生？

start_time = time.time()

Nodes = pd.read_csv(Csvfile)
fields = SelectCoordinates(Nodes,min_lon = 10, max_lon = 11, min_lat = 49, max_lat = 50)
ts = pd.read_csv(File to be read)

n=6 # number of steps of forecast
p = 2 # number of lags


L = 1000# size of data we will use
ts = ts.iloc[0*L:1*L]

Farms = ts.shape[1]
colnames = ts.columns.values.tolist()
ts=ts.values


# n = fstep # number of time steps ahead
size = int(len(ts) * 0.66) #Determining train and test set
train = np.zeros(shape=(size,Farms))
test = np.zeros(shape=(len(ts)-size,Farms))
train, test = ts[0:size], ts[size:len(ts)]

#=================Intialization================
G = np.zeros(shape=(Farms*p,Farms*p))#
Gt = np.zeros(shape=(Farms*p,Farms*p))#

gn = np.zeros(shape=(Farms*p,1))

b = np.zeros(shape=(Farms*p,Farms))
X = np.zeros(shape=(Farms*p,Farms)) # Coefficient Matrix
#x = np.ones(shape=(Farms*p,1)) # Coefficient Matrix
#x[0] = 1
lamda = 0.001
c = 0.1
v=0.999
T = len(train) 

workers =  MPIPoolExecutor(max_workers = 4)       
for t in range(p,T):

    result_list = []

    to_do = list()


    count = 0 
    for l in range(p):
        for k in range(4):
            gn[count]=train[t-l-1,k]
            count+=1
    G = G*v +  gn @ gn.T
    Gt = (1/(t-p+1))*G

    if __name__ == '__main__':
        for i in range(4):

            job = workers.submit(OULtraining, train[t,i], X[:,i,np.newaxis], b[:,i,np.newaxis], i, gn)

            job.add_done_callback(log_result)
            to_do.append(job)


        for future in cf.as_completed(to_do):
            result_list.append(future.result())

        result_list = sorted(result_list,key=itemgetter(2))
        for i  in range(4):
            X[:,i,np.newaxis], b[:,i,np.newaxis], s = result_list[i]

    if (t%100==0):
        print(t)
        print("--- %s seconds ---" % (time.time() - start_time)) 

我想实现一个递归并行算法，我希望只创建一次 MPIpool，并在每个时间步骤中执行一项工作，等待所有作业完成，然后使用输入和先前的输出再次调用进程，然后再次调用进程下一个时间步也是如此，等等

我也得到了4倍的打印品，但我只想由主人来打印。看起来整个程序甚至在我创建执行器之前就运行了 4 次。如果我使用具有相同代码的 ProcessPoolExecutor，一切正常!

还有最后一个问题。我可以使用 ProccesPoolExecutor 在具有不同机器的 HPC 中工作吗？或者我必须使用 MPIPoolExecutor？另外我发现在 OpenMPI 中，name == "main"不起作用，因为在所有进程中 name == "main"。 OpenMPI 等效项为:os.environ['OMPI_COMM_WORLD_RANK'] == '0'。但是当我使用这个时，我收到了这个错误:KeyError: 'OMPI_COMM_WORLD_RANK' 您有关于此的任何信息吗？

Answer 1

来自 http://mpi4py.readthedocs.io/en/stable/mpi4py.futures.html#mpipoolexecutor 的文档

MPIPoolExecutor takes advantage of the dynamic process management features introduced in the MPI-2 standard. In particular, the MPI.Intracomm.Spawn() method of MPI.COMM_SELF() is used in the master (or parent) process to spawn new worker (or child) processes running a Python interpreter. The master process uses a separate thread (one for each MPIPoolExecutor instance) to communicate back and forth with the workers. The worker processes serve the execution of tasks in the main (and only) thread until they are signaled for completion.

我的理解是，池的 MPI 任务是动态生成的，因此您实际上应该只启动一个 MPI 任务(例如 mpirun -np 1 ...)