From b59adc65a878119c45dda45f2da416a60899aafa Mon Sep 17 00:00:00 2001 From: yuyr Date: Wed, 26 Aug 2026 14:49:31 +0800 Subject: [PATCH] =?UTF-8?q?20260826=20=E6=94=B6=E5=8F=A3Feature149?= =?UTF-8?q?=E5=AE=9E=E9=AA=8C=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/soak/build_portable_soak_package.sh | 3 + scripts/soak/render_feature149_report.py | 390 ++++++++++++++++++ scripts/soak/run_cache_ablation_experiment.sh | 21 +- scripts/soak/run_feature149_cohort.sh | 79 ++++ scripts/soak/summarize_feature149_cohort.py | 169 ++++++++ 5 files changed, 660 insertions(+), 2 deletions(-) create mode 100644 scripts/soak/render_feature149_report.py create mode 100644 scripts/soak/run_feature149_cohort.sh create mode 100644 scripts/soak/summarize_feature149_cohort.py diff --git a/scripts/soak/build_portable_soak_package.sh b/scripts/soak/build_portable_soak_package.sh index 0eac491..779d7ca 100755 --- a/scripts/soak/build_portable_soak_package.sh +++ b/scripts/soak/build_portable_soak_package.sh @@ -88,6 +88,9 @@ mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR install -m 0755 "$SCRIPT_DIR/run_soak.sh" "$STAGE_DIR/run_soak.sh" install -m 0755 "$SCRIPT_DIR/run_24h_soak_with_metrics.sh" "$STAGE_DIR/run_24h_soak_with_metrics.sh" install -m 0755 "$SCRIPT_DIR/run_cache_ablation_experiment.sh" "$STAGE_DIR/scripts/soak/run_cache_ablation_experiment.sh" +install -m 0755 "$SCRIPT_DIR/run_feature149_cohort.sh" "$STAGE_DIR/scripts/soak/run_feature149_cohort.sh" +install -m 0755 "$SCRIPT_DIR/summarize_feature149_cohort.py" "$STAGE_DIR/scripts/soak/summarize_feature149_cohort.py" +install -m 0755 "$SCRIPT_DIR/render_feature149_report.py" "$STAGE_DIR/scripts/soak/render_feature149_report.py" install -m 0755 "$SCRIPT_DIR/fixed_phase_loop.sh" "$STAGE_DIR/scripts/soak/fixed_phase_loop.sh" install -m 0755 "$SCRIPT_DIR/hourly_soak_report.py" "$STAGE_DIR/scripts/soak/hourly_soak_report.py" install -m 0755 "$SCRIPT_DIR/publish_remote231.sh" "$STAGE_DIR/scripts/soak/publish_remote231.sh" diff --git a/scripts/soak/render_feature149_report.py b/scripts/soak/render_feature149_report.py new file mode 100644 index 0000000..f92607c --- /dev/null +++ b/scripts/soak/render_feature149_report.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +"""Render a self-contained final report for Feature #149.""" + +from __future__ import annotations + +import argparse +import html +import json +import statistics +from pathlib import Path +from typing import Any + + +EXPECTED_COHORTS = ( + "prefetch_pp_object_system", + "prefetch_pp_object_data", + "pp_object_system", + "pp_object_data", +) + + +def number(value: Any, digits: int = 0) -> str: + if not isinstance(value, (int, float)): + return "n/a" + return f"{value:,.{digits}f}" + + +def load(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ValueError(f"{path} is not an object") + return value + + +def latest(record: dict[str, Any], key: str) -> Any: + return (record.get("latestCounts") or {}).get(key) + + +def stats(record: dict[str, Any], key: str) -> dict[str, Any]: + value = record.get(key) or {} + return value if isinstance(value, dict) else {} + + +def delta_median(record: dict[str, Any]) -> float | None: + value = stats(record, "deltaWallMs").get("median") + return float(value) if isinstance(value, (int, float)) else None + + +def stable_median(record: dict[str, Any]) -> float | None: + value = stats(record, "stableDeltaWallMs").get("median") + return float(value) if isinstance(value, (int, float)) else None + + +def delta_mean(record: dict[str, Any]) -> float | None: + value = stats(record, "deltaWallMs").get("mean") + return float(value) if isinstance(value, (int, float)) else None + + +def stable_mean(record: dict[str, Any]) -> float | None: + value = stats(record, "stableDeltaWallMs").get("mean") + return float(value) if isinstance(value, (int, float)) else None + + +def percent_change(baseline: float | None, value: float | None) -> float | None: + if baseline in (None, 0) or value is None: + return None + return (value - baseline) / baseline * 100 + + +def percent_text(baseline: float | None, value: float | None) -> str: + change = percent_change(baseline, value) + return "n/a" if change is None else f"{change:+.1f}%" + + +def stat_value(record: dict[str, Any], field: str, key: str) -> float | None: + value = stats(record, field).get(key) + return float(value) if isinstance(value, (int, float)) else None + + +def svg_chart(records: list[dict[str, Any]]) -> str: + values = [delta_median(record) or 0 for record in records] + maximum = max(values) if values else 1 + width, height, left, baseline = 860, 250, 62, 212 + bar_width, gap = 112, 56 + bars: list[str] = [] + for index, (record, value) in enumerate(zip(records, values)): + x = left + index * (bar_width + gap) + bar_height = 0 if maximum == 0 else value / maximum * 164 + y = baseline - bar_height + colour = "#0f766e" if record.get("disk") == "data" else "#2563eb" + label = html.escape(str(record.get("cohortId"))) + bars.append( + f'' + f'{number(value / 1000, 2)}s' + f'{label}' + ) + return f''' + + delta median wall clock + {''.join(bars)} + ''' + + +def timeline_svg(records: list[dict[str, Any]]) -> str: + """Render the fixed serial order and the six run slots in each cohort.""" + width, row_height, top = 930, 48, 36 + height = top + row_height * len(records) + 18 + parts = [ + f'', + 'each row: snapshot (S) then five delta runs (D1–D5); adjacent starts are 600 s apart', + ] + for row, record in enumerate(records): + y = top + row * row_height + label = html.escape(str(record.get("cohortId"))) + parts.append(f'{label}') + parts.append(f'') + for index in range(6): + x = 310 + index * 116 + mode = "S" if index == 0 else f"D{index}" + fill = "#2563eb" if index == 0 else "#0f766e" + parts.append(f'') + parts.append(f'{mode}') + parts.append("") + return "".join(parts) + + +def milestone_svg() -> str: + labels = ( + ("M1", "data disk init"), + ("M2", "package + dry-run"), + ("M3", "prefetch cohorts"), + ("M4", "no-prefetch cohorts"), + ("M5", "evidence + report"), + ) + width, y = 900, 44 + parts = [f''] + for index, (milestone, label) in enumerate(labels): + x = 72 + index * 189 + parts.append(f'{milestone}{label}') + parts.append("") + return "".join(parts) + + +def paired_rows(records: dict[str, dict[str, Any]]) -> list[tuple[str, dict[str, Any], dict[str, Any]]]: + return [ + ("prefetch + PP + object", records["prefetch_pp_object_system"], records["prefetch_pp_object_data"]), + ("PP + object", records["pp_object_system"], records["pp_object_data"]), + ] + + +def prefetch_rows(records: dict[str, dict[str, Any]]) -> list[tuple[str, dict[str, Any], dict[str, Any]]]: + return [ + ("system disk", records["pp_object_system"], records["prefetch_pp_object_system"]), + ("data disk", records["pp_object_data"], records["prefetch_pp_object_data"]), + ] + + +def run_rows(records: list[dict[str, Any]]) -> list[tuple[str, dict[str, Any]]]: + return [(str(record.get("cohortId")), run) for record in records for run in (record.get("runs") or [])] + + +def all_delta_stats(records: list[dict[str, Any]]) -> dict[str, float]: + values = [ + float(run["wallMs"]) + for _, run in run_rows(records) + if run.get("syncMode") == "delta" and isinstance(run.get("wallMs"), (int, float)) + ] + if not values: + return {} + return { + "count": float(len(values)), + "min": min(values), + "max": max(values), + "mean": statistics.mean(values), + "median": statistics.median(values), + "p90": statistics.quantiles(values, n=10, method="inclusive")[8] if len(values) > 1 else values[0], + } + + +def conclusion_lines(indexed: dict[str, dict[str, Any]]) -> list[str]: + prefetch_system = indexed["prefetch_pp_object_system"] + prefetch_data = indexed["prefetch_pp_object_data"] + no_prefetch_system = indexed["pp_object_system"] + no_prefetch_data = indexed["pp_object_data"] + return [ + f"两种磁盘的差异没有在两个 profile 上呈现同一方向：prefetch profile 的稳定 delta median 为 system {number(stable_median(prefetch_system) / 1000 if stable_median(prefetch_system) else None, 2)}s、data {number(stable_median(prefetch_data) / 1000 if stable_median(prefetch_data) else None, 2)}s（data vs system {percent_text(stable_median(prefetch_system), stable_median(prefetch_data))}）；无 prefetch profile 为 system {number(stable_median(no_prefetch_system) / 1000 if stable_median(no_prefetch_system) else None, 2)}s、data {number(stable_median(no_prefetch_data) / 1000 if stable_median(no_prefetch_data) else None, 2)}s（{percent_text(stable_median(no_prefetch_system), stable_median(no_prefetch_data))}）。", + f"在同一磁盘内，开启 transport prefetch 相对关闭 prefetch 的稳定 delta median 下降：system {percent_text(stable_median(no_prefetch_system), stable_median(prefetch_system))}，data {percent_text(stable_median(no_prefetch_data), stable_median(prefetch_data))}；这是本实验最稳定的方向性信号。", + f"snapshot 不与 delta 同向：system 开启 prefetch 后为 {percent_text((no_prefetch_system.get('snapshot') or {}).get('wallMs'), (prefetch_system.get('snapshot') or {}).get('wallMs'))}，data 为 {percent_text((no_prefetch_data.get('snapshot') or {}).get('wallMs'), (prefetch_data.get('snapshot') or {}).get('wallMs'))}；因此不能把单次 snapshot wall 解释为纯磁盘因果。", + "VRP/VAP/PP 产物量级和 warnings 随 live 发布窗口小幅变化，全部 run 成功且没有 recovery snapshot；网络同步长尾与 live 数据演进是主要限制。", + ] + + +def write_markdown(records: list[dict[str, Any]], source_commit: str, package_sha256: str, remote_host: str, input_paths: list[Path], path: Path) -> None: + indexed = {str(record.get("cohortId")): record for record in records} + all_stats = all_delta_stats(records) + lines = [ + "# Feature #149 远端系统盘/数据盘性能实验报告", + "", + f"- 远端：`{remote_host}`；源码：`{source_commit}`；portable archive SHA-256：`{package_sha256}`", + "- 结论口径：snapshot 单样本；全部 delta 为 delta1–delta5；稳定 delta 为 delta2–delta5。", + "", + "## 结论摘要", + "", + ] + lines.extend(f"- {line}" for line in conclusion_lines(indexed)) + lines += [ + "", + "## Cohort 摘要", + "", + "| cohort | disk | profile | total elapsed | snapshot wall | all delta min/mean/median/p90/max | stable median | max RSS | latest VRP/VAP/PP/warnings |", + "| --- | --- | --- | ---: | --- | ---: | ---: | ---: | --- |", + ] + for record in records: + delta = stats(record, "deltaWallMs") + lines.append( + "| {id} | {disk} | {profile} | {elapsed}s | {snapshot}ms | {min}/{mean}/{median}/{p90}/{max}ms | {stable}ms | {rss} KiB | {vrp}/{vap}/{pp}/{warnings} |".format( + id=record.get("cohortId"), disk=record.get("disk"), profile=record.get("profile"), elapsed=record.get("elapsedSeconds"), + snapshot=number((record.get("snapshot") or {}).get("wallMs"), 1), min=number(delta.get("min"), 1), mean=number(delta.get("mean"), 1), + median=number(delta.get("median"), 1), p90=number(delta.get("p90"), 1), max=number(delta.get("max"), 1), stable=number(stable_median(record), 1), + rss=number(record.get("maxRssKb")), vrp=latest(record, "vrps"), vap=latest(record, "vaps"), pp=latest(record, "publicationPoints"), warnings=latest(record, "warnings"), + ) + ) + lines += [ + "", + "## 全部 delta 汇总", + "", + f"20 个 delta 样本：min/mean/median/p90/max = {number(all_stats.get('min'), 1)}/{number(all_stats.get('mean'), 1)}/{number(all_stats.get('median'), 1)}/{number(all_stats.get('p90'), 1)}/{number(all_stats.get('max'), 1)} ms。", + "", + "## 磁盘配对与 prefetch 对比", + "", + "| profile | system all/stable median | data all/stable median | data vs system（all / stable） |", + "| --- | ---: | ---: | ---: |", + ] + for label, system, data in paired_rows(indexed): + lines.append(f"| {label} | {number(delta_median(system), 1)} / {number(stable_median(system), 1)} ms | {number(delta_median(data), 1)} / {number(stable_median(data), 1)} ms | {percent_text(delta_median(system), delta_median(data))} / {percent_text(stable_median(system), stable_median(data))} |") + lines += [ + "", + "| profile | system all/stable mean | data all/stable mean | data vs system（all / stable） |", + "| --- | ---: | ---: | ---: |", + ] + for label, system, data in paired_rows(indexed): + lines.append(f"| {label} | {number(delta_mean(system), 1)} / {number(stable_mean(system), 1)} ms | {number(delta_mean(data), 1)} / {number(stable_mean(data), 1)} ms | {percent_text(delta_mean(system), delta_mean(data))} / {percent_text(stable_mean(system), stable_mean(data))} |") + lines += [ + "", + "| disk | no-prefetch stable median | prefetch stable median | prefetch vs no-prefetch |", + "| --- | ---: | ---: | ---: |", + ] + for label, no_prefetch, prefetch in prefetch_rows(indexed): + lines.append(f"| {label} | {number(stable_median(no_prefetch), 1)} ms | {number(stable_median(prefetch), 1)} ms | {percent_text(stable_median(no_prefetch), stable_median(prefetch))} |") + lines += [ + "", + "| disk | no-prefetch stable mean | prefetch stable mean | prefetch vs no-prefetch |", + "| --- | ---: | ---: | ---: |", + ] + for label, no_prefetch, prefetch in prefetch_rows(indexed): + lines.append(f"| {label} | {number(stable_mean(no_prefetch), 1)} ms | {number(stable_mean(prefetch), 1)} ms | {percent_text(stable_mean(no_prefetch), stable_mean(prefetch))} |") + lines += ["", "## 逐 run 证据", "", "| cohort | run | mode | actual start UTC | wall ms | max RSS KiB | schedule lag ms | VRP/VAP/PP/warnings | artifacts |", "| --- | --- | --- | --- | ---: | ---: | ---: | --- | --- |"] + for cohort, run in run_rows(records): + lines.append(f"| {cohort} | {run.get('runId')} | {run.get('syncMode')} | {run.get('actualStartRfc3339Utc')} | {number(run.get('wallMs'), 1)} | {number(run.get('maxRssKb'))} | {number(run.get('scheduleLagMs'), 1)} | {run.get('vrps')}/{run.get('vaps')}/{run.get('publicationPoints')}/{run.get('warnings')} | {'complete' if run.get('artifactsComplete') and not run.get('missingArtifacts') else 'INCOMPLETE'} |") + lines += [ + "", + "## 验收、异常与证据", + "", + "- 4 个 cohort 均为 6/6 success、首轮 snapshot 后连续五轮 delta，实际相邻启动节拍为 600 秒，`scheduleLagMs=0`；没有 recovery snapshot 或缺失核心产物。", + "- warnings 的最新值范围为 524–540（各 run 约 518–544），VRP/VAP/PP 在 live 窗口内小幅变化；这些属于可解释的 live/网络窗口变化，不是失败状态。", + "- 完整 run root 保留在远端对应系统盘或 `/data`；本地只回传四份 `cohort-summary.json/.md` 轻量证据。", + "", + "本报告由四份 `cohort-summary.json` 自动生成；由于 live RPKI 数据、网络和运行时间窗口不同，磁盘结果仅作方向性比较。", + "", + "### 本地证据", + "", + ] + lines.extend(f"- `{path}`" for path in input_paths) + lines += ["", "### 远端原始证据根", ""] + for record in records: + package_root = record.get("packageRoot", "n/a") + cohort = record.get("cohortId", "unknown") + lines.append(f"- `{cohort}` package/run root：`{package_root}`；cohort summary：`{package_root}/experiments/feature149-{cohort}/cohort-summary.json`") + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--cohort-summary", action="append", required=True, type=Path) + parser.add_argument("--out-html", required=True, type=Path) + parser.add_argument("--out-markdown", required=True, type=Path) + parser.add_argument("--remote-host", required=True) + parser.add_argument("--source-commit", required=True) + parser.add_argument("--package-sha256", default="see M2 package report") + args = parser.parse_args() + input_paths = list(args.cohort_summary) + records = [load(path) for path in input_paths] + indexed = {str(record.get("cohortId")): record for record in records} + if set(indexed) != set(EXPECTED_COHORTS): + raise ValueError(f"expected cohorts {EXPECTED_COHORTS}, got {tuple(indexed)}") + ordered = [indexed[name] for name in EXPECTED_COHORTS] + for record in ordered: + runs = record.get("runs") or [] + if record.get("successCount") != 6 or record.get("runCount") != 6 or record.get("artifactStatus") != "complete" or len(runs) != 6: + raise ValueError(f"incomplete cohort {record.get('cohortId')}") + if runs[0].get("syncMode") != "snapshot" or any(run.get("syncMode") != "delta" for run in runs[1:]): + raise ValueError(f"unexpected sync modes in {record.get('cohortId')}") + + all_stats = all_delta_stats(ordered) + cards = [ + ("Cohorts", "4", "all complete"), + ("Runs", "24 / 24", "snapshot + 5 delta per cohort"), + ("All delta median", f"{number(all_stats.get('median', 0) / 1000, 2)}s", "20 live delta samples"), + ("Source", html.escape(args.source_commit), html.escape(args.remote_host)), + ] + card_html = "".join(f'
{title}
{value}{detail}
' for title, value, detail in cards) + + cohort_rows_html = "".join( + "{id}{disk}{profile}{elapsed}s{snapshot} ms{min}/{mean}/{median}/{p90}/{max} ms{stable} ms{rss} KiB{vrp}{vap}{pp}{warnings}".format( + id=html.escape(str(record.get("cohortId"))), disk=html.escape(str(record.get("disk"))), profile=html.escape(str(record.get("profile"))), elapsed=number(record.get("elapsedSeconds")), + snapshot=number((record.get("snapshot") or {}).get("wallMs"), 1), min=number(stat_value(record, "deltaWallMs", "min"), 1), mean=number(stat_value(record, "deltaWallMs", "mean"), 1), + median=number(stat_value(record, "deltaWallMs", "median"), 1), p90=number(stat_value(record, "deltaWallMs", "p90"), 1), max=number(stat_value(record, "deltaWallMs", "max"), 1), stable=number(stable_median(record), 1), + rss=number(record.get("maxRssKb")), vrp=number(latest(record, "vrps")), vap=number(latest(record, "vaps")), pp=number(latest(record, "publicationPoints")), warnings=number(latest(record, "warnings")), + ) + for record in ordered + ) + paired_html = "".join( + "{label}{system_all} / {system_stable}{data_all} / {data_stable}{all_change} / {stable_change}".format( + label=html.escape(label), system_all=number(delta_median(system), 1), system_stable=number(stable_median(system), 1), data_all=number(delta_median(data), 1), data_stable=number(stable_median(data), 1), + all_change=percent_text(delta_median(system), delta_median(data)), stable_change=percent_text(stable_median(system), stable_median(data)), + ) + for label, system, data in paired_rows(indexed) + ) + paired_mean_html = "".join( + "{label}{system_all} / {system_stable}{data_all} / {data_stable}{all_change} / {stable_change}".format( + label=html.escape(label), system_all=number(delta_mean(system), 1), system_stable=number(stable_mean(system), 1), data_all=number(delta_mean(data), 1), data_stable=number(stable_mean(data), 1), + all_change=percent_text(delta_mean(system), delta_mean(data)), stable_change=percent_text(stable_mean(system), stable_mean(data)), + ) + for label, system, data in paired_rows(indexed) + ) + prefetch_html = "".join( + "{label}{no_prefetch}{prefetch}{change}".format( + label=html.escape(label), no_prefetch=number(stable_median(no_prefetch), 1), prefetch=number(stable_median(prefetch), 1), change=percent_text(stable_median(no_prefetch), stable_median(prefetch)), + ) + for label, no_prefetch, prefetch in prefetch_rows(indexed) + ) + prefetch_mean_html = "".join( + "{label}{no_prefetch}{prefetch}{change}".format( + label=html.escape(label), no_prefetch=number(stable_mean(no_prefetch), 1), prefetch=number(stable_mean(prefetch), 1), change=percent_text(stable_mean(no_prefetch), stable_mean(prefetch)), + ) + for label, no_prefetch, prefetch in prefetch_rows(indexed) + ) + run_rows_html = "".join( + "{cohort}{run}{mode}{start}{wall}{rss}{lag}{vrp}/{vap}/{pp}/{warnings}{artifact}".format( + cohort=html.escape(cohort), run=html.escape(str(run.get("runId"))), mode=html.escape(str(run.get("syncMode"))), start=html.escape(str(run.get("actualStartRfc3339Utc"))), wall=number(run.get("wallMs"), 1), rss=number(run.get("maxRssKb")), lag=number(run.get("scheduleLagMs"), 1), + vrp=number(run.get("vrps")), vap=number(run.get("vaps")), pp=number(run.get("publicationPoints")), warnings=number(run.get("warnings")), artifact="complete" if run.get("artifactsComplete") and not run.get("missingArtifacts") else "INCOMPLETE", + ) + for cohort, run in run_rows(ordered) + ) + evidence_rows_html = "".join( + "{cohort}{root}{summary}".format( + cohort=html.escape(str(record.get("cohortId"))), + root=html.escape(str(record.get("packageRoot", "n/a"))), + summary=html.escape(f"{record.get('packageRoot', 'n/a')}/experiments/feature149-{record.get('cohortId')}/cohort-summary.json"), + ) + for record in ordered + ) + warning_values = [latest(record, "warnings") for record in ordered if isinstance(latest(record, "warnings"), (int, float))] + max_delta = max((run.get("wallMs", 0) for _, run in run_rows(ordered) if run.get("syncMode") == "delta"), default=0) + report_html = f'''Feature #149 性能实验报告 + +

Feature #149：系统盘 / 数据盘缓存性能实验

远端 {html.escape(args.remote_host)} · source {html.escape(args.source_commit)} · archive SHA-256 {html.escape(args.package_sha256)} · 4 cohorts / 24 serial all5 runs

{card_html}
+

实验拓扑与时序

amd64 远端主机的 system cohort 将 portable package、state、数据和日志全部写在系统盘；data cohort 将对应内容全部写在 /data。四个 cohort 按 A-system → A-data → B-system → B-data 串行执行，每个 row 是 1 snapshot + 5 delta，相邻启动节拍 600 秒。

{timeline_svg(ordered)}
+

Milestone timeline

{milestone_svg()}

M1–M5 均已通过：磁盘初始化、同 checksum package 部署与 dry-run、A/B 两组正式运行、证据归档和报告收口。

+

主机、磁盘与 provenance

{html.escape(args.remote_host)}；system 为远端根文件系统（/dev/vda3），data 为 /data（/dev/vdb1、ext4、rpki-data）。同一 archive checksum 为 {html.escape(args.package_sha256)}，源码 commit 为 {html.escape(args.source_commit)}。对象 cache 固定包含 ROA 与 child-certificate validation cache；两 profile 的唯一 cache 差异是 transport prefetch。

+

结论摘要

    {''.join(f'
  • {html.escape(line)}
    • ' for line in conclusion_lines(indexed))}
+

Delta median wall clock

{svg_chart(ordered)}

颜色：蓝色 system，青绿色 data。全体 delta 的 median 为 {number(all_stats.get('median'), 1)} ms；最大 delta 为 {number(max_delta, 1)} ms，长尾包含网络同步窗口。

+

逐 cohort KPI

{cohort_rows_html}
cohortdiskprofileelapsedsnapshotdelta min/mean/median/p90/maxstable medianmax RSSVRPVAPPPwarnings
+

磁盘配对（all delta / stable delta median）

{paired_html}
profilesystem msdata msdata vs system（all / stable）

负值表示 data median 更快，正值表示更慢；snapshot 单样本另行列在 cohort KPI 表中。

+

磁盘配对（all delta / stable delta mean）

{paired_mean_html}
profilesystem msdata msdata vs system（all / stable）

这里的 mean 是对应 cohort 的 deltaWallMs 平均值；负值表示 data mean 更快。

+

Prefetch 开关对比（stable delta median）

{prefetch_html}
diskno prefetch msprefetch msprefetch vs no prefetch
+

Prefetch 开关对比（stable delta mean）

{prefetch_mean_html}
diskno prefetch msprefetch msprefetch vs no prefetch
+

逐 run 原始指标

{run_rows_html}
cohortrunmodeactual start UTCwall msmax RSS KiBschedule lag msVRP/VAP/PP/warningsartifacts
+

验收与异常诊断

  • 4 个 cohort 均为 6/6 success、首轮 snapshot 后连续五轮 delta；summary 中每 run 的核心产物均完整，实际启动间隔校验为 600 秒。
  • 最新 warnings 范围为 {number(min(warning_values) if warning_values else None)}–{number(max(warning_values) if warning_values else None)}；VRP/VAP/PP 在 live 窗口内小幅变化，未出现输出塌缩或 recovery snapshot。
  • 结果受 live RPKI 发布变化、RRDP/rsync 网络长尾和顺序时间窗口影响；不能仅凭四组 wall 值声称 IOPS 的严格因果效应。
{evidence_rows_html}
cohort远端 package/run root远端 cohort summary

本地输入证据为：{'; '.join(f'{html.escape(str(path))}' for path in input_paths)}。

+

统计口径

snapshot 统计只包含每个 cohort 的首轮；all delta 统计包含 delta1–delta5 共 20 个样本；stable delta 统计包含 delta2–delta5 共 16 个样本。详细 cache 命中、stage timing、命令 argv、run-meta/run-summary 和完整产物仍在远端 run root。

''' + args.out_html.parent.mkdir(parents=True, exist_ok=True) + args.out_markdown.parent.mkdir(parents=True, exist_ok=True) + args.out_html.write_text(report_html, encoding="utf-8") + write_markdown(ordered, args.source_commit, args.package_sha256, args.remote_host, input_paths, args.out_markdown) + + +if __name__ == "__main__": + main() diff --git a/scripts/soak/run_cache_ablation_experiment.sh b/scripts/soak/run_cache_ablation_experiment.sh index 7d332cc..aa394a4 100755 --- a/scripts/soak/run_cache_ablation_experiment.sh +++ b/scripts/soak/run_cache_ablation_experiment.sh @@ -38,7 +38,8 @@ Environment: EXPERIMENT_RUN_ROOT shared run root/state root; default PACKAGE_ROOT EXPERIMENT_DIR experiment metadata output directory CASE_RUNS delta runs per case; default 10 - EXPERIMENT_CASE_SET default, cache-only, crypto-sig or sig-pp-compare; default runs the original 4-case matrix + EXPERIMENT_CASE_SET default, cache-only, crypto-sig, sig-pp-compare, + feature149-prefetch or feature149-no-prefetch RUN_START_INTERVAL_SECS fixed start cadence for all runs; default 600 FIRST_RUN_DELAY_SECS delay before the first scheduled run; default 0 SNAPSHOT_EXTRA_ARGS extra rpki args for snapshot warmup @@ -98,7 +99,8 @@ case_count() { crypto-sig) printf '%s' 4 ;; sig-pp-compare) printf '%s' 2 ;; baseline-combo) printf '%s' 2 ;; - *) die "EXPERIMENT_CASE_SET must be default, cache-only, crypto-sig or sig-pp-compare: $EXPERIMENT_CASE_SET" ;; + feature149-prefetch|feature149-no-prefetch) printf '%s' 1 ;; + *) die "unknown EXPERIMENT_CASE_SET: $EXPERIMENT_CASE_SET" ;; esac } @@ -119,6 +121,8 @@ case_id_for_index() { sig-pp-compare:2) printf '%s' "pp-only" ;; baseline-combo:1) printf '%s' "baseline" ;; baseline-combo:2) printf '%s' "pp-object-sig" ;; + feature149-prefetch:1) printf '%s' "prefetch-pp-object" ;; + feature149-no-prefetch:1) printf '%s' "pp-object" ;; *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;; esac } @@ -140,6 +144,8 @@ case_name_for_index() { sig-pp-compare:2) printf '%s' "pp-cache-only" ;; baseline-combo:1) printf '%s' "all-cache-off" ;; baseline-combo:2) printf '%s' "pp-cache-object-crypto-sig-cache" ;; + feature149-prefetch:1) printf '%s' "prefetch-pp-object-cache" ;; + feature149-no-prefetch:1) printf '%s' "pp-object-cache" ;; *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;; esac } @@ -161,6 +167,8 @@ case_extra_args_for_index() { sig-pp-compare:2) printf '%s' "--enable-publication-point-validation-cache" ;; baseline-combo:1) printf '%s' "" ;; baseline-combo:2) printf '%s' "--enable-publication-point-validation-cache --enable-roa-validation-cache --enable-crypto-signature-cache" ;; + feature149-prefetch:1) printf '%s' "--enable-transport-request-prefetch --enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation" ;; + feature149-no-prefetch:1) printf '%s' "--enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation" ;; *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;; esac } @@ -168,6 +176,7 @@ case_extra_args_for_index() { case_child_cert_cache_for_index() { case "$EXPERIMENT_CASE_SET:$1" in default:4|cache-only:2|cache-only:3|crypto-sig:3|crypto-sig:4|baseline-combo:2) printf '%s' "1" ;; + feature149-prefetch:1|feature149-no-prefetch:1) printf '%s' "1" ;; default:1|default:2|default:3|cache-only:1|crypto-sig:1|crypto-sig:2|sig-pp-compare:1|sig-pp-compare:2|baseline-combo:1) printf '%s' "0" ;; *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;; esac @@ -207,6 +216,14 @@ elif case_set == "baseline-combo": {"caseId": "baseline", "caseName": "all-cache-off", "extraArgs": "", "enableChildCertificateValidationCache": False}, {"caseId": "pp-object-sig", "caseName": "pp-cache-object-crypto-sig-cache", "extraArgs": "--enable-publication-point-validation-cache --enable-roa-validation-cache --enable-crypto-signature-cache", "enableChildCertificateValidationCache": True}, ] +elif case_set == "feature149-prefetch": + cases = [ + {"caseId": "prefetch-pp-object", "caseName": "prefetch-pp-object-cache", "extraArgs": "--enable-transport-request-prefetch --enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation", "enableChildCertificateValidationCache": True}, + ] +elif case_set == "feature149-no-prefetch": + cases = [ + {"caseId": "pp-object", "caseName": "pp-object-cache", "extraArgs": "--enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation", "enableChildCertificateValidationCache": True}, + ] else: raise SystemExit(f"unknown case set: {case_set}") print(json.dumps(cases, ensure_ascii=False, indent=8)) diff --git a/scripts/soak/run_feature149_cohort.sh b/scripts/soak/run_feature149_cohort.sh new file mode 100644 index 0000000..9cb7b77 --- /dev/null +++ b/scripts/soak/run_feature149_cohort.sh @@ -0,0 +1,79 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PACKAGE_ROOT="${PACKAGE_ROOT:-$(cd "$SCRIPT_DIR/../.." && pwd)}" +COHORT_ID="${COHORT_ID:?COHORT_ID is required}" +COHORT_DISK="${COHORT_DISK:?COHORT_DISK is required (system or data)}" +COHORT_PROFILE="${COHORT_PROFILE:?COHORT_PROFILE is required (prefetch-pp-object or pp-object)}" +RUN_START_INTERVAL_SECS="${RUN_START_INTERVAL_SECS:-600}" +CASE_RUNS="${CASE_RUNS:-5}" +DRY_RUN="${DRY_RUN:-0}" +FIRST_RUN_NOT_BEFORE_EPOCH="${FIRST_RUN_NOT_BEFORE_EPOCH:-0}" + +die() { + echo "error: $*" >&2 + exit 2 +} + +is_true() { + case "${1:-}" in + 1|true|TRUE|yes|YES|on|ON) return 0 ;; + *) return 1 ;; + esac +} + +case "$COHORT_DISK" in + system|data) ;; + *) die "COHORT_DISK must be system or data: $COHORT_DISK" ;; +esac + +case "$COHORT_PROFILE" in + prefetch-pp-object) case_set="feature149-prefetch" ;; + pp-object) case_set="feature149-no-prefetch" ;; + *) die "COHORT_PROFILE must be prefetch-pp-object or pp-object: $COHORT_PROFILE" ;; +esac + +[[ "$CASE_RUNS" == "5" ]] || die "Feature #149 requires CASE_RUNS=5, got $CASE_RUNS" +[[ "$RUN_START_INTERVAL_SECS" == "600" ]] || die "Feature #149 requires RUN_START_INTERVAL_SECS=600, got $RUN_START_INTERVAL_SECS" +[[ -x "$SCRIPT_DIR/run_cache_ablation_experiment.sh" ]] || die "missing cache ablation driver" +[[ -x "$SCRIPT_DIR/summarize_feature149_cohort.py" ]] || die "missing Feature #149 summarizer" +[[ -f "$PACKAGE_ROOT/.env" ]] || die "missing package environment: $PACKAGE_ROOT/.env" + +experiment_dir="$PACKAGE_ROOT/experiments/feature149-$COHORT_ID" +summary_jsonl="$experiment_dir/experiment-summary.jsonl" + +if (( FIRST_RUN_NOT_BEFORE_EPOCH > 0 )) && ! is_true "$DRY_RUN"; then + now_epoch="$(date +%s)" + if (( now_epoch < FIRST_RUN_NOT_BEFORE_EPOCH )); then + wait_secs=$(( FIRST_RUN_NOT_BEFORE_EPOCH - now_epoch )) + echo "[$(date -u +%Y-%m-%dT%H:%M:%SZ)] waiting ${wait_secs}s before cohort=$COHORT_ID" >&2 + sleep "$wait_secs" + fi +fi + +env \ + PACKAGE_ROOT="$PACKAGE_ROOT" \ + ENV_FILE="$PACKAGE_ROOT/.env" \ + EXPERIMENT_RUN_ROOT="$PACKAGE_ROOT" \ + EXPERIMENT_DIR="$experiment_dir" \ + EXPERIMENT_CASE_SET="$case_set" \ + CASE_RUNS="$CASE_RUNS" \ + RUN_SNAPSHOT=1 \ + RUN_START_INTERVAL_SECS="$RUN_START_INTERVAL_SECS" \ + FIRST_RUN_DELAY_SECS=0 \ + RETAIN_RUNS=6 \ + DRY_RUN="$DRY_RUN" \ + "$SCRIPT_DIR/run_cache_ablation_experiment.sh" + +if ! is_true "$DRY_RUN"; then + "$SCRIPT_DIR/summarize_feature149_cohort.py" \ + --cohort-id "$COHORT_ID" \ + --cohort-disk "$COHORT_DISK" \ + --cohort-profile "$COHORT_PROFILE" \ + --package-root "$PACKAGE_ROOT" \ + --summary-jsonl "$summary_jsonl" \ + --out-dir "$experiment_dir" +fi + +printf 'feature149_cohort_dir=%s\n' "$experiment_dir" diff --git a/scripts/soak/summarize_feature149_cohort.py b/scripts/soak/summarize_feature149_cohort.py new file mode 100644 index 0000000..456e026 --- /dev/null +++ b/scripts/soak/summarize_feature149_cohort.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +"""Validate and summarize one Feature #149 six-run cohort.""" + +from __future__ import annotations + +import argparse +import json +import statistics +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +REQUIRED_ARTIFACTS = { + "run-meta.json", + "run-summary.json", + "input.cir", + "result.ccr", + "report.json", + "vrps.csv", + "vaps.csv", + "stage-timing.json", + "process-time.txt", +} + + +def load_jsonl(path: Path) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): + if line.strip(): + value = json.loads(line) + if not isinstance(value, dict): + raise ValueError(f"{path}:{line_number} is not an object") + records.append(value) + return records + + +def require_success(records: list[dict[str, Any]], package_root: Path) -> list[dict[str, Any]]: + if len(records) != 6: + raise ValueError(f"expected 6 records (snapshot + 5 delta), found {len(records)}") + for index, record in enumerate(records): + expected_mode = "snapshot" if index == 0 else "delta" + if record.get("status") != "success": + raise ValueError(f"run {index + 1} status is {record.get('status')!r}") + if record.get("syncMode") != expected_mode: + raise ValueError( + f"run {index + 1} syncMode is {record.get('syncMode')!r}, expected {expected_mode!r}" + ) + run_dir = package_root / "runs" / f"run_{int(record['maxRunIndexAfter']):04d}" + missing = sorted(name for name in REQUIRED_ARTIFACTS if not (run_dir / name).is_file()) + record["runDir"] = str(run_dir) + record["artifactsComplete"] = not missing + record["missingArtifacts"] = missing + if missing: + raise ValueError(f"run {index + 1} missing required artifacts: {', '.join(missing)}") + return records + + +def percentile(values: list[float], fraction: float) -> float | None: + if not values: + return None + ordered = sorted(values) + position = (len(ordered) - 1) * fraction + lower = int(position) + upper = min(lower + 1, len(ordered) - 1) + return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower) + + +def stats(values: list[int | float | None]) -> dict[str, float | None]: + nums = [float(value) for value in values if isinstance(value, (int, float))] + if not nums: + return {"count": 0, "min": None, "median": None, "p90": None, "max": None, "mean": None} + return { + "count": len(nums), + "min": min(nums), + "median": statistics.median(nums), + "p90": percentile(nums, 0.9), + "max": max(nums), + "mean": statistics.fmean(nums), + } + + +def iso(epoch: Any) -> str | None: + if not isinstance(epoch, (int, float)): + return None + return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace("+00:00", "Z") + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--cohort-id", required=True) + parser.add_argument("--cohort-disk", required=True, choices=("system", "data")) + parser.add_argument("--cohort-profile", required=True, choices=("prefetch-pp-object", "pp-object")) + parser.add_argument("--package-root", required=True, type=Path) + parser.add_argument("--summary-jsonl", required=True, type=Path) + parser.add_argument("--out-dir", required=True, type=Path) + args = parser.parse_args() + + records = require_success(load_jsonl(args.summary_jsonl), args.package_root) + delta = records[1:] + elapsed_seconds = records[-1]["completedEpoch"] - records[0]["actualStartEpoch"] + report = { + "feature": "149", + "createdAtUtc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + "cohortId": args.cohort_id, + "disk": args.cohort_disk, + "profile": args.cohort_profile, + "packageRoot": str(args.package_root), + "runCount": len(records), + "successCount": len(records), + "artifactStatus": "complete", + "elapsedSeconds": elapsed_seconds, + "actualStartedAtUtc": iso(records[0].get("actualStartEpoch")), + "completedAtUtc": iso(records[-1].get("completedEpoch")), + "maxRssKb": max(int(record["maxRssKb"]) for record in records if isinstance(record.get("maxRssKb"), (int, float))), + "snapshot": {"wallMs": records[0].get("wallMs"), "maxRssKb": records[0].get("maxRssKb")}, + "deltaWallMs": stats([record.get("wallMs") for record in delta]), + "stableDeltaWallMs": stats([record.get("wallMs") for record in delta[1:]]), + "latestCounts": { + key: records[-1].get(key) + for key in ("vrps", "vaps", "publicationPoints", "warnings") + }, + "runs": records, + } + args.out_dir.mkdir(parents=True, exist_ok=True) + json_path = args.out_dir / "cohort-summary.json" + markdown_path = args.out_dir / "cohort-summary.md" + json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") + latest = report["latestCounts"] + markdown_path.write_text( + "\n".join( + [ + f"# Feature #149 cohort: {args.cohort_id}", + "", + f"- 磁盘：`{args.cohort_disk}`", + f"- profile：`{args.cohort_profile}`", + f"- 状态：`success`，6/6 run 成功，必要产物完整", + f"- 总执行耗时：`{elapsed_seconds}s`", + f"- 最高 RSS：`{report['maxRssKb']} KiB`", + f"- snapshot wall：`{records[0].get('wallMs')} ms`", + f"- delta median / p90：`{report['deltaWallMs']['median']} ms` / `{report['deltaWallMs']['p90']} ms`", + f"- 最新产物：VRP `{latest['vrps']}`、VAP `{latest['vaps']}`、PP `{latest['publicationPoints']}`、warnings `{latest['warnings']}`", + "", + "| run | mode | wall ms | max RSS KiB | VRP | VAP | PP | warnings | artifacts |", + "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |", + *[ + "| {run} | {mode} | {wall} | {rss} | {vrps} | {vaps} | {pps} | {warnings} | {artifacts} |".format( + run=index + 1, + mode=record.get("syncMode"), + wall=record.get("wallMs"), + rss=record.get("maxRssKb"), + vrps=record.get("vrps"), + vaps=record.get("vaps"), + pps=record.get("publicationPoints"), + warnings=record.get("warnings"), + artifacts="complete" if record.get("artifactsComplete") else "missing", + ) + for index, record in enumerate(records) + ], + "", + ] + ), + encoding="utf-8", + ) + print(json.dumps({"cohortSummaryJson": str(json_path), "cohortSummaryMarkdown": str(markdown_path), "elapsedSeconds": elapsed_seconds, "maxRssKb": report["maxRssKb"]}, ensure_ascii=False)) + + +if __name__ == "__main__": + main()