python多处理池并不总是使用所有 worker

Question

问题:
当向 apply_async 发送 1000 个任务时，它们在所有 48 个 CPU 上并行运行，但有时运行的 CPU 越来越少，直到只剩下一个 CPU 正在运行，只有当最后一个 CPU 完成任务时，所有 CPU 才会再次继续运行每个人都有一个新任务。它不需要等待任何像这样的“任务批处理”..

我的(简化)代码:

from multiprocessing import Pool
pool = Pool(47)
tasks = [pool.apply_async(json2features, (j,)) for j in jsons]
feats = [t.get() for t in tasks]

jsons = [...] 是大约 1000 个已加载到内存并解析为对象的 JSON 的列表。
json2features(json) 对 json 执行一些 CPU 密集型工作，并返回一个数字数组。
此函数可能需要 1 秒到 15 分钟才能运行，因此我使用启发式 s.t. 对 json 进行排序。希望最长的任务在列表中排在第一位，因此最先开始。

json2features 函数还会打印任务何时完成以及花费了多长时间。它全部运行在具有 48 个内核的 ubuntu 服务器上，就像我上面所说的，它开始时很棒，使用了所有 47 个内核。然后随着任务的完成，运行的内核越来越少，起初听起来很正常，但事实并非如此，因为在最后一个内核完成后(当我看到它打印到标准输出时)，所有 CPU 都开始再次运行新任务，这意味着这并不是列表的末尾。它可能会再次做同样的事情，然后在列表的实际末尾再次做同样的事情。

有时它可能只使用一个内核 5 分钟，当任务最终完成时，它会再次开始使用所有内核来处理新任务。 (所以它不会卡在一些 IPC 开销上)

没有重复的 jsons，它们之间也没有任何依赖关系(都是静态的，来自磁盘的新鲜数据，没有引用等)，json2features 调用之间也没有任何依赖关系(没有全局状态或任何东西)除了他们使用相同的终端进行打印。

我怀疑问题是在调用 get 之前，worker 不会被释放，所以我尝试了以下代码:

from multiprocessing import Pool
pool = Pool(47)
tasks = [pool.apply_async(print, (i,)) for i in range(1000)]
# feats = [t.get() for t in tasks]

它确实打印了所有 1000 个数字，即使未调用 get。

我现在已经想不出可能是什么问题了。
这真的是 Pool 的正常行为吗？

非常感谢!

Answer 1

multiprocessing.Pool 依赖于单个 os.pipe 将任务传递给工作人员。

通常在 Unix 上，默认管道大小范围为 4 到 64 Kb。如果您传送的 JSON 很大，您可能会在任何给定时间点堵塞管道。

这意味着，当其中一名工作人员忙于从管道中读取大型 JSON 时，所有其他工作人员都将挨饿。

通过 IPC 共享大数据通常是一种不好的做法，因为它会导致性能不佳。这甚至在 multiprocessing programming guidelines 中带有下划线。 .

Avoid shared state

As far as possible one should try to avoid shifting large amounts of data between processes.

与其在主进程中读取 JSON 文件，不如将文件名发送给工作人员，让他们打开并读取内容。您肯定会注意到性能的提高，因为您也在并发域中移动了 JSON 加载阶段。

请注意，结果也是如此。单个 os.pipe 也用于将结果返回到主进程。如果一个或多个工作人员堵塞了结果管道，那么您将让所有进程等待主要进程将其排出。大的结果也应该写入文件。然后，您可以在主进程上利用多线程来快速从文件中读回结果。