keep frame shm open and skip overwritten frames

Every reader closed and reopened a frame's shared memory segment on each frame, so each process paid a page fault on every page of a fresh mapping (about 760 for a 1080p frame) per frame. Readers now keep segments mapped. Capture stamps each slot's frame time in a per-camera `{camera}_frame_times` segment, and `get()` returns None when the slot no longer holds the requested frame, reopening once by name so a mapping left over from a same-size camera recreate heals itself. Capture also reuses a slot when the frame queue is full, since no reader was told about it.

frigate/camera/maintainer.py

@@ -169,6 +169,10 @@ class CameraMaintainer(threading.Thread):
             frame_size = config.frame_shape_yuv[0] * config.frame_shape_yuv[1]
             self.frame_manager.create(f"{config.name}_frame{i}", frame_size)
 
+        # a leftover ledger from an unclean exit may be the wrong length
+        self.frame_manager.delete(f"{config.name}_frame_times")
+        self.frame_manager.create(f"{config.name}_frame_times", max(count, 1) * 8)
+
         capture_process = CameraCapture(
             config,
             count,

frigate/camera/state.py

@@ -349,7 +349,7 @@ class CameraState:
             return
 
         current_frame = self.frame_manager.get(
-            frame_name, self.camera_config.frame_shape_yuv
+            frame_name, self.camera_config.frame_shape_yuv, frame_time
         )
 
         tracked_objects = self.tracked_objects.copy()
@@ -603,9 +603,6 @@ class CameraState:
                 self.current_frame_time = frame_time
                 self._current_frame = np.copy(current_frame)
 
-                if self.previous_frame_id is not None:
-                    self.frame_manager.close(self.previous_frame_id)
-
             self.previous_frame_id = frame_name
 
     def send_mqtt_snapshot(self, new_obj: TrackedObject, object_type: str) -> None:

frigate/embeddings/maintainer.py

@@ -793,7 +793,7 @@ class EmbeddingMaintainer(threading.Thread):
         if topic is None:
             return
 
-        camera, frame_name, _, _, motion_boxes, _ = data
+        camera, frame_name, frame_time, _, motion_boxes, _ = data
 
         if not camera or camera not in self.config.cameras:
             return
@@ -817,7 +817,7 @@ class EmbeddingMaintainer(threading.Thread):
 
         try:
             yuv_frame = self.frame_manager.get(
-                frame_name, camera_config.frame_shape_yuv
+                frame_name, camera_config.frame_shape_yuv, frame_time
             )
         except FileNotFoundError:
             pass
@@ -841,8 +841,6 @@ class EmbeddingMaintainer(threading.Thread):
                     {"camera": camera, "motion": motion_boxes}, yuv_frame
                 )
 
-        self.frame_manager.close(frame_name)
-
     def _process_deferred_results(self) -> None:
         """Drain results from deferred processors and perform IPC side-effects."""
         for processor in self.realtime_processors:

frigate/object_detection/base.py

@@ -170,7 +170,6 @@ class DetectorRunner(FrigateProcess):
             mono_start = time.monotonic()
             detections = object_detector.detect_raw(input_frame)
             duration = time.monotonic() - mono_start
-            frame_manager.close(connection_id)
 
             if connection_id not in self.outputs:
                 self.create_output_shm(connection_id)
@@ -259,10 +258,6 @@ class AsyncDetectorRunner(FrigateProcess):
             ts = self.send_times.popleft()
             duration = time.perf_counter() - ts
 
-            # release input buffer
-            assert self._frame_manager is not None
-            self._frame_manager.close(connection_id)
-
             if connection_id not in self.outputs:
                 self.create_output_shm(connection_id)
 

frigate/output/output.py

@@ -241,7 +241,7 @@ class OutputProcess(FrigateProcess):
                 continue
 
             frame = frame_manager.get(
-                frame_name, self.config.cameras[camera].frame_shape_yuv
+                frame_name, self.config.cameras[camera].frame_shape_yuv, frame_time
             )
 
             if frame is None:
@@ -296,8 +296,6 @@ class OutputProcess(FrigateProcess):
                     frame,
                 )
 
-            frame_manager.close(frame_name)
-
         move_preview_frames("clips")
 
         while True:
@@ -309,21 +307,6 @@ class OutputProcess(FrigateProcess):
             if not topic or data is None:
                 break
 
-            (
-                camera,
-                frame_name,
-                frame_time,
-                current_tracked_objects,
-                motion_boxes,
-                regions,
-            ) = data
-
-            camera_config = self.config.cameras.get(camera)
-
-            if camera_config is not None:
-                frame_manager.get(frame_name, camera_config.frame_shape_yuv)
-                frame_manager.close(frame_name)
-
         detection_subscriber.stop()
 
         for jsmpeg in jsmpeg_cameras.values():

frigate/ptz/autotrack.py

@@ -124,7 +124,7 @@ class PtzMotionEstimator:
             )
 
             yuv_frame = self.frame_manager.get(
-                frame_name, self.camera_config.frame_shape_yuv
+                frame_name, self.camera_config.frame_shape_yuv, frame_time
             )
 
             if yuv_frame is None:
@@ -181,8 +181,6 @@ class PtzMotionEstimator:
             except Exception:
                 pass
 
-            self.frame_manager.close(frame_name)
-
         return self.coord_transformations
 
 

frigate/review/maintainer.py

@@ -544,7 +544,7 @@ class ReviewSegmentMaintainer(threading.Thread):
 
         try:
             yuv_frame = self.frame_manager.get(
-                frame_name, camera_config.frame_shape_yuv
+                frame_name, camera_config.frame_shape_yuv, frame_time
             )
         except FileNotFoundError:
             return
@@ -554,7 +554,6 @@ class ReviewSegmentMaintainer(threading.Thread):
             return
 
         pending.update_frame(camera_config, yuv_frame, objects)
-        self.frame_manager.close(frame_name)
 
     def update_existing_segment(
         self,
@@ -653,7 +652,7 @@ class ReviewSegmentMaintainer(threading.Thread):
                 try:
                     if should_update_image:
                         yuv_frame = self.frame_manager.get(
-                            frame_name, camera_config.frame_shape_yuv
+                            frame_name, camera_config.frame_shape_yuv, frame_time
                         )
 
                         if yuv_frame is None:
@@ -669,7 +668,6 @@ class ReviewSegmentMaintainer(threading.Thread):
                         activity.get_all_objects(),
                         prev_data,
                     )
-                    self.frame_manager.close(frame_name)
                 except FileNotFoundError:
                     return
 
@@ -677,7 +675,7 @@ class ReviewSegmentMaintainer(threading.Thread):
             if not segment.has_frame:
                 try:
                     yuv_frame = self.frame_manager.get(
-                        frame_name, camera_config.frame_shape_yuv
+                        frame_name, camera_config.frame_shape_yuv, frame_time
                     )
 
                     if yuv_frame is None:
@@ -685,7 +683,6 @@ class ReviewSegmentMaintainer(threading.Thread):
                         return
 
                     segment.save_full_frame(camera_config, yuv_frame)
-                    self.frame_manager.close(frame_name)
                     self._publish_segment_update(
                         segment, camera_config, None, [], prev_data
                     )
@@ -770,7 +767,7 @@ class ReviewSegmentMaintainer(threading.Thread):
 
                 try:
                     yuv_frame = self.frame_manager.get(
-                        frame_name, camera_config.frame_shape_yuv
+                        frame_name, camera_config.frame_shape_yuv, frame_time
                     )
 
                     if yuv_frame is None:
@@ -780,7 +777,6 @@ class ReviewSegmentMaintainer(threading.Thread):
                     new_segment.update_frame(
                         camera_config, yuv_frame, activity.get_all_objects()
                     )
-                    self.frame_manager.close(frame_name)
                     self._publish_segment_start(new_segment)
                 except FileNotFoundError:
                     return

frigate/test/test_review_flow.py

@@ -74,8 +74,8 @@ class ReviewFlowTestCase(unittest.TestCase):
         maintainer.config_subscriber.check_for_updates.return_value = {}
         maintainer.detection_subscriber = MagicMock()
         maintainer.frame_manager = MagicMock()
-        maintainer.frame_manager.get.side_effect = lambda _name, shape: np.zeros(
-            shape, np.uint8
+        maintainer.frame_manager.get.side_effect = lambda _name, shape, _t=None: (
+            np.zeros(shape, np.uint8)
         )
         return maintainer
 

frigate/test/test_shared_memory_frame_manager.py

@@ -6,13 +6,16 @@ in-process cache (e.g. TrackedObjectProcessor) still holds a ref to
 the old, smaller segment.
 """
 
+import queue
 import unittest
+from multiprocessing import Event, Value
 from types import SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import numpy as np
 
 from frigate.util.image import SharedMemoryFrameManager
+from frigate.video.ffmpeg import capture_frames
 
 
 def _fake_shm(size: int) -> SimpleNamespace:
@@ -152,5 +155,201 @@ class TestSharedMemoryFrameManagerGetRecreatesLargerSegment(unittest.TestCase):
         self.assertEqual(arr_large.shape, (1620, 1920))
 
 
+def _segments(**sizes: int) -> dict[str, SimpleNamespace]:
+    return {name: _fake_shm(size) for name, size in sizes.items()}
+
+
+def _open_from(segments: dict[str, SimpleNamespace]):
+    def open_segment(name: str, **_kwargs):
+        if name not in segments:
+            raise FileNotFoundError(name)
+        return segments[name]
+
+    return open_segment
+
+
+class TestSharedMemoryFrameManagerFrameTimes(unittest.TestCase):
+    def test_frame_times_is_float64_view_over_ledger_segment(self) -> None:
+        manager = SharedMemoryFrameManager()
+        segments = _segments(cam_frame_times=8 * 3)
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory",
+            side_effect=_open_from(segments),
+        ):
+            times = manager.frame_times("cam")
+
+        self.assertEqual(times.dtype, np.float64)
+        self.assertEqual(times.shape, (3,))
+        times[1] = 12.5
+        self.assertEqual(
+            np.frombuffer(segments["cam_frame_times"].buf, dtype=np.float64)[1], 12.5
+        )
+
+    def test_frame_times_returns_none_when_ledger_missing(self) -> None:
+        manager = SharedMemoryFrameManager()
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory", side_effect=FileNotFoundError
+        ):
+            self.assertIsNone(manager.frame_times("cam"))
+
+    def test_get_returns_frame_when_ledger_matches(self) -> None:
+        manager = SharedMemoryFrameManager()
+        segments = _segments(cam_frame0=2_500, cam_frame_times=8 * 2)
+        np.frombuffer(segments["cam_frame_times"].buf, dtype=np.float64)[0] = 100.25
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory",
+            side_effect=_open_from(segments),
+        ):
+            arr = manager.get("cam_frame0", (50, 50), frame_time=100.25)
+
+        self.assertIsNotNone(arr)
+        self.assertEqual(arr.shape, (50, 50))
+
+    def test_get_returns_none_when_slot_was_overwritten(self) -> None:
+        manager = SharedMemoryFrameManager()
+        segments = _segments(cam_frame0=2_500, cam_frame_times=8 * 2)
+        np.frombuffer(segments["cam_frame_times"].buf, dtype=np.float64)[0] = 200.0
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory",
+            side_effect=_open_from(segments),
+        ):
+            arr = manager.get("cam_frame0", (50, 50), frame_time=100.25)
+
+        self.assertIsNone(arr)
+
+    def test_get_returns_none_when_slot_index_is_beyond_ledger(self) -> None:
+        manager = SharedMemoryFrameManager()
+        segments = _segments(cam_frame5=2_500, cam_frame_times=8 * 2)
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory",
+            side_effect=_open_from(segments),
+        ):
+            arr = manager.get("cam_frame5", (50, 50), frame_time=100.25)
+
+        self.assertIsNone(arr)
+
+    def test_get_reopens_once_when_cached_ledger_is_stale(self) -> None:
+        """A camera recreated at the same resolution gets new segments under
+        the old names. A long-lived reader still mapped to the unlinked
+        ledger sees a mismatch, which must trigger one reopen by name before
+        the frame is declared gone."""
+        manager = SharedMemoryFrameManager()
+        stale = _segments(cam_frame0=2_500, cam_frame_times=8 * 2)
+        manager.shm_store.update(stale)
+        current = _segments(cam_frame0=2_500, cam_frame_times=8 * 2)
+        np.frombuffer(current["cam_frame_times"].buf, dtype=np.float64)[0] = 100.25
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory",
+            side_effect=_open_from(current),
+        ):
+            arr = manager.get("cam_frame0", (50, 50), frame_time=100.25)
+
+        self.assertIsNotNone(arr)
+        self.assertIs(manager.shm_store["cam_frame0"], current["cam_frame0"])
+        self.assertIs(manager.shm_store["cam_frame_times"], current["cam_frame_times"])
+
+    def test_get_without_frame_time_skips_ledger(self) -> None:
+        manager = SharedMemoryFrameManager()
+        segments = _segments(cam_frame0=2_500)
+
+        with patch(
+            "frigate.util.image.UntrackedSharedMemory",
+            side_effect=_open_from(segments),
+        ):
+            arr = manager.get("cam_frame0", (50, 50))
+
+        self.assertIsNotNone(arr)
+        self.assertNotIn("cam_frame_times", manager.shm_store)
+
+
+class TestCaptureFrames(unittest.TestCase):
+    """capture_frames stamps the ledger and only advances the slot index
+    once a frame has been handed to the queue."""
+
+    def _run(self, frames: list[bytes], put_results: list[Exception | None]):
+        frame_size = 4
+        slots = {f"cam_frame{i}": bytearray(frame_size) for i in range(3)}
+        times = np.zeros(3, dtype=np.float64)
+        frame_manager = MagicMock()
+        frame_manager.write.side_effect = lambda name: memoryview(slots[name])
+        frame_manager.frame_times.return_value = times
+
+        stop_event = Event()
+        reads = iter(frames)
+        self.times_during_read: list[np.ndarray] = []
+
+        def readinto(buffer):
+            self.times_during_read.append(times.copy())
+
+            try:
+                frame = next(reads)
+            except StopIteration:
+                stop_event.set()
+                return 0
+
+            buffer[:] = frame
+            return len(frame)
+
+        ffmpeg_process = MagicMock()
+        ffmpeg_process.stdout.readinto.side_effect = readinto
+        ffmpeg_process.poll.return_value = None
+
+        frame_queue = MagicMock()
+        frame_queue.put.side_effect = put_results
+        config = SimpleNamespace(name="cam", enabled=True)
+
+        capture_frames(
+            ffmpeg_process,
+            config,
+            3,
+            0,
+            (2, 2),
+            frame_manager,
+            frame_queue,
+            Value("d", 0.0),
+            Value("d", 0.0),
+            Value("d", 0.0),
+            stop_event,
+        )
+
+        return slots, times, [c.args[0] for c in frame_queue.put.call_args_list]
+
+    def test_advances_slot_and_stamps_ledger_on_success(self) -> None:
+        slots, times, puts = self._run([b"aaaa", b"bbbb"], [None, None])
+
+        self.assertEqual([name for name, _ in puts], ["cam_frame0", "cam_frame1"])
+        self.assertEqual(bytes(slots["cam_frame0"]), b"aaaa")
+        self.assertEqual(bytes(slots["cam_frame1"]), b"bbbb")
+        self.assertEqual(times[0], puts[0][1])
+        self.assertEqual(times[1], puts[1][1])
+        self.assertGreater(times[0], 0)
+
+    def test_reuses_slot_when_queue_is_full(self) -> None:
+        slots, times, puts = self._run(
+            [b"aaaa", b"bbbb", b"cccc"], [None, queue.Full(), None]
+        )
+
+        self.assertEqual(
+            [name for name, _ in puts], ["cam_frame0", "cam_frame1", "cam_frame1"]
+        )
+        self.assertEqual(bytes(slots["cam_frame1"]), b"cccc")
+        self.assertEqual(bytes(slots["cam_frame2"]), bytes(4))
+        self.assertEqual(times[1], puts[2][1])
+        self.assertEqual(times[2], 0)
+
+    def test_slot_never_matches_while_it_is_being_filled(self) -> None:
+        _, _, puts = self._run([b"aaaa", b"bbbb"], [queue.Full(), None])
+
+        # the second read refills slot 0, which the first frame had stamped
+        self.assertEqual([name for name, _ in puts], ["cam_frame0", "cam_frame0"])
+        self.assertEqual(self.times_during_read[1][0], 0)
+
+
 if __name__ == "__main__":
     unittest.main()

frigate/util/image.py

@@ -1079,7 +1079,37 @@ class SharedMemoryFrameManager(FrameManager):
             logger.info(f"the file {name} not found")
             return None
 
-    def get(self, name: str, shape) -> np.ndarray | None:
+    def frame_times(self, camera: str) -> np.ndarray | None:
+        """Per-slot write times for a camera's frame ring, stamped by capture."""
+        buf = self.write(f"{camera}_frame_times")
+
+        if buf is None:
+            return None
+
+        return np.ndarray((len(buf) // 8,), dtype=np.float64, buffer=buf)
+
+    def _holds(self, name: str, frame_time: float) -> bool:
+        camera, _, index = name.rpartition("_frame")
+        times = self.frame_times(camera)
+        return (
+            times is not None
+            and index.isdigit()
+            and int(index) < len(times)
+            and times[int(index)] == frame_time
+        )
+
+    def get(
+        self, name: str, shape, frame_time: float | None = None
+    ) -> np.ndarray | None:
+        if frame_time is not None and not self._holds(name, frame_time):
+            # a mapping left over from a same-size recreate of the camera
+            # looks the same as a lapped slot, so reopen by name once
+            self.close(name)
+            self.close(f"{name.rpartition('_frame')[0]}_frame_times")
+
+            if not self._holds(name, frame_time):
+                return None
+
         try:
             required = int(np.prod(shape))
             shm = self.shm_store.get(name)

frigate/video/detect.py

@@ -607,7 +607,6 @@ def process_frames(
             )
         # add to the queue if not full
         if detected_objects_queue.full():
-            frame_manager.close(frame_name)
             dropped_frames.dropped(time.monotonic())
             continue
         else:
@@ -624,7 +623,6 @@ def process_frames(
                 )
             )
             camera_metrics.detection_fps.value = object_detector.fps.eps()
-            frame_manager.close(frame_name)
 
     motion_detector.stop()
     requestor.stop()

frigate/video/ffmpeg.py

@@ -64,6 +64,7 @@ def capture_frames(
     frame_rate.start()
     skipped_eps = EventsPerSecond()
     skipped_eps.start()
+    frame_times = frame_manager.frame_times(config.name)
 
     while not stop_event.is_set():
         # CameraWatchdog applies enabled updates onto this same CameraConfig
@@ -78,6 +79,11 @@ def capture_frames(
         current_frame.value = datetime.now().timestamp()
         frame_name = f"{config.name}_frame{frame_index}"
         frame_buffer = frame_manager.write(frame_name)
+
+        # readinto fills the slot for the whole pipe wait, so readers must not
+        # match it until the frame is complete
+        frame_times[frame_index] = 0
+
         try:
             if ffmpeg_process.stdout.readinto(frame_buffer) != frame_size:
                 raise EOFError("incomplete frame")
@@ -96,16 +102,17 @@ def capture_frames(
 
             continue
 
+        frame_times[frame_index] = current_frame.value
         frame_rate.update()
 
         # don't lock the queue to check, just try since it should rarely be full
         try:
             # add to the queue
             frame_queue.put((frame_name, current_frame.value), False)
-            frame_manager.close(frame_name)
         except queue.Full:
-            # if the queue is full, skip this frame
+            # nobody was told about this slot, so the next frame reuses it
             skipped_eps.update()
+            continue
 
         frame_index = 0 if frame_index == shm_frame_count - 1 else frame_index + 1