#!/usr/bin/env python3 """Plot PNG line charts from measure_pause_resume_concurrency.py results. 读取一个或多个 sweep JSON(探针脚本 --output 的产物),按并发 level 画 折线并直接输出 PNG:每张图一个文件,另附一张总览拼图。依赖 matplotlib(pip install matplotlib)。脚本会自动挑选常见 CJK 字体渲染 中文标签;主机没有任何中文字体时中文会显示为方框(数字与坐标不受影响)。 图表(有数据才输出,编号即文件名前缀): 01 客户端 pause 时延(p50/p95/avg) —— p50 稳、p95 涨 = 服务端排队; 02 客户端 resume 时延; 03 吞吐(客户端 pause、服务端 pause/resume ops/s) —— 曲线走平即 服务端饱和能力 μ; 04 平均时延交叉验证(服务端 pause/resume avg、客户端 pause avg、 Little 估计) —— server avg 相对上一级 >2x 的最小 R 自动竖线标注 为 μ 拐点候选; 05 pause / resume in-flight 峰值; 06 进程组 CPU 核数(server / ublk / firecracker,有 proc 数据时); 07 进程组 RSS 峰值(MiB,有 proc 数据时); 08 --command 三层分解(客户端 p50/p95、guest 纯执行、服务端 envd avg,有 command 数据时)。 多数据集叠加时:颜色区分类别,线型 + 标记区分数据集(例如 2.46 干净 / 2.37 干净 / 2.37+脏页 三轮对比)。 用法: python3 scripts/plot_pause_resume_sweep.py sweep-clean.json python3 scripts/plot_pause_resume_sweep.py \ sweep-246.json sweep-237.json sweep-237-dirty.json \ --label 2.46-clean 2.37-clean 2.37-dirty --output-dir plots """ from __future__ import annotations import argparse import json import math from pathlib import Path from typing import Any, Callable import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt # noqa: E402 from matplotlib import font_manager # noqa: E402 # Okabe-Ito 色盲安全色板,每个图内按系列顺序取用。 PALETTE = [ "#0072B2", # 蓝 "#E69F00", # 橙 "#009E73", # 绿 "#D55E00", # 朱红 "#CC79A7", # 粉 "#56B4E9", # 天蓝 ] # 多数据集叠加时以线型 + 标记区分;颜色留给类别。 RUN_LINESTYLES = ["-", "--", ":", "-."] RUN_MARKERS = ["o", "s", "^", "D"] CJK_FONT_CANDIDATES = [ "Noto Sans CJK SC", "Noto Sans CJK JP", "Noto Sans SC", "WenQuanYi Zen Hei", "WenQuanYi Micro Hei", "Droid Sans Fallback", "AR PL UMing CN", ] LevelGetter = Callable[[dict[str, Any]], float | None] # --------------------------------------------------------------------------- # 数据加载与取值器。 # --------------------------------------------------------------------------- def load_run(path: Path, label: str) -> dict[str, Any]: data = json.loads(path.read_text(encoding="utf-8")) levels = data.get("levels") if not isinstance(levels, list) or not levels: raise SystemExit(f"{path}: no 'levels' array; not a sweep result JSON?") return { "label": label, "path": path, "config": data.get("config") or {}, "started_at": data.get("started_at"), "by_level": {entry["level"]: entry for entry in levels}, } def _num(value: Any) -> float | None: return float(value) if isinstance(value, (int, float)) else None def client_stat(kind: str, stat: str) -> LevelGetter: def getter(entry: dict[str, Any]) -> float | None: return _num((entry.get(kind) or {}).get(stat)) return getter def top_stat(key: str) -> LevelGetter: def getter(entry: dict[str, Any]) -> float | None: return _num(entry.get(key)) return getter def proc_stat(group: str, key: str) -> LevelGetter: def getter(entry: dict[str, Any]) -> float | None: return _num(((entry.get("proc") or {}).get(group) or {}).get(key)) return getter def build_charts() -> list[dict[str, Any]]: """(文件名 slug, 标题, y 轴标签, [(系列名, 取值器)], 是否标注 μ 拐点)。""" return [ { "name": "client-pause-latency", "title": "客户端 pause 时延 vs 并发", "caption": "p50 稳定而 p95 随 R 上升 => 请求在服务端排队(尾部先恶化)。", "y_label": "pause 时延 (s)", "specs": [ ("pause p50", client_stat("client_pause", "p50")), ("pause p95", client_stat("client_pause", "p95")), ("pause avg", client_stat("client_pause", "avg")), ], }, { "name": "client-resume-latency", "title": "客户端 resume 时延 vs 并发", "caption": "resume 走 connect();与 pause 同源排队时会一起恶化。", "y_label": "resume 时延 (s)", "specs": [ ("resume p50", client_stat("client_resume", "p50")), ("resume p95", client_stat("client_resume", "p95")), ("resume avg", client_stat("client_resume", "avg")), ], }, { "name": "throughput", "title": "吞吐 vs 并发", "caption": "曲线走平时的纵坐标即服务端饱和服务能力 μ。", "y_label": "ops/s", "specs": [ ("客户端 pause ops/s", top_stat("client_pause_ops_per_sec")), ("服务端 pause ops/s", top_stat("server_pause_ops_per_sec")), ("服务端 resume ops/s", top_stat("server_resume_ops_per_sec")), ], }, { "name": "avg-latency-crosscheck", "title": "平均时延交叉验证 vs 并发", "caption": "服务端 pause avg 相对上一级 >2x 的最小 R 是 μ 拐点候选(图中竖线);" "Little 估计(inflight 峰值 ÷ 吞吐)≈ 客户端 avg 即排队模型成立。", "y_label": "平均时延 (s)", "knee": True, "specs": [ ("服务端 pause avg", top_stat("server_pause_avg_sec")), ("服务端 resume avg", top_stat("server_resume_avg_sec")), ("客户端 pause avg", client_stat("client_pause", "avg")), ("Little 估计", top_stat("little_estimate_sec")), ], }, { "name": "inflight-peak", "title": "in-flight 峰值 vs 并发", "caption": "同时处于执行中的 pause/resume 数;闭合系统下上界为 R 本身。", "y_label": "峰值 in-flight", "specs": [ ("pause in-flight 峰值", top_stat("server_pause_max_inflight")), ("resume in-flight 峰值", top_stat("server_resume_max_inflight")), ], }, { "name": "proc-cpu", "title": "进程组 CPU vs 并发", "caption": "server / ublk daemon / firecracker 三组进程的平均 CPU 核数;" "接近物理核数即资源饱和。", "y_label": "CPU (核)", "specs": [ ("server", proc_stat("server", "cpu_cores_avg")), ("ublk daemon", proc_stat("ublk", "cpu_cores_avg")), ("firecracker", proc_stat("firecracker", "cpu_cores_avg")), ], }, { "name": "proc-rss-peak", "title": "进程组 RSS 峰值 vs 并发", "caption": "三组进程的 RSS 峰值(MiB);firecracker 随在线 VM 数近似线性。", "y_label": "RSS 峰值 (MiB)", "specs": [ ("server", proc_stat("server", "rss_mib_peak")), ("ublk daemon", proc_stat("ublk", "rss_mib_peak")), ("firecracker", proc_stat("firecracker", "rss_mib_peak")), ], }, { "name": "command-breakdown", "title": "--command 三层分解 vs 并发", "caption": "客户端全链路 p50/p95、guest 内纯执行(bash 纳秒计时)、" "服务端代理 envd RPC 平均;差值即代理/排队开销。", "y_label": "command 时延 (s)", "specs": [ ("客户端 cmd p50", client_stat("client_command", "p50")), ("客户端 cmd p95", client_stat("client_command", "p95")), ("guest 纯执行 p50", client_stat("client_command_guest", "p50")), ("服务端 envd avg", top_stat("server_envd_avg_sec")), ], }, ] def detect_knees(runs: list[dict[str, Any]]) -> list[dict[str, Any]]: """server pause avg 相对上一级 >2x 的最小 level,按数据集分别标注。""" annotations = [] for run in runs: prev: float | None = None for lvl in sorted(run["by_level"]): value = top_stat("server_pause_avg_sec")(run["by_level"][lvl]) if value is not None and prev is not None and prev > 0 and value > 2.0 * prev: annotations.append({"level": lvl, "label": run["label"]}) break if value is not None: prev = value return annotations # --------------------------------------------------------------------------- # 绘制。 # --------------------------------------------------------------------------- def configure_cjk_font() -> str | None: available = {f.name for f in font_manager.fontManager.ttflist} for name in CJK_FONT_CANDIDATES: if name in available: plt.rcParams["font.family"] = [name, "DejaVu Sans"] return name return None def draw_chart( ax: plt.Axes, chart: dict[str, Any], runs: list[dict[str, Any]], all_levels: list[int], annotations: list[dict[str, Any]], legend_fontsize: float = 8, ) -> bool: """把一张图画到 ax 上;该图无任何数据时返回 False。""" plotted = False for run_idx, run in enumerate(runs): linestyle = RUN_LINESTYLES[run_idx % len(RUN_LINESTYLES)] marker = RUN_MARKERS[run_idx % len(RUN_MARKERS)] suffix = f" ({run['label']})" if len(runs) > 1 else "" for spec_idx, (label, getter) in enumerate(chart["specs"]): values = [ getter(run["by_level"][lvl]) if lvl in run["by_level"] else None for lvl in all_levels ] if all(v is None for v in values): continue ys = [float("nan") if v is None else v for v in values] ax.plot( all_levels, ys, color=PALETTE[spec_idx % len(PALETTE)], linestyle=linestyle, marker=marker, markersize=5, linewidth=2.0 if linestyle == "-" else 1.6, label=f"{label}{suffix}", ) plotted = True if not plotted: return False ax.set_title(chart["title"], fontsize=12) ax.set_xlabel("并发循环沙箱数 R") ax.set_ylabel(chart["y_label"]) ax.grid(True, alpha=0.3, linewidth=0.6) if len(all_levels) <= 16: ax.set_xticks(all_levels) ax.legend(fontsize=legend_fontsize, loc="best", framealpha=0.9) if chart.get("knee"): for ann in annotations: ax.axvline( ann["level"], color="#7c3aed", linewidth=1.2, linestyle=(0, (4, 4)) ) ax.annotate( f"μ 拐点候选 ({ann['label']})", xy=(ann["level"], 1.0), xycoords=("data", "axes fraction"), xytext=(4, -6), textcoords="offset points", color="#7c3aed", fontsize=8, va="top", ) return True def print_run_cards(runs: list[dict[str, Any]]) -> None: for run in runs: config = run["config"] errors = sum(len(e.get("errors") or []) for e in run["by_level"].values()) command = config.get("command") or "(无)" if len(command) > 72: command = command[:72] + "…" print( f"[run] {run['label']} file={run['path'].name}\n" f" template={config.get('template')} " f"levels={config.get('levels')} rounds={config.get('rounds')} " f"think_sec={config.get('think_sec')}\n" f" command={command} errors={errors} " f"started={run['started_at']}" ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Plot PNG line charts from pause/resume concurrency sweep JSONs.", ) parser.add_argument( "inputs", nargs="+", type=Path, help="measure_pause_resume_concurrency.py 输出的 JSON(一个或多个)", ) parser.add_argument( "--label", nargs="*", default=[], help="每个输入一个数据集标签(默认用文件名)", ) parser.add_argument( "--output-dir", type=Path, default=None, help="PNG 输出目录(默认:第一个输入所在目录)", ) parser.add_argument( "--prefix", default=None, help="输出文件名前缀(默认:第一个输入的文件名去后缀)", ) parser.add_argument( "--no-overview", action="store_true", help="不生成把所有图拼在一起的总览大图", ) return parser def main() -> int: args = build_parser().parse_args() if len(args.label) > len(args.inputs): raise SystemExit("--label 数量多于输入文件") labels = list(args.label) + [ path.stem for path in args.inputs[len(args.label):] ] runs = [load_run(path, label) for path, label in zip(args.inputs, labels)] font = configure_cjk_font() plt.rcParams["axes.unicode_minus"] = False if font is None: print("warning: 未找到中文字体,中文标签可能显示为方框" "(安装 noto-sans-cjk 或 wqy-zenhei 可解决)") all_levels = sorted({lvl for run in runs for lvl in run["by_level"]}) annotations = detect_knees(runs) output_dir = args.output_dir or args.inputs[0].parent prefix = args.prefix or args.inputs[0].stem output_dir.mkdir(parents=True, exist_ok=True) print_run_cards(runs) rendered: list[dict[str, Any]] = [] for index, chart in enumerate(build_charts(), start=1): fig, ax = plt.subplots(figsize=(8, 4.5)) if draw_chart(ax, chart, runs, all_levels, annotations): fig.tight_layout() path = output_dir / f"{prefix}-{index:02d}-{chart['name']}.png" fig.savefig(path, dpi=150) rendered.append({"chart": chart, "path": path}) print(f"[ok] {path.name} # {chart['caption']}") plt.close(fig) if rendered and not args.no_overview: cols = 2 rows = math.ceil(len(rendered) / cols) fig, axes = plt.subplots( rows, cols, figsize=(8.2 * cols, 3.9 * rows), squeeze=False ) for slot, ax in enumerate(axes.flat): if slot < len(rendered): draw_chart( ax, rendered[slot]["chart"], runs, all_levels, annotations, legend_fontsize=7, ) else: ax.set_visible(False) fig.suptitle( "pause/resume 并发扫描总览 — " + " vs ".join(r["label"] for r in runs), fontsize=14, ) fig.tight_layout(rect=(0, 0, 1, 0.985)) path = output_dir / f"{prefix}-00-overview.png" fig.savefig(path, dpi=130) plt.close(fig) print(f"[ok] {path.name} # 全部图表总览") print( f"\n{len(rendered)} charts + overview written to {output_dir}" + (f" (μ 拐点标注: {annotations})" if annotations else "") ) return 0 if __name__ == "__main__": raise SystemExit(main())