同一进程混用Gstreamer与multiprocessing.Manager().Namespace()内存泄漏问题

问题根因 #

1. PyGObject封装的Gst Sample、Buffer对象没有显式释放，GObject原生引用计数和Python GC适配存在缺陷，会残留原生内存块
2. multiprocessing.Manager.Namespace本质是跨进程RPC代理，写入的对象会在Manager进程和当前进程各存一份副本，旧值替换时不会被立刻回收，长期运行就会出现内存堆积
3. buf.extract_dup返回的bytes临时对象未及时清理，累积占用内存

修复代码 #

修改后的测试代码如下：

import unittest
import gi
import traceback
import os
import psutil
import time
import gc
from multiprocessing import Process, Manager, Event

gi.require_version('Gst', '1.0')
from gi.repository import Gst


class RtpNamespaceTest(unittest.TestCase):
    pipeline_str = '''
        videotestsrc pattern=ball ! \
        appsink name=handle-app-sink \
        emit-signals=True \
        max-buffers=1 \
        drop=True \
        '''

    name_space = Manager().Namespace()
    pipeline = None
    event_interrupt: Event = Event()
    frame_cnt = 0

    def start(self):
        Gst.init(None)
        print(self.pipeline_str)
        self.pipeline = Gst.parse_launch(self.pipeline_str)
        self.pipeline.set_state(Gst.State.PLAYING)
        self.appsink = self.pipeline.get_by_name('handle-app-sink')
        self.appsink.connect("new-sample", self.on_new_buffer)

        bus = self.pipeline.get_bus()
        while not self.event_interrupt.is_set():
            bus.timed_pop_filtered(10000, Gst.MessageType.ANY)

        self.pipeline.set_state(Gst.State.NULL)

    def on_new_buffer(self, src):
        sample = src.emit("pull-sample")
        if not sample:
            return Gst.FlowReturn.OK
        buf = sample.get_buffer()
        if not buf:
            del sample
            return Gst.FlowReturn.OK
        # 提取帧数据
        frame_data = buf.extract_dup(0, buf.get_size())
        self.update_frame(frame_data)
        # 显式释放所有局部引用，触发GC清理临时对象
        del buf
        del sample
        del frame_data
        print(f"RAM = {psutil.Process(os.getpid()).memory_info().rss / 1024 / 1024:.2f} MB")
        return Gst.FlowReturn.OK
        
    def update_frame(self, frame: bytes):
        # 先删除旧值，主动释放Manager侧的旧对象引用
        if hasattr(self.name_space, 'frame'):
            del self.name_space.frame
        self.name_space.frame = frame
        # 每100帧主动触发一次全量GC，清理堆积的未回收对象
        self.frame_cnt += 1
        if self.frame_cnt % 100 == 0:
            gc.collect()

    def test_repository(self):
        while True:
            try:
                self.start()
            except Exception as ex:
                traceback.print_exc()
                time.sleep(10)

优化建议 #

如果长期运行还是有内存占用上涨的情况，建议放弃使用Manager.Namespace存储大体积帧数据，改用以下方案：

• 用multiprocessing.Queue传递帧数据，Queue的内存回收逻辑更完善
• 用multiprocessing.SharedMemory共享帧内存，避免多进程间多份数据副本，性能更高内存占用更低

内容的提问来源于stack exchange，提问作者Vladimir Yakovlev