使用 Python 多处理的高内存使用率答案

【问题标题】：High Memory Usage Using Python Multiprocessing使用 Python 多处理的高内存使用率
【发布时间】：2014-02-24 10:58:55
【问题描述】：

我看过几篇关于使用 Python 多处理模块的内存使用的帖子。然而，这些问题似乎并没有回答我在这里遇到的问题。我发布我的分析，希望有人可以帮助我。

问题

我正在使用多处理并行执行任务，我注意到工作进程的内存消耗无限增长。我有一个小的独立示例，应该可以复制我注意到的内容。

import multiprocessing as mp
import time

def calculate(num):
    l = [num*num for num in range(num)]
    s = sum(l)
    del l       # delete lists as an  option
    return s

if __name__ == "__main__":
    pool = mp.Pool(processes=2)
    time.sleep(5)
    print "launching calculation"
    num_tasks = 1000
    tasks =  [pool.apply_async(calculate,(i,)) for i in range(num_tasks)]
    for f in tasks:    
        print f.get(5)
    print "calculation finished"
    time.sleep(10)
    print "closing  pool"
    pool.close()
    print "closed pool"
    print "joining pool"
    pool.join()
    print "joined pool"
    time.sleep(5)

系统

我正在运行 Windows，我使用任务管理器来监控内存使用情况。我正在运行 Python 2.7.6。

观察

我总结了下面 2 个工作进程的内存消耗。

+---------------+----------------------+----------------------+
|  num_tasks    |  memory with del     | memory without del   |
|               | proc_1   | proc_2    | proc_1   | proc_2    |
+---------------+----------------------+----------------------+
| 1000          | 4884     | 4694      | 4892     | 4952      |
| 5000          | 5588     | 5596      | 6140     | 6268      |
| 10000         | 6528     | 6580      | 6640     | 6644      |
+---------------+----------------------+----------------------+

在上表中，我尝试更改任务数并观察在所有计算结束时和join-ing pool 之前消耗的内存。 'del' 和 'without del' 选项分别是我是否取消注释或注释 calculate(num) 函数内的 del l 行。计算前内存消耗在4400左右。

看起来手动清除列表会降低工作进程的内存使用率。我认为垃圾收集器会处理这个问题。有没有办法强制垃圾回收？
令人费解的是，随着任务数量的增加，两种情况下的内存使用量都在不断增长。有没有办法限制内存使用？

我有一个基于此示例的流程，旨在长期运行。我观察到这个工作进程在通宵运行后占用了大量内存（~4GB）。执行join 来释放内存不是一种选择，我正在尝试找出没有join-ing 的方法。

这似乎有点神秘。有没有人遇到过类似的事情？我该如何解决这个问题？

【问题讨论】：

似乎del 在这个测试用例中是多余的，因为l 在函数返回后被垃圾回收。内存使用量的增加可能源于[num*num for num in range(num)]，因为您将i 传递为num，而i 随着num_task 的增加而增加。
感谢您的评论。我希望在所有任务完成后，子进程的内存消耗会恢复到它开始时的状态（~4400）。
也许这个例子不足以解决你真正的问题。在您的实际过程中，您可以考虑使用生成器而不是列表。另外，gc.collect() 可能会转到handy。
我的真实应用程序有更复杂的对象，而不是列表。我试图用示例代码来模拟我的问题。我将与gc 一起玩，看看是否有帮助。你有一个关于正确使用gc 来释放内存的简单例子吗？谢谢！
我用 gc 尝试了这个示例代码，但它没有帮助:( 但是，我对其进行了一些更改。我没有创建一个可变大小的新列表，而是使用`range 创建一个新列表(1000000). It took about 20MB. After del l, python does no immediate gc. And explicit gc.collect()` 在函数calculate 中确实有帮助。gc.collect 的用法很简单，只需将其添加到子进程的末尾即可。但这会减慢您的进程很多，有条件地做手动gc。

标签： python performance memory multiprocessing

【解决方案1】：

我做了很多研究，但找不到解决问题本身的解决方案。但是有一个不错的解决方法可以以很小的成本防止内存爆裂，尤其是在服务器端长时间运行的代码上。

解决方案本质上是在完成固定数量的任务后重新启动单个工作进程。 python 中的Pool 类将maxtasksperchild 作为参数。您可以指定maxtasksperchild=1000，从而限制在每个子进程上运行 1000 个任务。达到maxtasksperchild 编号后，池刷新其子进程。为最大任务使用一个谨慎的数字，可以平衡消耗的最大内存与与重新启动后端进程相关的启动成本。 Pool 构造如下：

pool = mp.Pool(processes=2,maxtasksperchild=1000)

我将我的完整解决方案放在这里，以便对其他人有用！

import multiprocessing as mp
import time

def calculate(num):
    l = [num*num for num in range(num)]
    s = sum(l)
    del l       # delete lists as an  option
    return s

if __name__ == "__main__":

    # fix is in the following line #
    pool = mp.Pool(processes=2,maxtasksperchild=1000)

    time.sleep(5)
    print "launching calculation"
    num_tasks = 1000
    tasks =  [pool.apply_async(calculate,(i,)) for i in range(num_tasks)]
    for f in tasks:    
        print f.get(5)
    print "calculation finished"
    time.sleep(10)
    print "closing  pool"
    pool.close()
    print "closed pool"
    print "joining pool"
    pool.join()
    print "joined pool"
    time.sleep(5)

【讨论】：

+1 因为 maxtasksperschild 是解决此问题的方法。我已经多次处理这个问题，发现很难找到解决方案。
当我不使用 Pool 而是使用 proc = Process(target=func, args=args) 时有什么方法可以避免这个问题？
我已经为这个问题苦苦挣扎了很长时间，它只是省了一个巨大的头痛！
+1 来自我！谢谢你的帖子。我遇到了类似的问题，并尝试了许多其他修复，例如在子进程函数中使用 gc 等，但它们都不起作用，但这也达到了预期的效果，内存泄漏终于消失了！跨度>
1000 个任务需要多少内存！

【解决方案2】：

这里的一个潜在问题是结果可能以任何顺序返回，但是因为您是按顺序读取它们，所以它必须将从进程返回的所有结果存储在内存中。 num_tasks 越高，它可能需要存储在内存中等待您的 for f in tasks 循环处理结果的结果就越多。

在最坏的情况下，结果的计算顺序正好相反。在这种情况下，所有结果必须由多处理模块为您保存在内存中，然后您的 for f in tasks 循环将开始处理任何内容。

在这种情况下，他们使用的内存量似乎比我预期的要高（不仅仅是存储由 calculate() 函数返回的 1000-10000 个数字），但是可能每个工作人员结果的存储开销都很高。

您是否尝试将 callback 参数指定给 apply_async，以便在结果完成后立即处理结果，或使用 imap_unordered，所以它可以在结果准备好后立即给您返回结果？

【讨论】：