nik/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-18 07:29:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(axelera): pipeline the CPU postprocess behind the NPU run

Browse Source 
Split the single worker into a producer/consumer pipeline: the producer
letterboxes + quantizes + runs the model into a pooled, ping-ponged buffer
set, the consumer dequantises + postprocesses the previous frame's outputs
while the NPU is already busy on the next one (all run() calls stay on the
one producer thread, so in-flight frames never execute the model in
parallel). Measured on the Metis: ~24.9 ms/frame (~40 fps) vs ~28.9 ms
serial, giving real headroom over the ~35 fps offered by 7 cameras.
This commit is contained in:
patrykorwat 2026-08-23 21:07:11 +00:00
parent 44cc8bf4fc
commit 86637985ea
1 changed files with 106 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
frigate/detectors/plugins/axelera.py
View File

@ -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]