NN人物检测场景下，多进程保存IP摄像头视频流时进程被杀死

背景 #

我有一个用于视频人物检测的神经网络，通过IP摄像头获取视频流。最初用单线程实现功能正常，但耗时较长——逻辑是保存检测到人物的视频片段，以及人物消失后的5秒内容。

单线程实现代码 #

cap = cv2.VideoCapture(source)
width = cap.get(3)
height = cap.get(4)
fourcc = cv.VideoWriter_fourcc(*'XVID')
start = True
while True:
    ret,img = cap.read()
    if *NN part*:
        if start:
            out = cv2.VideoWriter(file, fourcc, 20.0, (width, height), True)
            start = False
        out.write(img)
        last_point = time.perf_counter()
    else:
        if time.perf_counter() - last_point < 5:
            out.write(img)
        else:
            out.release()

多进程优化尝试 #

为提升效率改用multiprocessing实现，通过队列向子进程传递帧数据和参数，但运行时进程被系统杀死。

多进程代码 #

import numpy as np
import os
import argparse
import sys
from ksnn.api import KSNN
from ksnn.types import *
import cv2 as cv
import time
from pathlib import Path
import json
from numba import njit
import multiprocessing as mp
from time import perf_counter

<...>

def save_video(q):
    while True:
        flag, data = q.get()
        if flag==0:                                             # Create new video-file
            frame, size, path, n, fps = data
            fourcc = cv.VideoWriter_fourcc(*'XVID')
            file = "{}vid_{}.avi".format(path, n)
            out = cv.VideoWriter(file, fourcc, fps, size, True)
            image = cv.resize(frame, size)
            out.write(image)
        elif flag==1:                                           # Add frame
            frame, size = data
            image = cv.resize(frame, size)
            out.write(image)
        elif flag==2:                                           # Stop and release
            file = "{}vid_{}.avi".format(path, n)
            out.release()
            print("Done. Video was saved in {}".format(file))
            break
        else:
            pass

def main():
    if is_video_file or is_url or webcam:
        cap = cv.VideoCapture(source)
        mp.set_start_method('spawn')
        flag1 = flag2 = True
        n = 0
        q = mp.Queue()
        while(1):
            start = time.perf_counter()
            ret,img = cap.read()
            if flag1:
                q.put([-1, None])
                p = mp.Process(target=save_video, args=[q,])
                p.start()
                flag1 = False

            boxes, scores, classes = *NN part*

            if boxes is not None:
                if flag2:
                    q.put([0, img, size, path, n, FPS])
                    flag2 = False
                else:
                    q.put([1, img, size])
                lost_object = perf_counter()
            elif boxes is None and not flag2:
                if perf_counter() - lost_object < 5:
                    q.put([1, img, size])
                else:
                    q.put([2, path, n])
                    flag1 = flag2 = True
                    n+=1

            end = time.perf_counter()
            print('1 frame per: {}s'.format(end - start))
        cap.release()

报错信息 #

[Sat Jan 27 21:10:22 2024] Killed process 201971 (python3) total-vm:4786480kB, anon-rss:2889920kB, file-rss:88kB, shmem-rss:16kB

问题排查与解决方案 #

核心原因 #

进程被系统杀死是**内存耗尽（OOM）**导致的，你的代码存在以下内存占用问题：

1. 队列帧堆积：主进程往队列塞帧的速度远快于子进程写入视频的速度，队列中积压的大量numpy帧会持续占用内存，最终撑爆系统内存。
2. 进程资源未回收：子进程完成任务后，主进程未调用join()回收进程资源，可能残留僵尸进程。
3. 子进程变量作用域错误：flag=2分支中path和n未定义，会引发隐性错误。
4. 帧传递冗余：直接传递完整numpy帧，单帧内存占用大，加剧内存消耗。

具体修复步骤 #

1. 限制队列大小，避免帧堆积

创建队列时设置最大容量，队列满时主进程自动阻塞，等待子进程处理：

q = mp.Queue(maxsize=30)  # 可根据实际帧率调整，比如设置为30帧

2. 优化帧传递，减少内存占用

将帧编码为JPEG字节流传递，子进程再解码，大幅降低单帧内存占用：

• 主进程传递时修改：

# 替换原q.put([0, img, size, path, n, FPS])
ret, buffer = cv.imencode('.jpg', img)
q.put([0, buffer.tobytes(), size, path, n, FPS])

# 替换原q.put([1, img, size])
ret, buffer = cv.imencode('.jpg', img)
q.put([1, buffer.tobytes(), size])

• 子进程接收时修改：

def save_video(q):
    current_path = ""
    current_n = 0
    out = None
    while True:
        flag, data = q.get()
        if flag==0:
            buffer, size, path, n, fps = data
            current_path = path
            current_n = n
            # 解码帧
            img_np = np.frombuffer(buffer, dtype=np.uint8)
            frame = cv.imdecode(img_np, cv.IMREAD_COLOR)
            # 后续写入逻辑不变
            fourcc = cv.VideoWriter_fourcc(*'XVID')
            file = "{}vid_{}.avi".format(path, n)
            out = cv.VideoWriter(file, fourcc, fps, size, True)
            image = cv.resize(frame, size)
            out.write(image)
        elif flag==1:
            buffer, size = data
            img_np = np.frombuffer(buffer, dtype=np.uint8)
            frame = cv.imdecode(img_np, cv.IMREAD_COLOR)
            image = cv.resize(frame, size)
            out.write(image)
        # 其他分支逻辑不变

3. 修复子进程变量作用域问题

在子进程中保存path和n的当前值，避免flag=2分支报错：

def save_video(q):
    current_path = ""
    current_n = 0
    out = None
    while True:
        flag, data = q.get()
        if flag==0:
            frame, size, path, n, fps = data
            current_path = path
            current_n = n
            # 原创建视频逻辑...
        elif flag==2:
            file = "{}vid_{}.avi".format(current_path, current_n)
            out.release()
            print("Done. Video was saved in {}".format(file))
            break

4. 正确回收进程资源

主进程在重置flag1前，等待子进程退出并回收资源：

elif boxes is None and not flag2:
    if perf_counter() - lost_object < 5:
        q.put([1, img, size])
    else:
        q.put([2, path, n])
        p.join()  # 等待子进程退出
        flag1 = flag2 = True
        n+=1

5. 复用子进程（可选优化）

让子进程完成一个视频后不退出，等待下一次任务，减少进程创建销毁的开销：

def save_video(q):
    current_path = ""
    current_n = 0
    out = None
    while True:
        flag, data = q.get()
        if flag==0:
            if out is not None:
                out.release()  # 释放之前的写入器
            # 原创建视频逻辑...
        elif flag==2:
            if out is not None:
                file = "{}vid_{}.avi".format(current_path, current_n)
                out.release()
                print("Done. Video was saved in {}".format(file))
                out = None
        elif flag==-1:  # 退出信号
            if out is not None:
                out.release()
            break

主进程退出前发送退出信号：

# 在main函数循环结束前（比如捕获Ctrl+C时）：
q.put([-1, None])
p.join()
cap.release()

额外优化建议 #

• 主进程读取帧后，先将帧缩放到神经网络需要的尺寸再检测，减少计算耗时。
• 加入内存监控，方便排查问题：

import psutil
# 在主进程循环中添加
print(f"当前内存占用: {psutil.Process().memory_info().rss / 1024 / 1024:.2f} MB")

内容的提问来源于stack exchange，提问作者Rodion Shkokov