Python的多处理和内存

Question

我正在使用 multiprocessing.imap_unordered 对值列表执行计算:

def process_parallel(fnc, some_list):
    pool = multiprocessing.Pool()
    for result in pool.imap_unordered(fnc, some_list):
        for x in result:
            yield x
    pool.terminate()

每次调用 fnc 都会返回一个巨大的对象，这是设计的结果。我可以将此类对象的 N 个实例存储在 RAM 中，其中 N ~ cpu_count，但不多(不是数百个)。

现在，使用这个函数会占用太多内存。内存完全用在主进程中，而不是在工作进程中。

imap_unordered 如何存储完成的结果？我的意思是工作人员已经返回但尚未传递给用户的结果。我认为它很聪明，只根据需要“懒惰地”计算它们，但显然不是。

看起来因为我不能足够快地消耗 process_parallel 的结果，池不断地从 fnc 某处在内部排队这些巨大的对象，然后爆炸。有没有办法避免这种情况？以某种方式限制其内部队列？

---

我正在使用 Python2.7。干杯。

Answer 1

您可以通过查看相应的源文件 (python2.7/multiprocessing/pool.py) 看到，IMapUnorderedIterator 使用 collections.deque 实例来存储结果。如果有新项目进来，它会在迭代中添加和删除。

正如你所建议的，如果在主线程仍在处理该对象时另一个巨大的对象进入，那么这些对象也将存储在内存中。

你可能会尝试是这样的:

it = pool.imap_unordered(fnc, some_list)
for result in it:
    it._cond.acquire()
    for x in result:
        yield x
    it._cond.release()

如果它试图将下一个对象放入双端队列，这应该会导致任务结果接收器线程在您处理项目时被阻塞。因此，内存中的巨大对象不应超过两个。如果这适用于您的情况，我不知道 ;)