feat(axelera): use Frigate's AsyncDetectorRunner; drop lock and keep-alives

The synchronous DetectorRunner path serialized run() and needed
20 Hz keep-alive zero frames just to keep the Voyager SDK pipeline
rotating: the SDK pipeline is clocked by the input arrival interval
(measured: 1 s gaps stretch a real frame's latency to ~4 s on the
depth-4 pipeline), and one-frame-in-flight caps multi-camera
throughput. AsyncDetectorRunner (as memryx) sends from the detect
thread and receives on the result thread, and closes the input SHM
buffer on the result side, so the frame copy in the plugin goes away.

- plugin implements send_input()/receive_output()/shutdown() and a
  detect_raw() stub (memryx precedent); content-based pixel-token
  pairing is preserved (primer results carry no token and are dropped).
- object_detection/base.py: route axelera to AsyncDetectorRunner.

With real frames clocking the pipeline, no keep-alives are needed;
the throughput ceiling imposed by one frame in flight is gone.

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

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,