python-3.x - 如何重用多处理池？

Question

底部是我现在拥有的代码。它似乎工作正常。但是，我并不完全理解它。我以为没有.join() ，我会冒着代码在池完成执行之前进入下一个 for 循环的风险。我们不需要那 3 行注释掉吗？

另一方面，如果我选择 .close()和 .join()方式，有什么方法可以“重新打开”那个关闭的池而不是 Pool(6)每次？

import multiprocessing as mp
import random as rdm
from statistics import stdev, mean
import time


def mesh_subset(population, n_chosen=5):
    chosen = rdm.choices(population, k=n_chosen)
    return mean(chosen)


if __name__ == '__main__':
    population = [x for x in range(20)]
    N_iteration = 10
    start_time = time.time()
    pool = mp.Pool(6)
    for i in range(N_iteration):
        print([round(x,2) for x in population])
        print(stdev(population))
        # pool = mp.Pool(6)
        population = pool.map(mesh_subset, [population]*len(population))
        # pool.close()
        # pool.join()
    print('run time:', time.time() - start_time)

Answer 1

工作人员池的设置成本相对较高，因此应该(如果可能)只设置一次，通常在脚本的开头。
pool.map命令块，直到所有任务完成。毕竟，它返回一个结果列表。除非 mesh_subset，否则它不能这样做已在所有输入上调用并为每个输入返回结果。相比之下，像 pool.apply_async 这样的方法不要阻止。 apply_async返回一个带有 get 的 ApplyResult 对象阻塞直到从工作进程获得结果的方法。
pool.close sets the worker handler's state关闭。这会导致处理程序 signal the workers终止。
pool.join阻塞直到所有工作进程都已终止。

所以你不需要打电话——事实上你不应该打电话——pool.close和 pool.join直到你完成游泳池。一旦 worker 被发送终止信号(通过 pool.close)，就无法“重新打开”它们。您将需要启动一个新池。

在您的情况下，由于您确实希望循环等待所有任务完成，因此使用 pool.apply_async 没有任何好处。而不是 pool.map .但是如果你要使用 pool.apply_async ，您可以通过调用 get 获得与之前相同的结果而不是求助于关闭和重新启动池:

# you could do this, but using pool.map is simpler
for i in range(N_iteration):
    apply_results = [pool.apply_async(mesh_subset, [population]) for i in range(len(population))]
    # the call to result.get() blocks until its worker process (running
    # mesh_subset) returns a value
    population = [result.get() for result in apply_results]

当循环完成时， len(population)不变。

如果您不希望每个循环在所有任务完成之前阻塞，您可以使用 apply_async的 callback特征:

N_pop = len(population)
result = []
for i in range(N_iteration):
    for i in range(N_pop):
        pool.apply_async(mesh_subset, [population]),
                         callback=result.append)
pool.close()
pool.join()
print(result)

现在，当任何 mesh_subset返回 return_value , result.append(return_value)叫做。调用 apply_async不要
阻止，所以 N_iteration * N_pop任务被推送到 pool任务
一下子排队。但由于池有 6 个 worker ，最多 6 个调用 mesh_subset在任何给定时间运行。当 worker 完成任务时，
哪个 worker 先完成调用 result.append(return_value) .所以 result 中的值是无序的。这与 pool.map 不同哪一个
返回一个列表，其返回值与其对应的顺序相同
参数列表。

除非有异常(exception)， result最终将包含 N_iteration * N_pop返回值一次
所有任务完成。以上， pool.close()和 pool.join()习惯了
等待所有任务完成。