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

Question

我已经看过几篇关于使用Python Multiprocessing模块的内存使用情况的帖子.然而问题似乎没有回答我在这里遇到的问题.我发布我的分析,希望有人可以帮助我.

问题

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

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'选项是我是否分别del l对calculate(num)函数内的行进行注释或注释.在计算之前,内存消耗大约是4400.

1. 看起来手动清除列表会导致工作进程的内存使用量降低.我以为垃圾收集者会照顾这个.有没有办法强制垃圾收集？
2. 令人费解的是,随着任务数量的增加,两种情况下的内存使用量都在不断增长.有没有办法限制内存使用？

我有一个基于这个例子的过程,并且意味着长期运行.我观察到这个工作进程在一夜之间运行后会占用大量内存(~4GB).做一个join释放内存不是一个选项,我试图找出一种没有join-ing 的方法.

这看起来有点神秘.有没有人遇到类似的东西？我该如何解决这个问题？

Answer 1

我做了很多研究,找不到解决问题的方法.但是有一个体面的工作可以以很小的代价防止内存井喷,特别是在服务器端长时间运行的代码上.

解决方案主要是在完成固定数量的任务后重新启动单个工作进程.Poolpython中的类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)