diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py new file mode 100644 index 0000000..d449947 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge.py @@ -0,0 +1,172 @@ +"""Binary input → existing full-graph bundles; source paths never enter this adapter.""" + +import hashlib +import socket +import threading +import time +from collections import Counter + +import numpy as np + +from k1link.perception.streaming_ingress import StreamingIngress +from k1link.perception.streaming_sensors import ( + CausalSensorWindow, + increment_identity, + milliseconds, + normalized_sensor, +) + + +class BinaryGraphBridge: + def __init__(self, runtime, decoder, source, report): + self.runtime, self.decoder, self.source, self.report = runtime, decoder, source, report + self.source_zero, self.wall_zero = source.source_zero, source.wall_zero + self.window = CausalSensorWindow(runtime.mailbox) + self.accepted = [] # Bounded pilot diagnostics, not a runtime history queue. + self.binding_reasons = Counter() + self.preroll = 0 + self.bgr_hashes = [] + self.left = right = None + try: + self.left, right = socket.socketpair() + self.receiver = StreamingIngress( + right, runtime, "recorded-acquisition", 1, self.consume, self.notice + ) + self.producer = threading.Thread( + target=source.run, args=(self.left,), daemon=True, name="binary-recorded-source" + ) + runtime.track_thread(self.producer) + except BaseException: + # No threads have started. A failed registration/lease must not + # strand the preallocated cache or either end of the socket. + for connection in (self.left, right): + if connection is not None: + connection.close() + self.window.close() + raise + + def start(self): + self.receiver.start() + self.producer.start() + + def notice(self, value): + # A gap is not a new codec epoch. This profile requires a new run/lease + # and fresh decoder state; never continue predictive decoding across it. + raise ValueError("binary full profile requires restart after an explicit source gap") + + def consume(self, event): + self.runtime.check_current(self.runtime.start) + stamp = event.received_monotonic_ns + self.window.advance(stamp) + if event.modality == "camera-init": + if event.source_id != "sensor.camera.right": + raise ValueError("camera source mapping changed") + self.decoder.configure(event.payload) + return + if event.modality in ("lidar", "pose"): + expected = { + "lidar": "normalized-map-point-increments", + "pose": "normalized-map-from-lidar-pose", + } + if event.source_id != expected[event.modality]: + raise ValueError("normalized sensor mapping changed") + self.window.append( + normalized_sensor(event.modality, event.source_sequence, stamp, event.payload) + ) + return + if event.modality != "camera-frame" or event.source_id != "sensor.camera.right": + raise ValueError("unsupported full-profile input") + # Freeze the causal cut BEFORE the decoder RPC. The synchronous ingress + # cannot consume future sensor messages while this callback is running. + binding = self.window.bind(stamp) + pose = self.window.pose + rolling = tuple(self.window.rolling) + fresh = binding.increments + count = sum(len(e.value[0]) for e in fresh) + rolling_count = sum(len(e.value[0]) for e in rolling) + size = 1440000 + count * 24 + rolling_count * 32 + 4096 + reservation = self.runtime.mailbox.reserve_ingress(size) + raw = image = points = rolling_points = rolling_times = bundle = None + transferred = False + decode_started = time.monotonic_ns() + try: + raw = bytearray(1440000) + decoded = self.decoder.decode(event.payload, raw) + image = np.frombuffer(raw, "u1").reshape(600, 800, 3) + image.setflags(write=False) + if decoded["frame_index"] != event.source_sequence: + raise ValueError("camera and decoder frame identity diverged") + points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3)) + rolling_points = ( + np.concatenate([e.value[0] for e in rolling]) if rolling else np.empty((0, 3)) + ) + rolling_times = np.empty(rolling_count, np.int64) + offset = 0 + for e in rolling: + length = len(e.value[0]) + rolling_times[offset : offset + length] = e.time_ns + offset += length + due = self.wall_zero + stamp - self.source_zero + bundle = { + "sequence": event.source_sequence, + "time_ns": stamp, + "source_ns": stamp - self.source_zero, + "due_ns": due, + "utc_ns": event.captured_at_epoch_ns, + "image": image, + "points": points, + "rolling_points": rolling_points, + "rolling_times": rolling_times, + "pose": pose.value if pose else None, + "available": binding.available, + "binding_age_ms": milliseconds(binding.binding_age_ns), + "sensor_binding": binding.document(), + "lineage": { + "camera_index_sequence": event.source_sequence + 1, + "camera_host_monotonic_ns": stamp, + "pose_sequence": pose.sequence if pose else None, + "pose_host_monotonic_ns": pose.time_ns if pose else None, + "point_increments": [increment_identity(e) for e in fresh], + "pose_age_ms": milliseconds(binding.pose_age_ns), + "newest_point_age_ms": milliseconds(binding.newest_point_age_ns), + "oldest_point_age_ms": milliseconds(binding.oldest_point_age_ns), + }, + "source_release_lag_ms": None, # Recorded independently by the source, not guessed. + "decode_ms": decoded["decode_ms"], + "decode_rpc_ms": (time.monotonic_ns() - decode_started) / 1e6, + "enqueued_ns": time.monotonic_ns(), + "payload_bytes": size, + } + self.bgr_hashes.append(hashlib.sha256(image).hexdigest()) + bundle["enqueued_ns"] = time.monotonic_ns() + transferred = self.runtime.admit_reserved(self.runtime.start, bundle, reservation) + self.accepted.append(event.source_sequence) + self.binding_reasons.update(binding.reasons) + self.preroll += sum(len(e.value[0]) for e in binding.history_only) + self.report["last_camera_due_ns"] = due + self.window.finish_camera(stamp) + finally: + raw = image = points = rolling_points = rolling_times = bundle = None + if not transferred: + reservation.release() + + def close(self): + if self.producer.ident is not None: + self.producer.join(timeout=2) + else: + self.left.close() + if self.receiver.thread.ident is None: + self.receiver.connection.close() + joined = self.receiver.join(timeout=2) if self.receiver.thread.ident is not None else True + self.report.update( + binary_ingress=self.receiver.snapshot(), + decoder_bgr_sha256=self.bgr_hashes, + failed_camera_sequences=sorted(set(self.source.released) - set(self.accepted)), + sensor_binding_reasons=dict(self.binding_reasons), + preroll_history_only_points=self.preroll, + ) + if joined and not self.producer.is_alive(): + self.window.close() + self.decoder.close() + return True + return False diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py new file mode 100644 index 0000000..ec261e6 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py @@ -0,0 +1,222 @@ +"""CPU-only decoded-bundle oracle before a full GPU graph run; bounded prefix.""" + +import argparse +import hashlib +import json +import os +import subprocess +import threading +import time +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace + +from pilot_binary_bridge import BinaryGraphBridge +from pilot_binary_source import RecordingSource + +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_decoder_client import StreamingDecoderClient +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_queue import StreamMailbox +from k1link.perception.worker_lease import WorkerLeaseError + + +def digest(bundle): + result = { + key: bundle[key] + for key in ( + "sequence", + "time_ns", + "utc_ns", + "lineage", + "sensor_binding", + "available", + "binding_age_ms", + ) + } + result["arrays"] = { + key: hashlib.sha256(bundle[key]).hexdigest() + for key in ("points", "rolling_points", "rolling_times") + } + result["pose"] = [a.tolist() for a in bundle["pose"]] if bundle["pose"] else None + return result + + +def run(args): + root = Path(args.output) + root.mkdir() + report = { + "created_utc": datetime.now(UTC).isoformat(), + "started_monotonic_ns": time.monotonic_ns(), + "scope": "CPU-only bundle parity", + "synthetic_decoder_timeout": args.fault_decoder, + } + source_report = {} + identity = StreamStart( + run_id=root.name, + source_id="RAVNOVES00-prefix", + worker_id="worker-006", + epoch_id="epoch-1", + lease_generation=1, + profile_sha256=hashlib.sha256(Path("/out/candidate-profile.json").read_bytes()).hexdigest(), + image_sha256="664824aa25de1db178f177d67a81b01541a938b479812b9383ddbf03f6b59dbe", + effective_config_sha256=hashlib.sha256(b"cpu-bundle-parity-v1").hexdigest(), + calibration_sha256=hashlib.sha256( + b"no-calibration-or-model-scene-in-this-probe" + ).hexdigest(), + clock_domain_id="original-host-arrival-clock", + input_mode="recorded-source-paced", + ) + runtime = StreamingLifecycle( + identity, Path("/tmp/cpu-bridge-lease"), StreamMailbox(), threading.Event() + ) + heartbeat_stop = threading.Event() + + def heartbeat(): + while not heartbeat_stop.wait(0.25): + try: + runtime.renew(identity) + except WorkerLeaseError: + # Stop/lease failure is already terminal in the common runtime. + # Do not turn an expected asynchronous shutdown into thread noise. + return + + renewer = threading.Thread(target=heartbeat, daemon=True) + renewer.start() + client = bridge = None + decoded = [] + with (root / "decoder.log").open("wb") as log: + try: + command = ["python3", "-B", "/probe/pilot_fragment_decoder.py"] + if args.fault_decoder: + command = [ + "python3", + "-B", + "-c", + "import sys,time; from pilot_ipc import receive,send; " + "send(sys.stdout.buffer,{'ready':'fragment-h264','pyav':'18.0.0'," + "'as_limit_mib':1024}); " + "receive(sys.stdin.buffer); send(sys.stdout.buffer,{'initialized':True}); " + "receive(sys.stdin.buffer); time.sleep(10)", + ] + child = runtime.spawn( + lambda: subprocess.Popen( + command, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=log, + start_new_session=True, + env={**os.environ, "CUDA_VISIBLE_DEVICES": ""}, + ) + ) + client = StreamingDecoderClient(runtime, child.stdout.fileno(), child.stdin.fileno()) + runtime.ready() + source = RecordingSource(args, runtime, source_report) + bridge = BinaryGraphBridge(runtime, client, source, source_report) + bridge.start() + while (bundle := runtime.mailbox.take()) is not None: + decoded.append(digest(bundle)) + runtime.mailbox.release(bundle) + if runtime.mailbox.error and not args.fault_decoder: + raise ValueError(runtime.mailbox.error) + finally: + heartbeat_stop.set() + renewer.join(timeout=1) + runtime.request_stop("completed" if len(decoded) == args.frames else "failed") + runtime.stop_children() + report["bridge_stopped"] = bridge.close() if bridge else True + if client: + client.close() + report["released"] = runtime.close() + report["runtime"] = runtime.snapshot() + report["source"] = source_report + report["peak_input_bytes"] = runtime.mailbox.peak_bytes + report["residual_input_bytes"] = runtime.mailbox.bytes + (root / "report.json").write_text(json.dumps(report, indent=2) + "\n") + if args.fault_decoder: + assert not decoded and report["released"] and not runtime.mailbox.bytes + assert "decoder RPC deadline" in source_report["binary_ingress"]["error"] + assert source_report["failed_camera_sequences"] + print( + json.dumps( + { + "synthetic_decoder_timeout": True, + "released": report["released"], + "source_failed": source_report["failed_camera_sequences"], + "input_bytes": runtime.mailbox.bytes, + } + ) + ) + return + # Independent oracle uses the old producer's exact assembly code AFTER the + # binary run. No prepared arrays feed the new decoder or graph. Zero camera + # bytes are used only here: image parity has its separate actual decoder test. + import run_joint_pilot as legacy + + reference = [] + + class Sink: + error = None + + def put(self, bundle): + reference.append(digest(bundle)) + return True + + def finish(self, error=None): + self.error = error + + sink = Sink() + old_send, old_receive = legacy.send, legacy.receive + image = bytes(1440000) + legacy.send = lambda *args: None + legacy.receive = lambda *args: ({"decode_ms": 0}, image) + try: + legacy.produce( + args, + SimpleNamespace(stdin=None, stdout=None), + sink, + SimpleNamespace(is_set=lambda: False, wait=lambda _: False), + {}, + ) + finally: + legacy.send, legacy.receive = old_send, old_receive + assert sink.error is None, sink.error + report.update( + bundle_count=len(decoded), + oracle_count=len(reference), + exact_sensor_bundles=decoded == reference, + model_runs=0, + network_qualified=False, + ) + (root / "decoded-bundles.json").write_text(json.dumps(decoded, indent=2) + "\n") + (root / "reference-bundles.json").write_text(json.dumps(reference, indent=2) + "\n") + (root / "report.json").write_text(json.dumps(report, indent=2) + "\n") + assert decoded == reference and len(decoded) == args.frames and report["released"] + print( + json.dumps( + { + k: report[k] + for k in ( + "bundle_count", + "exact_sensor_bundles", + "released", + "peak_input_bytes", + "residual_input_bytes", + ) + } + ) + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--output", required=True) + parser.add_argument("--frames", type=int, default=32) + parser.add_argument("--fault-decoder", action="store_true") + parser.add_argument("--camera-root", default="/camera") + parser.add_argument("--camera-index", default="/camera/index.jsonl") + parser.add_argument("--sensor-archive", default="/sensor-source.npz") + args = parser.parse_args() + if not 1 <= args.frames <= 128: + parser.error("CPU oracle is bounded to 128 cameras") + run(args) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py new file mode 100644 index 0000000..7e72ff1 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_binary_source.py @@ -0,0 +1,145 @@ +"""Recording adapter only: emits bounded camera/sensors at original 1x times. + +No model or decoder is called, and processing speed never sets source pacing. +Source inventory/count belong here, never to the binary consumer or full graph. +""" + +import hashlib +import struct +import time +import traceback +from collections import Counter +from pathlib import Path + +from pilot_source import SensorArchive, camera_events, merged_events + +from k1link.compute.live_perception import LiveIngressEvent +from k1link.perception.streaming_sender import StreamingSender + + +def read_member(path, root, maximum): + resolved = path.resolve(strict=True) + if not resolved.is_relative_to(root.resolve(strict=True)): + raise ValueError("source member escaped camera root") + with resolved.open("rb") as stream: + raw = stream.read(maximum + 1) + if not 0 < len(raw) <= maximum: + raise ValueError("source member exceeds bound") + return raw + + +class RecordingSource: + def __init__(self, args, runtime, report): + self.args, self.runtime, self.report = args, runtime, report + first = next(camera_events(args.camera_index, 1)).time_ns + self.source_zero = first - 500_000_000 + self.wall_zero = time.monotonic_ns() + 50_000_000 + self.released = [] # Diagnostic bounded prefix only, never runtime admission state. + report.update( + source_zero_ns=self.source_zero, + wall_zero_ns=self.wall_zero, + first_camera_source_ns=first, + source_clock_speed=1.0, + source_eof_required=False, + full_source_prepass=False, + ) + + def run(self, connection): + archive = None + sender = None + arrivals, skipped = Counter(), Counter() + lags = [] + clock = self.runtime.stop_event + try: + archive = SensorArchive(Path(self.args.sensor_archive)) + root = Path(self.args.camera_root) + sender = StreamingSender( + connection, + self.runtime.start, + "recorded-acquisition", + 1, + lambda: self.runtime.check_current(self.runtime.start), + ) + seq = 1 + sender.send( + LiveIngressEvent( + seq, + "recorded-acquisition", + 1, + "camera-init", + "sensor.camera.right", + 0, + 0, + self.source_zero, + read_member(root / "init.mp4", root, 65536), + ) + ) + for event in merged_events(archive, self.args.camera_index, self.args.frames): + if event.time_ns < self.source_zero: + skipped[event.channel] += 1 + continue + due = self.wall_zero + event.time_ns - self.source_zero + if clock.wait(max(0, (due - time.monotonic_ns()) / 1e9)): + break + arrivals[event.channel] += 1 + lags.append(max(0, time.monotonic_ns() - due) / 1e6) + utc = 0 # These normalized sensor rows have no UTC evidence. + if event.channel == "camera": + self.released.append(event.sequence) + row = event.value + raw = read_member(root / row["path"], root, 1024 * 1024) + if ( + len(raw) != row["length"] + or hashlib.sha256(raw).hexdigest() != row["sha256"] + ): + raise ValueError("camera fragment integrity changed") + modality, source_id, utc = ( + "camera-frame", + "sensor.camera.right", + row["host_epoch_ns"], + ) + elif event.channel == "points": + xyz, intensity = event.value + raw = struct.pack("= camera_time_ns: - raise ValueError("camera binding clock must increase") - if pose is not None and pose.time_ns > camera_time_ns: - raise ValueError("future pose cannot bind to camera") - if any(e.time_ns > camera_time_ns for e in increments): - raise ValueError("future points cannot bind to camera") - if any(a.time_ns > b.time_ns for a, b in zip(increments, increments[1:], strict=False)): - raise ValueError("point binding clock moved backwards") - - history_only = () - if previous_camera_time_ns is None: - # History used to warm rolling geometry is not one current increment. - # Preserve its identity separately; do not retimestamp or silently drop it. - selected, history = [], [] - for event in increments: - is_current = ( - pose is not None - and camera_time_ns - event.time_ns <= OLDEST_POINT_AGE_NS - and abs(event.time_ns - pose.time_ns) <= POINT_POSE_SKEW_NS - ) - (selected if is_current else history).append(event) - increments, history_only = tuple(selected), tuple(history) - - pose_age = None if pose is None else camera_time_ns - pose.time_ns - newest_age = None if not increments else camera_time_ns - increments[-1].time_ns - oldest_age = None if not increments else camera_time_ns - increments[0].time_ns - skew = ( - max(abs(e.time_ns - pose.time_ns) for e in increments) - if increments and pose is not None - else None - ) - reasons = [] - if pose is None: - pose_state = "unavailable" - reasons.append("pose-unavailable") - elif pose_age > POSE_AGE_NS: - pose_state = "stale" - reasons.append("pose-too-old") - else: - pose_state = ( - "held" - if previous_camera_time_ns is not None and pose.time_ns <= previous_camera_time_ns - else "current" - ) - if not increments or not any(len(e.value[0]) for e in increments): - points_state = "unavailable" - reasons.append("point-increment-unavailable") - else: - points_state = "current" - if newest_age > NEWEST_POINT_AGE_NS: - points_state = "stale" - reasons.append("newest-points-too-old") - if oldest_age > OLDEST_POINT_AGE_NS: - points_state = "stale" - reasons.append("oldest-points-too-old") - if skew is not None and skew > POINT_POSE_SKEW_NS: - reasons.append("point-pose-skew") - return SensorBinding( - increments, - history_only, - pose_age, - newest_age, - oldest_age, - skew, - pose_state, - points_state, - tuple(reasons), - ) +__all__ = [ + "NEWEST_POINT_AGE_NS", + "OLDEST_POINT_AGE_NS", + "POINT_POSE_SKEW_NS", + "POSE_AGE_NS", + "SensorBinding", + "bind_sensors", + "increment_identity", + "milliseconds", +] 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 0f86ffe..95204a9 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -35,6 +35,8 @@ from pilot_telemetry import NvmlSampler def distribution(values): + # Missing diagnostics are not zero; binary source lag is measured separately. + values = [value for value in values if value is not None] if not values: return None values = sorted(values) @@ -271,6 +273,7 @@ def run(args): "triton_verbose": args.triton_verbose, "costmap_freshness_mode": args.costmap_freshness, "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms, + "input_transport": args.input_transport, } children, logs, results, samples = [], [], [], [] stop = threading.Event() @@ -279,6 +282,7 @@ def run(args): producer = monitor = graph = gpu_stage = ddr_backend = None cpu_bundle = None controller = None + binary_bridge = decoder_client = None epoch_id = args.run_id def child(name, command, env=None): @@ -385,10 +389,28 @@ def run(args): ddr_backend.infer(tensor) decoder = child( "decoder", - [python, "-B", "/probe/pilot_model.py", "camera"], - {**os.environ, "PYTHONPATH": "/probe", "CUDA_VISIBLE_DEVICES": ""}, + ( + ["python3", "-B", "/probe/pilot_fragment_decoder.py"] + if args.input_transport == "binary-ipc" + else [python, "-B", "/probe/pilot_model.py", "camera"] + ), + { + **os.environ, + "PYTHONPATH": os.environ["PYTHONPATH"] + if args.input_transport == "binary-ipc" + else "/probe", + "CUDA_VISIBLE_DEVICES": "", + }, ) - report["decoder_ready"], _ = receive(decoder.stdout) + if args.input_transport == "binary-ipc": + from k1link.perception.streaming_decoder_client import StreamingDecoderClient + + decoder_client = StreamingDecoderClient( + controller.runtime, decoder.stdout.fileno(), decoder.stdin.fileno() + ) + report["decoder_ready"] = decoder_client.ready + else: + report["decoder_ready"], _ = receive(decoder.stdout) tgs = child("tgs", ["/usr/local/bin/pilot-tgs"]) from pilot_graph import JointGraph @@ -489,12 +511,22 @@ def run(args): from pilot_lifecycle import FencedIngress ingress = FencedIngress(controller) - producer = threading.Thread( - target=produce, args=(args, decoder, ingress, stop, source_report), daemon=True - ) - if controller: - controller.runtime.track_thread(producer) - producer.start() + if args.input_transport == "binary-ipc": + from pilot_binary_bridge import BinaryGraphBridge + from pilot_binary_source import RecordingSource + + source = RecordingSource(args, controller.runtime, source_report) + binary_bridge = BinaryGraphBridge( + controller.runtime, decoder_client, source, source_report + ) + binary_bridge.start() + else: + producer = threading.Thread( + target=produce, args=(args, decoder, ingress, stop, source_report), daemon=True + ) + if controller: + controller.runtime.track_thread(producer) + producer.start() if args.schedule == "overlap-cpu": gpu_stage = GpuStage(mailbox, compute_gpu, stop) if controller: @@ -666,6 +698,10 @@ def run(args): process.wait(timeout=1) if producer: producer.join(timeout=2) + if binary_bridge: + report["binary_bridge_stopped"] = binary_bridge.close() + if decoder_client and (not binary_bridge or report["binary_bridge_stopped"]): + decoder_client.close() if monitor: monitor.join(timeout=2) report["gpu_stage_stopped"] = gpu_stage.close() if gpu_stage else True @@ -813,6 +849,10 @@ if __name__ == "__main__": parser.add_argument("--output", required=True) parser.add_argument("--sensor-archive", default="/sensor-source.npz") parser.add_argument("--camera-index", default="/camera-index.jsonl") + parser.add_argument("--camera-root", default="/camera") + parser.add_argument( + "--input-transport", choices=("legacy-pilot", "binary-ipc"), default="legacy-pilot" + ) parser.add_argument( "--ddrnet-layout", choices=("reference", "channels-last"), default="reference" ) @@ -844,4 +884,6 @@ if __name__ == "__main__": parser.error("worker lease and launcher-verified image identity are required together") if args.stop_renew_after_sequence != -1 and not args.worker_lease_root: parser.error("lease-expiry injection requires a controller") + if args.input_transport == "binary-ipc" and not args.worker_lease_root: + parser.error("binary ingress requires the common lifecycle") raise SystemExit(run(args)) diff --git a/src/k1link/perception/streaming_decoder_client.py b/src/k1link/perception/streaming_decoder_client.py new file mode 100644 index 0000000..054e788 --- /dev/null +++ b/src/k1link/perception/streaming_decoder_client.py @@ -0,0 +1,64 @@ +"""Fenced decoder RPC; native execution lives in a separately supervised child.""" + +from __future__ import annotations + +import math +from typing import Any + +from .graph_contracts import GraphState +from .streaming_lifecycle import StreamingLifecycle +from .streaming_pipe_rpc import HEADER_SCRATCH, MAX_OUTPUT, BoundedPipeRpc, PipeRpcError + + +class StreamingDecoderClient: + def __init__(self, runtime: StreamingLifecycle, read_fd: int, write_fd: int) -> None: + if runtime.state != GraphState.STARTING: + raise PipeRpcError("decoder must be created during warmup") + self.runtime = runtime + self.reservation = runtime.mailbox.reserve_ingress(HEADER_SCRATCH) + self.closed = False + self.frames = 0 + try: + self.rpc = BoundedPipeRpc( + read_fd, write_fd, lambda: runtime.check_current(runtime.start, starting=True) + ) + self.ready = self.rpc.exchange(None, b"", bytearray(), timeout=10) + if self.ready != {"ready": "fragment-h264", "pyav": "18.0.0", "as_limit_mib": 1024}: + raise PipeRpcError("decoder startup identity changed") + self.rpc.check = lambda: runtime.check_current(runtime.start) + except BaseException: + runtime.request_stop("failed") + self.close() + raise + + def configure(self, raw: bytes) -> None: + if self.closed or not 0 < len(raw) <= 65536: + raise PipeRpcError("decoder init outside bound") + if self.rpc.exchange({"op": "init"}, raw, bytearray()) != {"initialized": True}: + raise PipeRpcError("decoder init response changed") + + def decode(self, raw: bytes, target: bytearray) -> dict[str, Any]: + # The caller has already reserved this exact BGR buffer in a decoded + # bundle ticket. RPC fills it directly, without an unaccounted copy. + if self.closed or len(target) != MAX_OUTPUT: + raise PipeRpcError("decoder output allocation changed") + result = self.rpc.exchange({"op": "decode"}, raw, target) + duration = result.get("decode_ms") + if ( + set(result) != {"decode_ms", "frame_index"} + or type(result["frame_index"]) is not int + or result["frame_index"] != self.frames + or not isinstance(duration, (int, float)) + or isinstance(duration, bool) + or not math.isfinite(duration) + or duration < 0 + ): + raise PipeRpcError("decoder frame identity or measurement changed") + self.frames += 1 + return result + + def close(self) -> None: + # Caller joins its callback and stops the owned child before closing. + if not self.closed: + self.closed = True + self.reservation.release() diff --git a/src/k1link/perception/streaming_lifecycle.py b/src/k1link/perception/streaming_lifecycle.py index a2c146b..6d9c488 100644 --- a/src/k1link/perception/streaming_lifecycle.py +++ b/src/k1link/perception/streaming_lifecycle.py @@ -21,7 +21,7 @@ from typing import Any, Literal from .graph_contracts import GraphState from .realtime_contract import StreamStart from .realtime_scene import _wire_integer -from .streaming_queue import StreamBundle, StreamMailbox +from .streaming_queue import IngressReservation, StreamBundle, StreamMailbox from .worker_lease import WorkerLease, WorkerLeaseError @@ -116,6 +116,13 @@ class StreamingLifecycle: self._check(start) return self.mailbox.put(bundle) + def admit_reserved( + self, start: StreamStart, bundle: StreamBundle, reservation: IngressReservation + ) -> bool: + with self._lock: + self._check(start) + return self.mailbox.put_reserved(bundle, reservation) + @contextmanager def work(self, start: StreamStart, lane: Literal["gpu", "cpu"]) -> Iterator[None]: with self._lock: @@ -133,9 +140,10 @@ class StreamingLifecycle: with self._lock: self._active[lane] -= 1 - def check_current(self, start: StreamStart) -> None: + def check_current(self, start: StreamStart, *, starting: bool = False) -> None: + """Only child startup handshakes may opt into the warmup state.""" with self._lock: - self._check(start) + self._check(start, starting=starting) def validate_result_binding(self, value: object) -> None: """Recheck at receipt/use; accepting a hash alone cannot renew authority.""" diff --git a/src/k1link/perception/streaming_pipe_rpc.py b/src/k1link/perception/streaming_pipe_rpc.py new file mode 100644 index 0000000..e05e7ee --- /dev/null +++ b/src/k1link/perception/streaming_pipe_rpc.py @@ -0,0 +1,130 @@ +"""Bounded supervised-child RPC with caller-owned output buffers, POSIX only. + +Same small header/binary framing as the existing pilot RPC. No process creation, +paths, models, queues or implicit allocation of image payloads. Caller reserves +the target buffer and HEADER_SCRATCH before using this transport. Native child +memory is separately constrained; timeout poisons this connection, never retries. +""" + +from __future__ import annotations + +import json +import os +import select +import struct +import threading +import time +from collections.abc import Callable, Mapping +from typing import Any + +HEADER_SCRATCH = 65536 +MAX_HEADER = 4096 +MAX_INPUT = 1024 * 1024 +MAX_OUTPUT = 1_440_000 + + +class PipeRpcError(ValueError): + pass + + +def _unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise PipeRpcError("duplicate RPC field") + result[key] = value + return result + + +class BoundedPipeRpc: + def __init__(self, read_fd: int, write_fd: int, check: Callable[[], None]) -> None: + self.read_fd, self.write_fd, self.check = read_fd, write_fd, check + os.set_blocking(read_fd, False) + os.set_blocking(write_fd, False) + self._lock = threading.Lock() + self.failed = False + + def _check(self, deadline: float) -> None: + self.check() + if time.monotonic() >= deadline: + raise PipeRpcError("decoder RPC deadline exceeded") + + def _read(self, target: memoryview, deadline: float) -> None: + offset = 0 + while offset < len(target): + self._check(deadline) + if not select.select([self.read_fd], [], [], 0.01)[0]: + continue + try: + count = os.readv(self.read_fd, [target[offset:]]) + except BlockingIOError: + continue + if not count: + raise PipeRpcError("truncated decoder RPC") + offset += count + self._check(deadline) + + def _write(self, data: bytes | memoryview, deadline: float) -> None: + view = memoryview(data) + while view: + self._check(deadline) + if not select.select([], [self.write_fd], [], 0.01)[1]: + continue + try: + count = os.write(self.write_fd, view) + except BlockingIOError: + continue + if not count: + raise PipeRpcError("decoder RPC write made no progress") + view = view[count:] + self._check(deadline) + + def exchange( + self, + header: Mapping[str, Any] | None, + payload: bytes, + target: bytearray, + *, + timeout: float = 0.25, + ) -> dict[str, Any]: + """None header reads a one-time startup message; no work is submitted.""" + if not 0 < timeout <= (10 if header is None else 0.25): + raise PipeRpcError("RPC timeout outside bound") + if len(payload) > MAX_INPUT or len(target) > MAX_OUTPUT: + raise PipeRpcError("RPC payload outside bound") + if not self._lock.acquire(blocking=False): + raise PipeRpcError("concurrent decoder RPC is forbidden") + try: + if self.failed: + raise PipeRpcError("decoder RPC is poisoned") + deadline = time.monotonic() + timeout + self._check(deadline) + if header is not None: + raw = json.dumps( + {**header, "payload_bytes": len(payload)}, allow_nan=False + ).encode() + if not 0 < len(raw) <= MAX_HEADER: + raise PipeRpcError("RPC header outside bound") + self._write(struct.pack(" bool: + """Atomically transfer preallocated decoded bytes into queue ownership. + + Success consumes the reservation. On rejection/error the caller still + owns it and must drop its buffers BEFORE releasing it. No unaccounted + interval or transient double charge; active inputs cannot be evicted. + """ + with self.condition: + size = self._ingress.get(reservation) + if size is None or type(bundle["payload_bytes"]) is not int: + raise ValueError("decoded handoff ownership mismatch") + if bundle["payload_bytes"] != size: + raise ValueError("decoded handoff size mismatch") + del self._ingress[reservation] + self.bytes -= size + accepted = False + try: + accepted = self.put(bundle) + return accepted + finally: + if not accepted: + self._ingress[reservation] = size + self.bytes += size + def _discard_pending(self, reason: str) -> None: old = self.pending.popleft() _, size = self._owned.pop(id(old)) diff --git a/src/k1link/perception/streaming_sensors.py b/src/k1link/perception/streaming_sensors.py new file mode 100644 index 0000000..bfa69da --- /dev/null +++ b/src/k1link/perception/streaming_sensors.py @@ -0,0 +1,239 @@ +"""Causal input binding with explicit preroll history and per-modality age. + +No new inference or interpolation. The caller retains all preroll points in its +bounded rolling window; only the initial current-pair selection is narrowed. +After the first camera, the original increment admission rules are unchanged. +""" + +from collections import deque +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +import numpy as np + +from .streaming_queue import StreamMailbox + + +@dataclass(frozen=True) +class SensorEvent: + time_ns: int + channel: str + sequence: int + value: Any + + +POSE_AGE_NS = 100_000_000 +NEWEST_POINT_AGE_NS = 100_000_000 +OLDEST_POINT_AGE_NS = 250_000_000 +POINT_POSE_SKEW_NS = 100_000_000 + + +def increment_identity(event: SensorEvent) -> dict[str, int]: + return { + "sequence": event.sequence, + "host_monotonic_ns": event.time_ns, + "points": len(event.value[0]), + } + + +def milliseconds(value: int | None) -> float | None: + return None if value is None else value / 1e6 + + +@dataclass(frozen=True) +class SensorBinding: + increments: tuple[SensorEvent, ...] + history_only: tuple[SensorEvent, ...] + pose_age_ns: int | None + newest_point_age_ns: int | None + oldest_point_age_ns: int | None + binding_age_ns: int | None + pose_state: str + points_state: str + reasons: tuple[str, ...] + + @property + def available(self) -> bool: + return not self.reasons + + def document(self) -> dict[str, Any]: + return { + "schema_version": "missioncore.pilot-sensor-binding/v1", + "pose_state": self.pose_state, + "points_state": self.points_state, + "current_pair_available": self.available, + "reason_codes": list(self.reasons), + "pose_age_ms": milliseconds(self.pose_age_ns), + "newest_point_age_ms": milliseconds(self.newest_point_age_ns), + "oldest_point_age_ms": milliseconds(self.oldest_point_age_ns), + "point_pose_skew_ms": milliseconds(self.binding_age_ns), + "preroll_history_only": [increment_identity(e) for e in self.history_only], + "preroll_history_disposition": "retained-in-bounded-rolling-window", + } + + +def bind_sensors( + camera_time_ns: int, + pose: SensorEvent | None, + increments: Sequence[SensorEvent], + *, + previous_camera_time_ns: int | None = None, +) -> SensorBinding: + increments = tuple(increments) + if previous_camera_time_ns is not None and previous_camera_time_ns >= camera_time_ns: + raise ValueError("camera binding clock must increase") + if pose is not None and pose.time_ns > camera_time_ns: + raise ValueError("future pose cannot bind to camera") + if any(e.time_ns > camera_time_ns for e in increments): + raise ValueError("future points cannot bind to camera") + if any(a.time_ns > b.time_ns for a, b in zip(increments, increments[1:], strict=False)): + raise ValueError("point binding clock moved backwards") + + history_only: tuple[SensorEvent, ...] = () + if previous_camera_time_ns is None: + # History used to warm rolling geometry is not one current increment. + # Preserve its identity separately; do not retimestamp or silently drop it. + selected: list[SensorEvent] = [] + history: list[SensorEvent] = [] + for event in increments: + is_current = ( + pose is not None + and camera_time_ns - event.time_ns <= OLDEST_POINT_AGE_NS + and abs(event.time_ns - pose.time_ns) <= POINT_POSE_SKEW_NS + ) + (selected if is_current else history).append(event) + increments, history_only = tuple(selected), tuple(history) + + pose_age = None if pose is None else camera_time_ns - pose.time_ns + newest_age = None if not increments else camera_time_ns - increments[-1].time_ns + oldest_age = None if not increments else camera_time_ns - increments[0].time_ns + skew = ( + max(abs(e.time_ns - pose.time_ns) for e in increments) + if increments and pose is not None + else None + ) + reasons = [] + if pose is None: + pose_state = "unavailable" + reasons.append("pose-unavailable") + elif pose_age is not None and pose_age > POSE_AGE_NS: + pose_state = "stale" + reasons.append("pose-too-old") + else: + pose_state = ( + "held" + if previous_camera_time_ns is not None and pose.time_ns <= previous_camera_time_ns + else "current" + ) + if not increments or not any(len(e.value[0]) for e in increments): + points_state = "unavailable" + reasons.append("point-increment-unavailable") + else: + points_state = "current" + if newest_age is not None and newest_age > NEWEST_POINT_AGE_NS: + points_state = "stale" + reasons.append("newest-points-too-old") + if oldest_age is not None and oldest_age > OLDEST_POINT_AGE_NS: + points_state = "stale" + reasons.append("oldest-points-too-old") + if skew is not None and skew > POINT_POSE_SKEW_NS: + reasons.append("point-pose-skew") + return SensorBinding( + increments, + history_only, + pose_age, + newest_age, + oldest_age, + skew, + pose_state, + points_state, + tuple(reasons), + ) + + +def normalized_sensor(modality: str, sequence: int, time_ns: int, raw: bytes) -> SensorEvent: + """Explicit existing normalized-map wire layout, not a vendor packet parser.""" + if modality == "lidar": + count = int.from_bytes(raw[:4], "little") + if len(raw) < 4 or not 0 <= count <= 50000 or len(raw) != 4 + count * 25: + raise ValueError("normalized point layout outside bound") + xyz = np.frombuffer(raw, " 0.01: + raise ValueError("invalid source pose") + return SensorEvent(time_ns, "pose", sequence, (values[:3], values[3:])) + + +class CausalSensorWindow: + """Single-consumer rolling/fresh caches; no async nearest/future lookup. + + Reserve the maximum union of two 64000-point sets before retaining raw + views. Metadata and pose allowance is separate; Python objects/native RSS + still need the enclosing process limit. Sensor selection rules above are + exactly the existing pilot rules, including the initial history-only cut. + """ + + CACHE_BYTES = 2 * 64000 * 25 + 16384 + + def __init__(self, mailbox: StreamMailbox) -> None: + self._reservation = mailbox.reserve_ingress(self.CACHE_BYTES) + self.rolling: deque[SensorEvent] = deque() + self.fresh: list[SensorEvent] = [] + self.pose: SensorEvent | None = None + self.previous_camera_time: int | None = None + self._last_time = -1 + self.closed = False + + def advance(self, time_ns: int) -> None: + if self.closed or time_ns < self._last_time: + raise ValueError("sensor stream is closed or moved backwards") + self._last_time = time_ns + while self.rolling and time_ns - self.rolling[0].time_ns > 1_000_000_000: + self.rolling.popleft() + + def append(self, event: SensorEvent) -> None: + self.advance(event.time_ns) + if event.channel == "pose": + self.pose = event + return + if event.channel != "points": + raise ValueError("unsupported normalized sensor channel") + for items in (self.rolling, self.fresh): + if ( + len(items) >= 64 + or sum(len(e.value[0]) for e in items) + len(event.value[0]) > 64000 + ): + raise ValueError("causal sensor cache exceeds bound") + self.rolling.append(event) + self.fresh.append(event) + + def bind(self, time_ns: int) -> SensorBinding: + self.advance(time_ns) + return bind_sensors( + time_ns, self.pose, self.fresh, previous_camera_time_ns=self.previous_camera_time + ) + + def finish_camera(self, time_ns: int) -> None: + if self.closed or time_ns != self._last_time: + raise ValueError("sensor window changed during camera binding") + self.previous_camera_time = time_ns + self.fresh.clear() + + def close(self) -> None: + if not self.closed: + self.rolling.clear() + self.fresh.clear() + self.pose = None + self.closed = True + self._reservation.release() diff --git a/tests/test_perception_binary_bridge.py b/tests/test_perception_binary_bridge.py new file mode 100644 index 0000000..175615d --- /dev/null +++ b/tests/test_perception_binary_bridge.py @@ -0,0 +1,229 @@ +"""Small synthetic memory/clock/pipe checks; no real decode or models on the Mac.""" + +import json +import socket +import struct +from contextlib import contextmanager +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pytest + +from k1link.perception.streaming_pipe_rpc import BoundedPipeRpc, PipeRpcError +from k1link.perception.streaming_queue import StreamMailbox +from k1link.perception.streaming_sensors import CausalSensorWindow, normalized_sensor + + +def bundle(seq=0, size=10): + return {"sequence": seq, "payload_bytes": size} + + +def test_reservation_handoff_is_atomic_and_does_not_double_charge(): + mailbox = StreamMailbox(byte_limit=10) + ticket = mailbox.reserve_ingress(10) + value = bundle() + assert mailbox.put_reserved(value, ticket) + assert mailbox.bytes == mailbox.peak_bytes == 10 + with pytest.raises(ValueError, match="ownership"): + ticket.release() + assert mailbox.take() is value + mailbox.cancel() + assert mailbox.bytes == 10 and not mailbox.quiescent + mailbox.release(value) + assert mailbox.quiescent + + +@pytest.mark.parametrize("case", ["closed", "size", "foreign", "sequence"]) +def test_rejected_handoff_keeps_caller_ownership(case): + mailbox = StreamMailbox(byte_limit=40) + ticket = mailbox.reserve_ingress(10) + value = bundle() + if case == "closed": + mailbox.cancel() + assert not mailbox.put_reserved(value, ticket) + else: + if case == "size": + value["payload_bytes"] = 11 + if case == "sequence": + mailbox.put(bundle(1)) + with pytest.raises(ValueError): + (StreamMailbox() if case == "foreign" else mailbox).put_reserved(value, ticket) + assert mailbox.bytes >= 10 + ticket.size = 99999 # Only the stored reservation size is authoritative. + ticket.release() + mailbox.cancel() + assert mailbox.bytes == 0 and mailbox.quiescent + + +def test_handoff_overflow_drops_pending_only(): + mailbox = StreamMailbox(byte_limit=40) + mailbox.put(bundle(0)) + active = mailbox.take() + mailbox.put(bundle(1, 5)) + mailbox.put(bundle(2, 5)) + ticket = mailbox.reserve_ingress(20) + assert mailbox.put_reserved(bundle(3, 20), ticket) + assert mailbox.dropped == [{"sequence": 1, "reason": "pending-overflow"}] + assert mailbox.bytes == 35 + mailbox.cancel() + assert mailbox.bytes == 10 + mailbox.release(active) + assert mailbox.quiescent + + +@contextmanager +def pipe(response=None, check=lambda: None): + client, server = socket.socketpair() + try: + if response: + server.sendall(response) + yield BoundedPipeRpc(client.fileno(), client.fileno(), check), server + finally: + client.close() + server.close() + + +def message(header, payload=b""): + raw = json.dumps({**header, "payload_bytes": len(payload)}).encode() + return struct.pack("