#!/usr/bin/env python3 """latency_peer.py - Link latency test peer for Phase 13 (M2). Server-driven protocol: the MCU (testTask, port 8080) sends one operation byte; the peer executes the corresponding traffic pattern. See docs/测试说明/专项验证/链路延迟测试说明.md §4.5 for the full protocol spec. Ops (MCU -> peer): 0x01 QUEUE_RT recv flood : peer sends 50 x 100B (5ms apart) 0x02 QUEUE_RT send drain : peer drains 50 x 100B 0x03 ECHO_RT : peer sends 10B request, expects 10B echo (timed) 0x04 FRAME_RT : peer sends 9B LCDMEM request, expects 3209B frame 0x05 CHUNK_WAIT send drain : peer drains 32 x 4096B 0xFF done : session complete The peer records host-side (TO_PC-side) timing: first byte / frame total / per-batch arrival curve / >=500ms spikes (TCP retransmission fingerprint). Both the serial log and this JSON carry HOST timestamps -> direct correlation. Usage: python latency_peer.py --port 8080 --json m2_net.json python latency_peer.py --mode net --ftp-stress ... (head-of-line repro) Pure stdlib (socket/json/time/argparse/threading). ASCII output only. """ import argparse import json import socket import sys import threading import time import bench_defaults # 台架连接默认值单一真源 # --- Protocol constants (mirror test/board/net_latency_test.c Phase 13 defines) --- HELLO = b"\x5a\xa5\x58\x01" OP_QUEUE_RT_RECV = 0x01 OP_QUEUE_RT_SEND = 0x02 OP_ECHO_RT = 0x03 OP_FRAME_RT = 0x04 OP_CHUNK_WAIT = 0x05 OP_DONE = 0xFF QUEUE_RT_CHUNKS = 20 # 2026-09-22 run1: 50 chunks saturated the chip RX buffer # (window close -> retransmit re-delivery -> RST); 20 @ 50ms # = 1KB/s quiet baseline (see 链路延迟测试说明.md §6.2) QUEUE_RT_CHUNK_LEN = 100 ECHO_RT_SIZE = 10 FRAME_RT_REQ_LEN = 9 FRAME_RT_FRAME_LEN = 3209 CHUNK_WAIT_SENDS = 16 # mirror firmware LAT_P13_CHUNK_WAIT_SENDS (run3: 32->16, 64 chunk samples) CHUNK_WAIT_LEN = 4096 SPIKE_MS = 500.0 # retransmission granularity fingerprint (D1) # --- Production RemoDispBus constants (mirror drivers/BSP/UI/ui_net_sink.h) --- REMO_TAG_CLIENT = 0xAA REMO_TAG_DEVICE = 0xBB REMO_CMD_KEEP = 0 REMO_CMD_INIT = 1 REMO_CMD_KEY = 2 REMO_CMD_LCDMEM = 3 REMO_MEMSIZE = 3200 # 160x160 1bpp REMO_LCDMEM_FRAME = 4 + 4 + REMO_MEMSIZE + 1 # 3209 (header + offset + bitmap + CRC) REMO_CYCLE_DEADLINE_S = 45.0 # per-cycle recv deadline (head-of-line stress: up to 30s) def now_ns(): return time.perf_counter_ns() def iso_now(): return time.strftime("%Y-%m-%d %H:%M:%S") def recv_exact(sock, n, deadline_ns): buf = bytearray() while len(buf) < n: left = deadline_ns - now_ns() if left <= 0: raise TimeoutError("recv_exact deadline (%d/%d bytes)" % (len(buf), n)) try: sock.settimeout(max(left, 1) / 1e9 / 5.0) d = sock.recv(n - len(buf)) except socket.timeout: continue if not d: raise ConnectionError("peer connection closed") buf.extend(d) return bytes(buf) def frame_rt_request(seq=0): """Production-shaped 9B LCDMEM request (ui_net_sink.c remo_handle): [TAG 0xAA][CMD 0x03][LEN=4 BE][offset u32 BE = 0][CRC = XOR of the 4 offset bytes]. LEN is the DATA-segment length (4), NOT the requested bitmap size; the offset field must be 0 (device only serves full-frame requests).""" data = bytes([0, 0, 0, 0]) crc = 0 for b in data: crc ^= b return bytes([0xAA, 0x03, 0x00, 0x04]) + data + bytes([crc]) def frame_rt_ok(data): """Validate the 3209B production LCDMEM response (ui_net_sink.c remo_send): [TAG 0xBB][CMD 0x03][LEN=3204 BE][offset u32 = 0][1bpp bitmap 3200B] [CRC = XOR of the 3204 data bytes] -> 4 + 3204 + 1 = 3209B.""" if len(data) != FRAME_RT_FRAME_LEN: return False if data[0] != 0xBB or data[1] != 0x03 or data[2] != 0x0C or data[3] != 0x84: return False payload = data[4:4 + 3204] crc = 0 for b in payload: crc ^= b if data[4 + 3204] != crc: return False if payload[:4] != bytes(4): return False for i in range(3200): if payload[4 + i] != (i & 0xFF): return False return True def pattern(n, seed=0): return bytes((seed + i) & 0xFF for i in range(n)) # ================= RemoDispBus production client (M3, :7003) ================= def xor_bytes(data): crc = 0 for b in data: crc ^= b return crc def remo_request(cmd, data=b""): """Client->device frame: [0xAA][CMD][LEN_H][LEN_L][DATA][CRC = XOR(DATA)].""" body = bytes(data) return (bytes([REMO_TAG_CLIENT, cmd, (len(body) >> 8) & 0xFF, len(body) & 0xFF]) + body + bytes([xor_bytes(body)])) def remo_lcdmem_request(): """9B full-frame LCDMEM request (offset 0) - what the production client sends.""" return remo_request(REMO_CMD_LCDMEM, bytes(4)) def recv_upto(sock, want, deadline_ns): """Bounded read: never consumes more than `want` bytes (no over-read of a following frame/op byte - the run10 lesson).""" left = deadline_ns - now_ns() if left <= 0: raise TimeoutError("frame deadline") try: sock.settimeout(max(left, 1) / 1e9 / 5.0) return sock.recv(want) except socket.timeout: return b"" def read_remo_frame(sock, deadline_ns, ack_probe=False): """Read exactly one device frame; returns tag/cmd/data/CRC + arrival curve. The first read is bounded to 8B so the segment-arrival curve stays fine-grained; subsequent reads are bounded by the remaining frame length (device never pipelines: one response per request). ack_probe=True: as soon as the frame is known to be incomplete, send one byte the device parser discards (0x00, not TAG 0xAA). The resulting TCP segment carries a piggybacked ACK for the data already received, which discriminates "sender holds the last partial segment waiting for an ACK (Nagle-style)" from "chip-internal fixed timer" - without any packet capture.""" buf = bytearray() batches = [] first_ns = None total = None while True: want = (8 - len(buf)) if total is None else (total - len(buf)) if want <= 0: break d = recv_upto(sock, min(want, 4096), deadline_ns) if not d: if total is not None: raise ConnectionError("frame truncated (%d/%d bytes)" % (len(buf), total)) continue # header still incomplete: wait until deadline t = now_ns() if first_ns is None: first_ns = t buf.extend(d) batches.append([t, len(d)]) if total is None and len(buf) >= 4: total = 4 + (((buf[2] << 8) | buf[3])) + 1 if ack_probe and total is not None and len(buf) < total: try: sock.sendall(b"\x00") # parser-safe: discarded while waiting for TAG 0xAA except OSError: pass data = bytes(buf[4:total - 1]) crc = buf[total - 1] return {"tag": buf[0], "cmd": buf[1], "data": data, "total": total, "crc_ok": crc == xor_bytes(data), "first_ns": first_ns, "batches": batches} def enable_immediate_ack(sock): """Windows: disable delayed ACK on this socket (SIO_TCP_SET_ACK_FREQUENCY=1). Root-cause fix for the ~50ms tail: the CH395F (MSS=1024) holds the final sub-MSS segment of a 3209B frame per Nagle until the peer ACKs; Windows' delayed ACK defers that ACK ~50ms. Setting ACK frequency=1 makes every segment ACKed immediately -> tail 52ms -> ~4ms (measured). Returns True on success. Other platforms: no-op (Linux uses TCP_QUICKACK per recv).""" if sys.platform != "win32": return False try: import ctypes ws2 = ctypes.WinDLL("ws2_32", use_last_error=True) WSAIoctl = ws2.WSAIoctl WSAIoctl.argtypes = [ctypes.c_void_p, ctypes.c_uint32, ctypes.c_void_p, ctypes.c_uint32, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(ctypes.c_uint32), ctypes.c_void_p, ctypes.c_void_p] WSAIoctl.restype = ctypes.c_int val = ctypes.c_uint32(1) # 1 = ACK every segment ret = ctypes.c_uint32(0) rc = WSAIoctl(ctypes.c_void_p(sock.fileno()), 0x98000017, ctypes.byref(val), 4, None, 0, ctypes.byref(ret), None, None) return rc == 0 except (OSError, AttributeError): return False def remo_bitmap_nontrivial(payload): """Same 'non-trivial frame' criterion as simulator/tests/check_remo_sink.py: the 1bpp bitmap must not be a single repeated byte (blank/broken frame).""" bmp = payload[4:4 + REMO_MEMSIZE] return len(set(bmp)) > 2 class RemoPeer: """Production-path peer (M3): drives PROD_CYCLE/RELISTEN against :7003 (device) or the simulator, measuring the full client-observed frame cycle.""" def __init__(self, host, port, wait_s, json_path, samples, cadence_ms, stress_from, nodelay, reconnect_rounds=1, ack_probe=False, ack_fast=False): self.host = host self.port = port self.wait_s = wait_s self.json_path = json_path self.samples = samples self.cadence_s = cadence_ms / 1000.0 self.stress_from = stress_from # cycle index where FTP stress starts self.nodelay = nodelay self.reconnect_rounds = max(1, reconnect_rounds) self.ack_probe = ack_probe # in-band ACK-release experiment (tail diagnosis) self.ack_fast = ack_fast # Windows immediate-ACK socket option self.ack_fast_ok = None self.sock = None self.t0_wall = None self.t0_ns = 0 self.connect_ms = None self.reconnect_ms = None self.reconnect_list = [] self.init_rtt_ms = None self.geom = {} self.frames = [] # --- connection --- def connect_once(self): t = now_ns() s = socket.create_connection((self.host, self.port), timeout=5.0) if self.nodelay: s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if self.ack_fast: self.ack_fast_ok = enable_immediate_ack(s) print("[peer] immediate-ACK (SIO_TCP_SET_ACK_FREQUENCY=1): %s" % ("applied" if self.ack_fast_ok else "not available")) self.sock = s return (now_ns() - t) / 1e6 def connect_with_retry(self): t_end = time.time() + self.wait_s last = None while time.time() < t_end: try: return self.connect_once() except OSError as e: last = e time.sleep(1.0) raise ConnectionError("cannot connect to %s:%d within %ds (%s)" % (self.host, self.port, self.wait_s, last)) def init_roundtrip(self, deadline_s=5.0): """INIT request -> geometry reply; returns round-trip ms.""" t = now_ns() self.sock.sendall(remo_request(REMO_CMD_INIT)) f = read_remo_frame(self.sock, t + int(deadline_s * 1e9)) ms = (now_ns() - t) / 1e6 if f["cmd"] == REMO_CMD_INIT and len(f["data"]) >= 6: # 8B (fixed): [W u16][H u16][MEMSIZE u32 BE]; 6B = 旧固件（u32 被截半 → 0） self.geom = {"w": (f["data"][0] << 8) | f["data"][1], "h": (f["data"][2] << 8) | f["data"][3], "framebytes": int.from_bytes(f["data"][4:8], "big") if len(f["data"]) >= 8 else ((f["data"][4] << 8) | f["data"][5]), "payload_len": len(f["data"])} return ms # --- FTP head-of-line stress --- def ftp_stress_worker(self, start_evt, stop_evt, hold_s): """Wait for the stress window to open, then hold an idle FTP login: lftpd read_line -> blocking net_recv(0) inside netTask (old 30s default budget NET_RECV_TIMEOUT_MS, removed in self-heal phase 1.5; today the wait is a CHUNK_WAIT pending recv -> this scenario must NOT freeze the queue anymore) -> every queued message (incl. the LCDMEM response) waits behind it.""" if not start_evt.wait(120.0): sys.stdout.write("[peer] ftp stress: window never opened\n") return try: s = socket.create_connection((self.host, 21), timeout=10.0) s.settimeout(10.0) s.sendall(b"USER latency\r\n") reply = b"" while b"\r\n" not in reply: d = s.recv(64) if not d: break reply += d sys.stdout.write("[peer] ftp stress armed (331: %r)\n" % reply[:40]) stop_evt.wait(hold_s) try: s.sendall(b"QUIT\r\n") except OSError: pass s.close() sys.stdout.write("[peer] ftp stress released\n") except OSError as e: sys.stdout.write("[peer] ftp stress FAILED: %s\n" % e) # --- session --- def one_session(self): self.t0_wall = iso_now() self.t0_ns = now_ns() self.connect_ms = self.connect_with_retry() print("[peer] connected %s:%d in %.1f ms" % (self.host, self.port, self.connect_ms)) self.init_rtt_ms = self.init_roundtrip() print("[peer] INIT rtt=%.1f ms geom=%s" % (self.init_rtt_ms, self.geom)) stress_start = threading.Event() stress_stop = threading.Event() stress_thread = None if self.stress_from is not None: # hold long enough to cover the remaining cycles (+15s margin) hold_s = self.cadence_s * max(1, self.samples - self.stress_from) + 15.0 stress_thread = threading.Thread( target=self.ftp_stress_worker, args=(stress_start, stress_stop, hold_s), daemon=True) stress_thread.start() for i in range(self.samples): if self.stress_from is not None and i == self.stress_from: print("[peer] === FTP stress ON (head-of-line) at cycle %d ===" % i) stress_start.set() time.sleep(0.5) # let the FTP half-login block netTask stressed = (self.stress_from is not None and i >= self.stress_from) t = now_ns() self.sock.sendall(remo_lcdmem_request()) f = read_remo_frame(self.sock, t + int(REMO_CYCLE_DEADLINE_S * 1e9), ack_probe=self.ack_probe) t_end = now_ns() first_ms = (f["first_ns"] - t) / 1e6 total_ms = (t_end - t) / 1e6 tail_ms = (f["batches"][-1][0] - f["batches"][0][0]) / 1e6 if len(f["batches"]) > 1 else 0.0 rec = { "cycle": i, "stress": stressed, "first_ms": round(first_ms, 3), "total_ms": round(total_ms, 3), "tail_lag_ms": round(tail_ms, 3), "batches": [[round((b[0] - t) / 1e6, 3), b[1]] for b in f["batches"]], "bytes": f["total"], "tag": f["tag"], "cmd": f["cmd"], "crc_ok": bool(f["crc_ok"]), "nontrivial": bool(f["cmd"] == REMO_CMD_LCDMEM and len(f["data"]) == 4 + REMO_MEMSIZE and remo_bitmap_nontrivial(f["data"])), } self.frames.append(rec) if (i % 10) == 0 or stressed: print("[peer] cycle %3d %-7s first=%7.1f total=%8.1f ms batches=%d %s" % (i, "STRESS" if stressed else "quiet", first_ms, total_ms, len(f["batches"]), "OK" if (rec["crc_ok"] and rec["nontrivial"]) else "BAD")) if i + 1 < self.samples: time.sleep(self.cadence_s) if stress_thread is not None: stress_stop.set() self.dump_json() # --- RELISTEN: close + reconnect rounds (each close drives the device auto-relisten # probe; each reconnect measures connect + INIT round trip) --- for r in range(self.reconnect_rounds): self.sock.close() print("[peer] round %d/%d: closed; reconnect in 1.5s (device relisten window) ..." % (r + 1, self.reconnect_rounds)) time.sleep(1.5) t = now_ns() self.connect_once() self.init_rtt_ms = self.init_roundtrip() ms = (now_ns() - t) / 1e6 self.reconnect_list.append(round(ms, 3)) self.reconnect_ms = ms print("[peer] round %d: reconnect + INIT = %.1f ms" % (r + 1, ms)) rep = self.dump_json() self.print_summary(rep) self.sock.close() # --- reporting --- def _split(self, key, stressed): return [f[key] for f in self.frames if f["stress"] is stressed] def _spikes(self, vals): return sum(1 for x in vals if x >= SPIKE_MS) def dump_json(self, error=None): all_total = [f["total_ms"] for f in self.frames] all_first = [f["first_ms"] for f in self.frames] all_tail = [f["tail_lag_ms"] for f in self.frames] quiet_total = self._split("total_ms", False) stress_total = self._split("total_ms", True) rep = { "mode": "remo", "host": self.host, "port": self.port, "ts_start": self.t0_wall, "ts_end": iso_now(), "duration_s": round((now_ns() - self.t0_ns) / 1e9, 2), "samples": self.samples, "cadence_ms": round(self.cadence_s * 1000.0, 1), "tcp_nodelay": bool(self.nodelay), "immediate_ack": (None if not self.ack_fast else bool(self.ack_fast_ok)), "in_band_ack_probe": bool(self.ack_probe), "stress_from": self.stress_from, "connect_ms": None if self.connect_ms is None else round(self.connect_ms, 3), "reconnect_ms": None if self.reconnect_ms is None else round(self.reconnect_ms, 3), "reconnect_list_ms": self.reconnect_list, "reconnect_stat": stat_block(self.reconnect_list), "init": {"rtt_ms": None if self.init_rtt_ms is None else round(self.init_rtt_ms, 3), "geom": self.geom}, "cycle_all": stat_block(all_total, self._spikes(all_total)), "cycle_quiet": stat_block(quiet_total, self._spikes(quiet_total)), "cycle_stress": stat_block(stress_total, self._spikes(stress_total)), "first_byte_all": stat_block(all_first, self._spikes(all_first)), "tail_lag_all": stat_block(all_tail, self._spikes(all_tail)), "crc": {"ok": sum(1 for f in self.frames if f["crc_ok"]), "fail": sum(1 for f in self.frames if not f["crc_ok"])}, "bitmap": {"nontrivial_ok": sum(1 for f in self.frames if f["nontrivial"]), "nontrivial_fail": sum(1 for f in self.frames if not f["nontrivial"])}, "frames": self.frames, } if error: rep["error"] = error if self.json_path: with open(self.json_path, "w", encoding="utf-8") as fh: json.dump(rep, fh, ensure_ascii=False, indent=1) print("[peer] JSON report -> %s%s" % (self.json_path, " (PARTIAL - error)" if error else "")) return rep def print_summary(self, rep): print("=== latency_peer summary (M3 / Phase 14, production RemoDispBus) ===") print("host=%s:%d samples=%d cadence=%.0fms nodelay=%d duration=%.1fs" % (self.host, self.port, rep["samples"], rep["cadence_ms"], int(rep["tcp_nodelay"]), rep["duration_s"])) print("connect=%s ms reconnect+INIT=%s ms geom=%s" % (rep["connect_ms"], rep["reconnect_ms"], rep["init"]["geom"])) for key in ("cycle_all", "cycle_quiet", "cycle_stress", "first_byte_all", "tail_lag_all"): b = rep[key] if b.get("n"): print("%-15s n=%-4d min=%9.1f med=%9.1f p95=%9.1f max=%9.1f ms spikes=%d" % (key, b["n"], b["min_ms"], b["med_ms"], b["p95_ms"], b["max_ms"], b.get("spikes500", 0))) else: print("%-15s (no samples)" % key) print("crc ok/fail = %d/%d bitmap nontrivial ok/fail = %d/%d" % (rep["crc"]["ok"], rep["crc"]["fail"], rep["bitmap"]["nontrivial_ok"], rep["bitmap"]["nontrivial_fail"])) def run(self): try: self.one_session() except (ConnectionError, TimeoutError, RuntimeError, OSError) as e: self.dump_json(error="%s: %s" % (type(e).__name__, e)) raise def percentile(sorted_vals, pct): if not sorted_vals: return None idx = (len(sorted_vals) * pct) // 100 if idx >= len(sorted_vals): idx = len(sorted_vals) - 1 return sorted_vals[idx] def stat_block(ms_vals, spikes=0): if not ms_vals: return {"n": 0} s = sorted(ms_vals) return { "n": len(s), "min_ms": round(s[0], 3), "med_ms": round(s[len(s) // 2], 3), "p95_ms": round(percentile(s, 95), 3), "max_ms": round(s[-1], 3), "spikes500": spikes, } def fresh_reports(): """Per-session measurement containers (reset on each retry attempt).""" return { "l6": {"samples": [], "echo_ok": 0, "echo_fail": 0}, "l7": {"frames": [], "first_ms": [], "total_ms": [], "spikes500": 0, "crc_ok": 0, "crc_fail": 0, "batches_per_frame": []}, "l5_recv": {"bytes": 0, "window_ms": None}, "l5_send": {"bytes": 0, "window_ms": None}, "l3": {"bytes": 0, "window_ms": None}, } class Peer: def __init__(self, host, port, wait_s, json_path, ftp_stress): self.host = host self.port = port self.wait_s = wait_s self.connect_wait_s = wait_s # per-attempt connect budget (retries shrink it) self.json_path = json_path self.ftp_stress = ftp_stress self.sock = None self.t0_wall = None self.t0_ns = 0 self.n_ops = 0 self.reports = fresh_reports() def connect_with_retry(self): t_end = time.time() + self.connect_wait_s while time.time() < t_end: try: s = socket.create_connection((self.host, self.port), timeout=5.0) s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) self.sock = s return except OSError: time.sleep(1.0) raise ConnectionError("cannot connect to %s:%d within %ds" % (self.host, self.port, self.wait_s)) def ftp_stress_worker(self, start_evt, stop_evt): """Hold an idle FTP login: lftpd read_line -> blocking net_recv(0) in netTask (30s timeout) -> all queued messages wait behind it.""" try: s = socket.create_connection((self.host, 21), timeout=10.0) s.settimeout(10.0) s.sendall(b"USER latency\r\n") reply = b"" while b"\r\n" not in reply: d = s.recv(64) if not d: break reply += d start_evt.set() # QUEUE_RT window starts under stress sys.stdout.write("[peer] ftp stress armed (331: %r)\n" % reply[:40]) stop_evt.wait(35.0) # > 30s: netTask read times out (full freeze) try: s.sendall(b"QUIT\r\n") except OSError: pass s.close() sys.stdout.write("[peer] ftp stress released\n") except OSError as e: sys.stdout.write("[peer] ftp stress FAILED: %s\n" % e) start_evt.set() def dump_json(self, error=None): """Build the report dict and write it (idempotent; error => partial).""" rep = { "mode": "net", "host": self.host, "port": self.port, "ts_start": self.t0_wall, "ts_end": iso_now(), "duration_s": round((now_ns() - self.t0_ns) / 1e9, 2), "ops_seen": self.n_ops, "ftp_stress": self.ftp_stress, "seg_l6": stat_block(self.reports["l6"]["samples"], 0) | {"echo_ok": self.reports["l6"]["echo_ok"], "echo_fail": self.reports["l6"]["echo_fail"]}, "seg_l7_first_byte": stat_block(self.reports["l7"]["first_ms"]), "seg_l7_total": stat_block(self.reports["l7"]["total_ms"], self.reports["l7"]["spikes500"]), "l7_crc": {"ok": self.reports["l7"]["crc_ok"], "fail": self.reports["l7"]["crc_fail"], "batches_per_frame": sorted(set(self.reports["l7"]["batches_per_frame"]))}, "l5_recv": self.reports["l5_recv"], "l5_send": self.reports["l5_send"], "l3_drain": self.reports["l3"], "frames_l7": self.reports["l7"]["frames"], } if error: rep["error"] = error if self.json_path: with open(self.json_path, "w", encoding="utf-8") as f: json.dump(rep, f, indent=1, ensure_ascii=True) print("[peer] JSON report -> %s%s" % (self.json_path, " (PARTIAL - error)" if error else "")) return rep def run(self, max_attempts=3): """整会话重试：芯片对空闲连接自行 RST（MCU 栈慢于芯片握手完成时，连接 建立后 15~30s 无流量即被芯片超时断开，2026-09-22 台架实证）→ 重连 + MCU 端 re-HELLO 重启 = 闭环。重试时缩短 connect 预算（设备已在运行）。""" last_err = None for attempt in range(1, max_attempts + 1): self.t0_wall = iso_now() self.t0_ns = now_ns() self.n_ops = 0 self.reports = fresh_reports() self.connect_wait_s = self.wait_s if attempt == 1 else min(self.wait_s, 60) try: self.one_session() return except (ConnectionError, TimeoutError, RuntimeError, OSError) as e: last_err = e self.close_quiet() if attempt < max_attempts: print("[peer] session attempt %d/%d failed: %s" % (attempt, max_attempts, e)) print("[peer] reconnect in 2s (MCU auto-relistens; re-HELLO restarts) ...") time.sleep(2.0) raise RuntimeError("all %d session attempts failed (last: %s)" % (max_attempts, last_err)) def close_quiet(self): try: if self.sock is not None: self.sock.close() except OSError: pass self.sock = None def one_session(self): t0_wall = self.t0_wall t0_ns = self.t0_ns print("[peer] connecting %s:%d (wait up to %ds) ..." % (self.host, self.port, self.connect_wait_s)) self.connect_with_retry() print("[peer] connected at %s" % iso_now()) queue_rt_start_evt = threading.Event() queue_rt_stop_evt = threading.Event() if self.ftp_stress: th = threading.Thread(target=self.ftp_stress_worker, args=(queue_rt_start_evt, queue_rt_stop_evt), daemon=True) th.start() self.sock.sendall(HELLO) print("[peer] HELLO sent, awaiting MCU ops ...") queue_rt_window_open = False queue_rt_t0 = 0 chunk_wait_t0 = 0 while True: op = recv_exact(self.sock, 1, t0_ns + 300 * 1000 * 1000000) op = op[0] self.n_ops += 1 if op == OP_DONE: print("[peer] op=done after %d ops" % self.n_ops) break if op == OP_QUEUE_RT_RECV: if self.ftp_stress: queue_rt_start_evt.wait(60.0) # start window only under full stress queue_rt_window_open = True queue_rt_t0 = now_ns() for i in range(QUEUE_RT_CHUNKS): self.sock.sendall(pattern(QUEUE_RT_CHUNK_LEN, seed=i)) time.sleep(0.05) # 50ms: keep 1KB/s below MCU drain rate self.reports["l5_recv"]["window_ms"] = (now_ns() - queue_rt_t0) / 1e6 self.reports["l5_recv"]["bytes"] = QUEUE_RT_CHUNKS * QUEUE_RT_CHUNK_LEN queue_rt_stop_evt.set() elif op == OP_QUEUE_RT_SEND: t = now_ns() data = recv_exact(self.sock, QUEUE_RT_CHUNKS * QUEUE_RT_CHUNK_LEN, t + 120 * 1000 * 1000000) self.reports["l5_send"]["bytes"] = len(data) self.reports["l5_send"]["window_ms"] = (now_ns() - t) / 1e6 elif op == OP_ECHO_RT: t0 = now_ns() req = pattern(ECHO_RT_SIZE, seed=self.n_ops) self.sock.sendall(req) resp = recv_exact(self.sock, ECHO_RT_SIZE, t0 + 40 * 1000 * 1000000) ms = (now_ns() - t0) / 1e6 self.reports["l6"]["samples"].append(ms) if resp == req: self.reports["l6"]["echo_ok"] += 1 else: self.reports["l6"]["echo_fail"] += 1 elif op == OP_FRAME_RT: t_send = now_ns() self.sock.sendall(frame_rt_request(self.n_ops)) got = bytearray() batches = [] first_ns = None while len(got) < FRAME_RT_FRAME_LEN: # bounded read: never consume the next op byte (ASCII 0x04) that # follows the frame in the stream (run10: unbounded recv ate it -> # 3210-byte "frame" -> length check failed -> peer closed) d = self.sock.recv(min(4096, FRAME_RT_FRAME_LEN - len(got))) if not d: raise ConnectionError("FRAME_RT frame truncated (%d/%d)" % (len(got), FRAME_RT_FRAME_LEN)) if first_ns is None: first_ns = now_ns() got.extend(d) batches.append([round((now_ns() - t_send) / 1e6, 3), len(d)]) t_end = now_ns() first_ms = (first_ns - t_send) / 1e6 if first_ns else None total_ms = (t_end - t_send) / 1e6 # inter-batch arrival gaps spikes = 0 for k in range(1, len(batches)): if batches[k][0] - batches[k - 1][0] >= SPIKE_MS: spikes += 1 self.reports["l7"]["first_ms"].append(first_ms if first_ms is not None else 0.0) self.reports["l7"]["total_ms"].append(total_ms) self.reports["l7"]["spikes500"] += spikes self.reports["l7"]["batches_per_frame"].append(len(batches)) if frame_rt_ok(bytes(got)): self.reports["l7"]["crc_ok"] += 1 else: self.reports["l7"]["crc_fail"] += 1 self.reports["l7"]["frames"].append( {"first_ms": round(first_ms, 3) if first_ms is not None else None, "total_ms": round(total_ms, 3), "batches": batches}) elif op == OP_CHUNK_WAIT: chunk_wait_t0 = now_ns() recv_exact(self.sock, CHUNK_WAIT_SENDS * CHUNK_WAIT_LEN, chunk_wait_t0 + 120 * 1000 * 1000000) self.reports["l3"]["bytes"] = CHUNK_WAIT_SENDS * CHUNK_WAIT_LEN self.reports["l3"]["window_ms"] = (now_ns() - chunk_wait_t0) / 1e6 else: raise RuntimeError("unknown op byte 0x%02X (protocol desync)" % op) # --- JSON report (also used on the error path: partial data kept) --- rep = self.dump_json() print("=== latency_peer summary (M2 / Phase 13) ===") print("ops=%d duration=%.1fs ftp_stress=%d" % (self.n_ops, rep["duration_s"], int(self.ftp_stress))) for key in ("seg_l6", "seg_l7_first_byte", "seg_l7_total"): b = rep[key] if b.get("n"): print("%-18s n=%-4d min=%9.1f med=%9.1f p95=%9.1f max=%9.1f ms spikes=%d" % (key, b["n"], b["min_ms"], b["med_ms"], b["p95_ms"], b["max_ms"], b.get("spikes500", 0))) else: print("%-18s (no samples)" % key) print("l7 crc ok/fail = %d/%d batches/frame = %s" % (rep["l7_crc"]["ok"], rep["l7_crc"]["fail"], rep["l7_crc"]["batches_per_frame"])) self.sock.close() def main(argv=None): ap = argparse.ArgumentParser( description="link latency test peer - M2 Phase 13 (net) / M3 Phase 14 (remo)") ap.add_argument("--mode", default="net", choices=["net", "remo"], help="net = Phase 13 test protocol :8080; " "remo = production RemoDispBus client :7003 (Phase 14 / simulator)") ap.add_argument("--host", default=bench_defaults.DEVICE_IP, help="device IP") ap.add_argument("--port", type=int, default=None, help="listen port (default: 8080 for net, 7003 for remo)") ap.add_argument("--wait", type=int, default=180, help="max seconds to retry connect (device boots after flash)") ap.add_argument("--json", default=None, help="write JSON report to this path") ap.add_argument("--ftp-stress", action="store_true", help="hold idle FTP login (head-of-line repro)") ap.add_argument("--samples", type=int, default=50, help="remo mode: LCDMEM frame cycles to measure (default 50)") ap.add_argument("--cadence-ms", type=int, default=100, help="remo mode: client poll cadence in ms (production client = 100)") ap.add_argument("--stress-from", type=int, default=None, help="remo mode: cycle index where FTP stress starts " "(default with --ftp-stress: half of --samples)") ap.add_argument("--nodelay", action="store_true", help="remo mode: set TCP_NODELAY on the client socket") ap.add_argument("--reconnect-rounds", type=int, default=1, help="remo mode: close+reconnect rounds at the end (each one drives " "the device auto-relisten probe; M3 uses 3)") ap.add_argument("--ack-probe", action="store_true", help="remo mode: when a frame is still incomplete, send one parser-safe " "byte so the TCP segment piggybacks an ACK (diagnoses whether the " "~50ms tail waits for an ACK; no packet capture needed)") ap.add_argument("--ack-fast", action="store_true", help="remo mode: Windows only - disable delayed ACK on this socket " "(SIO_TCP_SET_ACK_FREQUENCY=1); removes the ~50ms frame tail") args = ap.parse_args(argv) if args.mode == "remo": port = args.port if args.port else 7003 stress_from = args.stress_from if args.ftp_stress and stress_from is None: stress_from = max(1, args.samples // 2) peer = RemoPeer(args.host, port, args.wait, args.json, args.samples, args.cadence_ms, stress_from, args.nodelay, args.reconnect_rounds, args.ack_probe, args.ack_fast) else: peer = Peer(args.host, args.port or 8080, args.wait, args.json, args.ftp_stress) try: peer.run() return 0 except (ConnectionError, TimeoutError, RuntimeError, OSError) as e: print("[peer] ERROR: %s" % e) try: peer.dump_json(error=str(e)) # partial report: keep measured data except OSError as e2: print("[peer] (JSON write failed: %s)" % e2) return 1 if __name__ == "__main__": sys.exit(main())