perf(perception): borrow contiguous image buffers for local IPC

experiments/perception/worker/streaming_profile_stage1/pilot_binary_bridge_probe.py

@@ -85,6 +85,7 @@ def run(args):
     renewer.start()
     client = bridge = None
     decoded = []
+    timings = []
     with (root / "decoder.log").open("wb") as log:
         try:
             command = ["python3", "-B", "/probe/pilot_fragment_decoder.py"]
@@ -115,6 +116,14 @@ def run(args):
             bridge = BinaryGraphBridge(runtime, client, source, source_report)
             bridge.start()
             while (bundle := runtime.mailbox.take()) is not None:
+                timings.append(
+                    {
+                        "sequence": bundle["sequence"],
+                        "decode_ms": bundle["decode_ms"],
+                        "decode_rpc_ms": bundle["decode_rpc_ms"],
+                        "source_due_to_dequeue_ms": (time.monotonic_ns() - bundle["due_ns"]) / 1e6,
+                    }
+                )
                 decoded.append(digest(bundle))
                 runtime.mailbox.release(bundle)
             if runtime.mailbox.error and not args.fault_decoder:
@@ -132,6 +141,7 @@ def run(args):
             report["source"] = source_report
             report["peak_input_bytes"] = runtime.mailbox.peak_bytes
             report["residual_input_bytes"] = runtime.mailbox.bytes
+            report["timings"] = timings
             (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

experiments/perception/worker/streaming_profile_stage1/pilot_fragment_decoder.py

@@ -28,7 +28,7 @@ def main():
                         "decode_ms": (time.monotonic_ns() - begin) / 1e6,
                         "frame_index": decoder.frames - 1,
                     },
-                    image.tobytes(),
+                    memoryview(image),
                 )
                 del image
             else:

experiments/perception/worker/streaming_profile_stage1/pilot_ipc.py

@@ -32,10 +32,18 @@ def receive(stream, max_payload=MAX_PAYLOAD):
 
 
 def send(stream, header, payload=b""):
-    if len(payload) > MAX_PAYLOAD:
+    # Borrow a contiguous buffer until this synchronous write completes. Never
+    # concatenate the whole BGR payload with framing or make an implicit copy.
+    view = memoryview(payload).cast("B")
+    if len(view) > MAX_PAYLOAD:
         raise ValueError("IPC payload exceeds budget")
-    raw = json.dumps({**header, "payload_bytes": len(payload)}, allow_nan=False).encode()
+    raw = json.dumps({**header, "payload_bytes": len(view)}, allow_nan=False).encode()
     if len(raw) > 65536:
         raise ValueError("IPC header exceeds budget")
-    stream.write(struct.pack("<I", len(raw)) + raw + payload)
+    for part in (memoryview(struct.pack("<I", len(raw)) + raw), view):
+        while part:
+            count = stream.write(part)
+            if type(count) is not int or not 0 < count <= len(part):
+                raise EOFError("IPC write made invalid progress")
+            part = part[count:]
     stream.flush()

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -381,7 +381,7 @@ def run(args):
             zero = np.zeros((600, 800, 3), np.uint8)
             for _ in range(8):
                 if ddr:
-                    send(ddr.stdin, {"op": "preprocess"}, zero.tobytes())
+                    send(ddr.stdin, {"op": "preprocess"}, memoryview(zero))
                     _, raw = receive(ddr.stdout)
                     tensor = np.frombuffer(raw, "<f4").reshape(1, 3, 512, 512)
                 else:
@@ -441,12 +441,12 @@ def run(args):
             )
             if execute_ddrnet:
                 if args.ddrnet_runtime == "pytorch":
-                    send(ddr.stdin, {"op": "infer"}, bundle["image"].tobytes())
+                    send(ddr.stdin, {"op": "infer"}, memoryview(bundle["image"]))
                     ddr_result, raw = receive(ddr.stdout)
                     mask = np.frombuffer(raw, np.uint8).reshape(512, 512)
                 else:
                     if ddr:
-                        send(ddr.stdin, {"op": "preprocess"}, bundle["image"].tobytes())
+                        send(ddr.stdin, {"op": "preprocess"}, memoryview(bundle["image"]))
                         preprocess_result, raw = receive(ddr.stdout)
                         tensor = np.frombuffer(raw, "<f4").reshape(1, 3, 512, 512)
                     else:

tests/test_perception_joint_pilot.py

@@ -114,6 +114,43 @@ def test_ipc_preserves_binary_and_rejects_oversized_header(pilot):
         ipc.receive(io.BytesIO((65537).to_bytes(4, "little")))
 
 
+def test_ipc_borrows_payload_without_concatenation_and_handles_partial_writes(pilot):
+    ipc = pilot("pilot_ipc")
+    payload = np.arange(24, dtype=np.uint8).reshape(2, 4, 3)
+    calls, written = [], bytearray()
+
+    class Writer:
+        def write(self, value):
+            calls.append(value)
+            size = min(7, len(value))
+            written.extend(value[:size])
+            return size
+
+        def flush(self):
+            pass
+
+    ipc.send(Writer(), {"op": "infer"}, memoryview(payload))
+    body = payload.tobytes()
+    header = json.dumps({"op": "infer", "payload_bytes": len(body)}).encode()
+    assert written == len(header).to_bytes(4, "little") + header + body
+    assert any(value.obj is payload for value in calls)
+    assert all(isinstance(value, memoryview) for value in calls)
+    assert ipc.receive(io.BytesIO(written)) == ({"op": "infer"}, body)
+
+
+@pytest.mark.parametrize("result", [0, None, -1, True, 100000])
+def test_ipc_rejects_invalid_write_progress(pilot, result):
+    with pytest.raises(EOFError, match="progress"):
+        pilot("pilot_ipc").send(SimpleNamespace(write=lambda _: result), {}, b"x")
+
+
+def test_ipc_rejects_noncontiguous_payload_before_writing(pilot):
+    stream = io.BytesIO()
+    with pytest.raises(TypeError):
+        pilot("pilot_ipc").send(stream, {}, memoryview(np.arange(8, dtype=np.uint8)[::2]))
+    assert stream.getvalue() == b""
+
+
 def test_nearest_rank_tail_metrics_keep_outlier(pilot):
     metrics = pilot("run_joint_pilot").distribution([1] * 99 + [200])
     assert metrics["p99"] == 1