test_per_writer_block_size.py

1"""Isolated 32 B ordinary-write block-size sweep using the existing C++ harness."""2from pathlib import Path3import argparse4import csv5import datetime6import hashlib7import html8import json9import platform10import random11import statistics12import subprocess13import sys14import time1516ROOT = Path(__file__).resolve().parents[2]17GIB = 1024 ** 3181920def reader_capacity_bytes(writer_block, multiplier, requested_bytes, fixed=False):21    reader_block = min(writer_block*multiplier,requested_bytes) if fixed else writer_block*multiplier22    return (requested_bytes if fixed else max(requested_bytes,reader_block))//reader_block*reader_block232425def round_bytes_for(writer_block, multiplier, requested_bytes, override=0, fixed=False):26    limit = reader_capacity_bytes(writer_block,multiplier,requested_bytes,fixed)*4//527    if override < 0 or override > limit:28        raise ValueError("round exceeds 80% of actual reader capacity")29    return override or limit303132def save(path, value):33    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")343536def sha(path):37    return hashlib.sha256(path.read_bytes()).hexdigest()383940def probe(command):41    result = subprocess.run(command, capture_output=True, text=True)42    return dict(command=command, exit_code=result.returncode, stdout=result.stdout, stderr=result.stderr)434445def memory_state():46    if sys.platform == "darwin":47        return {"vm_stat": probe(["vm_stat"]), "swap": probe(["sysctl", "vm.swapusage"])}48    return {}495051def pool_summary(results, side):52    values = [x["pool_depth"][side] for x in results]53    count = sum(x["samples"] for x in values)54    return dict(samples=count,55                max_in_use=max(x["max_in_use"] for x in values),56                average_in_use=sum(x["in_use_sum"] for x in values)/count if count else None,57                max_allocated=max(x["max_allocated"] for x in values),58                average_allocated=sum(x["allocated_sum"] for x in values)/count if count else None,59                average_free=sum(x["free_sum"] for x in values)/count if count else None,60                max_in_use_bytes=max(x["max_in_use_bytes"] for x in values),61                max_allocated_bytes=max(x["max_allocated_bytes"] for x in values),62                runs=values)636465def summarize(rows, repetitions):66    output = []67    sizes = sorted({x["writer_block_bytes"] for x in rows})68    for size in sizes:69        measured = [x for x in rows if x["writer_block_bytes"] == size and x["stage"] == "measurement"]70        throughput = [x["packet"]["result"] for x in measured if x["mode"] == "throughput"]71        latency = [x["packet"]["result"] for x in measured if x["mode"] == "write_latency"]72        if len(throughput) != repetitions or len(latency) != repetitions:73            continue74        reference = measured[0]["packet"]75        result = dict(writer_block_bytes=size, reader_block_bytes=throughput[0]["config"]["output_block_size"],76                      block_size_test=reference["block_size_test"], parameters=throughput[0]["config"],77                      throughput_mps=statistics.fmean(x["m_messages_per_second"] for x in throughput),78                      throughput_runs_mps=[x["m_messages_per_second"] for x in throughput],79                      payload_gb_per_second=statistics.fmean(x["payload_gb_per_second"] for x in throughput),80                      peak_rss_mib=max(x["peak_rss_mib"] for x in throughput+latency),81                      writer_pool=pool_summary(throughput,"writer"), reader_pool=pool_summary(throughput,"reader"),82                      latency_writer_pool=pool_summary(latency,"writer"), latency_reader_pool=pool_summary(latency,"reader"))83        for metric in ("producer_seconds", "callback_processing_seconds", "callback_to_producer_ratio", "drain_seconds",84                       "failed_write_calls", "backpressured_messages", "producer_retry_seconds", "cpu_cores"):85            result[metric] = statistics.fmean(x[metric] for x in throughput)86        for metric in ("p50", "p99", "p999"):87            result[metric+"_ns"] = statistics.fmean(x["batch_average_write_ns"][metric] for x in latency)88            result[metric+"_runs_ns"] = [x["batch_average_write_ns"][metric] for x in latency]89        result["latency_samples"] = [x["latency_samples"] for x in latency]90        result["writer_pressure_events"] = statistics.fmean(x["library_backpressure"]["writer_pressure_events"] for x in throughput)91        result["copy_memory_full_events"] = statistics.fmean(x["library_backpressure"]["copy_memory_full_events"] for x in throughput)92        output.append(result)93    return output949596def size_label(value):97    for divisor, unit in ((GIB,"GiB"),(1024**2,"MiB"),(1024,"KiB")):98        if value >= divisor:99            return f"{value/divisor:g} {unit}"100    return f"{value} B"101102103def number(value, places=3):104    return "—" if value is None else f"{value:,.{places}f}"105106107def write_report(path, rows, environment):108    if not rows:109        return110    fastest = max(rows, key=lambda x:x["throughput_mps"])["writer_block_bytes"]111    lowest = min(rows, key=lambda x:x["p50_ns"])["writer_block_bytes"]112    performance, depth, allocated = [], [], []113    for row in rows:114        label = html.escape(size_label(row["writer_block_bytes"]))115        reader = html.escape(size_label(row["reader_block_bytes"]))116        classes = ' class="best"' if row["writer_block_bytes"] == fastest else ""117        p50_class = ' class="metric"' if row["writer_block_bytes"] == lowest else ""118        rr = row["throughput_runs_mps"]119        performance.append(f'<tr{classes}><th>{label}</th><td>{reader}</td>'120                           f'<td>{number(row["throughput_mps"])}</td><td>{number(row["payload_gb_per_second"])}</td>'121                           f'<td{p50_class}>{number(row["p50_ns"])}</td><td>{number(row["p99_ns"])}</td>'122                           f'<td>{number(row["p999_ns"])}</td><td>{number(min(rr))}–{number(max(rr))}</td>'123                           f'<td>{number(row["callback_to_producer_ratio"],2)}</td><td>{number(row["peak_rss_mib"],1)}</td></tr>')124        w, r = row["writer_pool"], row["reader_pool"]125        depth.append(f'<tr><th>{label}</th><td>{w["max_in_use"]:,}</td><td>{number(w["average_in_use"])}</td>'126                     f'<td>{number(w["max_in_use_bytes"]/1024**2,2)}</td><td>{w["samples"]:,}</td>'127                     f'<td>{r["max_in_use"]:,}</td><td>{number(r["average_in_use"])}</td>'128                     f'<td>{number(r["max_in_use_bytes"]/1024**2,2)}</td><td>{r["samples"]:,}</td>'129                     f'<td>{number(row["writer_pressure_events"],1)}</td><td>{number(row["copy_memory_full_events"],1)}</td></tr>')130        allocated.append(f'<tr><th>{label}</th><td>{w["max_allocated"]:,}</td><td>{number(w["average_allocated"])}</td>'131                         f'<td>{number(w["max_allocated_bytes"]/1024**2,2)}</td><td>{r["max_allocated"]:,}</td>'132                         f'<td>{number(r["average_allocated"])}</td><td>{number(r["max_allocated_bytes"]/1024**2,2)}</td>'133                         f'<td>{size_label(row["block_size_test"]["reader_cache_effective_limit_bytes"])}</td></tr>')134    settings = environment["settings"]135    round_budget_note = (f"每轮上限 {size_label(settings['round_storage_bytes'])}，不超过 reader 实际总容量的 80%。"136                         if settings['round_storage_bytes'] else137                         "每轮上限按 reader 实际总容量的 80% 计算，不按 writer 单块或 writer 总容量计算。")138    scheduling = (f"写线程绑定 CPU {settings['producer_cpu']}。" if settings['producer_cpu'] >= 0 else "未硬绑核。")139    scheduling += "采用高 QoS 调度偏好。" if settings['producer_qos'] == "user_interactive" else "保留平台默认调度。"140    scan_current = settings.get("is_scan_current", True)141    scan_label = "当前块扫描 1 ms" if scan_current else "不扫描当前块"142    publish_count = settings.get("publish_count", 1)143    publish_label = "仅交接块时发布 · UINT32_MAX" if publish_count == 4294967295 else "每条 release 发布"144    round_note = ("" if scan_current else145                  "关闭 current 扫描时，沿用框架在每轮计时结束后 finish_block 交接尾块并排空；交接与排空时间不计入写入吞吐，下一轮申请新块仍计时。发布次数见上方标签。")146    escaped = html.escape(json.dumps(environment,ensure_ascii=False,indent=2))147    page = f'''<!doctype html><html lang="zh-CN"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">148<title>TwoBufQueue · writer block size 遍历</title>149<style>body{{margin:0;background:#f4f7fb;color:#17243b;font:15px/1.65 system-ui,-apple-system,sans-serif}}main{{max-width:1450px;margin:auto;padding:32px 24px}}h1{{font-size:29px;margin:0 0 12px}}h2{{margin:28px 0 10px;font-size:20px}}.intro,.card{{background:white;padding:22px;border:1px solid #dce5ef;border-radius:12px;margin-bottom:18px}}.scroll{{overflow:auto;border:1px solid #dce5ef;border-radius:10px}}table{{border-collapse:collapse;width:100%;white-space:nowrap;font-variant-numeric:tabular-nums}}td,th{{padding:11px 13px;text-align:right;border-bottom:1px solid #dce5ef}}thead th{{background:#213b60;color:white;font-size:13px}}tbody th{{text-align:left}}tbody tr{{background:#edf5ff}}tbody tr:nth-child(even){{background:#e5effc}}tbody tr:hover{{background:#d5e7fb}}tr.best{{background:#d5f0e7!important;font-weight:650}}.metric{{color:#076644;font-weight:750}}.muted{{color:#53667f}}.tag{{display:inline-block;background:#e5effc;padding:4px 10px;margin:0 7px 8px 0;border-radius:6px}}pre{{white-space:pre-wrap;font-size:12px;overflow:auto}}a{{color:#1766ab}}.bestnote{{color:#076644}}</style>150<main><h1>writer 块大小与热路径性能</h1><div class="intro">151<span class="tag">32 B</span><span class="tag">{publish_label}</span><span class="tag">{scan_label}</span>152<span class="tag">回调循环 {settings['callback_iterations']}</span><span class="tag">{settings['repetitions']} 次平均</span>153<p>每项发送 {settings['messages']:,} 条。reader 单块按 writer 的 {settings['reader_block_multiplier']} 倍计算，固定容量模式下单块不超过 reader 总上限，初始各 {settings['initial_pool_blocks']} 块（受池上限约束），随后按需增长。154writer 容量上限 {size_label(settings['writer_cache_bytes'])}；reader 上限 {size_label(settings['reader_cache_bytes'])}；固定容量模式不扩大上限，旧模式才允许扩容。{round_budget_note}</p>155<p class="muted">沿用 PerformanceTest 的数据准备、普通 write、两层复制、独立回调线程、分轮发送及排空逻辑。数据准备与轮间排空不计入写入吞吐；统计换块的开销包含在测量中。156{scheduling}没有添加目标读回或额外清零优化。</p>157<p class="muted">{round_note}</p>158<p class="bestnote">吞吐最高：{size_label(fastest)}；p50 最低：{size_label(lowest)}。所有重复均保留，未挑选最快一次。</p></div>159<h2>写入性能</h2><p class="muted">p50/p99/p999 是每组 {settings['latency_batch_size']:,} 次成功写入的平均耗时分位数，再对各次运行取平均；不是单条端到端延时。160回调/写入比为每轮首条至末条回调的时间与写入时间之比，包含分发和等待。RSS 是进程实际驻留内存峰值。</p>161<div class="scroll"><table><thead><tr><th>writer 单块</th><th>reader 单块</th><th>M条/s</th><th>GB/s</th><th>p50 ns</th><th>p99 ns</th><th>p999 ns</th><th>吞吐范围 M条/s</th><th>回调/写入比</th><th>峰值 RSS MiB</th></tr></thead><tbody>{''.join(performance)}</tbody></table></div>162<h2>占用深度：已借出、尚未归还的块</h2><p class="muted">每次成功取新块后，在原有池锁内取得一致计数。均值按取块次数加权；最大值包含正式阶段起始快照。这里汇总吞吐测试的全部重复，延时测试的独立深度数据保存在 JSON。163没有取块样本时均值显示 —。writer/reader 两侧独立采样，不代表同一时刻的联合快照。</p>164<div class="scroll"><table><thead><tr><th>writer 单块</th><th>writer 最大块数</th><th>writer 平均块数</th><th>writer 最大占用 MiB</th><th>writer 样本数</th><th>reader 最大块数</th><th>reader 平均块数</th><th>reader 最大占用 MiB</th><th>reader 样本数</th><th>writer 背压/次</th><th>复制层背压/次</th></tr></thead><tbody>{''.join(depth)}</tbody></table></div>165<h2>已分配块数与容量</h2><p class="muted">已分配包括借出和池内空闲块。下面的容量是块数 × 单块字节数，包含尚未触碰的虚拟内存，不等于实际 RSS。166平均值仍按成功取块次数加权。</p>167<div class="scroll"><table><thead><tr><th>writer 单块</th><th>writer 最大分配块数</th><th>writer 平均分配块数</th><th>writer 最大分配 MiB</th><th>reader 最大分配块数</th><th>reader 平均分配块数</th><th>reader 最大分配 MiB</th><th>reader 容量上限</th></tr></thead><tbody>{''.join(allocated)}</tbody></table></div>168<div class="card" style="margin-top:24px"><details><summary>环境、二进制和源文件指纹</summary><pre>{escaped}</pre></details></div></main></html>'''169    path.parent.mkdir(parents=True,exist_ok=True)170    path.write_text(page,encoding="utf-8")171172173def main():174    parser = argparse.ArgumentParser(description=__doc__)175    parser.add_argument("--binary",type=Path,required=True)176    parser.add_argument("--output",type=Path,required=True)177    parser.add_argument("--report",type=Path)178    parser.add_argument("--min-block-bytes",type=int,default=1024)179    parser.add_argument("--max-block-bytes",type=int,default=GIB)180    parser.add_argument("--reader-block-multiplier",type=int,default=10)181    parser.add_argument("--writer-cache-bytes",type=int,default=2*GIB)182    parser.add_argument("--reader-cache-bytes",type=int,default=2*GIB)183    parser.add_argument("--reader-cache-fixed",choices=("true","false"),default="true")184    parser.add_argument("--initial-pool-blocks",type=int,default=2)185    parser.add_argument("--messages",type=int,default=10_000_000)186    parser.add_argument("--warmup-messages",type=int,default=50_000)187    parser.add_argument("--callback-iterations",type=int,default=0)188    parser.add_argument("--is-scan-current",choices=("true","false"),default="true")189    parser.add_argument("--publish-count",type=int,choices=(1,4294967295),default=None)190    parser.add_argument("--repetitions",type=int,default=3)191    parser.add_argument("--round-storage-bytes",type=int,default=0,help="0=80%% of actual reader pool capacity")192    parser.add_argument("--latency-batch-size",type=int,default=1000)193    parser.add_argument("--producer-cpu",type=int,default=-1)194    parser.add_argument("--producer-qos",choices=("user_interactive","unchanged"),195                        default="user_interactive" if sys.platform == "darwin" else "unchanged")196    parser.add_argument("--timeout-seconds",type=int,default=180)197    parser.add_argument("--order-seed",type=int,default=1729)198    args = parser.parse_args()199    args.is_scan_current = args.is_scan_current == "true"200    args.reader_cache_fixed = args.reader_cache_fixed == "true"201    if args.publish_count is None:202        args.publish_count = 1 if args.is_scan_current else 4294967295203    if (args.min_block_bytes <= 0 or args.min_block_bytes & (args.min_block_bytes-1) or204        args.max_block_bytes < args.min_block_bytes or args.max_block_bytes & (args.max_block_bytes-1)):205        parser.error("block bounds must be ordered powers of two")206    if min(args.reader_block_multiplier,args.writer_cache_bytes,args.reader_cache_bytes,args.initial_pool_blocks,207           args.messages,args.repetitions,args.latency_batch_size,args.timeout_seconds) <= 0:208        parser.error("counts, capacities and multiplier must be positive")209    if args.round_storage_bytes < 0:210        parser.error("round storage override must be nonnegative")211    if args.warmup_messages < 0 or args.callback_iterations < 0:212        parser.error("warmup and callback iterations must be nonnegative")213    args.binary = args.binary.resolve()214    args.output = args.output.resolve()215    args.report = (args.report or args.output/"test_per_writer_block_size.html").resolve()216    args.output.mkdir(parents=True,exist_ok=False)217    build = json.loads(subprocess.check_output([str(args.binary),"--build-info"],text=True))218    if not build["optimized"] or build["sanitized"] or not build["pool_depth_instrumented"]:219        raise ValueError("use an optimized, unsanitized, pool-instrumented block-size executable")220    power = probe(["pmset","-g","batt"]) if sys.platform == "darwin" else None221    if power and "AC Power" not in power["stdout"]:222        raise RuntimeError("connect AC power before running the block-size sweep")223    settings = {k:v for k,v in vars(args).items() if k not in ("binary","output","report")}224    sources = [ROOT/"CMakeLists.txt",ROOT/"benchmarks/spsc/spsc.cmake",Path(__file__).resolve(),225               ROOT/"benchmarks/block_size/test_per_writer_block_size.cpp",ROOT/"benchmarks/block_size/block_size_round_budget.hpp"]226    for directory in ("src","include","benchmarks/tunable"):227        sources += [x for x in (ROOT/directory).rglob("*") if x.is_file() and x.suffix in (".cpp",".hpp")]228    environment = dict(timestamp=datetime.datetime.now().astimezone().isoformat(),platform=platform.platform(),229                       machine=platform.machine(),build=build,binary=str(args.binary),binary_sha256=sha(args.binary),230                       settings=settings,power_before=power,memory_before=memory_state(),231                       source_sha256={str(x.relative_to(ROOT)):sha(x) for x in sources},232                       dependencies=json.loads((ROOT/"benchmarks/spsc/spsc_dependencies.json").read_text(encoding="utf-8")),233                       measurement=("32 B ordinary write, publish_count=" + str(args.publish_count) + ", " +234                                    ("current scan 1 ms" if args.is_scan_current else "completed blocks only") +235                                    ", dynamic pools"))236    save(args.output/"environment.json",environment)237    rows = []238    sizes = []239    size = args.min_block_bytes240    while size <= args.max_block_bytes:241        sizes.append(size)242        size *= 2243    started = time.monotonic()244    with (args.output/"raw.jsonl").open("w",encoding="utf-8") as raw:245        def run(size,mode,trial,stage):246            parameters = dict(source_block_size=size,reader_block_multiplier=args.reader_block_multiplier,247                              writer_cache_bytes=args.writer_cache_bytes,reader_cache_bytes=args.reader_cache_bytes,248                              reader_cache_fixed=args.reader_cache_fixed,249                              initial_pool_blocks=args.initial_pool_blocks,messages=args.messages,250                              warmup_messages=args.warmup_messages,callback_iterations=args.callback_iterations,251                              is_scan_current=args.is_scan_current,publish_count=args.publish_count,252                              mode=mode,verify_payload=stage == "validation",253                              round_storage_bytes=round_bytes_for(size,args.reader_block_multiplier,254                                                                 args.reader_cache_bytes,args.round_storage_bytes,args.reader_cache_fixed),255                              latency_batch_size=args.latency_batch_size,producer_cpu=args.producer_cpu,256                              producer_qos=args.producer_qos,timeout_seconds=args.timeout_seconds,random_seed=42+trial)257            command = [str(args.binary)]258            for key,value in parameters.items():259                command += ["--"+key,str(value).lower() if isinstance(value,bool) else str(value)]260            process = subprocess.run(command,capture_output=True,text=True,timeout=args.timeout_seconds+20)261            stem = f"{size}_{stage}_{mode}_{trial}"262            (args.output/(stem+".stdout.json")).write_text(process.stdout,encoding="utf-8")263            (args.output/(stem+".stderr.txt")).write_text(process.stderr,encoding="utf-8")264            if process.returncode:265                save(args.output/"failure.json",dict(command=command,exit_code=process.returncode,stderr=process.stderr))266                raise RuntimeError(f"{stem}: {process.stderr}")267            packet = json.loads(process.stdout)268            result = packet["result"]269            config = result["config"]270            assert config["payload_bytes"] == 32 and config["publish_count"] == args.publish_count271            assert config["is_scan_current"] == args.is_scan_current272            assert config["source_block_size"] == size and config["output_block_size"] == (min(size*args.reader_block_multiplier,args.reader_cache_bytes) if args.reader_cache_fixed else size*args.reader_block_multiplier)273            assert result["consumed"] == result["timed_callback_messages"] == args.messages274            assert result["timing_scope"] == "producer_rounds" and result["seconds"] == result["producer_seconds"]275            assert result["pool_depth"]["writer"]["max_in_use"] <= config["source_pool_max"]276            assert result["pool_depth"]["reader"]["max_in_use"] <= config["output_pool_max"]277            if args.producer_qos == "user_interactive":278                assert result["producer_qos_applied"]279            if mode == "write_latency":280                assert result["latency_samples"]*args.latency_batch_size+result["latency_tail_writes"] == args.messages281                assert result["migrated_samples"] == 0282            row = dict(writer_block_bytes=size,mode=mode,trial=trial,stage=stage,command=command,packet=packet)283            rows.append(row)284            raw.write(json.dumps(row,ensure_ascii=False)+"\n")285            raw.flush()286            print(f"{len(rows):3d}/{len(sizes)*(2*args.repetitions+1)} {stage:11s} {size_label(size):>8s} "287                  f"{mode:13s} {result['m_messages_per_second']:8.2f} M/s "288                  f"depth={result['pool_depth']['writer']['max_in_use']}/{result['pool_depth']['reader']['max_in_use']} "289                  f"RSS={result['peak_rss_mib']:.0f} MiB elapsed={time.monotonic()-started:.0f}s",flush=True)290        for size in sizes:291            run(size,"throughput",0,"validation")292        for trial in range(args.repetitions):293            jobs = [(size,mode) for size in sizes for mode in ("throughput","write_latency")]294            random.Random(args.order_seed+trial).shuffle(jobs)295            for size,mode in jobs:296                run(size,mode,trial,"measurement")297            summary = summarize(rows,args.repetitions)298            save(args.output/"summary.json",summary)299    summary = summarize(rows,args.repetitions)300    environment["power_after"] = probe(["pmset","-g","batt"]) if sys.platform == "darwin" else None301    environment["memory_after"] = memory_state()302    environment["elapsed_seconds"] = time.monotonic()-started303    environment["completed_runs"] = len(rows)304    save(args.output/"environment.json",environment)305    save(args.output/"summary.json",summary)306    columns = ["writer_block_bytes","reader_block_bytes","throughput_mps","payload_gb_per_second",307               "p50_ns","p99_ns","p999_ns","peak_rss_mib","writer_max_depth","writer_average_depth",308               "reader_max_depth","reader_average_depth","writer_depth_samples","reader_depth_samples"]309    with (args.output/"summary.csv").open("w",newline="",encoding="utf-8") as file:310        writer = csv.DictWriter(file,fieldnames=columns)311        writer.writeheader()312        for row in summary:313            value = {key:row[key] for key in columns[:8]}314            for side in ("writer","reader"):315                stats = row[side+"_pool"]316                value.update({side+"_max_depth":stats["max_in_use"],side+"_average_depth":stats["average_in_use"],317                              side+"_depth_samples":stats["samples"]})318            writer.writerow(value)319    write_report(args.report,summary,environment)320    print("Saved:",args.report,flush=True)321322323if __name__ == "__main__":324    main()