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.
This commit is contained in:
Josh Hawkins
2026-10-12 09:15:37 -05:00
parent ef540d88cb
commit 641a4ee0c6
7 changed files with 70 additions and 15 deletions
+3
View File
@@ -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()
+5
View File
@@ -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
+1
View File
@@ -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)
+2
View File
@@ -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)
@@ -243,26 +243,56 @@ class TestSharedMemoryFrameManagerFrameTimes(unittest.TestCase):
self.assertIsNone(arr)
def test_get_reopens_once_when_cached_ledger_is_stale(self) -> None:
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 trigger one reopen by name before
the frame is declared gone."""
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_frame_times=8 * 2)
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_frame_times=8 * 2)
np.frombuffer(current["cam_frame_times"].buf, dtype=np.float64)[0] = 100.25
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.assertIsNotNone(arr)
self.assertIs(manager.shm_store["cam_frame0"], current["cam_frame0"])
self.assertIs(manager.shm_store["cam_frame_times"], current["cam_frame_times"])
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)
)
manager.close_camera("cam")
self.assertEqual(set(manager.shm_store), {"other_frame0"})
def test_get_without_frame_time_skips_ledger(self) -> None:
manager = SharedMemoryFrameManager()
+1
View File
@@ -763,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)
+18 -5
View File
@@ -1003,6 +1003,10 @@ class FrameManager(ABC):
def close(self, name: str):
pass
@abstractmethod
def close_camera(self, camera: str) -> None:
pass
@abstractmethod
def delete(self, name: str):
pass
@@ -1110,7 +1114,8 @@ class SharedMemoryFrameManager(FrameManager):
"""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. Copy it to keep it longer.
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
@@ -1122,15 +1127,18 @@ class SharedMemoryFrameManager(FrameManager):
a different frame than frame_time
"""
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")
# 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)
@@ -1161,6 +1169,11 @@ 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 [n for n in self.shm_store if n.startswith(f"{camera}_frame")]:
self.close(name)
def delete(self, name: str):
if name in self.shm_store:
self.shm_store[name].close()