diff --git a/scripts/lib/usb_health.py b/scripts/lib/usb_health.py index 7228168..093396c 100644 --- a/scripts/lib/usb_health.py +++ b/scripts/lib/usb_health.py @@ -19,11 +19,26 @@ USB3 接口里 SuperSpeed 是一对独立的差分针脚，它们一旦接触不 一开始就报"协商速率不对"，能把这一类问题从"查算法"挪回"插紧线"。 +还有一种更彻底的坏法：**Hub 在内核里卡死** +-------------------------------------------- +2026-09-28 又实测到一次：不只是掉速，而是挂在同一个 Hub 下的**四台设备一起消失** +（腕部 D405 + 顶置 D435 + 左右两个视触觉）。此时 ``/sys/bus/usb/devices`` 里那个 +Hub 节点还在，但读它的属性会**永久阻塞**： + + /sys/bus/usb/devices/1-3.4/product ← 读 2 秒、20 秒、永不返回 + +原因是内核正在跟这个卡死的设备对话，拿不到应答就不返回。于是本脚本自己也挂住 —— +用户看到的是"体检卡死"，比报一行错更难判断。所以这里的 sysfs 读取**一律带超时**， +超时就当作一条 error 结论报出来（见 ``SysfsReader``）：宁可少读到一个字段， +也不能让体检本身失去意义。 + 判定口径（这里就是**唯一一处**） -------------------------------- · 5000 Mbps → USB3，OK（D435 可用 1280x720@30） · 480 Mbps → USB2，RealSense 会掉到 640x480@15 —— WARN 并给出拔插建议 · 12 / 1.5 → USB 全速/低速，串口转换器正常，不告警 + · 读属性超时 → Hub/设备内核级卡死，ERROR（软件复位要 root，现场断电重启该 Hub） + · 一台都没有 → ERROR 用法:: @@ -36,6 +51,7 @@ from __future__ import annotations import argparse import json +import threading import time from dataclasses import dataclass, asdict from pathlib import Path @@ -65,6 +81,70 @@ USB_RESEAT_HINT = ( "核对速率：lsusb -t（设备后面写 480M 还是 5000M）。" ) +#: 读 sysfs 属性的超时 [s]。正常读取是微秒级，1 秒足够宽松。 +DEFAULT_READ_TIMEOUT_S = 1.0 + + +class SysfsReader: + """带超时的 sysfs 读取 —— 卡死的 USB 设备会让读取**永久阻塞**。 + + 为什么不能直接用 ``Path.read_text``：内核在跟卡死的设备对话时， + ``/sys/bus/usb/devices/1-3.4/product`` 这类属性读下去不返回，Python 就 + 卡在系统调用里。对体检工具来说这是最坏的结果 —— 本该报出故障的那一次 + 恰恰什么都不输出。 + + 超时后读属性用的线程会一直留在后台（Python 没法强杀阻塞在系统调用里的 + 线程），所以同一个路径**只尝试一次**并记入 ``blocked``，后续直接跳过。 + 这样 ``--watch`` 反复采样也不会攒出几十个卡住的线程。 + """ + + def __init__(self, timeout_s: float = DEFAULT_READ_TIMEOUT_S) -> None: + self.timeout_s = timeout_s + #: 读取超时的 sysfs 路径（按出现顺序，去重） + self.blocked: List[str] = [] + self._tried_blocked: set[str] = set() + + def read(self, path: Path) -> str: + key = str(path) + if key in self._tried_blocked: + return "" + done = threading.Event() + box: Dict[str, str] = {} + + def _worker() -> None: + try: + box["v"] = path.read_text(encoding="utf-8", errors="replace").strip() + except OSError: + box["v"] = "" + finally: + done.set() + + threading.Thread(target=_worker, daemon=True).start() + if not done.wait(self.timeout_s): + self._tried_blocked.add(key) + if key not in self.blocked: + self.blocked.append(key) + return "" + return box.get("v", "") + + +def blocked_devices(blocked: Sequence[str]) -> List[str]: + """把阻塞的 sysfs 路径归到 USB 设备路径上（``/sys/.../1-3.4/product`` → ``1-3.4``）。 + + 归类是为了给出**该动哪一个**：报"1-3.4/product 读不了"没用， + 要能说出"1-3.4 这个 Hub 卡死了，挂在它下面的设备一起没了"。 + """ + devs: List[str] = [] + for raw in blocked: + parts = Path(raw).parts + if "devices" in parts: + name = parts[parts.index("devices") + 1] + else: + name = Path(raw).parent.name + if name and name not in devs: + devs.append(name) + return devs + @dataclass(frozen=True) class UsbDevice: @@ -131,37 +211,63 @@ def _parse_speed(raw: str) -> Optional[int]: return None -def read_devices(root: Path = SYS_USB) -> List[UsbDevice]: - """列出当前 USB 设备（不需要 root）。""" +def read_devices( + root: Path = SYS_USB, reader: Optional[SysfsReader] = None +) -> List[UsbDevice]: + """列出当前 USB 设备（不需要 root）。 + + ``reader`` 传入时，读取超时的路径会记在它上面（见 :class:`SysfsReader`）—— + 调用方据此把"Hub 卡死"报出来，而不是默默少列几台设备。 + """ + rd = reader or SysfsReader() out: List[UsbDevice] = [] if not root.is_dir(): return out for entry in sorted(root.iterdir()): - vid = _read(entry / "idVendor") + vid = rd.read(entry / "idVendor") if not vid: continue # 不是设备节点（接口/端口目录没有 idVendor） out.append( UsbDevice( path=entry.name, vid=vid, - pid=_read(entry / "idProduct"), - product=_read(entry / "product"), - speed_mbps=_parse_speed(_read(entry / "speed")), - serial=_read(entry / "serial"), + pid=rd.read(entry / "idProduct"), + product=rd.read(entry / "product"), + speed_mbps=_parse_speed(rd.read(entry / "speed")), + serial=rd.read(entry / "serial"), ) ) return out -def findings_for(devices: Sequence[UsbDevice]) -> List[Finding]: - """对一批设备给出体检结论（纯函数，便于单测）。""" +def findings_for( + devices: Sequence[UsbDevice], blocked: Sequence[str] = () +) -> List[Finding]: + """对一批设备给出体检结论（纯函数，便于单测）。 + + ``blocked`` 是读取超时的 sysfs 路径（``SysfsReader.blocked``）。 + """ out: List[Finding] = [] + # 卡死的 Hub 要**排在最前面**：它是根因，后面"一台相机都没枚举到"只是症状。 + for dev in blocked_devices(blocked): + out.append( + Finding( + "error", + f"USB 设备 {dev} 的 sysfs 属性读取超时 —— 内核正卡在跟它对话，" + f"这一台**及其下游设备**会一起从系统里消失" + f"（相机/触觉都挂它下面时，现象是「相机和触觉同时没了」）。" + f"软件复位需要 root；现场处置是给这个 Hub 断电重启：" + f"拔插它的上游线，或把设备改插到主板后面板直连的 USB 口" + f"（两个 RealSense 都走同一个总线供电的小 Hub 很容易把它拖垮）。", + ) + ) cams = [d for d in devices if d.is_realsense] if not cams: out.append( Finding( "error", - "一台 RealSense 都没枚举到。拔插一次；确认没有 " + "一台 RealSense 都没枚举到。先看上面有没有「sysfs 读取超时」那条" + "（有就是 Hub 卡死，不是相机的问题）；否则拔插一次，并确认没有 " "realsense-viewer / 其它进程占着相机。", ) ) @@ -214,13 +320,17 @@ def flap_events(snapshots: Sequence[Sequence[str]]) -> List[str]: def watch_realsense( - seconds: float, interval_s: float = 1.0, root: Path = SYS_USB + seconds: float, + interval_s: float = 1.0, + root: Path = SYS_USB, + reader: Optional[SysfsReader] = None, ) -> List[List[str]]: """在 ``seconds`` 秒内反复采样 RealSense 的设备路径。""" + rd = reader or SysfsReader() snaps: List[List[str]] = [] deadline = time.monotonic() + max(0.0, seconds) while True: - snaps.append(sorted(d.path for d in read_devices(root) if d.is_realsense)) + snaps.append(sorted(d.path for d in read_devices(root, rd) if d.is_realsense)) if time.monotonic() >= deadline: break time.sleep(interval_s) @@ -233,9 +343,12 @@ def main(argv: Optional[Sequence[str]] = None) -> int: ap.add_argument("--json", action="store_true", help="输出 JSON") args = ap.parse_args(argv) - devices = read_devices() - findings = findings_for(devices) - snaps = watch_realsense(args.watch) if args.watch > 0 else [] + # 整轮共用一个 reader：某条路径一旦确认卡死就不再重读， + # 否则 --watch 每次采样都会再攒一个卡在内核里的线程。 + reader = SysfsReader() + devices = read_devices(reader=reader) + findings = findings_for(devices, reader.blocked) + snaps = watch_realsense(args.watch, reader=reader) if args.watch > 0 else [] flaps = flap_events(snaps) if snaps else [] if flaps: findings.append( @@ -252,6 +365,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int: "devices": [asdict(d) for d in devices if d.is_realsense], "findings": [asdict(f) for f in findings], "flaps": flaps, + "blocked_sysfs": list(reader.blocked), } if args.json: print(json.dumps(payload, ensure_ascii=False, indent=2)) @@ -262,6 +376,10 @@ def main(argv: Optional[Sequence[str]] = None) -> int: f" {d.model or d.product:8} {d.path:8} " f"{str(d.speed_mbps) + ' Mbps':>10} vid:pid={d.vid}:{d.pid}" ) + if reader.blocked: + # 单独打一行：它意味着"下面那些设备根本没被列出来"， + # 不写清楚的话，上面的列表看起来就像"设备没了"而没有原因。 + print(f" （{len(reader.blocked)} 条 sysfs 属性读取超时，详见下面的 FAIL）") for f in findings: mark = {"ok": "OK ", "warn": "WARN", "error": "FAIL"}[f.level] print(f" [{mark}] {f.text}") diff --git a/scripts/lib/v4l_health.py b/scripts/lib/v4l_health.py index 3a20365..2e64b8d 100644 --- a/scripts/lib/v4l_health.py +++ b/scripts/lib/v4l_health.py @@ -15,8 +15,12 @@ ``sudo /usr/local/bin/orisys-usb-reset ``（unbind/bind 白名单脚本） 复位 —— 若 unbind 成功而 bind 卡住，就会变成现在这个状态： ``/sys/bus/usb/devices/1-3.4.1`` 还在、接口还挂在 ``uvcvideo`` 上， - 但 ``/dev/video*`` 一个都没有。**软件侧已经没救了**，只能： - 在真终端里带密码跑一次 ``orisys-usb-reset``，或拔插一次该相机。 + 但 ``/dev/video*`` 一个都没有。**软件侧重连已经没救**，只能复位： + ``sudo -n /usr/local/bin/orisys-usb-reset ``。 + 这条路径**不需要密码** —— sudoers 已对这一个脚本配了 NOPASSWD 白名单 + （``orisys_tactile.py`` 里的自动复位也是用 ``sudo -n`` 调的）， + 所以无人值守时可以直接跑；真要输密码说明白名单被改掉了。 + 复位仍不行（unbind 都不响应）就只剩拔插一次该相机。 3. ``可读但无帧`` —— 节点在、open 得到，但一帧都读不出来 （日志里 ``select() timeout``）。→ 多半是同一条 USB2 总线被别的流占满； @@ -112,9 +116,10 @@ class Check: if self.state == "wedged": return ( f"{self.name}：设备在总线上（{self.devpath}）但**没有 /dev/video* 节点** " - "—— 内核级卡死，软件侧救不回来。到真终端里跑 " - f"`sudo /usr/local/bin/orisys-usb-reset {self.devpath}`（要密码），" - "或把这只相机拔插一次；然后重启触觉节点。" + "—— 内核级卡死，软件侧重连救不回来。跑 " + f"`sudo -n /usr/local/bin/orisys-usb-reset {self.devpath}` 复位" + "（免密，sudoers 白名单；这条本来就是给无人值守用的），" + "复位仍不行就拔插一次这只相机；然后重启触觉节点。" ) return f"{self.name}：没有插（{self.by_path} 不存在）—— 插线/换 USB 口。" @@ -148,7 +153,11 @@ def check_devices( return out -def main(argv: Optional[Sequence[str]] = None) -> int: +def main( + argv: Optional[Sequence[str]] = None, + v4l_root: Path = V4L_BY_PATH, + usb_root: Path = SYS_USB, +) -> int: ap = argparse.ArgumentParser(description="V4L2 触觉相机存在性体检") ap.add_argument( "--dev", @@ -164,7 +173,7 @@ def main(argv: Optional[Sequence[str]] = None) -> int: for item in args.dev: name, _, path = str(item).partition("=") pairs.append((name.strip() or "v4l", path.strip())) - checks = check_devices(pairs) + checks = check_devices(pairs, v4l_root=v4l_root, usb_root=usb_root) if args.json: print(json.dumps([asdict(c) for c in checks], ensure_ascii=False, indent=2)) diff --git a/src/arm_application/test/test_usb_health.py b/src/arm_application/test/test_usb_health.py index f4cb5fc..47517ce 100644 --- a/src/arm_application/test/test_usb_health.py +++ b/src/arm_application/test/test_usb_health.py @@ -16,6 +16,8 @@ lib 的提示必须给出**同一个核对手段**（lsusb -t），否则两边 from __future__ import annotations import sys +import threading +import time from pathlib import Path import pytest @@ -152,3 +154,78 @@ def test_two_sources_of_advice_agree_on_lsusb() -> None: def test_finding_levels_are_from_a_fixed_vocabulary(level: str) -> None: """级别只能是这三个词 —— 脚本按它们判定成败，多一个词就没人处理。""" assert level in {"ok", "warn", "error"} + + +# --------------------------------------------------------------------------- +# 卡死的 Hub：读 sysfs 会永久阻塞，体检不能跟着挂 +# --------------------------------------------------------------------------- + + +class _BlockingPath: + """假装是一个"读下去不返回"的 sysfs 节点（内核跟卡死设备对话时的真实行为）。""" + + def __init__(self, name: str = "product") -> None: + self.name = name + self.reads = 0 + self._release = threading.Event() + + def read_text(self, encoding: str = "utf-8", errors: str = "replace") -> str: + self.reads += 1 + self._release.wait(30.0) # 永不 set —— 模拟内核不返回 + return "never" + + +def test_blocked_read_times_out_instead_of_hanging() -> None: + """这一条是本次实测的核心：体检工具自己不能挂在它要诊断的故障上。""" + reader = usb_health.SysfsReader(timeout_s=0.2) + t0 = time.monotonic() + assert reader.read(_BlockingPath()) == "" + elapsed = time.monotonic() - t0 + assert elapsed < 5.0, f"读取超时没生效（耗时 {elapsed:.1f}s）—— 体检会卡死" + assert len(reader.blocked) == 1 + + +def test_blocked_path_is_only_attempted_once() -> None: + """--watch 会反复采样；确认卡死的路径不再重试，否则会攒一堆卡住的线程。""" + reader = usb_health.SysfsReader(timeout_s=0.2) + path = _BlockingPath() + reader.read(path) + reader.read(path) + reader.read(path) + assert path.reads == 1 + assert reader.blocked == [str(path)] + + +def test_blocked_is_reported_as_error_and_says_what_to_do() -> None: + """报错要说清"动哪一个"以及现场怎么恢复，而不是只报"读不到"。""" + findings = findings_for( + [_cam("2-3.4.3", D405, 5000)], blocked=["/sys/bus/usb/devices/1-3.4/product"] + ) + assert findings[0].level == "error", "卡死的 Hub 必须是最前面的 error" + assert "1-3.4" in findings[0].text + assert "下游" in findings[0].text, "要说清它会带走下游设备" + assert "root" in findings[0].text or "断电" in findings[0].text + assert usb_health.has_errors(findings) + + +def test_blocked_devices_maps_attr_paths_to_device_paths() -> None: + devs = usb_health.blocked_devices( + [ + "/sys/bus/usb/devices/1-3.4/product", + "/sys/bus/usb/devices/1-3.4/speed", + "/sys/bus/usb/devices/2-1.2/serial", + ] + ) + assert devs == ["1-3.4", "2-1.2"], "同一台设备的多个属性只报一次" + + +def test_no_camera_error_points_at_the_hub_when_one_is_blocked() -> None: + """"一台相机都没有"有两种原因，处置完全不同，措辞要能区分开。""" + only_hub = findings_for([], blocked=["/sys/bus/usb/devices/1-3.4/product"]) + assert len(only_hub) == 2, "根因（Hub 卡死）与症状（没有相机）都要报" + assert "读取超时" in only_hub[0].text + assert "Hub 卡死" in only_hub[1].text + + unrelated = findings_for([], blocked=[]) + assert "拔插" in unrelated[0].text + diff --git a/src/arm_application/test/test_v4l_health.py b/src/arm_application/test/test_v4l_health.py index 2537675..14fe79b 100644 --- a/src/arm_application/test/test_v4l_health.py +++ b/src/arm_application/test/test_v4l_health.py @@ -105,7 +105,18 @@ def test_empty_by_path_is_unplugged_not_ok(tmp_path: Path) -> None: def test_main_exit_code_is_nonzero_on_error(tmp_path: Path, capsys) -> None: - """CLI 退出码必须能表示失败 —— 否则脚本里的 it 会被忽略掉。""" - code = v4l_health.main(["--dev", "左指=" + LEFT_BY_PATH]) + """CLI 退出码必须能表示失败 —— 否则脚本里的 it 会被忽略掉。 + + 这里必须**注入临时 sysfs**：真机上这只相机可能根本没插（unplugged）， + 也可能卡死（wedged），让用例去读真 sysfs 就会"换台机器就变红"。 + """ + usb = tmp_path / "usb" + usb.mkdir() + (usb / "1-3.4.1").mkdir() # 设备在总线上，但没有 /dev/video* → wedged + v4l = tmp_path / "v4l" + v4l.mkdir() + code = v4l_health.main( + ["--dev", "左指=" + LEFT_BY_PATH], v4l_root=v4l, usb_root=usb + ) assert code == 1 assert "orisys-usb-reset" in capsys.readouterr().out