#!/usr/bin/env python3 # -*- coding: utf-8 -*- """诊断 v2b: BUGGY 对照组 — 不加裁剪修复, 复现 PySide6 6.11.2 上 _log 裁剪 AttributeError 触发后的行为: tick 中止 / RX 处理滞后 ratchet / 事件循环漂移 / 点击->TX。指标同 v2, 另增帧处理年龄(frame_age)。""" import gc import os import sys import threading import time os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import filmforge_rs485 as ff class FakeSerial: def __init__(self, *a, **kw): self._buf = bytearray() self._lock = threading.Lock() self._alive = True self.write_log = [] self.report_ms = 0 self._next_tx = 0.0 self._th = threading.Thread(target=self._fw, daemon=True) self._th.start() FakeSerial.last = self def _fw(self): while self._alive: now = time.time() if self.report_ms and now >= self._next_tx: self._push(self._frame("34.6")) self._next_tx = now + self.report_ms / 1000.0 time.sleep(0.002) @staticmethod def _frame(payload, type_code=0xF0): body = bytes([type_code, len(payload)]) + payload.encode("ascii") return bytes([0xB5]) + body + bytes([ff.sum_crc(body)]) def _push(self, b): with self._lock: self._buf += b @property def in_waiting(self): with self._lock: return len(self._buf) def read(self, n): deadline = time.time() + 0.15 while True: with self._lock: if self._buf: out = bytes(self._buf[:n]) del self._buf[:n] return out if time.time() >= deadline: return b"" time.sleep(0.004) def write(self, data): self.write_log.append((time.time(), bytes(data))) cmd = bytes(data[3:-1]).decode("ascii", "replace").strip() threading.Timer(0.05, lambda: self._on_cmd(cmd)).start() return len(data) def _on_cmd(self, cmd): if cmd.startswith("M2S"): try: ms = int(cmd[3:]) except ValueError: ms = 0 self.report_ms = ms self._next_tx = time.time() + ms / 1000.0 self._push(self._frame("ok")) elif cmd in ("M1", "M2", "M3", "M4", "M999"): self._push(self._frame({"M1": "34.6", "M2": "OFF", "M3": "OFF", "M4": "35.0", "M999": "v41.0.0.1"}[cmd])) else: self._push(self._frame("ok")) def close(self): self._alive = False try: self._th.join(timeout=0.5) except Exception: pass ff.serial.Serial = FakeSerial from PySide6.QtCore import QTimer from PySide6.QtWidgets import QApplication app = QApplication([]) import app_fluent # 注意: 不打 setMaximumBlockCount 补丁 —— 原样复现 bug window = app_fluent.App(0) window.show() fake = None exception_count = [0] stats = {"tick": [0, 0.0, 0.0], "gc": [0, 0.0, 0.0], "drift": [], "qmax": [0], "click": [], "age": [0, 0.0, 0.0]} # age: n,total,max _orig_tick = app_fluent.App._tick def timed_tick(self): t0 = time.perf_counter() try: _orig_tick(self) except Exception as e: exception_count[0] += 1 if exception_count[0] <= 3: print("TICK EXCEPTION:", type(e).__name__, e, flush=True) dt = time.perf_counter() - t0 s = stats["tick"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt) app_fluent.App._tick = timed_tick _orig_handle = app_fluent.App._handle_frame def timed_handle(self, fr, t_rx): _orig_handle(self, fr, t_rx) age = time.time() - t_rx # 帧从收到到被主线程处理的滞后 s = stats["age"]; s[0] += 1; s[1] += age; s[2] = max(s[2], age) app_fluent.App._handle_frame = timed_handle _gc_t = [0.0] def gc_cb(phase, info): if phase == "start": _gc_t[0] = time.perf_counter() else: dt = time.perf_counter() - _gc_t[0] s = stats["gc"]; s[0] += 1; s[1] += dt; s[2] = max(s[2], dt) gc.callbacks.append(gc_cb) probe = {"expect": None} def probe_fire(): now = time.time() if probe["expect"] is None: probe["expect"] = now + 0.02 return stats["drift"].append((now, now - probe["expect"])) probe["expect"] += 0.02 if now - probe["expect"] > 0.5: probe["expect"] = now + 0.02 pt = QTimer(); pt.timeout.connect(probe_fire); pt.start(20) CLICK_PERIOD = 30 click = {"expect": None, "n": 0} def do_click_stop(): now = time.time() drift_ms = (now - click["expect"]) * 1000 if click["expect"] else 0.0 q0 = window._rx_q.qsize() window._send_set("M2S0", "report", ("OFF", None)) click["n"] += 1 stats["click"].append(dict(n=click["n"], t=now, drift_ms=drift_ms, qsize=q0, pending=len(window._pending), samples=len(window.samples_t), logblk=window.txt_log.blockCount())) QTimer.singleShot(3000, do_click_restart) def do_click_restart(): window._send_set("M2S100", "report", ("ON", 100)) click["expect"] = time.time() + CLICK_PERIOD QTimer.singleShot(CLICK_PERIOD * 1000, do_click_stop) def start(): global fake window.link.open("COMFAKE", 115200) def wait_conn(): if window.link.is_connected(): fake = FakeSerial.last click["expect"] = time.time() + 15 QTimer.singleShot(15000, do_click_stop) print("[t0] connected; stop-clicks every %ds from +15s (BUGGY mode)" % CLICK_PERIOD, flush=True) else: QTimer.singleShot(200, wait_conn) QTimer.singleShot(200, wait_conn) QTimer.singleShot(100, start) t_start = time.time() DURATION = float(sys.argv[1]) if len(sys.argv) > 1 else 320.0 INTERVAL = float(sys.argv[2]) if len(sys.argv) > 2 else 40.0 def monitor(): while time.time() - t_start < DURATION: time.sleep(INTERVAL) el = time.time() - t_start recent = [d for t, d in stats["drift"] if t >= time.time() - INTERVAL] rmax = max(recent) * 1000 if recent else 0 stats["qmax"][0] = max(stats["qmax"][0], window._rx_q.qsize()) n, tot, mx = stats["age"] print(f"[{el:6.1f}s] cpu={time.process_time():7.2f}s " f"samples={len(window.samples_t):6d} q={window._rx_q.qsize():5d} " f"(max {stats['qmax'][0]:5d}) logblk={window.txt_log.blockCount():5d} " f"frame_age avg={tot/max(n,1)*1000:7.1f}ms max={mx*1000:8.1f}ms " f"tick_max={stats['tick'][2]*1000:6.1f}ms drift_max={rmax:8.1f}ms " f"exc={exception_count[0]}", flush=True) print("==== CLICK 明细 ====", flush=True) for c in stats["click"]: tx = next((t for t, d in FakeSerial.last.write_log if d[3:-1].decode("ascii", "replace") == "M2S0" and c["t"] <= t <= c["t"] + 2.0), None) total = (tx - c["t"]) * 1000 if tx else -1 print(f"click#{c['n']} t=+{c['t']-t_start:6.1f}s 调度漂移={c['drift_ms']:8.1f}ms " f"click->TX={total:7.1f}ms q={c['qsize']:5d} pending={c['pending']} " f"samples={c['samples']:6d} logblk={c['logblk']}", flush=True) ds = sorted(d for _t, d in stats["drift"]) if ds: print(f"drift: n={len(ds)} p50={ds[len(ds)//2]*1000:.1f}ms " f"p99={ds[int(len(ds)*0.99)]*1000:.1f}ms max={ds[-1]*1000:.1f}ms " f">500ms次数={sum(1 for d in ds if d>0.5)}", flush=True) n, tot, mx = stats["age"] print(f"frame_age avg={tot/max(n,1)*1000:.1f}ms max={mx*1000:.1f}ms | " f"exceptions={exception_count[0]} | tick n={stats['tick'][0]} " f"max={stats['tick'][2]*1000:.1f}ms", flush=True) os._exit(0) threading.Thread(target=monitor, daemon=True).start() app.exec()