diag: 加邮箱楔死探针三件套, 移除 _repro_hang.py

_probe_wedge.py 逐环节定位死点: A 写 SM0 的 WKC → B 0x220 bit0 是否置位 → C SSC 是否清事件并置 SM1 就绪 → D SM1 是否滞留旧响应, 用于区分 ESC 硬件楔死与 SSC 软件死循环。 _probe_wedge2.py 按真实邮箱协议走: 查写邮箱空 → 写满 256B → 验事件/响应。 _probe_stuck.py 验"SM1 响应滞留 → 邮箱死锁"假说: 复现失败 → 手动读走 SM1 清 bit3 → 再读是否恢复。若恢复则主站侧可自愈。 _repro_hang.py 的闪退成因已定位(见 a1ec486, config_map 撞楔死邮箱), 且规避已内建到 op/start 入口, 该复现脚本失去用途, 删除。 README 同步脚本清单, 并记下 SM = 0x0800 + 8n 这个易错点。 .gitignore 排除 _analyze_log.py(硬编码已失效的临时日志路径)。
2026-09-03 02:36:07 +08:00
parent a1ec486c74
commit fc9e8f34e5
6 changed files with 283 additions and 79 deletions
@@ -15,6 +15,9 @@ env/
 *.pcapng
 sniff_*.txt

+# 一次性日志分析脚本(硬编码临时路径, 无留存价值)
+_analyze_log.py
+
 # 编辑器 / 系统
 .vscode/
 .idea/
@@ -25,7 +25,7 @@ venv\Scripts\python.exe -m pip install -r requirements.txt
 | `ecat_esc_regs.py` | 从**网络侧**直读 LAN9252 ESC 寄存器，绕开 GD32/PDI，用于判断 EEPROM 加载是否真的失败；支持 `--reload` |
 | `ecat_flash_eeprom.py` | 通过网线烧写 `gen_eeprom.py` 生成的 SII 镜像，逐字回读校验（烧完需断电重启让 ESC 重新加载） |
 | `ecat_sniff.py` | 旁路抓包，hexdump SOEM 自己发出的帧，用于对照裸帧格式 |
-| `_repro_hang.py` | 复现 op/start 卡住→闪退的最小脚本，带 faulthandler 看门狗 |
+| `_probe_wedge.py` | 邮箱楔死现场逐环节定位：A 写 SM0 的 WKC → B `0x220` bit0 是否置位 → C SSC 是否清事件并置 SM1 就绪 → D SM1 是否滞留旧响应，用于区分 ESC 硬件楔死与 SSC 软件死循环 |

 ## 启动方式

@@ -39,5 +39,7 @@ venv\Scripts\python.exe -m pip install -r requirements.txt

 ## 注意

+- ESC 的 SyncManager 寄存器是 `0x0800 + 8n`：邮箱是 SM0=`0x0800`、SM1=`0x0808`，
+  `0x0810`/`0x0818` 是 SM2/SM3（过程数据）。写诊断代码时别混用，否则抓到的现场会误导判断。
 - 前端 CSS/JS 改动后记得 bump `?v=` 缓存版本，否则浏览器会加载旧代码。
 - EEPROM 烧写后必须断电冷启动，ESC 才会重新加载配置。
@@ -0,0 +1,77 @@
+"""_probe_stuck.py — 验证"SM1 响应滞留 → 邮箱死锁"假说。
+板上现状: SM1 挂着一个未读走的响应(上一探针注入)。实验:
+  1. SOEM SDO 读 → 预期失败(复现楔死现象)
+  2. 手动读走 SM1 全长(清 bit3) → 模拟"主站把滞留响应读掉"
+  3. SOEM SDO 再读 → 是否恢复?
+若恢复 = 根治方向明确:邮箱死锁的物理本质是响应滞留,主站侧可自愈,从站侧可加超时复位。
+"""
+import struct
+import sys
+
+import ecat_server as es
+import pysoem
+
+r = es.api_open({"substr": "I226"})
+if not r.get("ok"):
+    sys.exit(1)
+m = es.g_master
+s = m.slaves[0]
+
+
+def fprd(addr, size):
+    try:
+        return bytes(s._fprd(addr, size, 20000))
+    except Exception:
+        return None
+
+
+def sm1_state(tag):
+    d = fprd(0x0808, 8)
+    if d is None:
+        print(f"[{tag}] SM1 读失败", flush=True)
+        return None
+    start, length = struct.unpack("<HH", d[:4])
+    st = d[5]
+    print(f"[{tag}] SM1 st=0x{st:02X}(bit3={st >> 3 & 1} 读邮箱满)", flush=True)
+    return {"start": start, "len": length, "st": st}
+
+
+m.state = pysoem.INIT_STATE | pysoem.STATE_ACK
+m.write_state()
+m.state_check(pysoem.INIT_STATE, 1_000_000)
+m.config_init()
+m.state = pysoem.PREOP_STATE | pysoem.STATE_ACK
+m.write_state()
+print("PREOP:", "OK" if m.state_check(pysoem.PREOP_STATE, 1_000_000) == pysoem.PREOP_STATE else "FAIL", flush=True)
+
+# --- 1. 现状: SM1 挂着滞留响应 ---
+sm1_state("实验前")
+
+print("\n--- SOEM SDO #1(预期失败) ---", flush=True)
+try:
+    d = bytes(s.sdo_read(0x1000, 0, 8))
+    print(f"  成功?! = {d.hex().upper()}", flush=True)
+except Exception as e:
+    print(f"  失败: {type(e).__name__}: {e}", flush=True)
+
+sm1_state("SDO#1 后")
+
+# --- 2. 读走 SM1 滞留响应(全长度读取,读到末字节才清 bit3) ---
+sm1 = sm1_state("读滞留前")
+if sm1 and (sm1["st"] & 0x08):
+    d = fprd(sm1["start"], sm1["len"])
+    print(f"  读走 SM1 {sm1['len']}B: {d[:16].hex().upper()}...", flush=True)
+    sm1_state("读走后")
+
+# --- 3. SOEM SDO 再试 ---
+print("\n--- SOEM SDO #2(读走滞留后) ---", flush=True)
+try:
+    d = bytes(s.sdo_read(0x1000, 0, 8))
+    print(f"  成功! = {d.hex().upper()}", flush=True)
+    print(">>> 假说成立:响应滞留是死锁根源,读走即恢复!", flush=True)
+except Exception as e:
+    print(f"  仍失败: {type(e).__name__}: {e}", flush=True)
+    sm1_state("SDO#2 后")
+
+es.api_close({})
+print("DONE", flush=True)
@@ -0,0 +1,107 @@
+"""_probe_wedge.py — 楔死现场逐环节定位(纠正之前 0x0810/0x0818=SM2/SM3 的标号错误,真邮箱 SM 是 0x0800/0x0808)。
+
+环节定位:
+  A. FPWR 写 SM0 RAM 的 WKC —— 写都进不去 = SM0 硬件层被拒(disabled/错址)
+  B. 写入后 0x220 bit0(SM0 写事件)是否置位 —— ESC 没置事件 = ESC 硬件楔死
+  C. 200ms 后 bit0 是否被 SSC 清掉 + SM1 status bit3(读邮箱满)是否置位 —— 没处理 = SSC 软件死了
+  D. SM1 RAM 里有没有滞留的旧响应(bSendMbxIsFull 死锁证据)
+"""
+import struct
+import sys
+import time
+
+import ecat_server as es
+import pysoem
+
+r = es.api_open({"substr": "I226"})
+if not r.get("ok"):
+    print("open 失败,板子可能已断电", flush=True)
+    sys.exit(1)
+m = es.g_master
+s = m.slaves[0]
+
+
+def fprd(addr, size):
+    try:
+        return bytes(s._fprd(addr, size, 20000))
+    except Exception:
+        return None
+
+
+def dump(tag):
+    al = fprd(0x0130, 8)
+    ev = fprd(0x0220, 2)
+    am = fprd(0x0204, 2)
+    sm0 = fprd(0x0800, 8)
+    sm1 = fprd(0x0808, 8)
+    def fmt(d):
+        if d is None:
+            return "ERR"
+        start, length = struct.unpack("<HH", d[:4])
+        return f"start=0x{start:04X} len={length} ctl=0x{d[4]:02X} st=0x{d[5]:02X} act=0x{d[6]:02X} pdi=0x{d[7]:02X}"
+    al_s = f"AL=0x{struct.unpack('<H', al[:2])[0]:04X} code=0x{struct.unpack('<H', al[4:6])[0]:04X}" if al else "AL=ERR"
+    ev_v = struct.unpack("<H", ev)[0] if ev else -1
+    am_v = struct.unpack("<H", am)[0] if am else -1
+    print(f"[{tag}] {al_s} 0x220=0x{ev_v:04X}(bit0={ev_v & 1}bit1={(ev_v >> 1) & 1}) 0x204=0x{am_v:04X}", flush=True)
+    print(f"        SM0(0x0800): {fmt(sm0)}", flush=True)
+    print(f"        SM1(0x0808): {fmt(sm1)}", flush=True)
+    return sm0, sm1, ev_v
+
+
+# --- 阶段 1: PREOP 下探(邮箱 SM 应处于使能态) ---
+m.state = pysoem.INIT_STATE | pysoem.STATE_ACK
+m.write_state()
+m.state_check(pysoem.INIT_STATE, 1_000_000)
+m.config_init()
+m.state = pysoem.PREOP_STATE | pysoem.STATE_ACK
+m.write_state()
+st = m.state_check(pysoem.PREOP_STATE, 1_000_000)
+print(f"PREOP state_check = 0x{st & 0x0F:02X}", flush=True)
+sm0, sm1, ev = dump("PREOP 稳态")
+
+if sm0 is None or sm1 is None:
+    print("SM0/SM1 读取失败", flush=True)
+    es.api_close({})
+    sys.exit(1)
+
+# SM0/SM1 RAM 内容(滞留响应证据)
+mbx0_addr, mbx0_len = struct.unpack("<HH", sm0[:4])
+mbx1_addr, mbx1_len = struct.unpack("<HH", sm1[:4])
+print(f"邮箱 RAM: SM0@0x{mbx0_addr:04X}({mbx0_len}B) SM1@0x{mbx1_addr:04X}({mbx1_len}B)", flush=True)
+for name, addr, ln in (("SM0", mbx0_addr, mbx0_len), ("SM1", mbx1_addr, mbx1_len)):
+    d = fprd(addr, 16)
+    if d:
+        print(f"        {name} RAM[0:16] = {d.hex().upper()}", flush=True)
+
+# --- 阶段 2: 手动注入一个合法 CoE SDO 读请求(0x1000:00 expedited upload) ---
+# 邮箱帧: len=0x000A addr=0x0000 pri=0 type=0x03(CoE, counter=0) | CANOpen=0x1000+8(UP_REQ) idx=0x1000 sub=0
+req = struct.pack("<HHBB", 10, 0, 0, 0x03) + struct.pack("<HBBB", 0x1000, 0x100 & 0xFF, 0x00, 0x00)[:5]
+req = bytes([0x0A, 0x00, 0x00, 0x00, 0x00, 0x03, 0x10, 0x00, 0x00, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
+print(f"\n--- 阶段A: FPWR 写 SM0 RAM(0x{mbx0_addr:04X}) {req.hex().upper()} ---", flush=True)
+try:
+    s._fpwr(mbx0_addr, req)
+    print("  写入 OK (WKC=1) —— SM0 硬件接受写入", flush=True)
+except Exception as e:
+    print(f"  写入失败: {type(e).__name__}: {e} —— SM0 硬件拒绝(环节A死点)", flush=True)
+
+ev2 = fprd(0x0220, 2)
+if ev2:
+    v = struct.unpack("<H", ev2)[0]
+    print(f"  阶段B: 0x220=0x{v:04X} bit0(SM0写事件)={v & 1} —— {'已置位' if v & 1 else '未置位!ESC 硬件楔死'}", flush=True)
+
+# --- 阶段 3: 给 SSC 200ms 处理时间 ---
+time.sleep(0.2)
+sm0b, sm1b, ev3 = dump("注入请求 200ms 后")
+sm1_st = sm1b[5] if sm1b else 0
+print(f"  阶段C: 0x220 bit0={ev3 & 1}(应=0,SSC 已读走) SM1 st=0x{sm1_st:02X}(bit3={sm1_st >> 3 & 1},应=1,响应就绪)", flush=True)
+if ev3 is not None and (ev3 & 1):
+    print("  >>> bit0 仍挂起:SSC 轮询没处理邮箱(环节C死点:MainLoop/ECAT_Main 邮箱分支死了)", flush=True)
+elif sm1_st & 0x08:
+    d = fprd(mbx1_addr, 16)
+    print(f"  阶段D: SM1 响应 RAM = {d.hex().upper() if d else 'ERR'}", flush=True)
+    print("  >>> SSC 活着!邮箱处理正常 —— 楔死只在特定条件下", flush=True)
+else:
+    print("  >>> bit0 被清了但 SM1 无响应:SSC 收到了但没回(处理链中途死)", flush=True)
+
+es.api_close({})
+print("DONE", flush=True)
@@ -0,0 +1,93 @@
+"""_probe_wedge2.py — 按真实邮箱协议探针: 检查写邮箱空 → 写满 256B → 查事件/响应。"""
+import struct
+import sys
+import time
+
+import ecat_server as es
+import pysoem
+
+r = es.api_open({"substr": "I226"})
+if not r.get("ok"):
+    sys.exit(1)
+m = es.g_master
+s = m.slaves[0]
+
+
+def fprd(addr, size):
+    try:
+        return bytes(s._fprd(addr, size, 20000))
+    except Exception:
+        return None
+
+
+def sm_reg(addr):
+    d = fprd(addr, 8)
+    if d is None:
+        return None
+    start, length = struct.unpack("<HH", d[:4])
+    return {"start": start, "len": length, "ctl": d[4], "st": d[5], "act": d[6],
+            "raw": d.hex().upper()}
+
+
+def show(tag):
+    sm0, sm1 = sm_reg(0x0800), sm_reg(0x0808)
+    ev = fprd(0x0220, 2)
+    al = fprd(0x0130, 8)
+    ev_v = struct.unpack("<H", ev)[0] if ev else -1
+    al_v = struct.unpack("<H", al[:2])[0] if al else -1
+    print(f"[{tag}] AL=0x{al_v:02X} 0x220=0x{ev_v:04X} SM0[st=0x{sm0['st']:02X} act=0x{sm0['act']:02X}] "
+          f"SM1[st=0x{sm1['st']:02X} act=0x{sm1['act']:02X}] (SM0.bit3={sm0['st'] >> 3 & 1}写满 SM1.bit3={sm1['st'] >> 3 & 1}读满)",
+          flush=True)
+    return sm0, sm1, ev_v
+
+
+m.state = pysoem.INIT_STATE | pysoem.STATE_ACK
+m.write_state()
+m.state_check(pysoem.INIT_STATE, 1_000_000)
+m.config_init()
+m.state = pysoem.PREOP_STATE | pysoem.STATE_ACK
+m.write_state()
+print("PREOP:", "OK" if m.state_check(pysoem.PREOP_STATE, 1_000_000) == pysoem.PREOP_STATE else "FAIL", flush=True)
+
+sm0, sm1, ev = show("稳态")
+
+# 合法 CoE SDO 读 0x1000:00(expedited), padded 到整个邮箱长度 256B
+req_hdr = bytes([0x0A, 0x00, 0x00, 0x00, 0x00, 0x03,  # len=10, addr=0, pri=0, type=CoE(cnt=0)
+                0x10, 0x00,  # CANOpen = 0x0010 (SDO req<<12 | 0x1000? 不对, 服务号 0x10?) —— 用标准 SOEM 格式
+                0x00, 0x10,  # Command = 0x40(up req, exp)
+                0x00, 0x10,  # Index = 0x1000
+                0x00, 0x00]) # sub=0, +pad
+# SOEM ecx_SDOread 的帧: CANOpen = (ECT_COES_SDOREQ<<12)=0x1000 | number; Command=0x40; Index; SubIndex
+req = bytes([0x0A, 0x00, 0x00, 0x00, 0x00, 0x03,
+             0x00, 0x10, 0x40, 0x00, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00])
+mbx = (req + bytes(sm0["len"]))[:sm0["len"]]
+print(f"\n写满 SM0 RAM {sm0['len']}B(合法 CoE 读 0x1000:00)...", flush=True)
+try:
+    s._fpwr(sm0["start"], mbx)
+    print("  写入 OK (WKC=1)", flush=True)
+except Exception as e:
+    print(f"  写入失败: {e}", flush=True)
+
+sm0b, sm1b, ev2 = show("写入后立即")
+time.sleep(0.2)
+sm0c, sm1c, ev3 = show("写入后 200ms")
+
+if sm1c and (sm1c["st"] & 0x08):
+    d = fprd(sm1c["start"], 32)
+    print(f"*** SSC 响应了! SM1 RAM = {d.hex().upper() if d else 'ERR'}", flush=True)
+    print(">>> ESC+SSC 完全正常,楔死另有其因", flush=True)
+else:
+    print("\nSSC 无响应。检查 0x220 bit0 事件是否置位过...", flush=True)
+    if sm0c and (sm0c["st"] & 0x08):
+        print(f"SM0.bit3(写满)={sm0c['st'] >> 3 & 1} —— 写满标志挂着(主站侧协议已触发)", flush=True)
+        # 模拟 SSC 解锁: 读 SM0 首 word + 末 word
+        fprd(sm0c["start"], 2)
+        fprd(sm0c["start"] + sm0c["len"] - 2, 2)
+        sm0d, sm1d, ev4 = show("模拟 SSC 解锁后")
+        if not (sm0d["st"] & 0x08):
+            print("解锁有效(SM0.bit3 清零) —— SM 状态机活着", flush=True)
+    else:
+        print(f"SM0.bit3(写满)={sm0c['st'] >> 3 & 1 if sm0c else '?'} —— 写满标志都没置!写到末字节也没触发 mailbox full", flush=True)
+
+es.api_close({})
+print("DONE", flush=True)
@@ -1,78 +0,0 @@
-"""_repro_hang.py — 复现 2026-09-03 01:40 "op/start 卡住→闪退"。
-
-复刻 ecat.html 的 UI 操作序列(100ms 周期 + 2s pollSync + 手动 stop/start 节奏):
-  open → 循环 N 轮{ op/start → 立即 cycle+sync_diag → cycle@100ms → ~2s 后 op/stop } → close
-faulthandler 45s 看门狗: 卡住时自动转储所有线程栈(op/start 有界 ~26s, 超过 45s = 真卡死)。
-退出码非 0xC0000005 之外的异常退出也会打印。
-"""
-import faulthandler
-import sys
-import time
-
-faulthandler.enable()  # 捕获原生访问违例(SEH)并打印 Python 栈 = 闪退根因现场
-faulthandler.dump_traceback_later(45, exit=False)
-
-import ecat_server as es
-
-
-def t():
-    return time.strftime("%H:%M:%S")
-
-
-def show(res):
-    for line in res.get("log", []):
-        print(f"  [{t()}] {line}", flush=True)
-
-
-def main():
-    rounds = int(sys.argv[1]) if len(sys.argv) > 1 else 30
-    r = es.api_open({"substr": "I226"})
-    show(r)
-    if not r.get("ok"):
-        print("open 失败, 退出", flush=True)
-        return 1
-
-    for i in range(rounds):
-        print(f"=== round {i} op/start ===", flush=True)
-        r = es.api_op_start({"dc_cycle_us": 0})
-        show(r)
-        if not r.get("ok"):
-            print(f"  [{t()}] op/start 失败, 继续下一轮", flush=True)
-            time.sleep(1.0)
-            continue
-        # 复刻 doCycle() 立即一次 + pollSync() 立即一次
-        r = es.api_op_cycle({"outputs": ""})
-        if not r.get("ok"):
-            show(r)
-        r = es.api_sync_diag({})
-        if not r.get("ok"):
-            show(r)
-        # 100ms 周期跑 ~2s, 中途 2s 处再 pollSync 一次
-        end = time.time() + 2.0
-        synced = False
-        while time.time() < end:
-            r = es.api_op_cycle({"outputs": ""})
-            if not r.get("ok"):
-                show(r)
-                break
-            if not synced and time.time() > end - 1.0:
-                synced = True
-                r = es.api_sync_diag({})
-                if not r.get("ok"):
-                    show(r)
-            time.sleep(0.1)
-        print(f"  [{t()}] op/stop", flush=True)
-        r = es.api_op_stop({})
-        show(r)
-        time.sleep(1.0)  # 用户点击间隔
-
-    es.api_close({})
-    print(f"[{t()}] 全部 {rounds} 轮完成, 无卡死", flush=True)
-    faulthandler.cancel_dump_traceback_later()
-    return 0
-
-
-if __name__ == "__main__":
-    rc = main()
-    print(f"EXIT rc={rc}", flush=True)
-    sys.exit(rc)