fix: drain the sample camera SUB socket to the newest frame

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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DVqbXPoeHyqrQq5Vc8EcYP
This commit is contained in:
2026-09-08 11:35:07 +02:00
co-authored by Claude Opus 5
parent 9f0122bd96
commit 37c9750206
+29 -3
View File
@@ -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