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)"