diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index 0aa3134925..67a1e0b29e 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -14,11 +14,27 @@ This is the only code path that is reliable across Metis hosts (raw device-level connections can wedge the board on some hosts), and it amortises model compilation across frames. +The plugin uses Frigate's ASYNC detector contract (`AsyncDetectorRunner`, +same as the memryx plugin): `send_input(connection_id, tensor)` is called +from the detect worker thread as frames arrive, and `receive_output()` from +the result worker thread returns `(connection_id, rows)` per completed frame. +Because the runner keeps submitting real frames as cameras feed them, the SDK +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. + Model configuration follows the hardware-backend convention: - - `model.path` = URL (downloaded to the model cache) | local file path | SDK - model-zoo preset name (e.g. ``yolox-s-coco-onnx``). When omitted, the SDK - model-zoo preset ``AX_DEFAULT_NETWORK`` (YOLOX, Apache-2.0 licensed) is - used; use `model.path: yolox-s-coco-onnx` to select it explicitly. + - `model.path` = URL | scheme-less host URL (``www.example.com/...``) | local + file path | SDK model-zoo preset name (e.g. ``yolox-s-coco-onnx``). When + omitted, the SDK model-zoo preset ``AX_DEFAULT_NETWORK`` (YOLOX, Apache-2.0 + licensed) is used. - `model.labelmap_path` / `model.labelmap`: labels are mapped to class ids BY NAME via the merged labelmap, independent of label ordering. - `model.width` / `model.height`: input frame dimensions for the network. @@ -28,8 +44,6 @@ import logging import os import queue import re -import threading -import time from typing import Literal import numpy as np @@ -48,17 +62,8 @@ DETECTOR_KEY = "axelera" # YOLOX (Megvii) is Apache-2.0 licensed; Ultralytics YOLO models are AGPL-3.0 # and cannot be referenced by Frigate (MIT), see blakeblackshear/frigate#10717. AX_DEFAULT_NETWORK = "yolox-s-coco-onnx" - -INFERENCE_TIMEOUT = 5.0 MAX_ROWS = 20 PRIMER_FRAMES = 3 -# Keep-alive spacing. The SDK pipeline is "clocked" by the input arrival -# interval (measured: pipeline depth x arrival gap, e.g. ~4 x 1 s = ~4 s real -# frame latency at 1 s keep-alive spacing, vs ~58 ms back-to-back). Feeding -# frames with a SHORT idle gap keeps the pipeline rotating, so a real frame -# gets its result in tens of ms even when requests are sparse. The zeros carry -# no request token and their results are dropped by the pump. -KEEPALIVE_INTERVAL = 0.05 # 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), @@ -204,41 +209,35 @@ class _AxeleraInference: """Feeds frames through a single Voyager SDK inference stream. The stream is built from a frame generator (``sources``), so frames can be - pushed by ``detect_raw`` while the consumer thread iterates the stream. + 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``. - A short primer of zero frames kick-starts model pipeline init (matching the - validated behaviour of ``create_inference_stream``). While no request is - pending, the feeder keeps yielding zero keep-alive frames so the SDK stream - never starves and is not torn down by its internal inference timeout during - quiet periods (e.g. a camera with no motion). + No keep-alives: the caller (Frigate's AsyncDetectorRunner) submits real + frames continuously as cameras feed them, so the SDK pipeline is clocked + by real frames (verified: 1 s input gaps stretch a real frame's latency to + ~4 s on this depth-4 pipeline, back-to-back frames round-trip in ~14 ms). 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, the pump + 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 - that frame's caller. Primer/keep-alive 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 + 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. """ def __init__( self, network, labelmap: dict, width: int, height: int, stream_factory=None - ): + ) -> None: self._labelmap = labelmap self._width, self._height = int(width), int(height) self._frame_q: queue.Queue = queue.Queue(maxsize=64) - self._cond = threading.Condition() - # request ids a caller is still waiting on (guarded by _cond); the pump - # only delivers results whose decoded token is in this set. The feeder - # also uses it as the "real frame outstanding" signal. - self._waiting: set = set() - self._results: dict[int, np.ndarray] = {} + # 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 - # Serializes run() calls: pairing is content-based (see _read_token), - # so concurrent frames would be perfectly matched too, but keeping at - # most ONE frame in flight gives fully deterministic behaviour and - # bounds the SDK's internal buffering. - self._run_lock = threading.Lock() if stream_factory is None: # lazy import: never import the SDK unless the detector is used @@ -253,42 +252,20 @@ class _AxeleraInference: stream_factory = create_inference_stream self._stream = stream_factory(network=network, sources=[self._feed()]) - self._thread = threading.Thread( - target=self._pump, daemon=True, name="axelera-pump" - ) - self._thread.start() - - @property - def alive(self) -> bool: - return self._thread.is_alive() def _feed(self): # primer frames: the SDK init pre-buffers several frames before the - # source yields (validated on hardware). Their results are dropped by - # the pump while the pending queue is empty. + # source yields (validated on hardware); they carry no token and their + # results are dropped by collect(). for _ in range(PRIMER_FRAMES): yield np.zeros((self._height, self._width, 3), np.uint8) while True: - frame = frame_id = None - try: - frame_id, frame = self._frame_q.get(timeout=KEEPALIVE_INTERVAL) - except queue.Empty: - # keep-alive: the SDK pipeline is clocked by the input arrival - # interval, so feed a zero frame on a SHORT cadence to keep it - # rotating (a 1 s idle gap makes a real frame's result take ~4 s - # on a depth-4 pipeline; a short gap keeps it in the tens of ms). - # These frames carry no request token, so their results are - # dropped by the pump. (Previously a 1 s gap ALSO starved the - # SDK input source, which waited/looped instead of yielding.) - frame_id, frame = None, np.zeros( - (self._height, self._width, 3), np.uint8 - ) + # Blocking: the SDK input side pulls the next frame when ready. + # Real frames (pushed by submit) clock the pipeline. + token, frame = self._frame_q.get() if frame is None: # shutdown sentinel from stop() return - if frame_id is not None: - # Tag the frame with its request id; the pump decodes it from - # the result image (content-based pairing, no ordering assumed). - _stamp_token(frame, frame_id) + _ = token # the token was already stamped in submit() yield frame def _detection_rows(self, frame_result) -> list: @@ -341,104 +318,54 @@ class _AxeleraInference: rows.sort(key=lambda r: r[1], reverse=True) return rows[:MAX_ROWS] - def _pump(self): - logger.info("axelera: inference stream started") - try: - for frame_result in self._stream: - out = _pad_rows(self._detection_rows(frame_result)) - token = _read_token(frame_result) - if token == TOKEN_NO_ID: - # primer / keep-alive result: it carries no request token, - # so it can NEVER be delivered to a caller, even if it - # surfaces while a request is pending. This is what makes a - # buffered/synthetic result provably unable to steal a - # request id (cf. prior FIFO pairing). - continue - with self._cond: - if token not in self._waiting: - # stale result for a frame that already timed out - continue - self._results[token] = out - self._cond.notify_all() - except Exception: - logger.exception( - "axelera: inference stream failed; detector requires restart" - ) - - def run(self, tensor_input, timeout: float = INFERENCE_TIMEOUT) -> np.ndarray: - """Run one inference and return this frame's detections (see detect_raw).""" - # At most ONE frame may be in flight at a time: with several frames in - # the stream the SDK can reorder/buffer their results (misattribution - # across cameras observed after many thousands of frames). Serialized, - # the next returned result can only belong to the one frame we pushed. - with self._run_lock: - return self._run_locked(tensor_input, timeout) - - def _run_locked(self, tensor_input, timeout: float) -> np.ndarray: - if tensor_input is None or getattr(tensor_input, "ndim", 0) < 3: - raise ValueError( - f"Unexpected frame shape: {getattr(tensor_input, 'shape', tensor_input)}" - ) - - frame = np.asarray(tensor_input) - if frame.ndim == 4: - # Frigate feeds a batched NHWC frame (1, H, W, C); drop the batch dim - if frame.shape[0] != 1: - raise ValueError(f"Unexpected batch dimension in frame: {frame.shape}") - frame = frame[0] - # Real copy: the shared-memory camera buffer is unmapped by Frigate as - # soon as detect_raw returns, and our queue can hold the frame past that - # point (timeout path, backed-up queue). NP views would let the SDK read - # a torn/unmapped segment. - frame = np.array(frame, dtype=np.uint8, order="C", copy=True) - - if not self._thread.is_alive(): - # The pump is gone (stream wedged): never block on the queue put, - # the SDK side is not draining it and the queue would fill up, - # permanently wedging this detector. - raise RuntimeError("Axelera inference thread stopped; restart required.") - - # Every submitted frame gets a unique id; its token is decoded from the - # result image by the pump, so we can wait for OUR frame's outcome - # exclusively (content-based pairing, no ordering assumptions). + 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 - frame_id = self._next_id - with self._cond: - self._waiting.add(frame_id) + token = self._next_id + _stamp_token(frame, token) + self._token_to_conn[token] = connection_id try: - self._frame_q.put((frame_id, frame), timeout=2.0) + self._frame_q.put_nowait((token, frame)) except queue.Full: - with self._cond: - self._waiting.discard(frame_id) + self._token_to_conn.pop(token, None) raise RuntimeError( "Axelera inference queue is full; restart required." ) from None - deadline = time.monotonic() + timeout - with self._cond: - while frame_id not in self._results: - remaining = deadline - time.monotonic() - if remaining <= 0: - break - self._cond.wait(remaining) - if frame_id in self._results: - result = self._results.pop(frame_id) - self._waiting.discard(frame_id) - return result - # timeout: drop the id so a late result is not delivered/stored - with self._cond: - self._waiting.discard(frame_id) - if self._thread.is_alive(): - raise RuntimeError( - "Axelera inference stream returned no result for the submitted " - f"frame within {timeout:.1f}s (stream stall); restart required." - ) - raise RuntimeError("Axelera inference thread stopped; restart required.") + def collect(self): + """Wait for the next completed frame; return (connection_id, rows). - def stop(self): - # cleanup must not raise + ``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. + """ + while True: + try: + frame_result = next(self._stream) + except StopIteration: + return None, None + except Exception: + logger.exception( + "axelera: inference stream failed; detector requires restart" + ) + 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. + 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 + + def stop(self) -> None: + """Stop the stream and release the Metis device. Never raises.""" try: - self._frame_q.put_nowait((None, None)) + self._frame_q.put_nowait((0, None)) except Exception: # noqa: S110, BLE001 pass try: @@ -493,14 +420,49 @@ class AxeleraDetector(DetectionApi): self.height, ) - def detect_raw(self, tensor_input) -> np.ndarray: - """Run the detector on one RGB frame and return (20, 6) float32 rows. + # --- async API (AsyncDetectorRunner, same contract as the memryx plugin) --- - Each row is (class_id, score, ymin, xmin, ymax, xmax) in NORMALIZED - coordinates [0..1] relative to the model input size. Unused rows are - zero-filled. Detections are strictly paired to THIS frame's inference. + def send_input(self, connection_id, tensor_input: np.ndarray) -> None: + """Preprocess (if needed) and send a frame to the Metis stream.""" + if tensor_input is None: + raise ValueError("[send_input] No image data provided for inference") + + frame = np.asarray(tensor_input) + if frame.ndim == 4: + # Frigate feeds a batched NHWC frame (1, H, W, C); drop the batch dim + 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. + frame = np.ascontiguousarray(frame, dtype=np.uint8) + self._inference.submit(connection_id, frame) + + def receive_output(self): + """Return (connection_id, rows) for the next completed inference. + + On a fatal stream error the detector process exits so Frigate restarts + it (a silent (None, None) would stall detection forever). """ - return self._inference.run(tensor_input) + try: + return self._inference.collect() + except Exception: + logger.exception("axelera: fatal stream error; exiting detector process") + try: + self._inference.stop() + except Exception: # noqa: S110, BLE001 + pass + os._exit(1) + + def shutdown(self) -> None: + """Gracefully shut down the Metis stream (called at detector stop).""" + self._inference.stop() + + def detect_raw(self, tensor_input: np.ndarray): + """Synchronous path removed: axelera uses the async detector contract.""" + return 0 # type: ignore[override] def close(self): """Stop the inference stream and release the Metis device.""" diff --git a/frigate/object_detection/base.py b/frigate/object_detection/base.py index bc7910e4d0..eca9b72d73 100644 --- a/frigate/object_detection/base.py +++ b/frigate/object_detection/base.py @@ -357,8 +357,8 @@ class ObjectDetectProcess: if (self.detect_process is not None) and self.detect_process.is_alive(): self.stop() - # Async path for MemryX - if self.detector_config.type == "memryx": + # Async path for MemryX and Axelera (stream-based send/receive threads) + if self.detector_config.type in ("memryx", "axelera"): self.detect_process = AsyncDetectorRunner( f"frigate.detector:{self.name}", self.detection_queue,