From 37c9750206bb49b35e8ec4d613ea77faa42ce1d3 Mon Sep 17 00:00:00 2001 From: Filip Leonarski Date: Tue, 8 Sep 2026 11:35:07 +0200 Subject: [PATCH] fix: drain the sample camera SUB socket to the newest frame MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SUB socket is strictly FIFO, so once the GUI falls behind at 50 Hz it stays behind: every recv returns the oldest queued frame and the lag never recovers on its own. zmq.CONFLATE cannot fix this because it keeps only the last *part* of a message, which shreds the multipart header+payload frames the producer sends. recv_multipart is atomic, though — a non-blocking recv either yields a whole message or raises Again — so draining in a loop is a multipart-safe conflate. The drain runs after the previous frame's decode, so it discards exactly the backlog that built up while we were busy, and the socket is consumed at line rate regardless of how long a frame takes to decode. Verified against a real PUB/SUB pair: 100 queued multipart messages collapse to the newest one with both parts intact, and RCVTIMEO is still honoured when the queue is empty. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01DVqbXPoeHyqrQq5Vc8EcYP --- src/aare/gui/threads/prediction_subscriber.py | 32 +++++++++++++++++-- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/src/aare/gui/threads/prediction_subscriber.py b/src/aare/gui/threads/prediction_subscriber.py index 85b5a24f..b5234948 100644 --- a/src/aare/gui/threads/prediction_subscriber.py +++ b/src/aare/gui/threads/prediction_subscriber.py @@ -50,6 +50,8 @@ class PredictionSubscriber(QThread): self._camera_available = False self._last_camera_error: str | None = None + self._dropped = 0 # stale frames skipped since the last fps report + if isinstance(topic, str): self._sock.setsockopt_string(zmq.SUBSCRIBE, topic) elif isinstance(topic, bytes): @@ -151,13 +153,30 @@ class PredictionSubscriber(QThread): sharpness = focus_measure_edges(gray, self._focus_mask) self.focus_measure.emit(sharpness) + def _recv_latest(self) -> list[bytes]: + """Return the newest queued message, discarding the backlog behind it. + + zmq.CONFLATE is unusable here: it keeps only the last *part*, which + shreds multipart messages. recv_multipart is atomic though — a + non-blocking recv either yields a whole message or raises Again — so + draining in a loop is a multipart-safe conflate. The drain runs after + the (slow) decode of the previous frame, so it eats exactly the + backlog that piled up while we were busy, and the stream is always + consumed at line rate no matter how long a frame takes to decode. + """ + parts = self._sock.recv_multipart() # honours RCVTIMEO, may raise zmq.Again + while True: + try: + parts = self._sock.recv_multipart(zmq.NOBLOCK) + except zmq.Again: + return parts + self._dropped += 1 + def run(self): - self._debug_last_log_ts = time.perf_counter() - self._debug_msg_count = 0 try: while self.running: try: - parts = self._sock.recv_multipart() + parts = self._recv_latest() except zmq.Again: now = time.perf_counter() elapsed = now - self._fps_window_start @@ -185,6 +204,13 @@ class PredictionSubscriber(QThread): if elapsed >= self._fps_emit_period_s: fps = self._fps_frame_count / elapsed if elapsed > 0 else 0.0 self.fps_measure.emit(float(fps)) + if self._dropped: + logger.debug( + "Sample camera: skipped %d stale frame(s), showing %.1f fps", + self._dropped, + fps, + ) + self._dropped = 0 self._fps_window_start = now self._fps_frame_count = 0