Pool/starmap 的 Python 多处理行为

Question

我有一个使用多处理库来计算一些东西的程序。大约有 10K 个输入需要计算，每个输入需要 0.2 秒到 10 秒的时间。

我当前的方法使用池:

# Inputs
signals = [list(s) for s in itertools.combinations_with_replacement(possible_inputs, 3)]

# Compute
with mp.Pool(processes = N) as p:
    p.starmap(compute_solutions, [(s, t0, tf, folder) for s in signals])
    print ("    | Computation done.")

我注意到，在最后检查 300/400 个输入时，程序变得慢了很多。我的问题是:Pool 和 starmap() 的行为如何？

根据我的观察，我相信如果我有 10K 个输入并且 N = 4(4 个进程)，那么前 2 500 个输入将分配给第一个进程，紧邻的 2 500 个输入将分配给第一个进程。其次，......每个进程都以串行方式处理其输入。这意味着如果某些进程先于其他进程清除了队列，它们就不会执行新任务。

这是正确的吗？

如果这是正确的，我怎样才能拥有一个可以用这个伪代码表示的更智能的系统:

workers = Initialize N workers
tasks = A list of the tasks to perform

for task in tasks:
    if a worker is free:
        submit task to this worker
    else:
        wait

感谢您的帮助:)

注意:不同的 map 功能之间有什么区别。我相信 map()、imap_unordered()、imap、starmap 存在。

它们之间有什么区别，我们什么时候应该使用其中一种？

Answer 1

Which means that if some processes have cleared the Queue before others, they do not get new tasks to perform.

Is this correct?

没有。 multiprocess.Pool() 的主要目的是将传递的工作负载分散到其工作池中 - 这就是它附带所有这些映射选项的原因 - 其各种方法之间的唯一区别在于如何工作量实际上是分配的，以及如何收集由此产生的返回。

在您的情况下，您使用 [(s, t0, tf,folder) for s in Signals] 生成的迭代将具有其每个元素(最终取决于 信号大小)发送到池中的下一个空闲工作线程(作为compute_solutions(s, t0, tf,folder)调用)，一次一个(或者如果则发送多个) chunksize 参数被传递)，直到整个可迭代被耗尽。不过，您无法控制哪个工作人员执行哪个部分。

工作负载也可能不均匀分布 - 一个工作人员可能比另一个工作人员处理更多的条目，具体取决于资源使用情况、执行速度、各种内部事件...

但是，使用 multiprocessing.Pool 的 map、imap 和 starmap 方法，您甚至会得到这样的错觉:并有序地传播，因为它们在内部同步每个工作线程的返回值以匹配 source 可迭代对象(即结果的第一个元素将包含被调用函数的返回结果与可迭代对象的第一个元素) )。如果您想了解下面实际发生的情况，您可以尝试这些方法的异步/无序版本。

因此，默认情况下您会获得更智能的系统，但您始终可以使用multiprocessing.Pool.apply_async()如果您想完全控制您的工作人员池。

顺便说一句，如果您正在考虑优化对可迭代本身的访问(因为池映射选项将消耗其中的很大一部分)，您可以检查 this answer .

最后，

What are the differences between them and when should we use one or the other?

不要在这里引用，请转到 official docs因为对它们之间的差异有很好的解释。