From 76431fd9849a0bfeeb3d91b6b8b6a557875f5140 Mon Sep 17 00:00:00 2001 From: Jehan Azad Date: Sun, 6 Sep 2026 15:55:26 +0000 Subject: [PATCH] Snapshot track history deques before off-thread iteration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The five `Track.history` buffers became `deque(maxlen=...)` in #25. A deque raises `RuntimeError: deque mutated during iteration` when it is appended to while a Python-level loop walks it — something the plain lists tolerated. Only the frame worker appends, but `get_recent_detections` and `to_dict` are called from other threads (feed flush, analytics, admin state snapshots), so those two readers were raising in production and dropping the tick that hit the race. `list(deque)` is a single C call and therefore an atomic snapshot under the GIL, so both readers now snapshot each buffer before iterating: - `get_recent_detections`: snapshot measurements and timestamps back to back, then run the same `zip(reversed(...), reversed(...))` reverse scan. The zip is deliberately unchanged — the mutator appends timestamps before measurements, so a reader landing mid-append can see one extra timestamp and zip truncates to the shorter buffer, exactly as it did when these were lists. - `to_dict`: snapshot all four emitted buffers, then truncate to their common length so the diagnostic arrays stay aligned with each other. tracker.py's own readers are untouched: they run on the mutating thread. New `tests/test_history_snapshot.py` stresses both readers from the main thread for ~0.5 s while a second thread drives `update()`/`mark_missed()`. It fails on 5 of 5 runs without this change. Co-Authored-By: Claude Fable 5.1 --- retina_tracker/track.py | 38 ++++++++++++--- tests/test_history_snapshot.py | 86 ++++++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 7 deletions(-) create mode 100644 tests/test_history_snapshot.py diff --git a/retina_tracker/track.py b/retina_tracker/track.py index 80cbacf..a361a39 100644 --- a/retina_tracker/track.py +++ b/retina_tracker/track.py @@ -794,8 +794,17 @@ def get_length_bucket(self): def get_recent_detections(self, n=20): """Return the last *n* non-None detections (reverse scan, early exit).""" + # Called off the frame worker (analytics, feed builders) while that worker + # may be appending, and a deque raises RuntimeError if it is mutated during + # Python-level iteration. list(deque) is a single C call, so under the GIL + # it is an atomic snapshot; take both buffers before iterating. The zip is + # unchanged: a reader that lands mid-append can see one more timestamp than + # measurement (the mutator appends timestamps first), and zip truncating to + # the shorter buffer is the same behaviour these buffers had as lists. + meas = list(self.history["measurements"]) + ts_hist = list(self.history["timestamps"]) result = [] - for m, ts in zip(reversed(self.history["measurements"]), reversed(self.history["timestamps"])): + for m, ts in zip(reversed(meas), reversed(ts_hist)): if m is not None: result.append( { @@ -812,6 +821,21 @@ def get_recent_detections(self, n=20): return result def to_dict(self): + # Served from admin/state-snapshot threads while the frame worker appends. + # list(deque) is atomic under the GIL; the comprehensions below are not, so + # snapshot every buffer first. The snapshots can differ in length by one + # (the mutator appends timestamps -> frames -> states -> measurements), so + # truncate to the common length to keep the emitted arrays aligned. + hist_timestamps = list(self.history["timestamps"]) + hist_states = list(self.history["states"]) + hist_measurements = list(self.history["measurements"]) + hist_state_status = list(self.history["state_status"]) + n_hist = min(len(hist_timestamps), len(hist_states), len(hist_measurements), len(hist_state_status)) + hist_timestamps = hist_timestamps[:n_hist] + hist_states = hist_states[:n_hist] + hist_measurements = hist_measurements[:n_hist] + hist_state_status = hist_state_status[:n_hist] + duration_sec = (self.death_timestamp - self.birth_timestamp) / 1000.0 avg_snr = self.total_snr / max(self.n_associated, 1) continuity = self.n_associated / max(self.n_frames, 1) @@ -835,11 +859,11 @@ def to_dict(self): "anomaly_types": list(self.anomaly_types), "anomaly_detections": self.anomaly_detections, "history": { - "timestamps": list(self.history["timestamps"]), - "states": [s.tolist() for s in self.history["states"]], - "delays": [m["delay"] if m else None for m in self.history["measurements"]], - "dopplers": [m["doppler"] if m else None for m in self.history["measurements"]], - "snrs": [m["snr"] if m else None for m in self.history["measurements"]], - "state_status": list(self.history["state_status"]), + "timestamps": hist_timestamps, + "states": [s.tolist() for s in hist_states], + "delays": [m["delay"] if m else None for m in hist_measurements], + "dopplers": [m["doppler"] if m else None for m in hist_measurements], + "snrs": [m["snr"] if m else None for m in hist_measurements], + "state_status": hist_state_status, }, } diff --git a/tests/test_history_snapshot.py b/tests/test_history_snapshot.py new file mode 100644 index 0000000..17dd2cc --- /dev/null +++ b/tests/test_history_snapshot.py @@ -0,0 +1,86 @@ +"""Off-thread readers of `Track.history` must not see a mutating deque. + +The five history buffers are `deque(maxlen=TRACK_HISTORY_MAX)`. Only the frame +worker appends to a given node's tracker, but several readers run on other +threads (the aircraft-feed flush executor, the analytics executor, the admin +state-snapshot routes). A deque raises `RuntimeError: deque mutated during +iteration` if it is appended to while a Python-level loop walks it, which a +plain list tolerated -- so `get_recent_detections` and `to_dict` must snapshot +each buffer with `list()` (one C call, atomic under the GIL) before iterating. +""" + +import threading + +from retina_tracker.config import get_config +from retina_tracker.tracker import Tracker + + +def make_detection(i): + return {"delay": 10.0 + i * 0.01, "doppler": 50.0, "snr": 20.0} + + +def make_track(): + """A live track with a real Kalman filter, via the normal Tracker path.""" + tracker = Tracker(config=get_config()) + ts = 0 + for i in range(20): + tracker.process_frame([make_detection(i)], ts) + ts += 1000 + return max(tracker.get_tracks(), key=lambda t: t.n_associated), ts + + +def test_readers_survive_concurrent_appends(): + track, start_ts = make_track() + + errors = [] + stop = threading.Event() + + def appender(): + ts = start_ts + i = track.n_frames + try: + while not stop.is_set(): + ts += 1000 + i += 1 + if i % 5 == 0: + track.mark_missed(ts, frame=i) + else: + track.update(make_detection(i), ts, frame=i) + except Exception as exc: # pragma: no cover - failure path + errors.append(("appender", exc)) + + writer = threading.Thread(target=appender, daemon=True) + writer.start() + try: + deadline = threading.Event() + timer = threading.Timer(0.5, deadline.set) + timer.start() + reads = 0 + try: + while not deadline.is_set(): + for _ in range(50): + recent = track.get_recent_detections(n=5) + for det in recent: + assert isinstance(det, dict) + assert det["timestamp"] is not None + assert det["delay"] is not None + assert det["doppler"] is not None + assert det["snr"] is not None + dumped = track.to_dict() + history = dumped["history"] + n = len(history["timestamps"]) + assert len(history["states"]) == n + assert len(history["delays"]) == n + assert len(history["dopplers"]) == n + assert len(history["snrs"]) == n + assert len(history["state_status"]) == n + reads += 1 + finally: + timer.cancel() + finally: + stop.set() + writer.join(timeout=2.0) + + assert not errors, errors + assert reads > 100, f"stress loop did too little work to be meaningful ({reads} reads)" + assert track.n_frames > 100, f"appender did too little work to be meaningful ({track.n_frames} frames)"