super_agent_bot
fc0d3e4b5c
第一次上真机把台架的两处软肋照出来了，都不是被测代码的问题： - peer 死在 Windows 的 ICMP 10054：板子还没 bind 时那一下端口不可达会被记在套接句上， 下一次 recvfrom 才抛出来。本机 CPython 3.14.7 既没有 socket.SIO_UDP_CONNRESET 也不接受 它的数值形式，所以主力防线改成把 10054 当「这一轮没收到」继续重试，并在回路自检里加 冷端口探针——探针走脚本真正在用的 _recv_maybe，另拿裸套接句证明这台机器确实会回 ICMP， 否则那条判定就没有牙。 - 单步预算错位：板端每步等 25 s，peer 每步只等 8 s，一步超时后整条梯子全灭。现在 peer 每步都重发、单步预算 --step-budget 默认 35 s（小于 26 s 打警告）；板端 udp_bench_expect 按载荷第 0 字节的步号自愈，比本步小的旧包丢掉继续等（上限 UDP_BENCH_STALE_MAX = 8）， 比本步大的直接报出来。DBG_WARN 在本工程不存在、字符串字面量必须纯 ASCII，两处已改。 真机已拿到的证据：TC-NET-1601a/b/c（建 socket、bind :6001、net_udp_get_mode 回 SERVER）与 TC-NET-1602（服务端形态收到 20 B、addrlen/sin_family/源端口全对、逐字节回显、形态翻成 CLIENT）全绿——剥 8 字节信息头与源地址回填在真芯片上成立；②③④⑤ 因板子当晚两次失电 （19:19、21:0x，SWD No target connected + ping 不通）未测，指南 §12.6 与覆盖率口径第 4 轮 的状态条据此改写。产品态（宏回默认）重新构建 0 Error 0 Warning，Code 249,124 / RO 29,836 / RW 2,052 / ZI 149,996 与本阶段之前逐字节一致，阶段 16 在产品里被预处理干净。
144 lines
6.2 KiB
Python
144 lines
6.2 KiB
Python
"""PC peer 自身的回路自检：不需要板子，用假板子把 udp_queue_bench.py 的步骤表跑一遍。
|
||
|
||
为什么要有这个文件：阶段 16 的真机台架是一次性的——板子通电、烧好台架镜像、PC 起对端，
|
||
对端脚本自己要是有 bug（取端口、步骤表、拆包预期），一趟台架时间就白费了。这里在
|
||
127.0.0.1 上放一个「按板端阶段 16 同样规矩说话」的假板子，把对端脚本的判定逻辑先验一遍。
|
||
|
||
假板子复刻的三条规矩（与 test/board/net_test_task.c 阶段 16 一致）：
|
||
1. 收到的数据报原样回显给发送方（tag 字节不动，所以源地址/剥头判据留在板端断言）；
|
||
2. tag 0x04 这一步板端故意只用 64 字节缓冲去读 200 字节的数据报，所以只回 64 字节；
|
||
3. 回显超过芯片发送缓冲（1KB）时按 1024 拆成多个数据报发出（NET_SEND_CHUNK_MAX）。
|
||
|
||
用法（在本仓库根目录）：
|
||
python test/host/udp_queue_bench_loopback.py
|
||
退出码：0 = 对端脚本判定逻辑全绿；1 = 有问题，先别上台架。
|
||
"""
|
||
|
||
import os
|
||
import socket
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
if HERE not in sys.path:
|
||
sys.path.insert(0, HERE)
|
||
|
||
import udp_queue_bench as peer # noqa: E402
|
||
|
||
CHIP_SEND_BUF = 1024 # NET_SEND_CHUNK_MAX
|
||
BOARD_RX_CAP = 1600 # 板端 s_udp_buf 尺寸 UDP_BENCH_BUF_SIZE
|
||
TRUNCATE_TAG = 0x04 # 这一步板端只用 64 字节缓冲
|
||
TRUNCATE_CAP = 64
|
||
RUNOUT_S = 90.0
|
||
LISTEN_PORTS = (6001, 6002) # UDP_BENCH_PORT / UDP_BENCH_PORT2
|
||
|
||
|
||
def echo_one(sock, deadline, stop):
|
||
sock.settimeout(0.5)
|
||
while (not stop.is_set()) and (time.time() < deadline):
|
||
try:
|
||
data, addr = sock.recvfrom(4096)
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
return
|
||
if not data:
|
||
continue
|
||
tag = data[0]
|
||
cap = TRUNCATE_CAP if tag == TRUNCATE_TAG else BOARD_RX_CAP
|
||
echo = peer.payload(tag, min(len(data), cap))
|
||
print("[board] rx 0x{0:02X} n={1} on :{2} -> echo {3}B".format(
|
||
tag, len(data), sock.getsockname()[1], len(echo)), flush=True)
|
||
for off in range(0, len(echo), CHIP_SEND_BUF):
|
||
sock.sendto(echo[off:off + CHIP_SEND_BUF], addr)
|
||
|
||
|
||
def check_cold_port() -> bool:
|
||
"""对端脚本必须先扛住「板子还没把 socket 打开」这段时间——真机第一次上台架就死在这。
|
||
|
||
往一个确定没人监听的端口发一颗包，Windows 会收下 ICMP 端口不可达并记在套接句上，
|
||
下一次 recvfrom 就抛 WinError 10054。这里走**脚本真正在用的那条路**（Peer._recv_maybe），
|
||
要求它返回 None（= 这一轮没收到，继续重试），而不是把异常抛穿。裸套接句那一段只是
|
||
证明这台机器确实会回 ICMP——不然这个探针没有牙，别把它当唯一证据。
|
||
"""
|
||
dead = 6099 # 假板子只绑 6001/6002，这个端口确定空着
|
||
ok = False
|
||
p = peer.Peer("127.0.0.1", 1.0)
|
||
s = peer.new_peer_socket()
|
||
try:
|
||
s.settimeout(1.0)
|
||
s.sendto(b"\x00", ("127.0.0.1", dead))
|
||
try:
|
||
r = p._recv_maybe(s, 1.0)
|
||
if r is None:
|
||
ok = True
|
||
else:
|
||
print("[selftest] 冷端口探针：竟然收到了数据报，探针失效", flush=True)
|
||
except ConnectionResetError:
|
||
print("[selftest] 冷端口探针：Peer._recv_maybe 把 WinError 10054 抛穿了"
|
||
"（真机上对端会当场死在第一条用例）", flush=True)
|
||
finally:
|
||
s.close()
|
||
except OSError as exc:
|
||
print("[selftest] 冷端口探针建句/发送失败：{0}".format(exc), flush=True)
|
||
|
||
raw = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
try:
|
||
raw.settimeout(1.0)
|
||
raw.sendto(b"\x00", ("127.0.0.1", dead))
|
||
try:
|
||
raw.recvfrom(64)
|
||
print("[selftest] 裸套接句也收到了数据报（探针无牙，仅记录）", flush=True)
|
||
except ConnectionResetError:
|
||
print("[selftest] 裸套接句抛 WinError 10054 —— 这台机器确实会踩，"
|
||
"上面那条判定有牙", flush=True)
|
||
except socket.timeout:
|
||
print("[selftest] 裸套接句超时（本机没回 ICMP，探针无牙，仅记录）", flush=True)
|
||
except OSError as exc:
|
||
print("[selftest] 裸套接句探针失败：{0}".format(exc), flush=True)
|
||
finally:
|
||
raw.close()
|
||
return ok
|
||
|
||
|
||
def main() -> int:
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
if not check_cold_port():
|
||
print("=== UDP Peer Loopback Self-Test: FAIL（冷端口探针没过，先修 new_peer_socket）===",
|
||
flush=True)
|
||
return 1
|
||
deadline = time.time() + RUNOUT_S
|
||
stop = threading.Event()
|
||
for listen_port in LISTEN_PORTS:
|
||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
try:
|
||
s.bind(("127.0.0.1", listen_port))
|
||
except OSError as exc:
|
||
print("无法绑定 127.0.0.1:{0}：{1}（端口被占用？先腾出来再跑）".format(listen_port, exc),
|
||
file=sys.stderr)
|
||
return 2
|
||
threading.Thread(target=echo_one, args=(s, deadline, stop), daemon=True).start()
|
||
time.sleep(0.3)
|
||
|
||
# --late 必须落在板端 7000~25000 ms 判定窗内；这里取 9 秒，比默认 12 秒省一点台架外时间
|
||
# --step-budget 8：假板子秒回，用不到真机那 35 秒预算；真机台架千万别照抄这个值
|
||
#（真机上必须盖过板端单步 25 秒等待，否则一次超时就整条梯子错位）。
|
||
sys.argv = ["udp_queue_bench.py", "--board", "127.0.0.1",
|
||
"--first-wait", "15", "--timeout", "6", "--step-budget", "8",
|
||
"--probe-silence", "2.0", "--late", "9.0"]
|
||
try:
|
||
rc = peer.main()
|
||
finally:
|
||
stop.set()
|
||
if rc == 0:
|
||
print("=== UDP Peer Loopback Self-Test: PASS（对端脚本判定逻辑可用，可上台架）===", flush=True)
|
||
else:
|
||
print("=== UDP Peer Loopback Self-Test: FAIL（先修对端脚本，别浪费台架时间）===", flush=True)
|
||
return rc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|