[BUG] Running picamera.start_stream(h264encoder, ffoutput) in child thread or process produces BrokenPipeError

[DanikKul]

Describe the bug
I have Flask app which manages cameras and HLS streams. Flask uses threads to process requests. So when I trying to execute that method I'm getting BrokenPipeError. (self.cameras contains Picamera2 objects). And class containing method provided below is singleton.

    def start_stream(self, in_cam: int):
        try:
            config = self.cameras[in_cam]['obj'].create_video_configuration(main={"size": (800, 600)})
            encoder = H264Encoder(bitrate=16000, qp=30)
            output = FfmpegOutput(f"-f hls -hls_time 4 -hls_list_size 5 -hls_flags delete_segments -hls_allow_cache 0 /tmp/hls/stream{in_cam}.m3u8")
            output.error_callback = lambda x: print(f"FFMpeg Error: {x}")
            self.cameras[in_cam]['obj'].configure(config)
            self.cameras[in_cam]['obj'].start_recording(encoder, output)
            self.cameras[in_cam]['busy'] = True
            print(f'Started streaming for camera {in_cam}')
        except Exception as e:
            print(e)
            return False
        return True

When I'm trying to put that method in different thread/process result is the same.

That code works if executed in main thread.

To Reproduce
Flask server, singleton with initialized cameras, HLS streaming using H264Encoder and FfmpegOutput. Or it can be reproduced with built-in python threading or multiprocessing.

# camera_manager.py
import os
import sys
import signal
import time
import cv2
import multiprocessing as mp
from picamera2 import Picamera2
from picamera2.encoders import H264Encoder
from picamera2.outputs import FfmpegOutput
from src.device_manager.camera.camera_stats import CameraStats


class CameraManager:
    def __init__(self):
        print('Creating cameramanager')
        self.stats = CameraStats()
        self.cameras = {}
        self.processes = {}
        brief = self.stats.get_brief_cam_info()
        for i in range(len(brief)):
            self.cameras[brief[i]['Num']] = {
                "obj": Picamera2(brief[i]['Num']),
                "busy": False,
            }

    def stop_stream(self, in_cam: int):
        try:
            self.cameras[in_cam]['obj'].stop_recording()
            self.cameras[in_cam]['busy'] = False
            self.processes[in_cam].terminate()
        except Exception as e:
            print(e)
            return False
        return True

    def start_stream(self, in_cam: int):
        process = mp.Process(target=self.__start_stream, args=(in_cam,))
        self.processes[in_cam] = process
        self.processes[in_cam].start()
        return True

    def __start_stream(self, in_cam: int):
        try:
            config = self.cameras[in_cam]['obj'].create_video_configuration(main={"size": (800, 600)})
            encoder = H264Encoder(bitrate=16000, qp=30)
            output = FfmpegOutput(f"-f hls -hls_time 4 -hls_list_size 5 -hls_flags delete_segments -hls_allow_cache 0 /tmp/hls/stream{in_cam}.m3u8")
            output.error_callback = lambda x: print(f"FFMpeg Error: {x}")
            self.cameras[in_cam]['obj'].configure(config)
            self.cameras[in_cam]['obj'].start_recording(encoder, output)
            self.cameras[in_cam]['busy'] = True
            print(f'Started streaming for camera {in_cam}')
        except Exception as e:
            print(e)
            return False
        return True
    
    def get_camera(self, in_cam):
        return self.cameras[in_cam]


if __name__ == "__main__":
    CM = CameraManager()
    CM.start_stream(0)
    time.sleep(10)
    CM.stop_stream(0)

import json
from pprint import pprint
from picamera2 import Picamera2


class CameraStats:
    def is_camera_busy(self, camera):
        return camera['busy']
    
    def get_brief_cam_info(self):
        return Picamera2.global_camera_info()

    def get_cam_info(self, cam):
        cameras = self.get_brief_cam_info()
        props = []
        for i in range(len(cameras)):
            if cameras[i]['Num'] == cam['obj'].camera_idx:
                props = cam['obj'].camera_properties | cameras[i]
                break
        return props
    
    def get_cam_controls(self, cam):
        return cam['obj'].camera_controls

Expected behaviour
FFMpeg writes HLS stream encoded by H264Encoder.

Console Output, Screenshots
[Errno 32] Broken pipe

Hardware :
Raspberry Pi 5 with M.2 HAT+, NoIR Camera V2, behavior is not changed with other raspberries or cameras

[davidplowman]

Hi, and thanks for the question. There's quite a lot going on here, let me try and pick out a few things.

When using multi-processing, you have to be very careful that you create and talk to the camera object from a single process. libcamera does not support communicating with the camera from more than one process, and doing so is likely to lead to lock-ups and crashes. In your case, I think that means not making Picamera2 objects in your main process.

Even then, there are some issues with the camera system and multi-processing. Apparently - and I didn't know this either - you have to put

    mp.set_start_method('spawn')

into your if __name__ == "__main__": block. I'm not 100% clear why, but the default method (forking) can leave some state linked between the two processes (if you've used Picamera2.global_camera_info()).

Finally, you need to make sure your process lives long enough to do some recording. In your example, your __start_stream method returns immediately, so your process will finish immediately. You need to have it wait, perhaps for an event to signal it to stop. Be a bit careful with terminate too, I think it may not give the process time to finish gracefully.

Threads should also work. There do seem to be some issues with the way the child (Ffmpeg) process is set up to quit when the parent process dies, and for reasons I don't understand this causes the pipe to be closed when the calling thread terminates. I honestly have no clue about that. But the workaround is not to let that thread quit - so in fact it would be working the same way as the multi-processing version. But with threads you do have a bit more leeway in who makes calls to the Picamera2 object - though you have to take care to synchronise things yourself.

I found it helpful to experiment with some very simple code.

import time
import multiprocessing as mp
from picamera2 import Picamera2
from picamera2.encoders import H264Encoder
from picamera2.outputs import FfmpegOutput

def process_func(cam_num, event):
    with Picamera2(cam_num) as picam2:
        encoder = H264Encoder()
        output = FfmpegOutput("-f hls -hls_time 4 -hls_list_size 5 -hls_flags delete_segments -hls_allow_cache 0 /tmp/stream0.m3u8")
        picam2.start_recording(encoder, output)
        event.wait()
        picam2.stop_recording()

if __name__ == "__main__":
    mp.set_start_method('spawn')
    event = mp.Event()

    print(Picamera2.global_camera_info())

    proc = mp.Process(target=process_func, args=(0, event))
    proc.start()
    time.sleep(10)
    event.set()
    proc.join()

[davidplowman]

Minor update. I've discovered that the mechanism whereby the Ffmpeg child is killed when the parent dies actually records the thread creating the process as the parent, not the actual parent process. Who knew. But I guess that explains the behaviour.

One could just remove that mechanism from the code, but then you risk getting zombie Ffmpeg processes all over the place if they are ever not terminated gracefully. I see there's a "child subreaper" mechanism that might provide a way to fix this, but honestly, at this point my head is spinning.

So for now the advice to treat the thread as though it were a sub-process (and not let it terminate until the recording is finished) probably stands...