40

这是程序：

#!/usr/bin/python

import multiprocessing

def dummy_func(r):
    pass

def worker():
    pass

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=16)
    for index in range(0,100000):
        pool.apply_async(worker, callback=dummy_func)

    # clean up
    pool.close()
    pool.join()

我发现内存使用量（VIRT 和 RES）一直增长到 close()/join()，有什么解决方案可以解决这个问题吗？我用 2.7 尝试了 maxtasksperchild，但它也没有帮助。

我有一个更复杂的程序，它调用 apply_async() ~6M 次，在 ~1.5M 点我已经获得了 6G+ RES，为了避免所有其他因素，我将程序简化为上述版本。

编辑：

原来这个版本效果更好，感谢大家的意见：

#!/usr/bin/python

import multiprocessing

ready_list = []
def dummy_func(index):
    global ready_list
    ready_list.append(index)

def worker(index):
    return index

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=16)
    result = {}
    for index in range(0,1000000):
        result[index] = (pool.apply_async(worker, (index,), callback=dummy_func))
        for ready in ready_list:
            result[ready].wait()
            del result[ready]
        ready_list = []

    # clean up
    pool.close()
    pool.join()

我没有放任何锁，因为我相信主进程是单线程的（回调或多或少像我阅读的每个文档的事件驱动的东西）。

我将 v1 的索引范围更改为 1,000,000，与 v2 相同并进行了一些测试 - 对我来说很奇怪 v2 甚至比 v1（33 秒对 37 秒）快约 10%，也许 v1 做了太多内部列表维护工作。v2绝对是内存使用的赢家，它从未超过300M（VIRT）和50M（RES），而v1曾经是370M/120M，最好的是330M/85M。所有数字只是3~4次测试，仅供参考。

4

6 回答 6

25

我最近遇到了内存问题，因为我多次使用多处理功能，所以它不断产生进程，并将它们留在内存中。

这是我现在使用的解决方案：

def myParallelProcess(ahugearray):
    from multiprocessing import Pool
    from contextlib import closing
    with closing(Pool(15)) as p:
        res = p.imap_unordered(simple_matching, ahugearray, 100)
    return res
于 2014-01-23T18:01:28.153 回答
10

只需在循环中创建池并在循环结束时使用 pool.close().

于 2018-01-23T13:42:13.300 回答
8

使用map_async而不是apply_async避免过多的内存使用。

对于您的第一个示例，更改以下两行：

for index in range(0,100000):
    pool.apply_async(worker, callback=dummy_func)

至

pool.map_async(worker, range(100000), callback=dummy_func)

它会在您看到它的内存使用情况之前在眨眼间完成top。将列表更改为更大的列表以查看差异。但是map_async，如果没有__len__方法，请注意首先将您传递给它的可迭代对象转换为列表以计算其长度。如果你有一个包含大量元素的迭代器，你可以用itertools.islice更小的块来处理它们。

我在具有更多数据的实际程序中遇到了内存问题，最后发现罪魁祸首是apply_async.

PS，在内存使用方面，你的两个例子没有明显的区别。

于 2013-12-07T08:17:05.487 回答
6

我正在处理一个非常大的 3d 点云数据集。我尝试使用多处理模块来加快处理速度，但我开始出现内存不足错误。经过一些研究和测试，我确定我正在填充要处理的任务队列，比子流程清空它的速度要快得多。我确定通过分块或使用 map_async 或其他可以调整负载的方法，但我不想对周围的逻辑进行重大更改。

我遇到的愚蠢解决方案是pool._cache间歇性地检查长度，如果缓存太大，则等待队列清空。

在我的主循环中，我已经有一个计数器和一个状态代码：

# Update status
count += 1
if count%10000 == 0:
    sys.stdout.write('.')
    if len(pool._cache) > 1e6:
        print "waiting for cache to clear..."
        last.wait() # Where last is assigned the latest ApplyResult

所以每 10k 插入池中，我检查是否有超过 100 万个操作排队（大约 1G 用于主进程的内存）。当队列已满时，我只需等待最后插入的作业完成。

现在我的程序可以运行几个小时而不会耗尽内存。当工作人员继续处理数据时，主进程偶尔会暂停。

顺便说一句，_cache 成员记录了多处理模块池示例：

#
# Check there are no outstanding tasks
#

assert not pool._cache, 'cache = %r' % pool._cache
于 2014-01-14T18:51:39.273 回答
2

我认为这与我发布的问题相似，但我不确定您是否有同样的延迟。我的问题是我从多处理池中产生结果的速度比我消耗它们的速度要快，所以它们在内存中建立起来。为了避免这种情况，我使用信号量来限制池中的输入，这样它们就不会比我消耗的输出领先太多。

于 2016-12-02T00:55:36.833 回答
0

您可以限制每个子进程的任务数

multiprocessing.Pool(maxtasksperchild=1)

maxtasksperchild是工作进程在退出并被新的工作进程替换之前可以完成的任务数，以释放未使用的资源。默认的 maxtasksperchild 是 None，这意味着工作进程将与池一样长。关联

于 2022-01-12T10:04:57.537 回答