78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
|
"""_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)
|