From f50a2dc1edd1890743a4f006dbbc57cc7e1e5247 Mon Sep 17 00:00:00 2001 From: patrykorwat Date: Sat, 5 Sep 2026 05:44:29 +0000 Subject: [PATCH] axelera: publish the run duty cycle through a shared Value Replace the tmp-file duty-cycle handoff (and the FRIGATE_TMP env var nothing else reads) with a third multiprocessing.Value alongside avg_inference_speed and detection_start: ObjectDetectProcess creates it, AsyncDetectorRunner passes it to the detector, and the plugin publishes busy/wall of the blocking ModelInstance.run over a 2s window, mirroring how memryx receives its shared state. No filesystem path, no staleness window, no atomic rename, and stats/hardware.py no longer imports the plugin module to find the file. The producer also ticks the window while the queue is empty, so the duty cycle falls to 0 when the card idles instead of sticking at its last value. Name it what it is: host-side occupancy of the run call, including DMA and driver time, not an AIPU core counter (the runtime exposes no device-side one). The NPU entry now reads temperature from the board controller plus the detector-reported run duty cycle folded in where the detector processes are in reach. --- frigate/detectors/plugins/axelera.py | 52 +++++++++++++++------------- frigate/object_detection/base.py | 15 +++++++- frigate/stats/hardware.py | 29 ++-------------- frigate/stats/util.py | 32 +++++++++++++++++ 4 files changed, 76 insertions(+), 52 deletions(-) diff --git a/frigate/detectors/plugins/axelera.py b/frigate/detectors/plugins/axelera.py index 6ae143790d..bf0305077a 100644 --- a/frigate/detectors/plugins/axelera.py +++ b/frigate/detectors/plugins/axelera.py @@ -36,7 +36,6 @@ import logging import os import queue import re -import tempfile import threading import time import zipfile @@ -56,12 +55,9 @@ logger = logging.getLogger(__name__) DETECTOR_KEY = "axelera" -# windowed duty cycle of the AIPU, written by the detector process and read -# by the main process's hardware poller (same tmpfs cache dir as the rest of -# the runtime's scratch state); " " text -AXELERA_USAGE_PATH = os.path.join( - os.environ.get("FRIGATE_TMP", tempfile.gettempdir()), "frigate-axelera-usage" -) +# window length over which the producer thread averages ModelInstance.run +# time before publishing the run duty cycle to the shared stats Value +USAGE_WINDOW_SECONDS = 2.0 # The Axelera runtime SDK is installed at first start rather than shipped in # the image. The wheels come from Axelera's public Artifactory PyPI index @@ -417,7 +413,10 @@ class _AxeleraRuntimeInference: self._producer_thread: threading.Thread | None = None self._consumer_thread: threading.Thread | None = None self._usage_busy = 0.0 - self._usage_window_start: float | None = None + self._usage_window_start = time.monotonic() + # multiprocessing.Value("d") handed in via set_duty_value; None means + # nobody reads the duty cycle (tests, standalone use) + self._duty_value = None if runtime_factory is None: try: @@ -653,6 +652,8 @@ class _AxeleraRuntimeInference: try: connection_id, frame = self._frame_q.get(timeout=0.25) except queue.Empty: + # keep the duty cycle window turning while the card idles + self._publish_duty(time.monotonic()) continue try: in_buf, out_bufs = self._pool.get(timeout=1.0) @@ -672,28 +673,25 @@ class _AxeleraRuntimeInference: self._push_result(connection_id, np.zeros((MAX_ROWS, 6), np.float32)) def _note_run(self, t0: float, t1: float) -> None: - """Accumulate card busy time and publish a windowed duty cycle. + """Accumulate run time and publish a windowed duty cycle. - ModelInstance.run blocks until the AIPU completes, so busy/wall over a - window is the card's true utilization; the runtime exposes no usage - counter to read instead. The main stats process reads the file. + ModelInstance.run blocks until the AIPU completes, so busy/wall over + a window is the detector thread's run duty cycle: host-side occupancy + of the blocking call (including DMA and driver time), which the + runtime exposes no device-side counter to read instead. The value + rides a multiprocessing.Value to the main process's stats poller. """ - now = time.monotonic() self._usage_busy += t1 - t0 - if self._usage_window_start is None: - self._usage_window_start = t0 - if now - self._usage_window_start < 2.0: + self._publish_duty(time.monotonic()) + + def _publish_duty(self, now: float) -> None: + if now - self._usage_window_start < USAGE_WINDOW_SECONDS: return window = now - self._usage_window_start - if window > 0: - pct = min(100.0, round(100.0 * self._usage_busy / window, 1)) - try: - tmp = AXELERA_USAGE_PATH + ".tmp" - with open(tmp, "w") as f: - f.write(f"{pct} {time.time():.3f}") - os.replace(tmp, AXELERA_USAGE_PATH) - except OSError: - pass + if window > 0 and self._duty_value is not None: + self._duty_value.value = min( + 100.0, round(100.0 * self._usage_busy / window, 1) + ) self._usage_window_start = now self._usage_busy = 0.0 @@ -951,6 +949,10 @@ class AxeleraDetector(DetectionApi): """Return (connection_id, rows) for the next completed inference.""" return self._inference.collect() + def set_duty_value(self, duty_value) -> None: + """Hand the shared stats Value to the producer thread's duty meter.""" + self._inference._duty_value = duty_value + def shutdown(self) -> None: """Gracefully shut down the runtime and release the Metis device.""" self._inference.stop() diff --git a/frigate/object_detection/base.py b/frigate/object_detection/base.py index e27e25a031..db5ae09a0b 100644 --- a/frigate/object_detection/base.py +++ b/frigate/object_detection/base.py @@ -45,6 +45,7 @@ class BaseLocalDetector(ObjectDetector): detector_config: BaseDetectorConfig | None = None, labels: str | None = None, stop_event: MpEvent | None = None, + duty_value: Any = None, ) -> None: self.fps = EventsPerSecond() if labels is None: @@ -65,6 +66,10 @@ class BaseLocalDetector(ObjectDetector): if hasattr(self.detect_api, "set_stop_event") and stop_event: self.detect_api.set_stop_event(stop_event) + # If the detector publishes a run duty cycle, hand it the shared Value + if hasattr(self.detect_api, "set_duty_value") and duty_value is not None: + self.detect_api.set_duty_value(duty_value) + def _transform_input(self, tensor_input: np.ndarray) -> np.ndarray: if self.input_transform: tensor_input = np.transpose(tensor_input, self.input_transform) @@ -196,6 +201,7 @@ class AsyncDetectorRunner(FrigateProcess): config: FrigateConfig, detector_config: BaseDetectorConfig, stop_event: MpEvent, + duty_value: Any = None, ) -> None: super().__init__(stop_event, PROCESS_PRIORITY_HIGH, name=name, daemon=True) self.detection_queue = detection_queue @@ -204,6 +210,7 @@ class AsyncDetectorRunner(FrigateProcess): self.start_time = start_time self.config = config self.detector_config = detector_config + self.duty_value = duty_value self.outputs: dict[str, Any] = {} self._frame_manager: SharedMemoryFrameManager | None = None self._publisher: ObjectDetectorPublisher | None = None @@ -282,7 +289,9 @@ class AsyncDetectorRunner(FrigateProcess): self._frame_manager = SharedMemoryFrameManager() self._publisher = ObjectDetectorPublisher() self._detector = AsyncLocalObjectDetector( - detector_config=self.detector_config, stop_event=self.stop_event + detector_config=self.detector_config, + stop_event=self.stop_event, + duty_value=self.duty_value, ) for name in self.cameras: @@ -330,6 +339,9 @@ class ObjectDetectProcess: self.detection_queue = detection_queue self.avg_inference_speed = Value("d", 0.01) self.detection_start = Value("d", 0.0) + # async detectors that time their accelerator runs publish a windowed + # duty cycle here (see AsyncDetectorRunner); read by the stats poller + self.run_duty_cycle = Value("d", -1.0) self.detect_process: FrigateProcess | None = None self.config = config self.detector_config = detector_config @@ -382,6 +394,7 @@ class ObjectDetectProcess: self.config, self.detector_config, self.stop_event, + duty_value=self.run_duty_cycle, ) else: self.detect_process = DetectorRunner( diff --git a/frigate/stats/hardware.py b/frigate/stats/hardware.py index 343c3f1789..4b9a2964be 100644 --- a/frigate/stats/hardware.py +++ b/frigate/stats/hardware.py @@ -196,39 +196,16 @@ def _poll_axengine(config: FrigateConfig) -> HardwarePollResult: return HardwarePollResult(npu={"axengine": npu_usage}) -def _axelera_board_utilization() -> float | None: - """Duty-cycle of the AIPU published by the detector process. - - The detector times each blocking ModelInstance.run and writes - busy/wall for the last window to a small file; a stale file means the - detector has not run inference recently and no number is reported. - """ - try: - from frigate.detectors.plugins.axelera import AXELERA_USAGE_PATH - - with open(AXELERA_USAGE_PATH) as f: - pct, stamp = f.read().split() - if time.time() - float(stamp) > 10: - return None - return max(0.0, float(pct)) - except (OSError, ValueError, ImportError): - return None - - def _poll_axelera(config: FrigateConfig) -> HardwarePollResult: - # the runtime exposes no device-side usage counter; the board - # controller temperature and the detector process's run-time duty cycle - # are the two honest readouts + # the runtime exposes no device-side usage counter; the board controller + # temperature is the only honest device readout. The run duty cycle is a + # detector-process stat, added to npu_usages in stats_snapshot board_temp = get_axelera_board_temp() if board_temp is None: return HardwarePollResult(ok=False) entry: dict[str, Any] = {"mem": "-%", "temp": board_temp} - util = _axelera_board_utilization() - - if util is not None: - entry["npu"] = util return HardwarePollResult(npu={"axelera": entry}) diff --git a/frigate/stats/util.py b/frigate/stats/util.py index 575903e8f7..df4eadfd38 100644 --- a/frigate/stats/util.py +++ b/frigate/stats/util.py @@ -64,6 +64,30 @@ def stats_init( return stats_tracking +def _axelera_run_duty_cycle( + detectors: dict[str, ObjectDetectProcess], +) -> float | None: + """Highest run duty cycle published by the configured axelera detectors. + + The detector process times each blocking ModelInstance.run and publishes + busy/wall for the last window through a shared Value (a negative value + means the detector has not produced a window yet and nothing is shown). + This is host-side occupancy of the run call, including DMA and driver + time, not an AIPU core counter: the runtime exposes no device-side one. + """ + cycles = [ + detector.run_duty_cycle.value + for detector in detectors.values() + if detector.detector_config.type == "axelera" + and detector.run_duty_cycle.value >= 0.0 + ] + + if not cycles: + return None + + return round(max(cycles), 1) + + def get_detector_stats( stats_tracking: StatsTrackingTypes, ) -> dict[str, dict[str, Any]]: @@ -247,6 +271,14 @@ def stats_snapshot( hardware_stats.update_stats(stats) + # the axelera run duty cycle is a detector-process stat (a shared Value, + # not a device counter); fold it into the polled NPU entry if one exists + axelera_npu = stats.get("npu_usages", {}).get("axelera") + if axelera_npu is not None: + duty = _axelera_run_duty_cycle(stats_tracking["detectors"]) + if duty is not None: + axelera_npu["npu"] = duty + if config.telemetry.stats.network_bandwidth: bandwidth_stats = get_bandwidth_stats(config)