如何组织多处理池来收集字典中的输出

Question

我试图重现我的"真实世界";使用下面可重复的小示例的问题。这个例子试图利用我在这里找到的功能。我的笔记本电脑有16个核心，而实际的例子是在一个单核上运行了16天，所以考虑到大多数核心，我希望能把运行时间缩短到1到2天。然而，首先，我需要理解我在下面的小示例中做错了什么。

该示例首先设置一个名为all_combos的元组列表。我们的想法是将all_combos中的每个元组传递给函数do_one_run()。我的目标是使用多处理并行化do_one_run()。不幸的是，下面这个可重复使用的小示例会返回我无法解决的错误消息。我怀疑我误解了Pool的工作方式，特别是将每个参数元组映射到do_one_run()的参数，或者我可能误解了如何收集do_one_run()的输出，或者更有可能两者兼有?

任何见解非常欢迎!

import random
import numpy as np
import multiprocessing as mp
slns = {}
var1 = [5, 6, 7]
var2 = [2, 3, 4]
var3 = [10, 9, 8]
all_combos = []
key = 0
for v1 in var1:
for v2 in var2:
for v3 in var3:
all_combos.append([key, v1, v2, v3])
key += 1
def example_func(v1_passed, v2_passed, v3_passed):
tmp = np.random.random((v1_passed, v2_passed, v3_passed))*100
my_arr = tmp.astype(int)
piece_arr = my_arr[1,:,1:3]
return piece_arr

def do_one_run(key, v1_passed, v2_passed, v3_passed):
results = example_func(v1_passed, v2_passed, v3_passed)
slns.update({key: [v1_passed, v2_passed, v3_passed, results]})
pool = mp.Pool(4)  # 4 cores devoted to job?
result = pool.starmap(do_one_run, all_combos)

Answer 1

不能通过多处理共享像slns这样的变量。您必须收集do_one_run函数的所有返回值:

import random
import numpy as np
import multiprocessing as mp
# slns = {}  <- Remove this line
...
# Return result
def do_one_run(key, v1_passed, v2_passed, v3_passed):
results = example_func(v1_passed, v2_passed, v3_passed)
return key, [v1_passed, v2_passed, v3_passed, results]
if __name__ == '__main__':
with mp.Pool(4) as pool:
results = pool.starmap(do_one_run, all_combos)  # <- Collect results
result = dict(itertools.chain(*map(dict.items, result))) # <- Merge results

>>> result
{0: [5,
2,
10,
array([[77, 90],
[34, 28]])],
1: [5,
2,
9,
array([[64, 43],
[45, 53]])],
2: [5,
2,
8,
array([[ 8, 78],
[39,  3]])],
...
}

Answer 2

把最后两行改成这样:-

if __name__ == '__main__':
mp.Pool().starmap(do_one_run, all_combos)
print('Done') # So you know when it's finished

你可能也会发现这个讨论很有帮助:- python multiprocessing on windows, if __name__ == "__main__">

还要注意，本例中Pool()的构造没有参数。通过这种方式，底层实现将最好地利用它运行的CPU体系结构