From 34f13ba59ed41ac1261b10e33dc6a265955c7068 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 02:52:11 +0300 Subject: [PATCH] feat(perception): add bounded triton runtime diagnostics --- .../compare_ddrnet_triton.py | 142 ++++++++ .../ddrnet-config.pbtxt | 6 + .../ddrnet-cuda-graphs-config.pbtxt | 7 + .../ddrnet-cuda-graphs-spin-config.pbtxt | 7 + .../export_ddrnet_onnx.py | 99 ++++++ .../pilot_ddrnet_numpy.py | 68 ++++ .../pilot_ddrnet_preprocess.py | 64 ++++ .../pilot_ddrnet_triton.py | 86 +++++ .../pilot_telemetry.py | 88 +++++ .../pilot_triton_stats.py | 45 +++ .../profile_ddrnet_tensor.py | 58 ++++ .../run_joint_pilot.py | 313 +++++++++++++----- tests/test_perception_joint_pilot.py | 116 +++++++ 13 files changed, 1023 insertions(+), 76 deletions(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/compare_ddrnet_triton.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/ddrnet-config.pbtxt create mode 100644 experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-config.pbtxt create mode 100644 experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-spin-config.pbtxt create mode 100644 experiments/perception/worker/streaming_profile_stage1/export_ddrnet_onnx.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_numpy.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_preprocess.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_triton.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_triton_stats.py create mode 100644 experiments/perception/worker/streaming_profile_stage1/profile_ddrnet_tensor.py diff --git a/experiments/perception/worker/streaming_profile_stage1/compare_ddrnet_triton.py b/experiments/perception/worker/streaming_profile_stage1/compare_ddrnet_triton.py new file mode 100644 index 0000000..fd144bb --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/compare_ddrnet_triton.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Compare the pinned PyTorch DDRNet mask with its FP32 TensorRT export.""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import json +import os +import signal +import subprocess +import time +from pathlib import Path + +import cv2 +import numpy as np +from pilot_ddrnet_preprocess import load_preprocess, preprocess_bgr +from pilot_ddrnet_triton import TritonDdrnetMaskHttpInferenceBackend +from pilot_ipc import receive, send + + +def wait_ready(process): + deadline = time.monotonic() + 90 + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError("Triton exited before DDRNet readiness") + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1) + try: + connection.request("GET", "/v2/models/ddrnet_goose_fp32_mask/ready") + if connection.getresponse().status == 200: + return + except OSError: + pass + finally: + connection.close() + time.sleep(0.1) + raise TimeoutError("DDRNet Triton readiness exceeded budget") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--video", default="/source.mp4") + parser.add_argument("--frames", type=int, default=128) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + signal.alarm(240) + started = time.monotonic_ns() + args.output.parent.mkdir(parents=True, exist_ok=True) + log = args.output.with_suffix(".triton.log").open("wb") + triton = subprocess.Popen( + [ + "tritonserver", + "--model-repository=/models", + "--model-control-mode=explicit", + "--load-model=ddrnet_goose_fp32_mask", + "--allow-grpc=false", + "--allow-metrics=false", + "--http-address=127.0.0.1", + ], + stdout=log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + reference_log = args.output.with_suffix(".reference.log").open("wb") + reference = candidate = capture = None + try: + wait_ready(triton) + reference = subprocess.Popen( + [ + "/opt/conda/envs/goose/bin/python", + "-B", + "/probe/pilot_model.py", + "ddrnet", + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=reference_log, + env={**os.environ, "PYTHONPATH": "/probe"}, + ) + ready, _ = receive(reference.stdout) + if ready.get("ready") != "ddrnet": + raise RuntimeError("PyTorch DDRNet reference did not become ready") + candidate = TritonDdrnetMaskHttpInferenceBackend() + preprocess = load_preprocess() + zero = np.zeros((600, 800, 3), np.uint8) + for _ in range(8): + candidate.infer(preprocess_bgr(preprocess, zero)) + capture = cv2.VideoCapture(args.video) + reference_hash = hashlib.sha256() + candidate_hash = hashlib.sha256() + rows = [] + for sequence in range(args.frames): + ok, bgr = capture.read() + if not ok or bgr is None or bgr.shape != (600, 800, 3): + raise RuntimeError("DDRNet parity source ended or changed shape") + send(reference.stdin, {"op": "infer"}, bgr.tobytes()) + reference_result, raw = receive(reference.stdout) + reference_mask = np.frombuffer(raw, np.uint8).reshape(512, 512) + candidate_started = time.monotonic_ns() + candidate_mask = candidate.infer(preprocess_bgr(preprocess, bgr)) + candidate_ms = (time.monotonic_ns() - candidate_started) / 1e6 + different = int(np.count_nonzero(reference_mask != candidate_mask)) + reference_hash.update(reference_mask.tobytes()) + candidate_hash.update(candidate_mask.tobytes()) + rows.append( + { + "sequence": sequence, + "different_pixels": different, + "reference_component_ms": reference_result["component_ms"], + "candidate_preprocess_infer_ms": candidate_ms, + } + ) + report = { + "schema_version": "missioncore.ddrnet-tensorrt-parity/v1", + "frames": args.frames, + "exact_frames": sum(row["different_pixels"] == 0 for row in rows), + "different_pixels": sum(row["different_pixels"] for row in rows), + "reference_mask_sequence_sha256": reference_hash.hexdigest(), + "candidate_mask_sequence_sha256": candidate_hash.hexdigest(), + "elapsed_ms": (time.monotonic_ns() - started) / 1e6, + "rows": rows, + } + args.output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") + print(json.dumps({key: report[key] for key in report if key != "rows"}), flush=True) + finally: + if capture is not None: + capture.release() + if candidate is not None: + candidate.close() + if reference is not None and reference.poll() is None: + send(reference.stdin, {"op": "stop"}) + reference.wait(timeout=5) + if triton.poll() is None: + triton.terminate() + triton.wait(timeout=5) + log.close() + reference_log.close() + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/worker/streaming_profile_stage1/ddrnet-config.pbtxt b/experiments/perception/worker/streaming_profile_stage1/ddrnet-config.pbtxt new file mode 100644 index 0000000..4a31960 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/ddrnet-config.pbtxt @@ -0,0 +1,6 @@ +name: "ddrnet_goose_fp32_mask" +platform: "tensorrt_plan" +max_batch_size: 0 +input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }] +output [{ name: "mask" data_type: TYPE_UINT8 dims: [1,512,512] }] +instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }] diff --git a/experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-config.pbtxt b/experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-config.pbtxt new file mode 100644 index 0000000..01fc732 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-config.pbtxt @@ -0,0 +1,7 @@ +name: "ddrnet_goose_fp32_mask" +platform: "tensorrt_plan" +max_batch_size: 0 +input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }] +output [{ name: "mask" data_type: TYPE_UINT8 dims: [1,512,512] }] +instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }] +optimization { cuda { graphs: true busy_wait_events: false } } diff --git a/experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-spin-config.pbtxt b/experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-spin-config.pbtxt new file mode 100644 index 0000000..08b2116 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/ddrnet-cuda-graphs-spin-config.pbtxt @@ -0,0 +1,7 @@ +name: "ddrnet_goose_fp32_mask" +platform: "tensorrt_plan" +max_batch_size: 0 +input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }] +output [{ name: "mask" data_type: TYPE_UINT8 dims: [1,512,512] }] +instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }] +optimization { cuda { graphs: true busy_wait_events: true } } diff --git a/experiments/perception/worker/streaming_profile_stage1/export_ddrnet_onnx.py b/experiments/perception/worker/streaming_profile_stage1/export_ddrnet_onnx.py new file mode 100644 index 0000000..277674e --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/export_ddrnet_onnx.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Export the pinned FP32 DDRNet logits graph for a bounded TensorRT probe.""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.util +import json +import signal +import time +from pathlib import Path + +CHECKPOINT_SHA256 = "b99c2838051bcd7b092fd3970aa62a77d5c0bbb809c9b9afb2ff4b0ebdaa4ee6" +RUNNER_SHA256 = "b18ad60f277eea69a240a28f290611b94627fb9707faf1bb3e6e22102dad67c1" + + +def digest(path): + value = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + value.update(chunk) + return value.hexdigest() + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument( + "--runner", + type=Path, + default=Path("/probe/run_goose_vegetation_benchmark.py"), + ) + parser.add_argument("--checkpoint", type=Path, default=Path("/checkpoint.pth")) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--output-kind", choices=("logits", "mask"), default="logits") + args = parser.parse_args() + signal.alarm(180) + if digest(args.runner) != RUNNER_SHA256 or digest(args.checkpoint) != CHECKPOINT_SHA256: + raise RuntimeError("DDRNet export input identity changed") + spec = importlib.util.spec_from_file_location("pinned_goose_export_runner", args.runner) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + import onnx + import torch + + class LogitsOnly(torch.nn.Module): + def __init__(self, model): + super().__init__() + self.model = model + + def forward(self, value): + logits = module.logits_from_output(self.model(value)) + if args.output_kind == "logits": + return logits + return torch.argmax(torch.sigmoid(logits), dim=1).to(torch.uint8) + + started = time.monotonic_ns() + model, model_name, failures = module.load_model("ddrnet", args.checkpoint) + wrapper = LogitsOnly(model).eval() + example = torch.zeros((1, 3, 512, 512), dtype=torch.float32, device="cuda") + with torch.inference_mode(): + output = wrapper(example) + expected = ( + ((1, 64, 512, 512), torch.float32, "logits") + if args.output_kind == "logits" + else ((1, 512, 512), torch.uint8, "mask") + ) + if tuple(output.shape) != expected[0] or output.dtype != expected[1]: + raise RuntimeError("DDRNet export output contract changed") + args.output.parent.mkdir(parents=True, exist_ok=True) + torch.onnx.export( + wrapper, + example, + str(args.output), + input_names=["input"], + output_names=[expected[2]], + opset_version=17, + do_constant_folding=True, + ) + graph = onnx.load(str(args.output)) + onnx.checker.check_model(graph) + result = { + "schema_version": "missioncore.ddrnet-onnx-export/v1", + "model": model_name, + "precision": "fp32", + "input": {"name": "input", "shape": [1, 3, 512, 512]}, + "output": {"name": expected[2], "shape": list(expected[0])}, + "checkpoint_sha256": CHECKPOINT_SHA256, + "runner_sha256": RUNNER_SHA256, + "onnx_sha256": digest(args.output), + "onnx_bytes": args.output.stat().st_size, + "elapsed_ms": (time.monotonic_ns() - started) / 1e6, + "load_failures": failures, + } + print("STAGE1_RESULT=" + json.dumps(result, sort_keys=True), flush=True) + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_numpy.py b/experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_numpy.py new file mode 100644 index 0000000..13a041a --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_ddrnet_numpy.py @@ -0,0 +1,68 @@ +"""Fixed KB4 DDRNet preprocessing without a second Python runtime or IPC.""" + +from __future__ import annotations + +import numpy as np + +# Pinned runner: center crop 800x600 -> 600x600, Pillow NEAREST -> 512x512. +# Integer pixel-center coordinates avoid a float rounding convention change. +_NEAREST = ((2 * np.arange(512) + 1) * 600) // (2 * 512) + + +def preprocess_bgr(bgr): + if bgr.shape != (600, 800, 3) or bgr.dtype != np.uint8: + raise ValueError("DDRNet native image contract changed") + rgb = bgr[_NEAREST[:, None], (100 + _NEAREST)[None, :], ::-1] + normalized = np.asarray(rgb, dtype=np.float32) / 255.0 + return np.ascontiguousarray(normalized.transpose(2, 0, 1)[None], dtype=" MAX_RESPONSE_BYTES: + self.close() + raise RuntimeError("DDRNet Triton response exceeds byte budget") + received = time.monotonic_ns() + self.last_timing_ms = { + "http_encode_ms": (encoded - started) / 1e6, + "http_send_ms": (sent - encoded) / 1e6, + "http_wait_headers_ms": (responded - sent) / 1e6, + "http_read_body_ms": (received - responded) / 1e6, + } + if response.status != 200: + raise RuntimeError(f"DDRNet Triton inference failed: HTTP {response.status}") + try: + header_length = int(response.getheader("Inference-Header-Content-Length") or "") + if not 0 < header_length <= 65536: + raise ValueError("invalid header length") + outputs = json.loads(payload[:header_length])["outputs"] + if not isinstance(outputs, list) or len(outputs) != 1: + raise ValueError("invalid output count") + output = outputs[0] + valid = ( + output["name"] == "mask" + and output["datatype"] == "UINT8" + and output["shape"] == [1, 512, 512] + and output["parameters"]["binary_data_size"] == 512 * 512 + ) + except (KeyError, TypeError, ValueError, json.JSONDecodeError) as error: + raise RuntimeError("DDRNet Triton response contract is invalid") from error + if not valid or len(payload) != header_length + 512 * 512: + raise RuntimeError("DDRNet Triton mask contract changed") + return np.frombuffer(payload, np.uint8, offset=header_length).reshape(512, 512).copy() diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py b/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py new file mode 100644 index 0000000..c3acec5 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py @@ -0,0 +1,88 @@ +"""Low-overhead Worker GPU telemetry through the already loaded NVML library.""" + +from __future__ import annotations + +import ctypes + + +class _MemoryInfo(ctypes.Structure): + _fields_ = [ + ("total", ctypes.c_ulonglong), + ("free", ctypes.c_ulonglong), + ("used", ctypes.c_ulonglong), + ] + + +class _Utilization(ctypes.Structure): + _fields_ = [("gpu", ctypes.c_uint), ("memory", ctypes.c_uint)] + + +class NvmlSampler: + """Keep one NVML session instead of spawning ``nvidia-smi`` per sample.""" + + def __init__(self, library=None): + self.library = library or ctypes.CDLL("libnvidia-ml.so.1") + self.handle = ctypes.c_void_p() + self.closed = False + self._configure() + self._check(self.library.nvmlInit_v2(), "initialize") + try: + self._check( + self.library.nvmlDeviceGetHandleByIndex_v2(0, ctypes.byref(self.handle)), + "select device", + ) + except Exception: + self.library.nvmlShutdown() + self.closed = True + raise + + def _configure(self): + self.library.nvmlInit_v2.argtypes = [] + self.library.nvmlInit_v2.restype = ctypes.c_int + self.library.nvmlShutdown.argtypes = [] + self.library.nvmlShutdown.restype = ctypes.c_int + self.library.nvmlDeviceGetHandleByIndex_v2.argtypes = [ + ctypes.c_uint, + ctypes.POINTER(ctypes.c_void_p), + ] + self.library.nvmlDeviceGetHandleByIndex_v2.restype = ctypes.c_int + self.library.nvmlDeviceGetMemoryInfo.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(_MemoryInfo), + ] + self.library.nvmlDeviceGetMemoryInfo.restype = ctypes.c_int + self.library.nvmlDeviceGetUtilizationRates.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(_Utilization), + ] + self.library.nvmlDeviceGetUtilizationRates.restype = ctypes.c_int + + @staticmethod + def _check(code, operation): + if code != 0: + raise RuntimeError(f"NVML failed to {operation}: code {code}") + + def sample(self): + memory = _MemoryInfo() + utilization = _Utilization() + self._check( + self.library.nvmlDeviceGetMemoryInfo(self.handle, ctypes.byref(memory)), + "read memory", + ) + self._check( + self.library.nvmlDeviceGetUtilizationRates( + self.handle, + ctypes.byref(utilization), + ), + "read utilization", + ) + return { + "gpu_used_mib": int((memory.used + 1048575) // 1048576), + "gpu_utilization": int(utilization.gpu), + } + + def close(self): + if self.closed: + return + self.closed = True + self._check(self.library.nvmlShutdown(), "shutdown") diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_triton_stats.py b/experiments/perception/worker/streaming_profile_stage1/pilot_triton_stats.py new file mode 100644 index 0000000..2a934a7 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_triton_stats.py @@ -0,0 +1,45 @@ +"""Warmup-excluded server statistics; queried outside the measured stream.""" + +from __future__ import annotations + +import http.client +import json + + +def read_stats(model_ids): + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=5) + result = {} + try: + for model_id in model_ids: + connection.request("GET", f"/v2/models/{model_id}/versions/1/stats") + response = connection.getresponse() + payload = response.read(1024 * 1024 + 1) + if response.status != 200 or len(payload) > 1024 * 1024: + raise RuntimeError("Triton statistics unavailable or unbounded") + rows = json.loads(payload)["model_stats"] + if len(rows) != 1 or rows[0]["name"] != model_id or rows[0]["version"] != "1": + raise RuntimeError("Triton statistics model identity changed") + result[model_id] = rows[0]["inference_stats"] + finally: + connection.close() + return result + + +def stats_delta(before, after): + result = {} + for model_id, stages in before.items(): + result[model_id] = {} + for name in ( + "success", "fail", "queue", "compute_input", "compute_infer", "compute_output" + ): + first, last = stages[name], after[model_id][name] + count = int(last["count"]) - int(first["count"]) + duration_ns = int(last["ns"]) - int(first["ns"]) + if count < 0 or duration_ns < 0: + raise RuntimeError("Triton statistics counters went backwards") + result[model_id][name] = { + "count": count, + "total_ms": duration_ns / 1e6, + "mean_ms": duration_ns / count / 1e6 if count else None, + } + return result diff --git a/experiments/perception/worker/streaming_profile_stage1/profile_ddrnet_tensor.py b/experiments/perception/worker/streaming_profile_stage1/profile_ddrnet_tensor.py new file mode 100644 index 0000000..ae4fa47 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/profile_ddrnet_tensor.py @@ -0,0 +1,58 @@ +"""Bounded component diagnostic on one real frame; not a realtime proof.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import time +from pathlib import Path + +import cv2 +from pilot_ddrnet_numpy import preprocess_bgr + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + args.output.mkdir(parents=True, exist_ok=False) + started = time.monotonic_ns() + capture = cv2.VideoCapture("/source.mp4") + try: + ok, image = capture.read() + if not ok: + raise ValueError("real tensor source unavailable") + payload = preprocess_bgr(image).tobytes() + finally: + capture.release() + tensor_path = args.output / "frame-000000.f32" + tensor_path.write_bytes(payload) + command = [ + "/usr/bin/trtexec", "--loadEngine=/model.plan", + f"--loadInputs=input:{tensor_path}", "--warmUp=500", "--duration=0", + "--iterations=64", "--avgRuns=64", "--dumpProfile", + f"--exportProfile={args.output / 'layers.json'}", + ] + with (args.output / "trtexec.log").open("wb") as log: + result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, timeout=30) + report = { + "schema_version": "missioncore.ddrnet-real-tensor-component-profile/v1", + "source_sequence": 0, + "repeated_same_frame": True, + "source_paced": False, + "transport_included": False, + "input_sha256": hashlib.sha256(payload).hexdigest(), + "input_bytes": len(payload), + "command": command, + "returncode": result.returncode, + "elapsed_ms": (time.monotonic_ns() - started) / 1e6, + } + (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n") + print(json.dumps(report), flush=True) + return result.returncode + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py index e94e014..8c74d8a 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -29,6 +29,7 @@ from pilot_queue import Mailbox from pilot_scheduler import GpuStage from pilot_sensor_binding import bind_sensors, increment_identity, milliseconds from pilot_source import SensorArchive, camera_events, merged_events +from pilot_telemetry import NvmlSampler def distribution(values): @@ -53,6 +54,14 @@ def payload_size(bundle): ) +def layer_refresh_due(last_source_ns, current_source_ns, minimum_interval_ms): + if last_source_ns is None: + return True + if current_source_ns < last_source_ns: + raise ValueError("layer source clock moved backwards") + return current_source_ns - last_source_ns >= minimum_interval_ms * 1_000_000 + + def produce(args, decoder, mailbox, stop, report): archive = SensorArchive(Path(args.sensor_archive)) first_camera = next(camera_events(args.camera_index, args.frames)).time_ns @@ -172,52 +181,51 @@ def produce(args, decoder, mailbox, stop, report): def telemetry(stop, samples): - while not stop.is_set(): - sample = {"monotonic_ns": time.monotonic_ns()} - try: - result = subprocess.run( - [ - "nvidia-smi", - "--query-gpu=memory.used,utilization.gpu", - "--format=csv,noheader,nounits", - ], - capture_output=True, - text=True, - timeout=2, - check=True, - ) - mem, util = result.stdout.strip().split(",") - sample.update(gpu_used_mib=int(mem), gpu_utilization=int(util)) - sample["cgroup_memory_mib"] = ( - int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576 - ) - sample["cpu_stat"] = { - key: int(value) - for key, value in ( - line.split() - for line in Path("/sys/fs/cgroup/cpu.stat").read_text().splitlines() + sampler = None + try: + sampler = NvmlSampler() + while not stop.is_set(): + sample = {"monotonic_ns": time.monotonic_ns()} + try: + sample.update(sampler.sample()) + sample["cgroup_memory_mib"] = ( + int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576 ) - } - except Exception as error: - sample["error"] = str(error) - samples.append(sample) - stop.wait(0.5) + sample["cpu_stat"] = { + key: int(value) + for key, value in ( + line.split() + for line in Path("/sys/fs/cgroup/cpu.stat").read_text().splitlines() + ) + } + except Exception as error: + sample["error"] = str(error) + samples.append(sample) + stop.wait(0.5) + except Exception as error: + samples.append({"monotonic_ns": time.monotonic_ns(), "error": str(error)}) + finally: + if sampler is not None: + sampler.close() -def wait_triton(process, seconds=90): +def wait_triton(process, model_ids, seconds=90): deadline = time.monotonic() + seconds while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError("Triton exited during initialization") - connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1) - try: - connection.request("GET", "/v2/models/rf_detr_large_native_kb4/ready") - if connection.getresponse().status == 200: - return - except OSError: - pass - finally: - connection.close() + ready = True + for model_id in model_ids: + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1) + try: + connection.request("GET", f"/v2/models/{model_id}/ready") + ready = ready and connection.getresponse().status == 200 + except OSError: + ready = False + finally: + connection.close() + if ready: + return time.sleep(0.1) raise TimeoutError("Triton initialization exceeded budget") @@ -242,13 +250,20 @@ def run(args): "clock_quality": "original host arrival best effort; not hardware synchronization", "ddrnet_postprocess_layout": args.ddrnet_layout, "ddrnet_execution": args.ddrnet_execution, + "ddrnet_runtime": args.ddrnet_runtime, + "ddrnet_backend_status": ( + "reference" if args.ddrnet_runtime == "pytorch" else "experimental-unqualified" + ), + "ddrnet_preprocess": args.ddrnet_preprocess, "schedule": args.schedule, + "telemetry_mode": args.telemetry_mode, + "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms, } children, logs, results, samples = [], [], [], [] stop = threading.Event() mailbox = Mailbox(capacity=2) source_report = {} - producer = monitor = graph = gpu_stage = None + producer = monitor = graph = gpu_stage = ddr_backend = None def child(name, command, env=None): log = (output / (name + ".log")).open("wb") @@ -270,39 +285,78 @@ def run(args): previous = signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(240) try: + triton_models = ["rf_detr_large_native_kb4"] + if args.ddrnet_runtime == "triton": + triton_models.append("ddrnet_goose_fp32_mask") + triton_command = [ + "tritonserver", + "--model-repository=/models", + "--model-control-mode=explicit", + *[f"--load-model={model_id}" for model_id in triton_models], + "--allow-grpc=false", + "--allow-metrics=false", + "--http-address=127.0.0.1", + "--pinned-memory-pool-byte-size=16777216", + "--cuda-memory-pool-byte-size=0:16777216", + ] triton = child( "triton", - [ - "tritonserver", - "--model-repository=/models", - "--model-control-mode=explicit", - "--load-model=rf_detr_large_native_kb4", - "--allow-grpc=false", - "--allow-metrics=false", - "--http-address=127.0.0.1", - "--pinned-memory-pool-byte-size=16777216", - "--cuda-memory-pool-byte-size=0:16777216", - ], + triton_command, ) # Triton logs stdout too: drain directly to the same bounded-run log. threading.Thread(target=lambda: drain(triton.stdout, logs[0]), daemon=True).start() - wait_triton(triton) + wait_triton(triton, triton_models) + report["triton_config_sha256"] = { + model_id: hashlib.sha256( + Path(f"/models/{model_id}/config.pbtxt").read_bytes() + ).hexdigest() + for model_id in triton_models + } python = "/opt/conda/envs/goose/bin/python" - ddr = child( - "ddrnet", - [ - python, - "-B", - "/probe/pilot_model.py", + if args.ddrnet_runtime == "pytorch": + ddr = child( "ddrnet", - "--ddrnet-layout", - args.ddrnet_layout, - "--ddrnet-execution", - args.ddrnet_execution, - ], - {**os.environ, "PYTHONPATH": "/probe"}, - ) - report["ddrnet_ready"], _ = receive(ddr.stdout) + [ + python, + "-B", + "/probe/pilot_model.py", + "ddrnet", + "--ddrnet-layout", + args.ddrnet_layout, + "--ddrnet-execution", + args.ddrnet_execution, + ], + {**os.environ, "PYTHONPATH": "/probe"}, + ) + report["ddrnet_ready"], _ = receive(ddr.stdout) + else: + from pilot_ddrnet_numpy import preprocess_bgr as native_preprocess + from pilot_ddrnet_triton import TritonDdrnetMaskHttpInferenceBackend + + ddr = None + if args.ddrnet_preprocess == "pinned-child": + ddr = child( + "ddrnet-preprocess", + [python, "-B", "/probe/pilot_ddrnet_preprocess.py"], + { + **os.environ, + "PYTHONPATH": "/probe", + "CUDA_VISIBLE_DEVICES": "", + }, + ) + report["ddrnet_ready"], _ = receive(ddr.stdout) + else: + report["ddrnet_ready"] = {"ready": "native-numpy-preprocess"} + ddr_backend = TritonDdrnetMaskHttpInferenceBackend() + zero = np.zeros((600, 800, 3), np.uint8) + for _ in range(8): + if ddr: + send(ddr.stdin, {"op": "preprocess"}, zero.tobytes()) + _, raw = receive(ddr.stdout) + tensor = np.frombuffer(raw, " 250 if scene["stale_at_publication"]: scene["policy_actions"] = [2] * len(scene["policy_actions"]) @@ -412,6 +547,8 @@ def run(args): "finished_monotonic_ns": finished, "timing_ms": timing, "available": bundle["available"], + "ddrnet_state": ddr_layer["state"], + "ddrnet_source_age_ms": ddr_layer["source_age_ms"], "surface_state": scene["surface_state"], "proposal_count": len(scene["proposals"]), "observation_count": len(scene["observations"]), @@ -447,6 +584,9 @@ def run(args): ) if mailbox.error: raise RuntimeError(mailbox.error) + report["triton_statistics_delta"] = stats_delta( + stats_before, read_stats(triton_models) + ) report["execution_complete"] = True except Exception: report["execution_complete"] = False @@ -474,6 +614,8 @@ def run(args): report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True if graph: graph.backend.close() + if ddr_backend: + ddr_backend.close() for log in logs: log.close() report["stop_ms"] = (time.monotonic_ns() - shutdown_start) / 1e6 @@ -511,6 +653,15 @@ def run(args): "dropped": len(mailbox.dropped), "unaccounted": camera_count - len(results) - len(mailbox.dropped), } + ddrnet_executed = sum(r.get("ddrnet_state") == "current" for r in results) + report["model_cadence"] = { + "ddrnet_executed": ddrnet_executed, + "ddrnet_reused": len(results) - ddrnet_executed, + "ddrnet_source_age_ms": distribution( + [r["ddrnet_source_age_ms"] for r in results] + ), + "detector_executed": len(results), + } latency = report["distributions_ms"].get("source_due_to_receiver_ms") queue_lags = [ r["timing_ms"]["queue_wait_ms"] + r["timing_ms"]["gpu_to_cpu_queue_wait_ms"] @@ -552,6 +703,8 @@ def run(args): and max(s.get("cgroup_memory_mib", 99999) for s in samples) <= 8192, "all_modalities_fresh": len(results) == args.frames and all(r["available"] for r in results), + "layer_source_age_250ms": bool(results) + and all(r["ddrnet_source_age_ms"] <= 250 for r in results), "network_end_to_end_qualified": False, } report["profile_realtime_qualified"] = False @@ -594,8 +747,16 @@ if __name__ == "__main__": "--ddrnet-layout", choices=("reference", "channels-last"), default="reference" ) parser.add_argument("--ddrnet-execution", choices=("eager", "cuda-graph"), default="eager") + parser.add_argument("--ddrnet-runtime", choices=("pytorch", "triton"), default="pytorch") + parser.add_argument( + "--ddrnet-preprocess", choices=("pinned-child", "native-numpy"), default="pinned-child" + ) parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial") + parser.add_argument("--telemetry-mode", choices=("nvml", "none"), default="nvml") + parser.add_argument("--ddrnet-min-source-interval-ms", type=float, default=0.0) args = parser.parse_args() if not 1 <= args.frames <= 256: parser.error("pilot window must be 1..256 frames") + if not 0 <= args.ddrnet_min_source_interval_ms <= 250: + parser.error("DDRNet source interval must be 0..250 ms") raise SystemExit(run(args)) diff --git a/tests/test_perception_joint_pilot.py b/tests/test_perception_joint_pilot.py index 084e9dc..a049cc0 100644 --- a/tests/test_perception_joint_pilot.py +++ b/tests/test_perception_joint_pilot.py @@ -98,6 +98,122 @@ def test_nearest_rank_tail_metrics_keep_outlier(pilot): assert pilot("run_joint_pilot").distribution([1, 200])["p99"] == 200 +def test_layer_refresh_cadence_is_source_clocked_and_fail_closed(pilot): + refresh = pilot("run_joint_pilot").layer_refresh_due + + assert refresh(None, 1_000_000_000, 50) + assert not refresh(1_000_000_000, 1_049_999_999, 50) + assert refresh(1_000_000_000, 1_050_000_000, 50) + with pytest.raises(ValueError, match="backwards"): + refresh(1_000_000_000, 999_999_999, 50) + + +def test_native_ddrnet_preprocess_matches_pillow_nearest_exactly(pilot): + from PIL import Image + + bgr = np.random.default_rng(41).integers(0, 256, (600, 800, 3), dtype=np.uint8) + rgb = Image.fromarray(bgr[:, :, ::-1]).crop((100, 0, 700, 600)) + resized = rgb.resize((512, 512), Image.Resampling.NEAREST) + expected = (np.asarray(resized, dtype=np.float32) / 255.0).transpose(2, 0, 1)[None] + actual = pilot("pilot_ddrnet_numpy").preprocess_bgr(bgr) + np.testing.assert_array_equal(actual, expected) + assert actual.flags.c_contiguous and actual.dtype == np.dtype("