SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(system): instrument worker pipeline stages

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-07-27 21:41:40 +03:00
parent 67e12a47a4
commit 667db00b52
10 changed files with 446 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
experiments/perception/worker/e15_shadow_runtime.py
+29 -20
View File
@@ -6,6 +6,7 @@ import threading
import time
from collections import deque
from collections.abc import Callable
from contextlib import AbstractContextManager, nullcontext
from dataclasses import dataclass
from typing import Any
@@ -168,6 +169,7 @@ class PersistentFmp4Decoder:
height: int = 600,
maximum_buffer_bytes: int = 8 * 1024 * 1024,
metadata_capacity: int = 16,
measure_decode: Callable[[int], AbstractContextManager[None]] | None = None,
) -> None:
if width < 1 or height < 1:
raise ValueError("decoder resolution is invalid")
@@ -176,6 +178,7 @@ class PersistentFmp4Decoder:
self._height = height
self._media = IncrementalMediaBuffer(maximum_buffer_bytes)
self._metadata = CameraMetadataQueue(metadata_capacity)
self._measure_decode = measure_decode
self._thread = threading.Thread(
target=self._decode,
name="lab-e15-fmp4-decoder",
@@ -262,27 +265,33 @@ class PersistentFmp4Decoder:
try:
for frame in container.decode(video=0):
metadata = self._metadata.take()
image = frame.to_ndarray(format="rgb24")
if image.shape != (self._height, self._width, 3):
raise ShadowRuntimeError("decoded camera resolution changed")
image.setflags(write=False)
decoded_monotonic = time.perf_counter()
self._on_frame(
DecodedCameraFrame(
frame_index=self._decoded_frames,
metadata=metadata,
image=image,
decoded_monotonic=decoded_monotonic,
decode_age_ms=max(
0.0,
(
decoded_monotonic
- metadata.worker_received_monotonic
)
* 1000,
),
)
measurement = (
self._measure_decode(self._decoded_frames)
if self._measure_decode is not None
else nullcontext()
)
with measurement:
image = frame.to_ndarray(format="rgb24")
if image.shape != (self._height, self._width, 3):
raise ShadowRuntimeError("decoded camera resolution changed")
image.setflags(write=False)
decoded_monotonic = time.perf_counter()
self._on_frame(
DecodedCameraFrame(
frame_index=self._decoded_frames,
metadata=metadata,
image=image,
decoded_monotonic=decoded_monotonic,
decode_age_ms=max(
0.0,
(
decoded_monotonic
- metadata.worker_received_monotonic
)
* 1000,
),
)
)
self._decoded_frames += 1
finally:
container.close()
experiments/perception/worker/run_e15_shadow_inference.py
+196 -98
View File
@@ -20,7 +20,8 @@ import sys
import threading
import time
from collections import Counter, deque
from contextlib import suppress
from collections.abc import Iterator
from contextlib import contextmanager, suppress
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
@@ -115,6 +116,17 @@ FUSION_SCHEMA = "missioncore.e15-shadow-fusion-frame/v1"
WORLD_SCHEMA = "missioncore.live-perception-world-state/v1"
SEMANTIC_SCHEMA = "missioncore.e15-shadow-semantic-frame/v1"
PIPELINE_ID = "shadow-fmp4-yolox-eomt-kb4-amodal-world-state/v1"
PIPELINE_STAGE_IDS = (
"source-ingress",
"camera-decode",
"preprocessing",
"detector",
"semantic-model",
"sensor-fusion",
"tracking",
"temporal-state",
"result-publication",
)
def arguments() -> argparse.Namespace:
@@ -778,6 +790,76 @@ class _RuntimeTelemetry:
return summary
class _StageExecutionTelemetry:
"""Measure named pipeline spans without pretending they are OS processes."""
def __init__(self, stage_ids: tuple[str, ...] = PIPELINE_STAGE_IDS) -> None:
if not stage_ids or len(stage_ids) != len(set(stage_ids)):
raise RuntimeError("pipeline stage identities are invalid")
self._stage_ids = stage_ids
self._lock = threading.Lock()
self._next_token = 0
self._active: dict[int, tuple[str, float, int | None]] = {}
self._elapsed_seconds = dict.fromkeys(stage_ids, 0.0)
self._activations = dict.fromkeys(stage_ids, 0)
self._last_frame_index: int | None = None
@contextmanager
def measure(
self,
stage_id: str,
frame_index: int | None = None,
) -> Iterator[None]:
if stage_id not in self._elapsed_seconds:
raise RuntimeError(f"unknown pipeline stage: {stage_id}")
started = time.perf_counter()
with self._lock:
self._next_token += 1
token = self._next_token
self._active[token] = (stage_id, started, frame_index)
self._activations[stage_id] += 1
if frame_index is not None:
self._last_frame_index = frame_index
try:
yield
finally:
finished = time.perf_counter()
with self._lock:
active = self._active.pop(token, None)
if active is not None:
self._elapsed_seconds[stage_id] += max(0.0, finished - active[1])
def snapshot(self) -> dict[str, Any]:
now = time.perf_counter()
with self._lock:
elapsed = dict(self._elapsed_seconds)
active_rows = list(self._active.values())
for stage_id, started, _frame_index in active_rows:
elapsed[stage_id] += max(0.0, now - started)
total = sum(elapsed.values())
active_stages = list(
dict.fromkeys(stage_id for stage_id, _started, _frame in active_rows)
)
current_stage = active_rows[-1][0] if active_rows else None
return {
"current_stage": current_stage,
"active_stages": active_stages,
"active_frame_index": self._last_frame_index,
"stages": {
stage_id: {
"elapsed_seconds": round(elapsed[stage_id], 6),
"activations": self._activations[stage_id],
"share_percent": (
round(elapsed[stage_id] / total * 100, 6)
if total > 0
else None
),
}
for stage_id in self._stage_ids
},
}
def _send_client_binary_frame(stream: Any, payload: bytes) -> None:
if not payload or len(payload) > LIVE_RESULT_MAX_PAYLOAD_BYTES + 256 * 1024 + 8:
raise ShadowRuntimeError("live result websocket frame exceeds the bound")
@@ -810,6 +892,7 @@ def _receiver(
sensor_decode_ms: dict[str, list[float]],
result_queue: queue.Queue[bytes],
result_complete: threading.Event,
stage_telemetry: _StageExecutionTelemetry,
) -> None:
import select
@@ -871,7 +954,8 @@ def _receiver(
continue
if opcode != 0x2:
raise ShadowRuntimeError(f"unexpected websocket opcode: {opcode}")
header, payload = _decode_event(frame)
with stage_telemetry.measure("source-ingress"):
header, payload = _decode_event(frame)
sequence = int(header["ingress_sequence"])
if state.last_ingress_sequence is not None:
if sequence <= state.last_ingress_sequence:
@@ -1154,17 +1238,16 @@ def run(
if not token or len(token) < 40:
raise RuntimeError("LAB E15 shadow token is missing")
def report_stage(stage: str, frame_index: int | None = None) -> None:
if runtime_state is None:
return
runtime_state["current_stage"] = stage
runtime_state["stage_observed_at_utc"] = (
datetime.now(UTC).isoformat().replace("+00:00", "Z")
)
if frame_index is not None:
runtime_state["active_frame_index"] = frame_index
stage_telemetry = (
runtime_state.get("_stage_telemetry")
if runtime_state is not None
else None
)
if not isinstance(stage_telemetry, _StageExecutionTelemetry):
stage_telemetry = _StageExecutionTelemetry()
if runtime_state is not None:
runtime_state["_stage_telemetry"] = stage_telemetry
report_stage("preprocessing")
common = _common(args) if loaded is None else loaded.common
live = common["live"]
e14 = common["e14"]
@@ -1231,7 +1314,6 @@ def run(
def on_decoded(frame: DecodedCameraFrame) -> None:
nonlocal decoded_frame_count
try:
report_stage("camera-decode", frame.frame_index)
if not first_camera_epoch_ns:
first_camera_epoch_ns.append(frame.metadata.captured_at_epoch_ns)
last_camera_epoch_ns[:] = [frame.metadata.captured_at_epoch_ns]
@@ -1269,6 +1351,10 @@ def run(
height=600,
maximum_buffer_bytes=int(live["transport"]["maximum_media_buffer_bytes"]),
metadata_capacity=int(live["transport"]["camera_metadata_capacity"]),
measure_decode=lambda frame_index: stage_telemetry.measure(
"camera-decode",
frame_index,
),
)
transport = _TransportState(Counter(), Counter())
result_queue: queue.Queue[bytes] = queue.Queue(maxsize=2)
@@ -1378,6 +1464,11 @@ def run(
self.count += 1
completed_semantics = SemanticResultStream()
def monitored_semantic_inference(image: Any) -> Any:
with stage_telemetry.measure("semantic-model"):
return infer_semantic(image)
semantic_thread = threading.Thread(
target=semantic_worker,
kwargs={
@@ -1386,7 +1477,7 @@ def run(
"valid_mask": valid_mask,
"target_lut": target_lut,
"target_names": target_names,
"infer": infer_semantic,
"infer": monitored_semantic_inference,
"latency": semantic_latency,
"completed": completed_semantics,
"failures": semantic_errors,
@@ -1400,7 +1491,6 @@ def run(
)
semantic_thread.start()
decoder.start()
report_stage("source-ingress")
receiver_thread = threading.Thread(
target=_receiver,
kwargs={
@@ -1418,6 +1508,7 @@ def run(
"sensor_decode_ms": sensor_decode_ms,
"result_queue": result_queue,
"result_complete": result_complete,
"stage_telemetry": stage_telemetry,
},
name="lab-e15-shadow-receiver",
daemon=True,
@@ -1447,14 +1538,22 @@ def run(
latency["decode_age_ms"].append(float(envelope.decode_ms))
latency["queue_wait_ms"].append(max(0.0, (started - envelope.decoded_monotonic) * 1000))
try:
report_stage("preprocessing", envelope.frame_index)
detector_started = time.perf_counter()
tensor = _preprocess(envelope.image, valid_mask, detector)
report_stage("detector", envelope.frame_index)
output_tensor, _request_ms = _infer(args.triton_url, detector["model"], tensor)
detections, _rejected = _detections(output_tensor, detector, valid_mask)
report_stage("tracking", envelope.frame_index)
tracks = tracker.update(detections, envelope.frame_index)
with stage_telemetry.measure("preprocessing", envelope.frame_index):
tensor = _preprocess(envelope.image, valid_mask, detector)
with stage_telemetry.measure("detector", envelope.frame_index):
output_tensor, _request_ms = _infer(
args.triton_url,
detector["model"],
tensor,
)
detections, _rejected = _detections(
output_tensor,
detector,
valid_mask,
)
with stage_telemetry.measure("tracking", envelope.frame_index):
tracks = tracker.update(detections, envelope.frame_index)
latency["detector_ms"].append((time.perf_counter() - detector_started) * 1000)
frame_seconds = float(envelope.timeline["session_seconds"])
@@ -1484,35 +1583,35 @@ def run(
).reshape((-1, 3))
position = binding.pose.position_xyz
quaternion = binding.pose.orientation_xyzw
report_stage("sensor-fusion", envelope.frame_index)
projection_started = time.perf_counter()
pixels, depths, source_indices, points_lidar = project_points(
points_map,
position,
quaternion,
projection,
)
latency["projection_ms"].append(
(time.perf_counter() - projection_started) * 1000
)
association_started = time.perf_counter()
fusions = fuse_tracks(
tracks=[_track_document(track) for track in tracks],
semantic_map=current_semantic.mask,
pixels=pixels,
depths=depths,
source_indices=source_indices,
points_map=points_map,
points_lidar=points_lidar,
association=e14["association"],
distance_history=history,
completion_tracker=completion_tracker,
sensor_position_map=position,
session_seconds=frame_seconds,
)
latency["association_ms"].append(
(time.perf_counter() - association_started) * 1000
)
with stage_telemetry.measure("sensor-fusion", envelope.frame_index):
projection_started = time.perf_counter()
pixels, depths, source_indices, points_lidar = project_points(
points_map,
position,
quaternion,
projection,
)
latency["projection_ms"].append(
(time.perf_counter() - projection_started) * 1000
)
association_started = time.perf_counter()
fusions = fuse_tracks(
tracks=[_track_document(track) for track in tracks],
semantic_map=current_semantic.mask,
pixels=pixels,
depths=depths,
source_indices=source_indices,
points_map=points_map,
points_lidar=points_lidar,
association=e14["association"],
distance_history=history,
completion_tracker=completion_tracker,
sensor_position_map=position,
session_seconds=frame_seconds,
)
latency["association_ms"].append(
(time.perf_counter() - association_started) * 1000
)
fusion_state = "fused"
fused_frames += 1
fusion_state_counts[fusion_state] += 1
@@ -1555,21 +1654,21 @@ def run(
if temporal_stabilizer is None:
fusion_objects = raw_fusion_objects
else:
report_stage("temporal-state", envelope.frame_index)
temporal_started = time.perf_counter()
fusion_objects = temporal_stabilizer.update(
frame_index=envelope.frame_index,
session_seconds=frame_seconds,
objects=raw_fusion_objects,
)
world = stabilize_world_state(
world,
fusion_objects,
temporal_world_memory,
)
latency["temporal_2d_3d_ms"].append(
(time.perf_counter() - temporal_started) * 1000
)
with stage_telemetry.measure("temporal-state", envelope.frame_index):
temporal_started = time.perf_counter()
fusion_objects = temporal_stabilizer.update(
frame_index=envelope.frame_index,
session_seconds=frame_seconds,
objects=raw_fusion_objects,
)
world = stabilize_world_state(
world,
fusion_objects,
temporal_world_memory,
)
latency["temporal_2d_3d_ms"].append(
(time.perf_counter() - temporal_started) * 1000
)
stabilized_cuboids += sum(
str(item.get("cuboid_status", "")).startswith("accepted-")
for item in fusion_objects
@@ -1628,30 +1727,29 @@ def run(
quality=80,
optimize=False,
)
live_result = encode_live_perception_result(
frame_index=envelope.frame_index,
source_frame_index=int(envelope.timeline["source_frame_index"]),
session_seconds=frame_seconds,
captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]),
image_jpeg=encoded_image.getvalue(),
segmentation_mask=(
current_semantic.mask
if current_semantic is not None and semantic_status == "fresh"
else None
),
objects=fusion_objects,
delivery=world["delivery"],
)
report_stage("result-publication", envelope.frame_index)
try:
result_queue.put_nowait(live_result)
except queue.Full:
with suppress(queue.Empty):
result_queue.get_nowait()
result_queue.task_done()
transport.results_dropped += 1
result_queue.put_nowait(live_result)
report_stage("source-ingress", envelope.frame_index)
with stage_telemetry.measure("result-publication", envelope.frame_index):
live_result = encode_live_perception_result(
frame_index=envelope.frame_index,
source_frame_index=int(envelope.timeline["source_frame_index"]),
session_seconds=frame_seconds,
captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]),
image_jpeg=encoded_image.getvalue(),
segmentation_mask=(
current_semantic.mask
if current_semantic is not None and semantic_status == "fresh"
else None
),
objects=fusion_objects,
delivery=world["delivery"],
)
try:
result_queue.put_nowait(live_result)
except queue.Full:
with suppress(queue.Empty):
result_queue.get_nowait()
result_queue.task_done()
transport.results_dropped += 1
result_queue.put_nowait(live_result)
except Exception:
detector_failures += 1
raise
@@ -1907,6 +2005,7 @@ def run(
},
"gpu_telemetry": gpu.summary(),
"runtime_telemetry": runtime_summary,
"stage_telemetry": stage_telemetry.snapshot(),
"process_cpu_seconds": time.process_time() - process_cpu_started,
"process_peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024,
"cuda_peak_memory_allocated_mib": torch.cuda.max_memory_allocated() / 2**20,
@@ -2094,8 +2193,7 @@ def serve(args: argparse.Namespace) -> int:
"failed_runs": 0,
"active_request_id": None,
"active_frame_index": None,
"current_stage": None,
"stage_observed_at_utc": None,
"_stage_telemetry": _StageExecutionTelemetry(),
}
class Handler(BaseHTTPRequestHandler):
@@ -2124,6 +2222,7 @@ def serve(args: argparse.Namespace) -> int:
if self.path != "/health":
self._send(404, {"ok": False, "error": "not-found"})
return
stage_snapshot = state["_stage_telemetry"].snapshot()
self._send(
200,
{
@@ -2134,9 +2233,10 @@ def serve(args: argparse.Namespace) -> int:
"completed_runs": state["completed_runs"],
"failed_runs": state["failed_runs"],
"active_request_id": state["active_request_id"],
"active_frame_index": state["active_frame_index"],
"current_stage": state["current_stage"],
"stage_observed_at_utc": state["stage_observed_at_utc"],
"active_frame_index": stage_snapshot["active_frame_index"],
"current_stage": stage_snapshot["current_stage"],
"active_stages": stage_snapshot["active_stages"],
"stage_metrics": stage_snapshot["stages"],
"authority": common["live"]["authority"],
"gpu": torch.cuda.get_device_name(),
},
@@ -2167,6 +2267,7 @@ def serve(args: argparse.Namespace) -> int:
run_args = _persistent_run_arguments(args, document)
document["token"] = None
state["active_request_id"] = request_id
state["_stage_telemetry"] = _StageExecutionTelemetry()
exit_code = run(run_args, loaded, state)
state["completed_runs"] += 1
self._send(
@@ -2203,9 +2304,6 @@ def serve(args: argparse.Namespace) -> int:
finally:
state["busy"] = False
state["active_request_id"] = None
state["active_frame_index"] = None
state["current_stage"] = None
state["stage_observed_at_utc"] = None
run_lock.release()
server = ThreadingHTTPServer((args.listen_host, args.listen_port), Handler)