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

fix(axelera): pair results by recorded FIFO order instead of pixel tokens

Browse Source 
Replace the content-based pixel-token pairing with the reviewer's scheme:
the feeder registers every yielded frame on a pending deque (None for
primer/keep-alive synthetics, the connection id for real frames) and
collect() pops exactly one entry per result, so synthetics are dropped and
real results always reach the connection that owns them - without reading
the returned image at all.

Validated on hardware first: the SDK InferenceStream returns results
strictly in input order, one per yielded frame (200/200 frames in order,
no drops or duplicates, synthetics interleaved included), and the per-frame
image read (img.asarray + band decode) measured 0.58 ms against a 14.36 ms
back-to-back round-trip, i.e. not the driver of the System Metrics latency.

A desync guard raises (detector restarts) if the SDK ever emits a result
with no registered frame, instead of silently misattributing everything
after it (the P1 failure mode). The pixel-token path and its 200/200 FIFO
probe stay in git history for rollback.
This commit is contained in:
patrykorwat 2026-08-22 08:56:39 +00:00
parent bc90ab343b
commit 3959640861
1 changed files with 81 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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