From ff9eb8f18430198548d9c9d2cf138ed9ac7b526b Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Mon, 12 Oct 2026 11:21:08 -0500 Subject: [PATCH] Keep frame shared memory mapped (#24644) * 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. * verify the frame for embeddings object updates Tracked object events only carried the object's own frame_time, which stays at its last detection when the object isn't detected in a frame, so embeddings couldn't check the slot it read and could crop face or plate data from a frame that had already been overwritten. Events now carry the time of the frame they name, and embeddings passes it to `get()`. * log when a frame slot was overwritten before it was read A lapped read and a missing segment ended up as the same debug line at every caller, so there was no way to tell a slow reader from a missing frame. * bring the FrameManager interface in line with its implementation The abstract get() still had a timeout_ms signature from an older implementation and frame_times() wasn't declared, though capture and detect are typed against the interface. get() also didn't say how long the array it returns stays valid. * drop a camera's cached frame mappings when it is removed or recreated A reader that kept its mappings open stayed on a removed camera's segments, which pinned their shared memory until restart, and after a same-size recreate the ledger reopen only refreshed the slot being read while the other cached slots still returned the old camera's pixels. Readers now drop every cached slot for a camera on its remove update, and get() does the same when a ledger mismatch turns out to be a replaced ledger rather than a lapped slot. * match a camera's frame names exactly in close_camera The prefix match on `{camera}_frame` also caught another camera whose name starts with it, so removing `cam` dropped the cached ring of a camera named `cam_frame`. --- frigate/camera/maintainer.py | 4 + frigate/camera/state.py | 13 +- frigate/comms/events_updater.py | 20 +- frigate/embeddings/maintainer.py | 15 +- frigate/events/maintainer.py | 2 +- frigate/object_detection/base.py | 5 - frigate/output/output.py | 24 +- frigate/ptz/autotrack.py | 4 +- frigate/review/maintainer.py | 13 +- frigate/test/test_camera_lifecycle.py | 2 + frigate/test/test_review_flow.py | 4 +- .../test/test_shared_memory_frame_manager.py | 323 +++++++++++++++++- frigate/track/object_processing.py | 26 +- frigate/util/image.py | 71 +++- frigate/video/detect.py | 2 - frigate/video/ffmpeg.py | 11 +- 16 files changed, 472 insertions(+), 67 deletions(-) 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..764ef10f61 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() @@ -404,7 +404,7 @@ class CameraState: self.send_mqtt_snapshot(new_obj, object_type) for c in self.callbacks["start"]: - c(self.name, new_obj, frame_name) + c(self.name, new_obj, frame_name, frame_time) for id in updated_ids: updated_obj = tracked_objects[id] @@ -416,7 +416,7 @@ class CameraState: if autotracker_update or significant_update: for c in self.callbacks["autotrack"]: - c(self.name, updated_obj, frame_name) + c(self.name, updated_obj, frame_name, frame_time) if thumb_update and current_frame is not None: # ensure this frame is stored in the cache @@ -466,7 +466,7 @@ class CameraState: ): # call event handlers for c in self.callbacks["update"]: - c(self.name, updated_obj, frame_name) + c(self.name, updated_obj, frame_name, frame_time) updated_obj.last_published = frame_time # send MQTT snapshot when object first enters a required zone, @@ -488,7 +488,7 @@ class CameraState: removed_obj.obj_data["end_time"] = frame_time logger.debug(f"{self.name}: end callback for object {id}") for c in self.callbacks["end"]: - c(self.name, removed_obj, frame_name) + c(self.name, removed_obj, frame_name, frame_time) # TODO: can i switch to looking this up and only changing when an event ends? # maintain best objects @@ -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/comms/events_updater.py b/frigate/comms/events_updater.py index cfd958d2c7..2071c0ae13 100644 --- a/frigate/comms/events_updater.py +++ b/frigate/comms/events_updater.py @@ -8,7 +8,16 @@ from .zmq_proxy import Publisher, Subscriber class EventUpdatePublisher( - Publisher[tuple[EventTypeEnum, EventStateEnum, str | None, str, dict[str, Any]]] + Publisher[ + tuple[ + EventTypeEnum, + EventStateEnum, + str | None, + str, + dict[str, Any], + float | None, + ] + ] ): """Publishes events (objects, audio, manual).""" @@ -19,7 +28,14 @@ class EventUpdatePublisher( def publish( self, - payload: tuple[EventTypeEnum, EventStateEnum, str | None, str, dict[str, Any]], + payload: tuple[ + EventTypeEnum, + EventStateEnum, + str | None, + str, + dict[str, Any], + float | None, + ], sub_topic: str = "", ) -> None: super().publish(payload, sub_topic) diff --git a/frigate/embeddings/maintainer.py b/frigate/embeddings/maintainer.py index f1055f26f5..e95a78212b 100644 --- a/frigate/embeddings/maintainer.py +++ b/frigate/embeddings/maintainer.py @@ -332,6 +332,9 @@ class EmbeddingMaintainer(threading.Thread): """Apply camera config updates and register newly enabled processors.""" updated_topics = self.config_updater.check_for_updates() + for camera in updated_topics.get("remove", []): + self.frame_manager.close_camera(camera) + if updated_topics.keys() & GENAI_UPDATE_TOPICS: self._sync_genai_processors() @@ -517,7 +520,7 @@ class EmbeddingMaintainer(threading.Thread): if update is None: return - source_type, event_type, camera, frame_name, data = update + source_type, event_type, camera, frame_name, data, frame_time = update logger.debug( f"Received update - source_type: {source_type}, camera: {camera}, data label: {data.get('label') if data else 'None'}" @@ -546,7 +549,7 @@ class EmbeddingMaintainer(threading.Thread): # Create our own thumbnail based on the bounding box and the frame time 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: logger.debug(f"Frame {frame_name} not found for camera {camera}") @@ -583,8 +586,6 @@ class EmbeddingMaintainer(threading.Thread): PostProcessDataEnum.tracked_object, ) - self.frame_manager.close(frame_name) - def _process_finalized(self) -> None: """Process the end of an event.""" while True: @@ -793,7 +794,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 +818,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 +842,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/events/maintainer.py b/frigate/events/maintainer.py index cc7a9aa10e..c677d4f58f 100644 --- a/frigate/events/maintainer.py +++ b/frigate/events/maintainer.py @@ -94,7 +94,7 @@ class EventProcessor(threading.Thread): if update == None: continue - source_type, event_type, camera, _, event_data = update + source_type, event_type, camera, _, event_data, _ = update logger.debug( f"Event received: {source_type} {event_type} {camera} {event_data['id']}" 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..3a76478da4 100644 --- a/frigate/output/output.py +++ b/frigate/output/output.py @@ -145,6 +145,7 @@ class OutputProcess(FrigateProcess): CameraConfigUpdateEnum.birdseye, CameraConfigUpdateEnum.enabled, CameraConfigUpdateEnum.record, + CameraConfigUpdateEnum.remove, ], ) birdseye_config_subscriber = ConfigSubscriber("config/birdseye", exact=True) @@ -208,6 +209,10 @@ class OutputProcess(FrigateProcess): birdseye, ) + if CameraConfigUpdateEnum.remove in updates: + for camera in updates["remove"]: + frame_manager.close_camera(camera) + _result = detection_subscriber.check_for_update(timeout=1) if _result is None: continue @@ -241,7 +246,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 +301,6 @@ class OutputProcess(FrigateProcess): frame, ) - frame_manager.close(frame_name) - move_preview_frames("clips") while True: @@ -309,21 +312,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..d9c31871cb 100644 --- a/frigate/review/maintainer.py +++ b/frigate/review/maintainer.py @@ -504,6 +504,7 @@ class ReviewSegmentMaintainer(threading.Thread): def _handle_camera_removed(self, camera: str) -> None: """Close out a deleted camera's segment so a reused name cannot inherit it.""" self.forcibly_end_segment(camera) + self.frame_manager.close_camera(camera) self.indefinite_events.pop(camera, None) self.recent_classification_state_changes.pop(camera, None) @@ -544,7 +545,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 +555,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 +653,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 +669,6 @@ class ReviewSegmentMaintainer(threading.Thread): activity.get_all_objects(), prev_data, ) - self.frame_manager.close(frame_name) except FileNotFoundError: return @@ -677,7 +676,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 +684,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 +768,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 +778,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_camera_lifecycle.py b/frigate/test/test_camera_lifecycle.py index 1b5b8920ac..6917a7bccd 100644 --- a/frigate/test/test_camera_lifecycle.py +++ b/frigate/test/test_camera_lifecycle.py @@ -183,10 +183,12 @@ class TestReviewMaintainerRemoval(unittest.TestCase): "deleted_cam": [{"model": "gate", "from": "a", "to": "b", "timestamp": 1.0}] } maintainer.forcibly_end_segment = MagicMock() + maintainer.frame_manager = MagicMock() maintainer._handle_camera_removed("deleted_cam") maintainer.forcibly_end_segment.assert_called_once_with("deleted_cam") + maintainer.frame_manager.close_camera.assert_called_once_with("deleted_cam") self.assertNotIn("deleted_cam", maintainer.indefinite_events) self.assertNotIn("deleted_cam", maintainer.recent_classification_state_changes) 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..786fc0563d 100644 --- a/frigate/test/test_shared_memory_frame_manager.py +++ b/frigate/test/test_shared_memory_frame_manager.py @@ -6,13 +6,20 @@ in-process cache (e.g. TrackedObjectProcessor) still holds a ref to the old, smaller segment. """ +import queue +import threading 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.embeddings.maintainer import EmbeddingMaintainer +from frigate.events.types import EventStateEnum, EventTypeEnum +from frigate.track.object_processing import TrackedObjectProcessor from frigate.util.image import SharedMemoryFrameManager +from frigate.video.ffmpeg import capture_frames def _fake_shm(size: int) -> SimpleNamespace: @@ -152,5 +159,319 @@ 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), + ), + self.assertLogs("frigate.util.image", "DEBUG") as logs, + ): + arr = manager.get("cam_frame0", (50, 50), frame_time=100.25) + + self.assertIsNone(arr) + self.assertIn( + "cam_frame0 no longer holds the frame from 100.25", logs.output[0] + ) + + 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_every_cached_slot_when_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 drop every cached slot of that camera, not + only the one being read, or later reads return the old pixels.""" + manager = SharedMemoryFrameManager() + stale = _segments(cam_frame0=2_500, cam_frame1=2_500, cam_frame_times=8 * 2) + manager.shm_store.update(stale) + current = _segments(cam_frame0=2_500, cam_frame1=2_500, cam_frame_times=8 * 2) + current["cam_frame1"].buf[0] = 7 + np.frombuffer(current["cam_frame_times"].buf, dtype=np.float64)[:] = ( + 100.25, + 100.5, + ) + + with patch( + "frigate.util.image.UntrackedSharedMemory", + side_effect=_open_from(current), + ): + first = manager.get("cam_frame0", (50, 50), frame_time=100.25) + second = manager.get("cam_frame1", (50, 50), frame_time=100.5) + + self.assertIsNotNone(first) + self.assertEqual(second[0, 0], 7) + self.assertIs(manager.shm_store["cam_frame1"], current["cam_frame1"]) + + def test_get_keeps_cached_slots_when_only_lapped(self) -> None: + manager = SharedMemoryFrameManager() + segments = _segments(cam_frame0=2_500, cam_frame1=2_500, cam_frame_times=8 * 2) + manager.shm_store.update(segments) + 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) + self.assertIs(manager.shm_store["cam_frame1"], segments["cam_frame1"]) + + def test_close_camera_drops_only_that_camera(self) -> None: + manager = SharedMemoryFrameManager() + manager.shm_store.update( + _segments(cam_frame0=1, cam_frame1=1, cam_frame_times=8, other_frame0=1) + ) + # a camera named cam_frame shares the cam_frame prefix + manager.shm_store.update( + {"cam_frame_frame0": _fake_shm(1), "cam_frame_frame_times": _fake_shm(8)} + ) + + manager.close_camera("cam") + + self.assertEqual( + set(manager.shm_store), + {"other_frame0", "cam_frame_frame0", "cam_frame_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) + + +class TestTrackedObjectEventsCarryFrameTime(unittest.TestCase): + """Tracked object events carry the time of the frame they name, which can + differ from the object's own frame_time when it wasn't detected in that + frame.""" + + def _callbacks(self) -> tuple[dict, MagicMock]: + processor = TrackedObjectProcessor.__new__(TrackedObjectProcessor) + processor.config = SimpleNamespace(cameras={"cam": SimpleNamespace(name="cam")}) + processor.event_sender = MagicMock() + processor.dispatcher = MagicMock() + processor.ptz_autotracker_thread = MagicMock() + processor.camera_states = {} + processor.camera_states_lock = threading.Lock() + processor.frame_manager = MagicMock() + processor.should_save_snapshot = MagicMock(return_value=False) + processor.should_retain_recording = MagicMock(return_value=False) + callbacks: dict = {} + + with patch("frigate.track.object_processing.CameraState") as state_cls: + state_cls.return_value.on.side_effect = lambda name, fn: callbacks.update( + {name: fn} + ) + processor.create_camera_state("cam") + + return callbacks, processor.event_sender + + def test_start_update_end_publish_frame_time(self) -> None: + callbacks, sender = self._callbacks() + obj = MagicMock() + obj.to_dict.return_value = {"id": "1", "frame_time": 10.0} + obj.obj_data = {"frame_time": 10.0} + obj.previous = {"false_positive": True} + obj.false_positive = True + + callbacks["start"]("cam", obj, "cam_frame3", 12.5) + callbacks["update"]("cam", obj, "cam_frame4", 13.0) + callbacks["end"]("cam", obj, "cam_frame5", 13.5) + + payloads = [c.args[0] for c in sender.publish.call_args_list] + self.assertEqual( + [(p[1], p[3], p[5]) for p in payloads], + [ + (EventStateEnum.start, "cam_frame3", 12.5), + (EventStateEnum.update, "cam_frame4", 13.0), + (EventStateEnum.end, "cam_frame5", 13.5), + ], + ) + + +class TestEmbeddingsObjectUpdateVerifiesFrame(unittest.TestCase): + def test_object_update_reads_frame_at_event_frame_time(self) -> None: + maintainer = EmbeddingMaintainer.__new__(EmbeddingMaintainer) + maintainer.config = SimpleNamespace( + cameras={"cam": SimpleNamespace(frame_shape_yuv=(6, 4))}, + semantic_search=SimpleNamespace(enabled=False), + ) + maintainer.event_subscriber = MagicMock() + maintainer.event_subscriber.check_for_update.return_value = ( + EventTypeEnum.tracked_object, + EventStateEnum.update, + "cam", + "cam_frame4", + {"id": "1", "label": "person", "frame_time": 10.0}, + 13.0, + ) + maintainer.realtime_processors = [MagicMock()] + maintainer.post_processors = [] + maintainer.frame_manager = MagicMock() + + maintainer._process_updates() + + maintainer.frame_manager.get.assert_called_once_with("cam_frame4", (6, 4), 13.0) + maintainer.frame_manager.close.assert_not_called() + + if __name__ == "__main__": unittest.main() diff --git a/frigate/track/object_processing.py b/frigate/track/object_processing.py index eb334e95f4..1183811651 100644 --- a/frigate/track/object_processing.py +++ b/frigate/track/object_processing.py @@ -118,7 +118,9 @@ class TrackedObjectProcessor(threading.Thread): def create_camera_state(self, camera: str) -> None: """Creates a new camera state.""" - def start(camera: str, obj: TrackedObject, frame_name: str) -> None: + def start( + camera: str, obj: TrackedObject, frame_name: str, frame_time: float + ) -> None: self.event_sender.publish( ( EventTypeEnum.tracked_object, @@ -126,10 +128,13 @@ class TrackedObjectProcessor(threading.Thread): camera, frame_name, obj.to_dict(), + frame_time, ) ) - def update(camera: str, obj: TrackedObject, frame_name: str) -> None: + def update( + camera: str, obj: TrackedObject, frame_name: str, frame_time: float + ) -> None: obj.has_snapshot = self.should_save_snapshot( camera_state.camera_config, obj ) @@ -149,13 +154,18 @@ class TrackedObjectProcessor(threading.Thread): camera, frame_name, obj.to_dict(), + frame_time, ) ) - def autotrack(camera: str, obj: TrackedObject, frame_name: str) -> None: + def autotrack( + camera: str, obj: TrackedObject, frame_name: str, frame_time: float + ) -> None: self.ptz_autotracker_thread.autotrack_object(camera, obj) - def end(camera: str, obj: TrackedObject, frame_name: str) -> None: + def end( + camera: str, obj: TrackedObject, frame_name: str, frame_time: float + ) -> None: # populate has_snapshot obj.has_snapshot = self.should_save_snapshot( camera_state.camera_config, obj @@ -186,6 +196,7 @@ class TrackedObjectProcessor(threading.Thread): camera, frame_name, obj.to_dict(), + frame_time, ) ) @@ -599,6 +610,7 @@ class TrackedObjectProcessor(threading.Thread): "type": source_type, "draw": draw, }, + None, ) ) @@ -664,6 +676,7 @@ class TrackedObjectProcessor(threading.Thread): "recognized_license_plate": plate, "recognized_license_plate_score": score, }, + None, ) ) @@ -692,6 +705,7 @@ class TrackedObjectProcessor(threading.Thread): None, "", {"id": event_id, "end_time": end_time}, + None, ) ) @@ -713,13 +727,14 @@ class TrackedObjectProcessor(threading.Thread): def force_end_all_events(self, camera: str, camera_state: CameraState) -> None: """Ends all active events on camera when disabling.""" last_frame_name = camera_state.previous_frame_id + last_frame_time = camera_state.current_frame_time for obj_id, obj in list(camera_state.tracked_objects.items()): if "end_time" not in obj.obj_data: logger.debug(f"Camera {camera} disabled, ending active event {obj_id}") obj.obj_data["end_time"] = datetime.datetime.now().timestamp() # end callbacks for callback in camera_state.callbacks["end"]: - callback(camera, obj, last_frame_name) + callback(camera, obj, last_frame_name, last_frame_time) # camera activity callbacks for callback in camera_state.callbacks["camera_activity"]: @@ -748,6 +763,7 @@ class TrackedObjectProcessor(threading.Thread): continue camera_state.shutdown() + self.frame_manager.close_camera(camera) with self.camera_states_lock: self.camera_states.pop(camera) diff --git a/frigate/util/image.py b/frigate/util/image.py index 37c2e0c895..f62ffed065 100644 --- a/frigate/util/image.py +++ b/frigate/util/image.py @@ -990,13 +990,23 @@ class FrameManager(ABC): pass @abstractmethod - def get(self, name: str, timeout_ms: int = 0): + def get( + self, name: str, shape: tuple[int, ...], frame_time: float | None = None + ) -> np.ndarray | None: + pass + + @abstractmethod + def frame_times(self, camera: str) -> np.ndarray | None: pass @abstractmethod def close(self, name: str): pass + @abstractmethod + def close_camera(self, camera: str) -> None: + pass + @abstractmethod def delete(self, name: str): pass @@ -1079,7 +1089,56 @@ 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: tuple[int, ...], frame_time: float | None = None + ) -> np.ndarray | None: + """Map a frame slot as an array without copying it. + + The array points into shared memory, so it is only valid until the next + close(), delete(), or get() of the same name, and capture overwrites its + pixels once the ring wraps back to this slot. Copy it to keep it longer. + + Args: + name: Slot name, such as front_door_frame3 + shape: Array shape, which must match the slot's size + frame_time: Time of the frame the caller was told is in the slot + + Returns: + The frame, or None when the slot is missing, the wrong size, or holds + a different frame than frame_time + """ + if frame_time is not None and not self._holds(name, frame_time): + # a ledger left over from a same-size recreate of the camera looks + # the same as a lapped slot, so reopen it by name once + camera = name.rpartition("_frame")[0] + self.close(f"{camera}_frame_times") + + if not self._holds(name, frame_time): + logger.debug("%s no longer holds the frame from %s", name, frame_time) + return None + + # the ledger was replaced, so every cached slot maps the old segments + self.close_camera(camera) + try: required = int(np.prod(shape)) shm = self.shm_store.get(name) @@ -1110,6 +1169,14 @@ class SharedMemoryFrameManager(FrameManager): self.shm_store[name].close() del self.shm_store[name] + def close_camera(self, camera: str) -> None: + """Drop every cached mapping for a camera's frame ring and ledger.""" + for name in list(self.shm_store): + owner, _, suffix = name.rpartition("_frame") + + if owner == camera and (suffix.isdigit() or suffix == "_times"): + self.close(name) + def delete(self, name: str): if name in self.shm_store: self.shm_store[name].close() 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