wandering1
13001f2ae4
task-32 裁定落成可执行件（道F 交付）。不是二选一而是**分权**：静态最坏链 margin 仍是唯一拦提交项（由 `stack_report.py` 守，它本就把运行时峰值取进 max）；新增 `stack_gate.py` 管台架侧——趋势只出 WARN， **绝对底线**（free_min < 25% cfg）是运行时唯一有 FAIL 权限的一档，**采集健康门不过 = VOID / exit 2**， 不红不绿、不进台账。把「daemon 死了 status 仍报正常、rx_lines=0」那类事故机械化：门不过与设备故障 exit 1 分开记，按码聚合的人不会把端口没开当成又崩了一次栈。 容差是推导的：tol = 上取整到 16 B 的 max(128 B, 2×实测摆动)，实测 80 B ⇒ 统一 160 B，登记进 `scripts/stack_tolerance.tsv` 带来源列；**改数权限写死**——只有台架重测后才允许回填，禁止为让某轮变绿 而调大容差（被这把尺考核的人不得当场决定尺子的松紧），且这句话**紧贴数值写**。 `--selftest` 14 条能红，开发中抓到作者三个真错（身份戳正则把 __DATE__ 月份写成三位数字、一度把跨轮戳 不同当 VOID 会让判据永不可用、文档 §4.0.2 被 doc-links 抓跨文档串号）。**不接 pre-commit**：它的输入 是台架串口采集，提交时既无设备也无当轮日志，挂上去要么永远 SKIP、要么拿上一轮日志判这一轮的码（与 体积门禁那条假凭证同族）。
311 lines
17 KiB
Python
311 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""stack_gate.py - 栈水位趋势判据（task-32 定稿的可执行形态，纯标准库）。
|
||
|
||
模块名称：台架栈水位判据器（趋势 + 采集健康门）
|
||
模块功能：把 N 段启动日志里的 [STACK] 水位汇成"本轮最大值"，与基线轮最大值按每任务
|
||
容差比较，输出四档结论；**先过采集健康门，再谈比较**。
|
||
创建日期：2026-10-27（裁定见 docs/内存管理/RAM使用分析与优化方案.md §4.0.6 第③条）
|
||
|
||
为什么要有它：旧判据「栈水位不劣于上一轮」拿单次运行时峰值跨轮比较，而同一份镜像自己
|
||
就会摆（实测 uiTask 4116~4188 = 72 B、adcTask 920~1000 = 80 B），于是既假红又漏报。
|
||
本脚本把那条判据拆成三件事，等级不同、互不越权：
|
||
1) 硬线（唯一拦提交项）= scripts/stack_report.py 的每任务 verdict：
|
||
margin = cfg - max(静态最坏链, 运行时峰值) >= max(isr_margin, cfg*30%)
|
||
本脚本不重算也不替代它，只在末尾提示其退出码（接不接 pre-commit 由 Lead 定）。
|
||
2) 趋势（告警级，永不单独拦提交）= 同一身份戳、>=3 圈取最大值，与基线轮最大值比，
|
||
超每任务容差 -> WARN。容差按字节，来源见 scripts/stack_tolerance.tsv。
|
||
3) 作废（VOID，优先于前两者）= 采集健康门不过：本轮既不判过也不判不过，也不得写进台账
|
||
结论。判据不许依赖"采集看起来很健康"这种自我感觉——2026-10-27 实证：常驻 daemon
|
||
死后 status 会自动拉起一个新 daemon，仍报"正常"而 rx_lines=0。
|
||
|
||
退出码（与 verify_log.py 同族口径：测量问题绝不伪装成设备故障）：
|
||
0 = PASS（含通过但有趋势 WARN）
|
||
1 = FAIL（运行时绝对底线破线：free_min < 25% of cfg —— 设备侧真消息）
|
||
2 = VOID_MEASUREMENT（采集健康门未过：本轮不可用，也不算设备的账）
|
||
3 = 用法错误
|
||
"""
|
||
|
||
import argparse
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
RUNTIME_FLOOR_PCT = 25 # 运行时绝对底线（与静态 30% 硬线错开一档：峰值只是下界证据）
|
||
MIN_CIRCLES = 3 # 少于 3 圈 = 噪声样本，不做跨轮比较
|
||
DEFAULT_TOL = 160 # 未登记任务的兜底容差（字节），推导见 scripts/stack_tolerance.tsv 头注
|
||
|
||
STACK_RE = re.compile(r"\[STACK\]\s+(\w+)\s+peak_used=(\d+)/(\d+)\s+free_min=(\d+)\s*\((\d+)%\)")
|
||
# __DATE__ 的月份是**英文名**（"Oct 26 2026 22:52:09"），不是数字——早先写成 \d{3} 会让
|
||
# 每一份真实日志都判成"没有身份戳"而整轮作废（判据全红比判据漏更响，但也别拿假红当纪律）。
|
||
STAMP_RE = re.compile(r"fw build:\s*([A-Za-z]{3}\s+\d{1,2}\s+\d{4}\s+\d{2}:\d{2}:\d{2})\s+suite=(\S+)\s+dstr=(\S+)")
|
||
ANCHOR = "reset cause:"
|
||
|
||
|
||
def parse_block(text):
|
||
"""取一段日志的 (每任务峰值, 身份戳, 锚点数, 有效行数)。
|
||
|
||
同一任务在一块里出现多次时取最大值——栈峰值的语义就是"最坏那一次"，不做平均。
|
||
"""
|
||
payload = [l for l in text.splitlines() if l.strip() and not l.lstrip().startswith("#")]
|
||
peaks = {}
|
||
for ln in payload:
|
||
m = STACK_RE.search(ln)
|
||
if m:
|
||
name = m.group(1)
|
||
used, cfg, free = int(m.group(2)), int(m.group(3)), int(m.group(4))
|
||
prev = peaks.get(name)
|
||
if prev is None or used > prev["used"]:
|
||
peaks[name] = {"used": used, "cfg": cfg, "free": free}
|
||
stamps = STAMP_RE.findall("\n".join(payload))
|
||
uniq = sorted(set(stamps))
|
||
stamp = None
|
||
if uniq:
|
||
stamp = uniq[0] if len(uniq) == 1 else ("MIXED:" + str(len(uniq)))
|
||
return {"peaks": peaks, "anchors": sum(1 for l in payload if ANCHOR in l),
|
||
"stamp": stamp, "lines": len(payload)}
|
||
|
||
|
||
def measurement_gate(blocks, label):
|
||
"""(ok, reasons)。任一条不满足就整轮作废：宁可不出结论，也不出错结论。"""
|
||
reasons = []
|
||
if len(blocks) < MIN_CIRCLES:
|
||
reasons.append("%s: only %d circle(s), need >= %d for a peak-vs-peak comparison"
|
||
% (label, len(blocks), MIN_CIRCLES))
|
||
stamps = set()
|
||
for i, b in enumerate(blocks):
|
||
n = i + 1
|
||
if b["anchors"] != 1:
|
||
reasons.append("%s block#%d: boot anchor count=%d (want exactly 1) -> capture did "
|
||
"not start at a reset, or the windows overlap" % (label, n, b["anchors"]))
|
||
if b["lines"] < 20:
|
||
reasons.append("%s block#%d: only %d payload line(s) -> the collector was not "
|
||
"carrying this boot (a respawned daemon still answers 'ok')"
|
||
% (label, n, b["lines"]))
|
||
if not b["peaks"]:
|
||
reasons.append("%s block#%d: zero [STACK] lines (a missing line is not zero used)"
|
||
% (label, n))
|
||
if not b["stamp"]:
|
||
reasons.append("%s block#%d: no 'fw build:' identity stamp -> cannot prove which "
|
||
"image these numbers belong to" % (label, n))
|
||
elif str(b["stamp"]).startswith("MIXED"):
|
||
reasons.append("%s block#%d: %s identity stamps inside one block -> the capture "
|
||
"crossed two images" % (label, n, b["stamp"]))
|
||
else:
|
||
stamps.add(b["stamp"])
|
||
# 同一轮内部出现多个身份戳 = 采集窗跨了一次重烧，这轮的数字不属于任何一份镜像 -> 作废。
|
||
# 注意：**跨轮**戳不同是正常的（那正是要比的两份镜像），不能按作废处理，见 evaluate()。
|
||
if len(stamps) > 1:
|
||
reasons.append("%s: this round mixes %d identity stamps %s -> one circle was reflashed "
|
||
"mid-round, these peaks belong to no single image"
|
||
% (label, len(stamps), sorted(stamps)))
|
||
return (not reasons), reasons
|
||
|
||
|
||
def load_tolerances(path):
|
||
"""容差表：每行 task<TAB>bytes[<TAB>来源]，# 开头为注释。缺文件则全用默认值。"""
|
||
tol = {}
|
||
if not path or not os.path.isfile(path):
|
||
return tol, "no tolerance file -> default %d B for every task" % DEFAULT_TOL
|
||
for raw in open(path, "r", encoding="utf-8").read().splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
parts = line.split("\t")
|
||
if len(parts) >= 2 and parts[1].isdigit():
|
||
tol[parts[0]] = int(parts[1])
|
||
return tol, "loaded %d task tolerance(s) from %s" % (len(tol), os.path.basename(path))
|
||
|
||
|
||
def worst_of(blocks):
|
||
"""把本轮各圈压成"每任务最大值"（峰值就该取最坏一次）。"""
|
||
agg = {}
|
||
for b in blocks:
|
||
for name, info in b["peaks"].items():
|
||
prev = agg.get(name)
|
||
if prev is None or info["used"] > prev["used"]:
|
||
agg[name] = dict(info)
|
||
return agg
|
||
|
||
|
||
def decide(cur, base, tol_map):
|
||
"""跨轮趋势：只出 WARN。FAIL 的权限只给绝对底线，硬线归 stack_report.py。"""
|
||
rows, warns = [], []
|
||
for name in sorted(cur):
|
||
info = cur[name]
|
||
bl = base.get(name)
|
||
tol = tol_map.get(name, DEFAULT_TOL)
|
||
pct = (100 * info["free"] // info["cfg"]) if info["cfg"] else 0
|
||
row = {"task": name, "cur_max": info["used"], "cfg": info["cfg"], "free_min": info["free"],
|
||
"floor_pct": pct, "tol": tol, "base_max": (bl or {}).get("used"),
|
||
"delta": (info["used"] - bl["used"]) if bl else None, "level": "PASS"}
|
||
if pct < RUNTIME_FLOOR_PCT:
|
||
row["level"] = "FAIL"
|
||
warns.append("%s free_min=%d/%d (%d%%) below the %d%% runtime floor"
|
||
% (name, info["free"], info["cfg"], pct, RUNTIME_FLOOR_PCT))
|
||
elif bl is not None and (info["used"] - bl["used"]) > tol:
|
||
row["level"] = "WARN"
|
||
warns.append("%s peak rose %d B over baseline max (%d -> %d, tol=%d B): trend only, "
|
||
"the blocking decision is stack_report.py's margin verdict"
|
||
% (name, info["used"] - bl["used"], bl["used"], info["used"], tol))
|
||
rows.append(row)
|
||
return rows, warns
|
||
|
||
|
||
def evaluate(cur_blocks, base_blocks, tol_map):
|
||
"""(exit_code, verdict, 报告行)。顺序：先门，再绝对底线，最后才是趋势告警。"""
|
||
out = []
|
||
ok_cur, why_cur = measurement_gate(cur_blocks, "current")
|
||
if not ok_cur:
|
||
out.append("[VOID_MEASUREMENT] this round is not usable:")
|
||
out.extend(" - " + r for r in why_cur)
|
||
out.append(" -> fix the measurement first (open the port and watch rx_lines grow,")
|
||
out.append(" capture from a reset, prove the identity stamp). A dead collector must")
|
||
out.append(" not become a stack FAIL, and this round must NOT enter the ledger.")
|
||
return 2, "VOID_MEASUREMENT", out
|
||
if base_blocks:
|
||
ok_base, why_base = measurement_gate(base_blocks, "baseline")
|
||
if not ok_base:
|
||
out.append("[WARN] baseline round failed its own gate -> trend skipped this run:")
|
||
out.extend(" - " + r for r in why_base)
|
||
out.append(" (we refuse to compare against numbers we cannot trust; the current")
|
||
out.append(" round still gets its absolute-floor check)")
|
||
base_blocks = []
|
||
cur = worst_of(cur_blocks)
|
||
base = worst_of(base_blocks) if base_blocks else {}
|
||
rows, warns = decide(cur, base, tol_map)
|
||
cur_stamp = cur_blocks[0]["stamp"]
|
||
base_stamp = base_blocks[0]["stamp"] if base_blocks else None
|
||
cross_image = bool(base_blocks) and cur_stamp != base_stamp
|
||
out.append("%-9s %8s %9s %6s %7s %6s %5s %s"
|
||
% ("task", "cur_max", "base_max", "delta", "cfg", "free%", "tol", "level"))
|
||
for r in rows:
|
||
out.append("%-9s %8d %9s %6s %7d %5d%% %5d %s" % (
|
||
r["task"], r["cur_max"],
|
||
"-" if r["base_max"] is None else r["base_max"],
|
||
"-" if r["delta"] is None else ("%+d" % r["delta"]),
|
||
r["cfg"], r["floor_pct"], r["tol"], r["level"]))
|
||
fails = [w for w in warns if "floor" in w]
|
||
trend = [w for w in warns if "floor" not in w]
|
||
for w in fails:
|
||
out.append("[FAIL] " + w)
|
||
for w in trend:
|
||
out.append("[WARN] " + w)
|
||
if not base_blocks:
|
||
out.append("note: no usable baseline -> trend comparison off this run (levels come from "
|
||
"the absolute runtime floor only)")
|
||
elif cross_image:
|
||
out.append("note: CROSS-IMAGE comparison (current=%s vs baseline=%s). Different stamps "
|
||
"are normal here -- but a delta is only attributable to a given change if the "
|
||
"source diff between those two builds is attached; write BOTH stamps into the "
|
||
"ledger row, never the commit SHA alone (SHAs moved three times on 2026-10-26)."
|
||
% (cur_stamp[0], base_stamp[0]))
|
||
else:
|
||
out.append("note: same-image comparison (stamps match: %s) -> a delta over tolerance is "
|
||
"measurement noise or an unexplained regression; re-measure before writing it up."
|
||
% (cur_stamp[0],))
|
||
out.append("hard line unchanged: run scripts/stack_report.py; its per-task verdict "
|
||
"(margin >= max(isr_margin, 30% of cfg), computed from the larger of the static "
|
||
"chain and the runtime peak) is the only stack criterion that blocks a commit.")
|
||
if fails:
|
||
return 1, "FAIL", out
|
||
if trend:
|
||
return 0, "PASS_WITH_WARN", out
|
||
return 0, "PASS", out
|
||
|
||
|
||
def synth(stamp, anchors, entries, filler=24):
|
||
"""造一段日志文本（selftest 用）。entries = [(task, used, cfg, free)]。"""
|
||
L = ["[RX] 2026-10-27T10:00:%02d.000+08:00 [PROF] filler=%d" % (i % 60, i) for i in range(filler)]
|
||
if stamp:
|
||
L.append("[RX] 2026-10-27T10:00:00.000+08:00 [2026-10-27 02:00:00][APP] fw build: "
|
||
+ stamp + " suite=NONE dstr=off")
|
||
for _ in range(anchors):
|
||
L.append("[RX] 2026-10-27T10:00:00.100+08:00 [2026-10-27 02:00:00][APP] reset cause: "
|
||
"software (NVIC_SystemReset) (CSR=0x14000000)")
|
||
for task, used, cfg, free in entries:
|
||
L.append("[RX] 2026-10-27T10:00:01.000+08:00 [2026-10-27 02:00:01][STACK] %s "
|
||
"peak_used=%d/%d free_min=%d (%d%%)" % (task, used, cfg, free, 100 * free // cfg))
|
||
return "\n".join(L) + "\n"
|
||
|
||
|
||
def selftest(out=print):
|
||
"""证明四档都能真的亮——不能失败的判据等于没有判据。"""
|
||
cases = []
|
||
|
||
def chk(name, got, want):
|
||
cases.append((name, got == want, "got=%r want=%r" % (got, want)))
|
||
|
||
tasks = ("default", "net", "ftp", "ui", "timesync", "adc", "iec104", "protect", "test")
|
||
clean = [(t, 1000, 4096, 3096) for t in tasks]
|
||
S = "Oct 26 2026 22:52:09"
|
||
b1 = parse_block(synth(S, 1, clean))
|
||
b2 = parse_block(synth(S, 1, [(t, 1040, 4096, 3056) for t in tasks]))
|
||
b3 = parse_block(synth(S, 1, [(t, 990, 4096, 3106) for t in tasks]))
|
||
chk("three clean circles -> PASS", evaluate([b1, b2, b3], [], {})[1], "PASS")
|
||
base72 = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 4116, 16384, 12268)]))]
|
||
cur72 = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 4188, 16384, 12196)]))] * 3
|
||
chk("the measured 72 B swing is not a red flag", evaluate(cur72, base72, {"ui": 160})[1], "PASS")
|
||
cur_big = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 4388, 16384, 11996)]))] * 3
|
||
base72r = base72 * 3 # 基线轮也要 >=3 圈：单圈基线是噪声样本，不许拿它比
|
||
chk("delta over tolerance = WARN not FAIL", evaluate(cur_big, base72r, {"ui": 160})[1], "PASS_WITH_WARN")
|
||
chk("single-circle baseline is refused (trend skipped, no false red)",
|
||
evaluate(cur_big, base72, {"ui": 160})[1], "PASS")
|
||
chk("cross-round different stamps still compares (with attribution note)",
|
||
0, evaluate([parse_block(synth("Oct 26 2026 22:52:09", 1, [("ui", 4400, 16384, 11984)]))] * 3,
|
||
base72r, {"ui": 160})[0])
|
||
low = [parse_block(synth("Oct 26 2026 19:43:23", 1, [("ui", 13000, 16384, 3384)]))] * 3
|
||
chk("runtime floor breach = FAIL", evaluate(low, [], {})[1], "FAIL")
|
||
dead = parse_block(synth(S, 0, []))
|
||
chk("dead collector -> VOID (not FAIL)", evaluate([dead] * 3, [], {})[1], "VOID_MEASUREMENT")
|
||
two = parse_block(synth(S, 2, clean))
|
||
chk("two boot anchors -> VOID", evaluate([two, two, two], [], {})[1], "VOID_MEASUREMENT")
|
||
nostamp = parse_block(synth("", 1, clean))
|
||
chk("missing identity stamp -> VOID", evaluate([nostamp] * 3, [], {})[1], "VOID_MEASUREMENT")
|
||
mixed = parse_block(synth(S, 1, clean) + synth("Oct 25 2026 09:00:00", 1, clean))
|
||
chk("mixed stamps in one block -> VOID", evaluate([mixed] * 3, [], {})[1], "VOID_MEASUREMENT")
|
||
cross = parse_block(synth("Oct 25 2026 09:00:00", 1, clean))
|
||
chk("round that mixes two stamps -> VOID", evaluate([b1, b2, cross], [], {})[1], "VOID_MEASUREMENT")
|
||
chk("only two circles -> VOID", evaluate([b1, b2], [], {})[1], "VOID_MEASUREMENT")
|
||
chk("unusable baseline -> trend skipped, floor still enforced",
|
||
evaluate([b1, b2, b3], [dead], {})[1], "PASS")
|
||
chk("peak takes the worst circle, not the last", worst_of([b1, b3, b2])["ui"]["used"], 1040)
|
||
|
||
bad = 0
|
||
for name, ok, detail in cases:
|
||
out("%-5s %s%s" % ("[OK]" if ok else "[FAIL]", name, "" if ok else " " + detail))
|
||
if not ok:
|
||
bad += 1
|
||
out("SELFTEST: %d case(s), %d failure(s) -> %s" % (len(cases), bad, "PASS" if bad == 0 else "FAIL"))
|
||
return 0 if bad == 0 else 1
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="bench stack water-level trend gate (task-32 ruling)")
|
||
ap.add_argument("--block", action="append", default=[],
|
||
help="current-round boot log block, repeatable (need >= %d)" % MIN_CIRCLES)
|
||
ap.add_argument("--baseline-block", action="append", default=[],
|
||
help="previous-round boot log block, repeatable; omit = trend comparison off")
|
||
ap.add_argument("--tolerance-tsv", default="scripts/stack_tolerance.tsv",
|
||
help="per-task tolerance table (task TAB bytes); missing task -> %d B" % DEFAULT_TOL)
|
||
ap.add_argument("--selftest", action="store_true", help="run built-in classification cases and exit")
|
||
a = ap.parse_args(argv)
|
||
if a.selftest:
|
||
return selftest()
|
||
if not a.block:
|
||
print("usage error: give at least one --block (or --selftest)")
|
||
return 3
|
||
tol_map, note = load_tolerances(a.tolerance_tsv)
|
||
blocks = [parse_block(open(p, "r", encoding="utf-8", errors="replace").read()) for p in a.block]
|
||
base = [parse_block(open(p, "r", encoding="utf-8", errors="replace").read()) for p in a.baseline_block]
|
||
print("stack trend gate -- %s (floor=%d%% of cfg, min circles=%d)" % (note, RUNTIME_FLOOR_PCT, MIN_CIRCLES))
|
||
code, verdict, lines = evaluate(blocks, base, tol_map)
|
||
for ln in lines:
|
||
print(ln)
|
||
print("VERDICT: %s" % verdict)
|
||
return code
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|