qbuissondebon
7f3cfcf058
Insert a "cmpunlocker unlock (install)" step after unlock verification: y/N confirmation (default n) or --unlock runs install.sh from the harness clone (main branch, ff-only pull first) with live terminal output and full log to cards/<serial>/install_unlocker.log. On success the Vulkan stress is skipped until after the cold reboot. Score now enforces the post-unlock invariant: when verify reports unlocked-ok, any final link other than Gen2 x16 fails with a red [ERROR] and exit 1. README safety section and AGENTS.md harness cloning invariant updated to match.
324 lines
14 KiB
Python
Executable File
324 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""CMP 170HX single-card checker - ONE card per run, no queue.
|
|
|
|
Per run: wait for card -> auto serial -> VBIOS backup -> PCIe link/AER/config
|
|
checks -> retrain + rescan tests -> dmesg scrape -> cmpunlocker verify.sh
|
|
(read-only) -> opt-in cmpunlocker unlock (install.sh, y/N or --unlock; a
|
|
successful install skips VRAM stress and asks for a cold reboot, then you
|
|
re-run this script for the final Gen2 x16 PCIe check)
|
|
-> optional 5-min Vulkan VRAM stress (last step, ask yes/no)
|
|
-> cards/<serial>/report.json.
|
|
|
|
Scoring: FAIL on no-device/link-down/lost-after-rescan, and on a final
|
|
post-unlock link other than Gen2 x16; WARN on correctable AER/Xid/width< x16/
|
|
verify issues/failed unlock install.
|
|
|
|
Exit codes: 0 = PASS/WARN, 1 = FAIL, 2 = no card, 3 = preflight blocking.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
|
|
import cxlib as X
|
|
from cxlib import C, ask, detect_gpus, err, info, now_iso, ok, sanitize, step, warn
|
|
|
|
|
|
# ---------------------------------------------------------------- scoring
|
|
|
|
def score(rep):
|
|
fails, warns = [], []
|
|
ver = rep.get("cmpunlocker_verify", {})
|
|
unlocked = ver.get("status") == "unlocked-ok"
|
|
if not rep.get("bdf"):
|
|
fails.append("no CMP 170HX device detected on the bus")
|
|
link = rep.get("link", {})
|
|
if rep.get("bdf"):
|
|
if link.get("lspci_rc") not in (0, None):
|
|
fails.append("lspci could not read the device")
|
|
if link.get("lnksta_width") in ("x0", "x1", None) or not link.get("lnksta_speed"):
|
|
fails.append(f"link down or untrained: {link.get('lnksta_raw') or 'no LnkSta'}")
|
|
width = link.get("lnksta_width")
|
|
gen = link.get("lnksta_gen")
|
|
if unlocked:
|
|
# Post-unlock final acceptance: only Gen2 x16 passes. The pre-unlock
|
|
# width/Gen1 warns are suppressed here - the error replaces them.
|
|
if width != "x16" or gen is None or gen < 2:
|
|
fails.append(f"[ERROR] final PCIe link after unlock is "
|
|
f"{link.get('lnksta_speed') or '?'} {width or '?'} "
|
|
f"- expected Gen2 x16")
|
|
else:
|
|
if width and width != "x16":
|
|
warns.append(f"link width {width} (capacitor mod expects x16 - mod likely "
|
|
f"missing/incomplete on this card)")
|
|
if gen == 1:
|
|
warns.append("link negotiated Gen1 only (cmpunlocker Gen2 patch/service not active?)")
|
|
if rep.get("rescan", {}).get("ok") is False:
|
|
fails.append("device lost after bus rescan")
|
|
if rep.get("retrain", {}).get("ok") is False:
|
|
fails.append("link did not survive retrain")
|
|
aer = rep.get("aer_after", {})
|
|
aer_before = rep.get("aer_before", {})
|
|
if (aer.get("ce_active") or aer.get("parent_ce_active")
|
|
or (aer.get("aer_correctable_sum", 0) > aer_before.get("aer_correctable_sum", 0))):
|
|
warns.append("correctable AER errors present/incremented")
|
|
if aer.get("ue_active"):
|
|
fails.append(f"uncorrectable AER errors active: {aer['ue_active']}")
|
|
dm = rep.get("dmesg", {})
|
|
if dm.get("errors"):
|
|
warns.append(f"{len(dm['errors'])} kernel error line(s) related to this card (see dmesg.log)")
|
|
vu = rep.get("vulkan_test", {})
|
|
if vu.get("requested") and vu.get("ran"):
|
|
if vu.get("tool_errors"):
|
|
warns.append("errors reported during Vulkan VRAM stress")
|
|
if dm.get("errors"):
|
|
warns.append("kernel errors around VRAM stress window")
|
|
if not rep.get("bios", {}).get("ok"):
|
|
warns.append("VBIOS backup failed/incomplete")
|
|
if ver.get("ran") and ver.get("status") not in ("unlocked-ok",):
|
|
warns.append(f"cmpunlocker verify status: {ver.get('status')}")
|
|
unlock = rep.get("unlock", {})
|
|
if unlock.get("attempted") and not unlock.get("ok"):
|
|
warns.append("cmpunlocker install failed (see install_unlocker.log)")
|
|
|
|
result = "FAIL" if fails else ("WARN" if warns else "PASS")
|
|
return result, fails, warns
|
|
|
|
|
|
# ---------------------------------------------------------------- card detect
|
|
|
|
def wait_for_gpu():
|
|
"""Wait for a single CMP 170HX (with rescan retries). Returns BDF or None."""
|
|
while True:
|
|
for _ in range(5):
|
|
bdfs = detect_gpus()
|
|
if bdfs:
|
|
if len(bdfs) > 1:
|
|
warn(f"multiple CMP 170HX devices present: {', '.join(bdfs)} - using {bdfs[0]}")
|
|
return bdfs[0]
|
|
time.sleep(1)
|
|
ans = ask("CMP 170HX not detected - insert/reseat the card, press Enter to rescan "
|
|
"(q=quit)", "")
|
|
if ans.lower().startswith("q"):
|
|
return None
|
|
if not X.rescan_bus():
|
|
warn("bus rescan not permitted (no root/sudo) - reseat the card physically")
|
|
|
|
|
|
# ---------------------------------------------------------------- one card
|
|
|
|
def process_card(args):
|
|
step("Card check")
|
|
bdf = wait_for_gpu()
|
|
if not bdf:
|
|
err("no card detected - nothing to do")
|
|
return None
|
|
devid = ""
|
|
for line in X.run(["lspci", "-D", "-s", bdf, "-nn"])["out"].splitlines():
|
|
if "[10de:" in line:
|
|
i = line.index("[10de:")
|
|
devid = line[i + 1:i + 10].rstrip("]")
|
|
ok(f"device found at {bdf} ({devid})")
|
|
|
|
nf = X.nvflash_bin(args)
|
|
serial, serial_src = X.get_serial(nf, args)
|
|
folder = X.CARDS_DIR / sanitize(serial)
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
info(f"folder: {folder}")
|
|
|
|
rep = {
|
|
"serial": serial, "serial_source": serial_src, "started_at": now_iso(),
|
|
"bdf": bdf, "device_id": devid, "host": X.run(["uname", "-n"])["out"].strip(),
|
|
"kernel": X.run(["uname", "-r"])["out"].strip(),
|
|
}
|
|
|
|
step("VBIOS backup")
|
|
rep["bios"] = X.save_bios(nf, folder / "bios.rom", args)
|
|
(ok if rep["bios"]["ok"] else warn)(f"bios: {rep['bios']['method'] or 'FAILED'} "
|
|
f"{rep['bios']['size']} bytes - {rep['bios']['note']}")
|
|
|
|
step("PCIe link + config")
|
|
link, lspci_txt = X.check_link(bdf)
|
|
rep["link"] = link
|
|
(folder / "lspci.txt").write_text(lspci_txt)
|
|
(folder / "config_space.hex").write_text(X.dump_config(bdf))
|
|
info(f"LnkSta: {link.get('lnksta_raw') or 'n/a'} (setpci gen={link.get('setpci_current_gen')})")
|
|
rep["aer_before"] = X.aer_snapshot(bdf)
|
|
|
|
step("retrain + rescan tests")
|
|
rep["retrain"] = X.retrain_test(bdf)
|
|
info(f"retrain: ok={rep['retrain']['ok']} - {rep['retrain']['note']}")
|
|
rep["rescan"] = X.rescan_test(bdf)
|
|
info(f"rescan: ok={rep['rescan']['ok']} - {rep['rescan']['note']}")
|
|
rep["aer_after"] = X.aer_snapshot(bdf)
|
|
link2, lspci2 = X.check_link(bdf)
|
|
rep["link_after"] = link2
|
|
if lspci2:
|
|
(folder / "lspci_after.txt").write_text(lspci2)
|
|
|
|
step("kernel log scrape")
|
|
dm = X.dmesg_capture(bdf)
|
|
rep["dmesg"] = dm
|
|
(folder / "dmesg.log").write_text("\n".join(dm.get("related", [])))
|
|
info(f"{dm.get('note')} (source: {dm.get('source')})")
|
|
for e in dm.get("errors", [])[-5:]:
|
|
warn(e.strip()[:160])
|
|
|
|
step("nvidia-smi + cmpunlocker verify")
|
|
rep["nvidia_smi"] = X.nvidia_smi_info()
|
|
rep["cmpunlocker_verify"] = X.run_verify_unlocker()
|
|
v = rep["cmpunlocker_verify"]
|
|
(folder / "verify_unlocker.log").write_text(
|
|
"\n".join(v.get("tail", [])) or (v.get("note") or "not run"))
|
|
info("verify: " + (f"rc={v.get('rc')} -> {v.get('status')}" if v.get("ran")
|
|
else f"skipped - {v.get('note')}"))
|
|
|
|
step("cmpunlocker unlock (install)")
|
|
unlock_installed = False
|
|
if rep["cmpunlocker_verify"].get("status") == "unlocked-ok":
|
|
info("memory already unlocked - skipping install")
|
|
rep["unlock"] = {"attempted": False, "ok": False, "note": "already unlocked"}
|
|
elif getattr(args, "demo", False):
|
|
# --demo never prompts: force the unlock install path off.
|
|
rep["unlock"] = {"attempted": False, "ok": False, "note": "demo"}
|
|
info("unlock install skipped (demo mode)")
|
|
elif not X.INSTALL_SH.exists():
|
|
rep["unlock"] = {"attempted": False, "ok": False,
|
|
"note": "harness/cmpunlocker not cloned - run --preflight"}
|
|
warn("harness/cmpunlocker not cloned - run --preflight (unlock install skipped)")
|
|
elif args.unlock or ask("Run cmpunlocker install.sh now? (y/N)", "n").lower().startswith("y"):
|
|
X.update_cmpunlocker()
|
|
rep["unlock"] = X.run_installer(folder / "install_unlocker.log")
|
|
unlock_installed = bool(rep["unlock"].get("attempted") and rep["unlock"].get("ok"))
|
|
if unlock_installed:
|
|
# Kernel modules are replaced but the card is not unlocked until the
|
|
# cold reboot - VRAM stress against the half-applied state is unsafe.
|
|
rep["vulkan_test"] = {"requested": False, "ran": False,
|
|
"note": "skipped - unlock installed, reboot first"}
|
|
else:
|
|
warn("cmpunlocker install failed - continuing with the Vulkan step")
|
|
else:
|
|
rep["unlock"] = {"attempted": False, "ok": False, "note": "operator declined"}
|
|
info("unlock install skipped")
|
|
|
|
step("optional Vulkan VRAM test (last step)")
|
|
if unlock_installed:
|
|
info("vulkan test skipped - unlock installed, cold reboot first")
|
|
elif args.vulkan == "never":
|
|
rep["vulkan_test"] = {"requested": False, "ran": False, "note": "disabled via --vulkan never"}
|
|
elif args.vulkan == "always":
|
|
rep["vulkan_test"] = X.run_vulkan_test()
|
|
else:
|
|
if ask("Run 5-min Vulkan VRAM stress test now? (y/N)", "n").lower().startswith("y"):
|
|
rep["vulkan_test"] = X.run_vulkan_test()
|
|
else:
|
|
rep["vulkan_test"] = {"requested": False, "ran": False, "note": "operator declined"}
|
|
info("vulkan test skipped")
|
|
|
|
result, fails, warns = score(rep)
|
|
rep.update(finished_at=now_iso(), result=result, failures=fails, warnings=warns)
|
|
(folder / "report.json").write_text(json.dumps(rep, indent=2))
|
|
|
|
print()
|
|
if result == "PASS":
|
|
ok(f"{serial}: PASS")
|
|
elif result == "WARN":
|
|
warn(f"{serial}: WARN")
|
|
else:
|
|
err(f"{serial}: FAIL")
|
|
for f in fails:
|
|
err(" FAIL: " + f)
|
|
for w in warns:
|
|
warn(" WARN: " + w)
|
|
info(f"report: {folder / 'report.json'}")
|
|
return rep
|
|
|
|
|
|
# ---------------------------------------------------------------- demo
|
|
|
|
def demo(args):
|
|
"""Validate report pipeline without hardware: writes cards/DEMO-*."""
|
|
step("Demo mode (no hardware)")
|
|
args.vulkan = "never"
|
|
args.serial = "DEMO-SERIAL-001"
|
|
args.skip_bios = True
|
|
|
|
orig_wait = globals()["wait_for_gpu"]
|
|
orig_detect, orig_check = X.detect_gpus, X.check_link
|
|
orig_aer, orig_dm, orig_cfg = X.aer_snapshot, X.dmesg_capture, X.dump_config
|
|
orig_retrain, orig_rescan = X.retrain_test, X.rescan_test
|
|
orig_smi, orig_verify = X.nvidia_smi_info, X.run_verify_unlocker
|
|
orig_update, orig_installer = X.update_cmpunlocker, X.run_installer
|
|
globals()["wait_for_gpu"] = lambda: "0000:0a:00.0"
|
|
X.detect_gpus = lambda: ["0000:0a:00.0"]
|
|
X.check_link = lambda b: ({"lnksta_speed": "5GT/s", "lnksta_width": "x4", "lnksta_gen": 2,
|
|
"lnkcap_speed": "8GT/s", "lnkcap_width": "x16",
|
|
"lnksta_raw": "LnkSta: Speed 5GT/s, Width x4, TrErr-",
|
|
"setpci_current_gen": 2, "lspci_rc": 0}, "demo lspci output\n")
|
|
X.aer_snapshot = lambda b: {"aer_sysfs_correctable": None, "aer_sysfs_fatal": None,
|
|
"aer_correctable_sum": 0, "ce_active": [], "ue_active": [],
|
|
"parent_bdf": None, "parent_ce_active": []}
|
|
X.dmesg_capture = lambda b: {"source": "demo", "related": ["demo line"], "errors": [],
|
|
"note": "demo"}
|
|
X.dump_config = lambda b: "demo config space dump\n"
|
|
X.retrain_test = lambda b: {"attempted": True, "ok": True, "note": "x4 -> x4",
|
|
"width_before": "x4", "width_after": "x4"}
|
|
X.rescan_test = lambda b: {"attempted": True, "ok": True, "note": "re-enumerated"}
|
|
X.nvidia_smi_info = lambda: {"available": False}
|
|
X.run_verify_unlocker = lambda: {"ran": False, "note": "demo"}
|
|
X.update_cmpunlocker = lambda: {"ok": True, "note": "demo"}
|
|
X.run_installer = lambda log_path: {"attempted": False, "ok": False, "rc": None,
|
|
"note": "demo", "log": str(log_path)}
|
|
try:
|
|
rep = process_card(args)
|
|
finally:
|
|
globals()["wait_for_gpu"] = orig_wait
|
|
X.detect_gpus, X.check_link = orig_detect, orig_check
|
|
X.aer_snapshot, X.dmesg_capture, X.dump_config = orig_aer, orig_dm, orig_cfg
|
|
X.retrain_test, X.rescan_test = orig_retrain, orig_rescan
|
|
X.nvidia_smi_info, X.run_verify_unlocker = orig_smi, orig_verify
|
|
X.update_cmpunlocker, X.run_installer = orig_update, orig_installer
|
|
return rep
|
|
|
|
|
|
# ---------------------------------------------------------------- main
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="CMP 170HX single-card checker (one card per run)")
|
|
ap.add_argument("--preflight", action="store_true",
|
|
help="tool check + package install (requires sudo) + harness clone, then exit")
|
|
ap.add_argument("--demo", action="store_true", help="run pipeline with fake hardware (self-test)")
|
|
ap.add_argument("--nvflash", help="path to nvflash binary")
|
|
ap.add_argument("--serial", help="override serial (skips nvflash auto-read)")
|
|
ap.add_argument("--skip-bios", action="store_true", help="skip VBIOS backup")
|
|
ap.add_argument("--vulkan", choices=("ask", "always", "never"), default="ask",
|
|
help="5-min Vulkan VRAM test policy (default ask, runs as last step)")
|
|
ap.add_argument("--unlock", action="store_true",
|
|
help="run cmpunlocker install.sh without prompting (default: ask y/N)")
|
|
ap.add_argument("--no-install", action="store_true",
|
|
help="do not auto-install packages (skips the sudo precondition)")
|
|
ap.add_argument("--no-clone", action="store_true", help="do not clone harness repos")
|
|
args = ap.parse_args()
|
|
|
|
pf = X.preflight(auto_install=not args.no_install, do_clone=not args.no_clone)
|
|
if args.preflight:
|
|
return 0 if pf.get("preflight_ok") else 3
|
|
if args.demo:
|
|
demo(args)
|
|
return 0
|
|
|
|
X.CARDS_DIR.mkdir(exist_ok=True)
|
|
rep = process_card(args)
|
|
if not rep:
|
|
return 2
|
|
return 1 if rep.get("result") == "FAIL" else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except KeyboardInterrupt:
|
|
print()
|
|
warn("interrupted")
|
|
sys.exit(130)
|