在嵌套循环中使用multiprocessor.Pool的正确方法

Question

我正在使用multiprocessor.Pool（）模块加速“令人尴尬的并行”循环。实际上，我有一个嵌套循环，并且正在使用multiprocessor.Pool加速内部循环。例如，在不并行循环的情况下，我的代码如下：

outer_array=[random_array1]
inner_array=[random_array2]
output=[empty_array]    

for i in outer_array:
    for j in inner_array:
        output[j][i]=full_func(j,i)

并行化：

import multiprocessing
from functools import partial

outer_array=[random_array1]
inner_array=[random_array2]
output=[empty_array]    

for i in outer_array:
    partial_func=partial(full_func,arg=i)     
    pool=multiprocessing.Pool() 
    output[:][i]=pool.map(partial_func,inner_array)
    pool.close()

我的主要问题是，这是否正确，我应该在循环中包含multiprocessing.Pool（），或者我应该在循环外创建池，即：

pool=multiprocessing.Pool() 
for i in outer_array:
     partial_func=partial(full_func,arg=i)     
     output[:][i]=pool.map(partial_func,inner_array)

另外，我不确定是否应该在上面的第二个示例中的每个循环的末尾包含行“pool.close（）”；这样做的好处是什么？

谢谢！

Answer 1

import time
from pathos.parallel import stats
from pathos.parallel import ParallelPool as Pool


def work(x, y):
    return x * y


pool = Pool(5)
pool.ncpus = 4
pool.servers = ('localhost:5654',)
t1 = time.time()
results = pool.imap(work, range(1, 2), range(1, 11))
print("INFO: List is: %s" % list(results))
print(stats())
t2 = time.time()
print("TIMER: Function completed time is: %.5f" % (t2 - t1))

Answer 2

import itertools
import multiprocessing as mp

def job(params):
    a = params[0]
    b = params[1]
    return a*b

def multicore():
    a = range(1000)
    b = range(2000)
    paramlist = list(itertools.product(a,b))
    print(paramlist[0])
    pool = mp.Pool(processes = 4)
    res=pool.map(job, paramlist)
    for i in res:
        print(i)

if __name__=='__main__':
    multicore()

这个怎么样？

Answer 3

理想情况下，应该只调用一次Pool()构造函数，而不是再次调用。创建工作进程时有大量的开销，每次调用Pool()时都要支付这些开销。由单个Pool()调用创建的进程保持在附近！当他们完成你在程序的某个部分交给他们的工作时，他们会一直呆在那里，等待着有更多的工作要做。

至于Pool.close()，您应该在-并且只有在-您永远不会向Pool实例提交更多工作时调用它。所以Pool.close()通常在主程序的可并行部分完成时调用。当所有已分配的工作完成时，工作进程将终止。

调用Pool.join()等待工作进程终止也是一个很好的实践。除其他原因外，通常没有好的方法可以在并行代码中报告异常（异常发生在与主程序正在执行的操作模糊相关的上下文中），并且Pool.join()提供了一个同步点，可以报告在工作进程中发生的一些异常，否则您将看不到这些异常。

玩得开心：-）