diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index 5b73a00667..4c564f799b 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -23,12 +23,14 @@ pipeline is continuously clocked by real frames and no keep-alives are needed (measured: back-to-back real frames round-trip in ~14 ms, while interleaving 1 s keep-alives stretched a real frame's result to ~4 s). -Results are paired to requests by CONTENT: each submitted frame carries a -32-bit pixel token in its top-left band (see `_stamp_token`); the SDK returns -the source frame back 1:1 (verified on hardware), so `receive_output` -decodes the token and delivers the result only to the matching connection id. -Primer frames (used only to kick-start pipeline init) carry no token and their -results are provably dropped even if they surface while a request is pending. +Results are paired to requests by RECORDED ORDER: the feeder registers every +frame it yields on a pending deque (``None`` for primer/keep-alive synthetic +frames, the connection id for real frames) and ``collect`` pops exactly one +entry per completed result. The SDK stream returns results strictly in input +order, one per yielded frame (verified on hardware: 200/200 frames in order, +no drops or duplicates, including with synthetics interleaved), so a real +frame always gets its own result, synthetics pop a ``None`` and are dropped, +and no frame id is ever assumed from an unrecorded ordering. Model configuration follows the hardware-backend convention: - `model.path` = URL | scheme-less host URL (``www.example.com/...``) | local @@ -44,6 +46,7 @@ import logging import os import queue import re +from collections import deque from typing import Literal import numpy as np @@ -71,23 +74,18 @@ PRIMER_FRAMES = 3 # A short spacing matters when they DO fire: the SDK pipeline is clocked by the # input arrival interval (depth 4 x gap), so a 1 s gap would stretch the next # real frame's result to ~4 s; 0.1 s keeps that penalty in the tens/hundreds -# of ms. These zeros carry no request token and are dropped by collect(). +# of ms. These zeros carry no request id and their results are dropped by collect(). KEEPALIVE_INTERVAL = 0.1 -# Per-frame tag: the SDK returns each source frame back 1:1 in the result -# (verified on hardware: BGR pass-through is exact, image dimension preserved), -# but it provides NO per-frame id (image_id is always empty, src_timestamp is a -# pipeline-clock value, not an id). We therefore write a 32-bit request token -# into a small pixel band in the top-left corner of every REAL frame and decode -# it from the result image, so results are paired to requests by CONTENT. This -# is immune to the SDK buffering, priming or reordering results: a buffered -# primer/keep-alive result carries token 0 and is dropped, and a real result -# can only be delivered to the frame that carries its token. -TOKEN_BITS = 32 -TOKEN_ROWS = (TOKEN_BITS + 2) // 3 # 11 rows x 3 channel-cells = 33 (one spare) -TOKEN_X0 = 4 -TOKEN_CELL_W = 4 -TOKEN_NO_ID = 0 +# Request pairing is done by RECORDED ORDER, not by content: the SDK's +# InferenceStream returns each source frame back in a result and yields results +# strictly in input order, one per yielded frame (verified on hardware: 200/200 +# frames in order, no drops or duplicates, synthetics interleaved included). +# The feeder therefore records connection ids (or None for primer/keep-alive +# synthetic frames) on a pending deque at each yield, and collect() pops exactly +# one entry per result: a synthetic entry is dropped, a real one delivers the +# result to its connection id. This mirrors how Frigate's own AsyncDetectorRunner +# pairs its send/recv timestamps, and how the memryx plugin pairs its queues. # ----------------- Utility Functions ----------------- # @@ -173,44 +171,6 @@ def _pad_rows(rows) -> np.ndarray: return out -def _stamp_token(frame: np.ndarray, token: int) -> None: - """Write a 32-bit request token into the top-left pixel band of `frame`. - - Band geometry must match `_read_token`. The values 0/255 survive the SDK's - BGR->BGRA pass-through exactly (verified on hardware). - """ - band = frame[0:TOKEN_ROWS, TOKEN_X0 : TOKEN_X0 + TOKEN_CELL_W] - for i in range(TOKEN_BITS): - row, ch = divmod(i, 3) - val = 255 if (token >> (TOKEN_BITS - 1 - i)) & 1 else 0 - band[row, :, ch] = val - - -def _read_token(frame_result) -> int: - """Decode the request token stamped on a stream result's source frame. - - The SDK returns the source frame 1:1 (BGRA); we read the matching band from - the first 3 channels. Returns TOKEN_NO_ID for tagless results (primer / - keep-alive, or a frame_result whose image is unavailable). - """ - img = getattr(frame_result, "image", None) - if img is None: - return TOKEN_NO_ID - try: - arr = img.asarray() - except Exception: # noqa: BLE001 - return TOKEN_NO_ID - if getattr(arr, "ndim", 0) < 3: - return TOKEN_NO_ID - band = arr[0:TOKEN_ROWS, TOKEN_X0 : TOKEN_X0 + TOKEN_CELL_W, 0:3] - token = 0 - for i in range(TOKEN_BITS): - row, ch = divmod(i, 3) - bit = 1 if band[row, :, ch].mean() > 127 else 0 - token = (token << 1) | bit - return token - - # ----------------- AxeleraInference (engine) ----------------- # @@ -220,8 +180,8 @@ class _AxeleraInference: The stream is built from a frame generator (``sources``), so frames can be pushed by ``submit`` while ``collect`` (called from the result thread) pulls one completed frame at a time. A short primer of zero frames - kick-starts model pipeline init; their results carry no token and are - dropped by ``collect``. + kick-starts model pipeline init; they are registered as synthetic (None) + and their results are dropped by ``collect``. In steady state real frames clock the SDK pipeline back-to-back: the caller (Frigate's AsyncDetectorRunner) submits frames continuously as @@ -230,15 +190,16 @@ class _AxeleraInference: latency to ~4 s on this depth-4 pipeline). A sparse 0.1 s keep-alive zero is emitted only while the incoming queue is momentarily empty (stream init, scheduling gaps, true idle) so the SDK source never starves; these - zeros carry no token and are dropped. + zeros are registered as synthetic (None) and their results are dropped. - Results are STRICTLY paired to requests by CONTENT: every real frame is - stamped with a 32-bit request token in its top-left pixel band, ``collect`` - decodes the token from the result image and delivers the outcome only to - the connection id that owns it. Primer frames carry no token, so their - results are provably dropped even if the SDK buffers or reorders them; no - FIFO ordering assumption remains, and a buffered result can never be - attributed to a different camera or crop. + Results are STRICTLY paired by RECORDED ORDER: ``_feed`` registers each + yielded frame on ``self._pending`` (``None`` for primer/keep-alive zeros, + the connection id for real frames) and ``collect`` pops exactly one entry + per result. The SDK yields results in input order, one per frame (verified + on hardware: 200/200 in order, no drops or duplicates, synthetics + interleaved included), so a synthetic result pops a ``None`` and is + dropped, and a real result always reaches the connection id that owns it; + no content read of the returned image is involved. """ def __init__( @@ -247,10 +208,10 @@ class _AxeleraInference: self._labelmap = labelmap self._width, self._height = int(width), int(height) self._frame_q: queue.Queue = queue.Queue(maxsize=64) - # token -> connection_id for frames in flight (written by submit on the - # detect thread, consumed by collect on the result thread) - self._token_to_conn: dict[int, str] = {} - self._next_id = 0 + # request ids (or None) for frames in flight, in the exact order they + # are yielded to the SDK (appended by _feed on the feeder thread, + # popped once per result by collect on the result thread) + self._pending: deque[str | None] = deque() if stream_factory is None: # lazy import: never import the SDK unless the detector is used @@ -271,23 +232,30 @@ class _AxeleraInference: def _feed(self): # primer frames: the SDK init pre-buffers several frames before the - # source yields (validated on hardware); they carry no token and their - # results are dropped by collect(). + # source yields (validated on hardware); they are synthetic, so each + # one is registered as None and its result is dropped by collect(). for _ in range(PRIMER_FRAMES): + self._pending.append(None) yield np.zeros((self._height, self._width, 3), np.uint8) while True: frame = None + connection_id = None try: - token, frame = self._frame_q.get(timeout=KEEPALIVE_INTERVAL) - _ = token # the token was already stamped in submit() + connection_id, frame = self._frame_q.get( + timeout=KEEPALIVE_INTERVAL + ) except queue.Empty: # safety net: the SDK source must never starve (stream init, - # scheduling gaps, true idle). spawn a token-less zero frame; - # its result is dropped by collect(). In steady state real - # frames keep the queue populated, so this rarely fires. + # scheduling gaps, true idle). spawn a synthetic zero frame; + # registered as None, its result is dropped by collect(). In + # steady state real frames keep the queue populated, so this + # rarely fires. frame = np.zeros((self._height, self._width, 3), np.uint8) if frame is None: # shutdown sentinel from stop() return + # record this yield's request id (None for synthetics) so collect() + # can pop results back to the right connection in order. + self._pending.append(connection_id) yield frame def _detection_rows(self, frame_result) -> list: @@ -341,15 +309,14 @@ class _AxeleraInference: return rows[:MAX_ROWS] def submit(self, connection_id: str, frame: np.ndarray) -> None: - """Send one frame for inference, tagged with a unique pixel token.""" - self._next_id += 1 - token = self._next_id - _stamp_token(frame, token) - self._token_to_conn[token] = connection_id + """Send one frame for inference, owned by `connection_id`. + + The frame is queued as-is (no pixel writes); `_feed` records the + connection id in yield order and `collect` pops it with the result. + """ try: - self._frame_q.put_nowait((token, frame)) + self._frame_q.put_nowait((connection_id, frame)) except queue.Full: - self._token_to_conn.pop(token, None) raise RuntimeError( "Axelera inference queue is full; restart required." ) from None @@ -358,8 +325,8 @@ class _AxeleraInference: """Wait for the next completed frame; return (connection_id, rows). ``rows`` is a fixed (20, 6) float32 numpy array of normalized - detections. Primer/synthetic results (token 0) are dropped internally. - Returns ``(None, None)`` when the stream is stopped. + detections. Primer/synthetic results (recorded as None) are dropped + internally. Returns ``(None, None)`` when the stream is stopped. """ while True: try: @@ -372,22 +339,30 @@ class _AxeleraInference: ) raise out = _pad_rows(self._detection_rows(frame_result)) - token = _read_token(frame_result) - if token == TOKEN_NO_ID: - # primer / keep-alive result: no request token, so it can never - # be delivered to a caller, even if it surfaces while a request - # is in flight. + try: + entry = self._pending.popleft() + except IndexError: + # a result with no registered yield: the SDK diverged from the + # recorded input order. Fail fast so the detector process is + # restarted, rather than silently misattribute every following + # result to the wrong connection (the P1 failure mode). + logger.error( + "axelera: inference result with no matching frame; " + "pairing desynced, restarting detector" + ) + raise RuntimeError( + "Axelera result/input pairing desynced" + ) from None + if entry is None: + # primer / keep-alive result: no request id recorded for it, so + # it is dropped and can never be delivered to a caller. continue - connection_id = self._token_to_conn.pop(token, None) - if connection_id is None: - # stale result for a frame we already returned or that errored - continue - return connection_id, out + return entry, out def stop(self) -> None: """Stop the stream and release the Metis device. Never raises.""" try: - self._frame_q.put_nowait((0, None)) + self._frame_q.put_nowait((None, None)) except Exception: # noqa: S110, BLE001 pass try: @@ -455,10 +430,12 @@ class AxeleraDetector(DetectionApi): if frame.shape[0] != 1: raise ValueError(f"Unexpected batch dimension in frame: {frame.shape}") frame = frame[0] - # Contiguous uint8 BGR for the SDK. If `frame` is still a view of the - # detect process's shared-memory buffer, stamping it in place is safe: - # AsyncDetectorRunner closes the input buffer only after our - # receive_output() returns, i.e. after the SDK has consumed its pixels. + # Contiguous uint8 BGR for the SDK. No pixel writes happen here: the + # frame is queued and fed as-is (pairing is by recorded order, not by + # content). If `frame` is a view of the detect process's shared-memory + # buffer and already contiguous, `ascontiguousarray` returns it + # unchanged, so the SDK reads the shared buffer directly; the runner + # keeps it valid until our receive_output() returns. frame = np.ascontiguousarray(frame, dtype=np.uint8) self._inference.submit(connection_id, frame)