具有多处理的 Python itertools - 巨大的列表与迭代器的低效 CPU 使用

Question

我处理 n 个元素(下面称为“对”)的变体，并将重复用作我函数的参数。显然，只要“r”列表不够大，无法消耗所有内存，一切都可以正常工作。问题是我最终必须为 6 个元素重复 16 次以上。为此，我在云中使用 40 核系统。

代码如下所示:

if __name__ == '__main__':
  pool = Pool(39)
  r = itertools.product(pairs,repeat=16)
  pool.map(f, r)

我相信我应该使用迭代器而不是预先创建巨大的列表，问题就在这里开始了..

我尝试使用以下代码解决问题:

if __name__ == '__main__':
  pool = Pool(39)
  for r in itertools.product(pairs,repeat=14):
    pool.map(f, r)

内存问题消失了，但 CPU 使用率大约为每个内核 5%。现在代码的单核版本比这个更快。

如果你能指导我一点，我将不胜感激..

谢谢。

Answer 1

您的原始代码不是在您自己的代码中预先创建一个list(itertools.product 返回一个生成器)，而是pool.map 正在实现整个生成器(因为它假定如果您可以存储所有输出，那么您也可以存储所有输入)。

不要在这里使用pool.map。如果您需要有序结果，请使用 pool.imap，或者如果结果顺序不重要，请使用 pool.imap_unordered。迭代任一调用的结果(不要包装在 list 中)，并在结果出现时对其进行处理，内存应该不是问题:

if __name__ == '__main__':
    pool = Pool(39)
    for result in pool.imap(f, itertools.product(pairs, repeat=16)):
        print(result)

如果您使用 pool.map 来处理副作用，那么您只需要运行它直到完成，但结果和顺序并不重要，您可以通过使用 imap_unordered 并使用 collections.deque 有效地耗尽“结果”而不实际存储任何东西(deque 的 maxlen 0 是强制迭代器运行完成而不存储结果的最快、内存最少的方法):

from collections import deque

if __name__ == '__main__':
    pool = Pool(39)
    deque(pool.imap_unordered(f, itertools.product(pairs, repeat=16)), 0)

最后，我对指定 39 个 Pool worker 有点怀疑； multiprocessing 在很大程度上有利于 CPU 密集型任务；如果您使用的 worker 数量超过 CPU 内核数量并获得 yield ，则 multiprocessing 可能会在 IPC 中让您付出更多的代价而不是获得的 yield ，并且使用更多的 worker 只是通过缓冲更多来掩盖问题数据。

如果您的工作主要受 I/O 限制，您可以尝试使用基于线程的池，这将避免 pickling 和 unpickling 的开销，以及父进程和子进程之间的 IPC 成本。与基于进程的池不同，Python 线程受制于 GIL。问题，所以你的 CPU 绑定(bind)在 Python 中工作(不包括 GIL 释放 I/O 调用，ctypes 调用 .dll/.so 文件，以及某些第三方扩展，如 numpy为繁重的 CPU 工作释放 GIL)仅限于单个内核(在 Python 2.x 中，对于 CPU 绑定(bind)工作，您经常浪费大量的解决 GIL 争用和执行上下文切换；Python 3 消除了大部分浪费) .但是，如果您的工作主要受 I/O 限制，则阻塞 I/O 会释放 GIL 以允许其他线程运行，因此您可以拥有许多线程，只要它们中的大多数延迟 I/O。切换也很容易(只要您没有将程序设计为依赖于每个工作人员的单独地址空间，假设您可以写入“共享”状态并且不影响其他工作人员或父进程)，只需更改:

from multiprocessing import Pool

到:

from multiprocessing.dummy import Pool

然后你得到 multiprocessing.dummy池的版本，基于线程而不是进程。