diff --git a/frigate/camera/maintainer.py b/frigate/camera/maintainer.py index 9634b3803c..6994759855 100644 --- a/frigate/camera/maintainer.py +++ b/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, diff --git a/frigate/camera/state.py b/frigate/camera/state.py index c459098f7c..33bc25d4d4 100644 --- a/frigate/camera/state.py +++ b/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: diff --git a/frigate/embeddings/maintainer.py b/frigate/embeddings/maintainer.py index f1055f26f5..ae51579e65 100644 --- a/frigate/embeddings/maintainer.py +++ b/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: diff --git a/frigate/object_detection/base.py b/frigate/object_detection/base.py index e43be54f46..14805086c9 100644 --- a/frigate/object_detection/base.py +++ b/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) diff --git a/frigate/output/output.py b/frigate/output/output.py index 69cd9eb6dc..940ed9caf7 100644 --- a/frigate/output/output.py +++ b/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(): diff --git a/frigate/ptz/autotrack.py b/frigate/ptz/autotrack.py index 8049b29434..2a75a33011 100644 --- a/frigate/ptz/autotrack.py +++ b/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 diff --git a/frigate/review/maintainer.py b/frigate/review/maintainer.py index f97940d4a3..08fa4f32e9 100644 --- a/frigate/review/maintainer.py +++ b/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 diff --git a/frigate/test/test_review_flow.py b/frigate/test/test_review_flow.py index 2a1b5ee097..d71884ba7f 100644 --- a/frigate/test/test_review_flow.py +++ b/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 diff --git a/frigate/test/test_shared_memory_frame_manager.py b/frigate/test/test_shared_memory_frame_manager.py index 63c96f732d..298ebeef3d 100644 --- a/frigate/test/test_shared_memory_frame_manager.py +++ b/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() diff --git a/frigate/util/image.py b/frigate/util/image.py index 37c2e0c895..9a688b8510 100644 --- a/frigate/util/image.py +++ b/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) diff --git a/frigate/video/detect.py b/frigate/video/detect.py index f55c1c9635..a2877f83c0 100644 --- a/frigate/video/detect.py +++ b/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() diff --git a/frigate/video/ffmpeg.py b/frigate/video/ffmpeg.py index 16e3afcdb5..d56671b520 100644 --- a/frigate/video/ffmpeg.py +++ b/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