在嵌套循环中使用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

理想情况下，您应该只调用Pool()构造函数一次，而不是一遍又一遍。在创建工作进程时会产生大量开销，并且每次调用Pool()时都要支付这些开销。由单个Pool()调用创建的进程将保留不变！当他们完成了你在程序的某一部分交给他们的工作时，他们会留下来，等待更多的工作要做。

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

调用Pool.join()等待工作进程终止也是一种很好的做法。除了其他原因之外，通常没有很好的方法来报告并行化代码中的异常(异常发生在与您的主程序正在执行的操作模糊相关的上下文中)，并且Pool.join()提供了一个同步点，可以报告在工作进程中发生的一些您从未见过的异常。

玩得开心:-)

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

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