Python - 解决内存泄漏问题

32 投票

4 回答

29727 浏览

提问于 2025-04-15 15:28

我有一个Python程序，它会进行一系列实验，每次实验之间不需要保存任何数据。我的代码里有一个内存泄漏的问题，我完全找不到它在哪里（我也看过关于内存泄漏的其他讨论）。因为时间有限，我不得不放弃寻找这个泄漏，但如果我能把每个实验隔离开来，程序可能会运行得足够久，从而产生我需要的结果。

把每个测试放在一个单独的线程里会有帮助吗？
还有其他方法可以隔离泄漏的影响吗？

具体情况说明

我的代码分为两部分：一个是实验运行器，另一个是实际的实验代码。
虽然运行所有实验的代码和每个实验使用的代码之间没有共享的全局变量，但一些类和函数是必须共享的。
实验运行器并不是一个简单的for循环，不能轻易放到一个shell脚本里。它首先根据配置参数决定需要运行哪些测试，然后执行这些测试，最后以特定的方式输出数据。
我尝试手动调用垃圾回收器，以防问题只是因为垃圾回收没有运行，但这并没有解决问题。

更新

Gnibbler的回答让我发现，我的ClosenessCalculation对象存储了每次计算中使用的所有数据，但它们并没有被销毁。然后我用这个信息手动删除了一些链接，这似乎解决了内存问题。

垃圾回收内存泄漏资源管理数据清理计算优化线程隔离实验运行器类和函数共享

4 个回答

我之前也遇到过一个第三方的C语言库有内存泄漏的问题。我想到的一个比较干净的解决办法就是使用分叉（fork）和等待（wait）。这样做的好处是，你不需要在每次运行后都创建一个新的进程。你可以自己定义每次处理的数据量。

这里有一个通用的解决方案（如果你发现了内存泄漏，只需要把run()改成调用run_single_process()，而不是run_forked()，就可以解决问题了）：

import os,sys
batchSize = 20

class Runner(object):
    def __init__(self,dataFeedGenerator,dataProcessor):
        self._dataFeed = dataFeedGenerator
        self._caller = dataProcessor

    def run(self):
        self.run_forked()

    def run_forked(self):
        dataFeed = self._dataFeed
        dataSubFeed = []
        for i,dataMorsel in enumerate(dataFeed,1):
            if i % batchSize > 0:
                dataSubFeed.append(dataMorsel)
            else:
                self._dataFeed = dataSubFeed
                self.fork()
                dataSubFeed = []
                if self._child_pid is 0:
                    self.run_single_process()
                self.endBatch()

    def run_single_process(self)
        for dataMorsel in self._dataFeed:
            self._caller(dataMorsel)

    def fork(self):
        self._child_pid = os.fork()

    def endBatch(self):
        if self._child_pid is not 0:
            os.waitpid(self._child_pid, 0)
        else:
            sys.exit() # exit from the child when done

这样做可以把内存泄漏限制在子进程里。而且它的泄漏次数不会超过你设定的batchSize变量的值。

回答于 2025-04-15 由 Python大师

分享举报

线程并不能解决这个问题。如果你真的放弃了寻找内存泄漏的办法，那么唯一能控制其影响的办法就是定期运行一个新的进程（比如，当某个测试导致整体内存使用过高时，你可以通过查看Linux系统中的/proc/self/status来轻松确定虚拟内存的大小，其他操作系统也有类似的方法）。

确保整个脚本可以接受一个可选参数，告诉它从哪个测试编号（或其他测试标识）开始，这样当脚本的一个实例觉得占用的内存太多时，它可以告诉下一个实例从哪里重新开始。

或者，更稳妥的方法是，确保每完成一个测试，它的标识会被追加到一个大家都知道名字的文件里。当程序启动时，它会先读取这个文件，从而知道哪些测试已经完成。这种架构更稳固，因为它也能处理程序在测试过程中崩溃的情况；当然，为了完全自动化从崩溃中恢复，你需要一个单独的监控程序和进程，负责在检测到之前的测试程序崩溃时启动一个新的实例（可以使用subprocess来实现这个目的——它还需要一种方法来判断测试序列是否完成，比如正常退出测试程序可以表示完成，而任何崩溃或退出状态不为0则意味着需要启动一个新的实例）。

如果这些架构让你感兴趣，但你在实现上需要进一步的帮助，可以在这个回答下评论，我很乐意提供示例代码——我不想提前提供，以免出现你尚未表达的问题，使得这些架构不适合你。（知道你需要运行的平台也可能会有所帮助）。

回答于 2025-04-15 由 Python大师

分享举报

你可以用类似这样的代码来帮助找出内存泄漏的问题。

>>> from collections import defaultdict
>>> from gc import get_objects
>>> before = defaultdict(int)
>>> after = defaultdict(int)
>>> for i in get_objects():
...     before[type(i)] += 1 
...

现在假设测试过程中出现了一些内存泄漏。

>>> leaked_things = [[x] for x in range(10)]
>>> for i in get_objects():
...     after[type(i)] += 1
... 
>>> print [(k, after[k] - before[k]) for k in after if after[k] - before[k]]
[(<type 'list'>, 11)]

因为我们泄漏了一个列表，这个列表里面还有10个其他的列表，所以总共泄漏了11个。

回答于 2025-04-15 由 Python大师

分享举报

Python - 解决内存泄漏问题

4 个回答

撰写回答