wandering1
7efa58d9a1
分两段（用户裁定 2026-11-02）：先做地址迁移，再把同类字面量收敛成
单一真源，以后再改地址只动一处。
一、IP 迁移 192.168.1.100 → 192.168.31.100（gw 同步 → .31.1）
- 工厂默认：Drivers/BSP/NETCFG/net_cfg.c 的 NET_CFG_DEF_IP/_GW
- 注释/文档示例、主机与板级测试、台架 py 脚本、两份 sim_config.ini
- 字节序教学点同步换算 hex：0x6401A8C0 → 0x641FA8C0（内存
C0 A8 1F 64），旧版示例引文与历史 Trap 记录按冻结纪律不动
- UI 黄金基准：网口设置 4 态 + shot_info 共 5 张 .bin 由
check_content_pages.py --update 重刷（渲染值 IP 变了，属预期变更）
- 保留判定：客户点表引用（IEC104/继电保护规划）、芯片手册摘录、
日期冻结的变更记录行、已退役台架档案 MDK-ARM/retracted_bench_*
二、单一真源收口（A+B+C+D，用户批准全做）
- 新建 test/host/bench_defaults.py：台架 py 默认值唯一定义点
（DEVICE_IP/DEVICE_GW/PC_IP/NET_BCAST），22 个脚本改为引用；
MDK-ARM 两探针与 scripts/board_ui_nav.py 经 sys.path shim 引入
- test/board/net_test_task.h 新增 TEST_PC_PEER_IP，板测三处
PHASE*_REMOTE_IP 改引该宏（保留 -D 覆盖）；Phase1 断言改
memcmp 数组比较，数组各只出现一次
- simulator/tests/check_content_pages.py 从 net_cfg.c 正则解析
NET_CFG_DEF_* 推导期望值（解析失败即报错、不静默兜底），
脚本内台架 IP 字面量清零
- 装饰性注释去具体值（net_cfg.h/net_socket.h/net_socket.c/
net_socket_sim.c）；字节序教学 hex 对刻意保留为唯一写值点
- 刻意不动：主机单测夹具 IP（逐文件钉住的任意值是正确测试
实践）、文档正文 IP（那是文档内容本身）
顺带：scripts/gen_task_table.py 规范串清掉「，B5」——编号清退提交
72b4b094 改了受管区没重跑生成器，task-table 门禁因此一直红
验证：产品 @build 0/0；CH395F 套件开关构建 0/0 后还原产品态；
6 套主机单测 rc=0；content_pages（含宏推导路径）/菜单 8 态/
推流位比对全 PASS；bom-drift 99 无漂移；doc-links/coding-standard/
cppcheck 绿；py_compile 23 文件 + --help 冒烟（含 shim 绝对路径）
841 lines
36 KiB
Python
841 lines
36 KiB
Python
#!/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())
|