diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index e4480645b8..0ba5ca41da 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -80,6 +80,11 @@ HEAD_DEQUANT = ( CONF_THRESHOLD = 0.25 NMS_IOU_THRESHOLD = 0.45 QUEUE_MAXSIZE = 32 +# ping-pong buffer sets for the producer/consumer pipeline: the producer +# writes one set while the decoder reads a previous one, so the NPU stays +# saturated. 3 sets bound the in-flight frames (pool size must equal the +# decode queue bound). +BUFFER_SETS = 3 # ----------------- Utility Functions ----------------- # @@ -212,16 +217,20 @@ def _nms_xyxy(boxes, scores, iou_threshold: float) -> list: class _AxeleraRuntimeInference: - """Runs a compiled model via ``axelera.runtime`` in a dedicated worker thread. + """Runs a compiled model via ``axelera.runtime`` on a two-thread pipeline. Frigate's AsyncDetectorRunner calls ``submit(connection_id, frame)`` from - the detect thread and ``collect()`` from the result thread. A single worker - thread pops frames strictly in order, letterboxes + quantizes, runs the - model (``ModelInstance.run``, synchronous and flat: ~21 ms measured), feeds - the six int8 heads through the SDK's baked ``postprocess_graph.onnx`` and - pushes ``(connection_id, rows)``. Because one thread processes frames FIFO, - results are paired to requests by construction - there is no pipeline - reorder to guard against (contrast: the create_inference_stream variant). + the detect thread and ``collect()`` from the result thread. Two engine + threads process frames strictly in order: a producer letterboxes + + quantizes + runs the model (``ModelInstance.run``, synchronous and flat: + ~21 ms measured), and a consumer feeds the six int8 heads through the SDK's + baked ``postprocess_graph.onnx`` while the NPU is already busy on the next + frame (measured throughput ~25 ms/frame, ~40 fps, vs ~29 ms serial). All + ``run`` calls stay on the single producer thread, so in-flight frames never + execute the model in parallel (a known card-wedge condition). Because both + threads consume their queues FIFO, results are paired to requests by + construction - there is no pipeline reorder to guard against (contrast: + the create_inference_stream variant). """ def __init__( @@ -241,7 +250,8 @@ class _AxeleraRuntimeInference: self._frame_q: queue.Queue = queue.Queue(maxsize=QUEUE_MAXSIZE) self._result_q: queue.Queue = queue.Queue(maxsize=QUEUE_MAXSIZE) self._stop = threading.Event() - self._thread: threading.Thread | None = None + self._producer_thread: threading.Thread | None = None + self._consumer_thread: threading.Thread | None = None if runtime_factory is None: try: @@ -281,14 +291,22 @@ class _AxeleraRuntimeInference: f"{len(HEAD_REAL_CHANNELS)} inputs, got {len(self._pp_input_names)}" ) - # pre-allocated input/output buffers (reused; do not allocate per frame) - self._in_buf = np.full( - (1, PADDED_H, PADDED_W, 4), INPUT_Q_ZERO, np.int8 - ) - self._out_bufs = [ - np.zeros((1, MODEL_INPUT_H // s, MODEL_INPUT_W // s, 128), np.int8) - for s in (8, 8, 16, 16, 32, 32) - ] + # pooled letterbox-input + raw-int8-output sets, ping-ponged between + # the producer (runs the model) and the decoder (postprocesses the + # previous frame's outputs while the NPU is busy on the next one). + self._pool: queue.Queue = queue.Queue(maxsize=BUFFER_SETS) + for _ in range(BUFFER_SETS): + in_buf = np.full( + (1, PADDED_H, PADDED_W, 4), INPUT_Q_ZERO, np.int8 + ) + out_bufs = [ + np.zeros( + (1, MODEL_INPUT_H // s, MODEL_INPUT_W // s, 128), np.int8 + ) + for s in (8, 8, 16, 16, 32, 32) + ] + self._pool.put((in_buf, out_bufs)) + self._decode_q: queue.Queue = queue.Queue(maxsize=BUFFER_SETS) # letterbox geometry from the frame coords (width x height) into 640x640 self._scale = min( MODEL_INPUT_W / self._width, MODEL_INPUT_H / self._height @@ -314,11 +332,19 @@ class _AxeleraRuntimeInference: raise RuntimeError( "Axelera inference queue is full; restart required." ) from None - if self._thread is None or not self._thread.is_alive(): - self._thread = threading.Thread( - target=self._worker, name="axelera-runtime", daemon=True + self._ensure_threads() + + def _ensure_threads(self) -> None: + if self._consumer_thread is None or not self._consumer_thread.is_alive(): + self._consumer_thread = threading.Thread( + target=self._consumer, name="axelera-decode", daemon=True ) - self._thread.start() + self._consumer_thread.start() + if self._producer_thread is None or not self._producer_thread.is_alive(): + self._producer_thread = threading.Thread( + target=self._producer, name="axelera-runtime", daemon=True + ) + self._producer_thread.start() def collect(self): """Return (connection_id, rows) for the next completed inference. @@ -333,8 +359,11 @@ class _AxeleraRuntimeInference: return None, None def stop(self) -> None: - """Stop the worker, release the Metis device. Never raises.""" + """Stop the workers, release the Metis device. Never raises.""" self._stop.set() + for t in (self._producer_thread, self._consumer_thread): + if t is not None and t.is_alive(): + t.join(timeout=2) try: self._instance.free() except Exception: # noqa: BLE001, S110 @@ -343,27 +372,69 @@ class _AxeleraRuntimeInference: # -- internals --------------------------------------------------------- - def _worker(self) -> None: + def _push_result(self, connection_id: str, rows: np.ndarray) -> None: + try: + self._result_q.put_nowait((connection_id, rows)) + except queue.Full: + logger.error("axelera: result queue full; dropping %s", connection_id) + + def _producer(self) -> None: + """Letterbox + quantize + run(); hand raw outputs to the decoder. + + All ``ModelInstance.run`` calls happen on THIS single thread, so the + three in-flight buffer sets never execute the model in parallel (a + known card-wedge condition) while still overlap with the decoder's + CPU postprocessing. + """ while not self._stop.is_set(): try: connection_id, frame = self._frame_q.get(timeout=0.25) except queue.Empty: continue try: - rows = self._process(frame) + in_buf, out_bufs = self._pool.get(timeout=1.0) + except queue.Empty: + logger.error( + "axelera: no free buffers; dropping %s", connection_id + ) + self._push_result( + connection_id, np.zeros((MAX_ROWS, 6), np.float32) + ) + continue + try: + self._preprocess(frame, in_buf) + self._instance.run([in_buf], out_bufs) + self._decode_q.put_nowait((connection_id, in_buf, out_bufs)) except Exception: logger.exception( "axelera: inference failed for %s", connection_id ) - rows = np.zeros((MAX_ROWS, 6), np.float32) - try: - self._result_q.put_nowait((connection_id, rows)) - except queue.Full: - logger.error("axelera: result queue full; dropping frame") + self._pool.put((in_buf, out_bufs)) + self._push_result( + connection_id, np.zeros((MAX_ROWS, 6), np.float32) + ) - def _process(self, frame: np.ndarray) -> np.ndarray: - """Letterbox + quantize, run, decode, threshold + NMS -> (20,6) rows.""" - # letterbox the source (H, W, 3) into 640x640 and quantize to int8 + def _consumer(self) -> None: + """Decode the raw int8 heads (dequant + onnx + NMS) to (20,6) rows.""" + while not self._stop.is_set(): + try: + connection_id, in_buf, out_bufs = self._decode_q.get( + timeout=0.25 + ) + except queue.Empty: + continue + try: + rows = self._decode(out_bufs) + except Exception: + logger.exception( + "axelera: decode failed for %s", connection_id + ) + rows = np.zeros((MAX_ROWS, 6), np.float32) + self._pool.put((in_buf, out_bufs)) + self._push_result(connection_id, rows) + + def _preprocess(self, frame: np.ndarray, buf: np.ndarray) -> None: + """Letterbox the source (H, W, 3) into the padded input buffer.""" resized = cv2.resize( frame, (self._new_w, self._new_h), @@ -372,7 +443,6 @@ class _AxeleraRuntimeInference: # uint8 - 128 wraps mod 256, then cast to int8 == px - 128 (exactly), # in one pass over the resized frame (no int16/clip intermediates) q = (resized - np.uint8(128)).astype(np.int8) - buf = self._in_buf buf.fill(INPUT_Q_ZERO) y0, y1 = self._pad_t, self._pad_t + self._new_h x0, x1 = self._pad_l, self._pad_l + self._new_w @@ -381,12 +451,12 @@ class _AxeleraRuntimeInference: buf[0, y0 + 1 : y1 + 1, x0 + 1 : x1 + 1, 0:3] = q[..., ::-1] buf[0, y0 + 1 : y1 + 1, x0 + 1 : x1 + 1, 3] = 127 - self._instance.run([buf], self._out_bufs) - + def _decode(self, out_bufs) -> np.ndarray: + """Dequantise heads + postprocess graph + conf threshold + NMS -> rows.""" # dequantize + crop to the real channel count + transpose NHWC -> NCHW feeds = {} for i, (name, obuf) in enumerate( - zip(self._pp_input_names, self._out_bufs) + zip(self._pp_input_names, out_bufs) ): scale, zp = self._head_dequant[i] real = HEAD_REAL_CHANNELS[i]