run_release_benchmarks.py

1"""Reproducible fixed-parameter ordinary-write / WriteSession reports (no downloads)."""2from pathlib import Path3import argparse, os, datetime, hashlib, json, platform, random, shutil, statistics, subprocess, sys, tarfile, time4import tune_spsc as tune5from spsc_html_report import write_html_report6R = Path(__file__).resolve().parents[2]78def save(path, value):9    path.write_text(json.dumps(value, ensure_ascii=False, indent=2)+'\n', encoding='utf-8')10def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest()11def power():12    return subprocess.check_output(['pmset','-g','batt'], text=True) if sys.platform=='darwin' else 'Unavailable on this platform\n'13def source_files():14    files=[R/'CMakeLists.txt']15    for directory in ('src','include','support','test','examples','cmake'):16        files += [p for p in (R/directory).rglob('*') if p.is_file() and17                  (p.suffix in ('.hpp','.cpp','.py','.json','.cmake','.in') or p.name=='CMakeLists.txt')]18    files += [R/'LICENSE', R/'NOTICE']19    files += [p for p in (R/'benchmarks').rglob('*') if p.is_file() and p.suffix in ('.py','.cpp','.hpp','.cmake','.json')]20    files += list((R/'benchmarks/tunable').glob('*.hpp')) + list((R/'benchmarks/tunable').glob('*.cpp'))21    files += [R/'.github/workflows/ci.yml', R/'benchmarks/profiles/spsc_release.json']22    return sorted(set(files))2324parser=argparse.ArgumentParser(description=__doc__)25parser.add_argument('--binary', type=Path, required=True)26parser.add_argument('--deps', type=Path, default=R/'third_party/spsc')27parser.add_argument('--output', type=Path, required=True, help='New immutable results directory')28parser.add_argument('--report-dir', type=Path, default=R/'docs/reports/comparison')29parser.add_argument('--profile', type=Path, default=R/'benchmarks/profiles/spsc_release.json')30parser.add_argument('--timer', choices=['auto','steady'], default='auto')31parser.add_argument('--write-mode', choices=['normal','session','both'], default='both', help='Only regenerate the selected reports; native controls are freshly measured')32parser.add_argument('--callback-iterations', type=int, nargs=4, metavar=('B8','B16','B32','B64'), help='Fixed 256 MiB loop counts for 8/16/32/64 B; skip recalibration; 8 MiB remains zero')33parser.add_argument('--hardware-info', type=Path, help='Optional independently collected hardware metadata JSON')34parser.add_argument('--copy-sensitivity', type=Path, help='Optional independent copy-constructor control summary')35parser.add_argument('--resume', action='store_true', help='Requires identical executable, sources and settings')36parser.add_argument('--messages', type=int, default=10000000)37parser.add_argument('--repetitions', type=int, default=3)38parser.add_argument('--cache', type=int, nargs='+', choices=[8,256], default=[8,256])39parser.add_argument('--payload', type=int, nargs='+', choices=[8,16,32,64], default=[8,16,32,64])40parser.add_argument('--producer-cpu', type=int, default=-1)41parser.add_argument('--producer-qos', choices=['unchanged','user_interactive'], default='user_interactive' if sys.platform=='darwin' else 'unchanged')42args=parser.parse_args()43if not __debug__: raise RuntimeError("Run without Python -O; audit assertions must remain enabled")44if args.messages<1000 or args.repetitions<1: parser.error('messages >= 1000 and repetitions >= 1 required')45if args.callback_iterations is not None and min(args.callback_iterations)<0: parser.error('callback iterations must be nonnegative')46write_modes=('normal','session') if args.write_mode=='both' else (args.write_mode,)47fixed_loops=dict(zip((8,16,32,64),args.callback_iterations)) if args.callback_iterations is not None else None48B=args.binary.resolve(); ROOT=args.output.resolve(); args.report_dir=args.report_dir.resolve()49ROOT.mkdir(parents=True,exist_ok=args.resume); args.report_dir.mkdir(parents=True,exist_ok=True)50STAMP=ROOT.name51settings=json.loads(args.profile.read_text(encoding='utf-8'))52# Older saved profiles predate this option. Make its unchanged legacy default53# explicit so the complete serialized configuration matches recorded arguments.54for cache_configs in settings['caches'].values():55    for config in cache_configs.values():56        config.setdefault('fixed_round_boundaries', False)57if sys.platform=='darwin' and 'AC Power' not in power(): raise RuntimeError('Connect AC power before formal measurements')58deps=json.loads((R/'benchmarks/spsc/spsc_dependencies.json').read_text(encoding='utf-8'))59for dep, definition in deps.items():60    for file in definition['files']:61        if sha(args.deps/dep/file['path'])!=file['sha256']: raise ValueError('Dependency hash mismatch: '+dep+'/'+file['path'])62build=json.loads(subprocess.check_output([str(B),'--build-info'],text=True))63if not build['optimized'] or build['sanitized']: raise ValueError('Formal measurements require optimized, unsanitized executable')64files=source_files(); sources={str(p.relative_to(R)):sha(p) for p in files}65try:66    commit=subprocess.check_output(['git','rev-parse','HEAD'],cwd=R,text=True,stderr=subprocess.DEVNULL).strip()67    dirty=bool(subprocess.check_output(['git','status','--porcelain'],cwd=R,text=True,stderr=subprocess.DEVNULL))68except (OSError,subprocess.CalledProcessError): commit=None;dirty=None69cli={k:v for k,v in vars(args).items() if k not in ('binary','deps','output','report_dir','profile','resume','hardware_info','copy_sensitivity')}70hardware={}71if sys.platform=='darwin':72    for key in ('machdep.cpu.brand_string','hw.memsize','hw.physicalcpu','hw.logicalcpu'):73        query=subprocess.run(['sysctl','-n',key],capture_output=True,text=True)74        hardware[key]=query.stdout.strip() if query.returncode==0 else {'unavailable':query.stderr.strip()}75if args.hardware_info:76    hardware=json.loads(args.hardware_info.read_text(encoding='utf-8'))77    if not isinstance(hardware,dict): raise ValueError('hardware-info must be a JSON object')78copy_sensitivity=json.loads(args.copy_sensitivity.read_text(encoding='utf-8')) if args.copy_sensitivity else None79env=dict(copy_sensitivity_sha256=sha(args.copy_sensitivity) if args.copy_sensitivity else None,hardware_input_sha256=sha(args.hardware_info) if args.hardware_info else None,hardware=hardware,timestamp=datetime.datetime.now().astimezone().isoformat(),platform=platform.platform(),machine=platform.machine(),build=build,binary_sha256=sha(B),source_commit=commit,source_worktree_dirty=dirty,source_sha256=sources,dependencies=deps,settings=cli,profile_sha256=sha(args.profile),measurement_kind='fixed_matched_parameters_normal_vs_single_block_session')80if args.resume:81    previous=json.loads((ROOT/'environment.json').read_text(encoding='utf-8'))82    for key in ('binary_sha256','source_sha256','settings','profile_sha256','hardware_input_sha256','copy_sensitivity_sha256'):83        if previous[key]!=env[key]: raise ValueError('Cannot resume changed '+key)84    env=previous85else:86    (ROOT/'profile-input.json').write_text(args.profile.read_text(encoding='utf-8'),encoding='utf-8')87    if args.hardware_info: shutil.copy2(args.hardware_info, ROOT/'hardware-input.json')88    if args.copy_sensitivity: shutil.copy2(args.copy_sensitivity, ROOT/'copy-sensitivity.json')89    save(ROOT/'environment.json',env); (ROOT/'power-before.txt').write_text(power())90    with tarfile.open(ROOT/'source.tar.gz','w:gz') as archive:91        for path in files: archive.add(path,arcname=str(path.relative_to(R)))92    build_dir=B.parent.parent if B.parent.name in ('Release','Debug') else B.parent93    for name in ('CMakeCache.txt','compile_commands.json'):94        if (build_dir/name).exists(): shutil.copy2(build_dir/name,ROOT/name)95    save(ROOT/'invocation.json',dict(command=[sys.executable,*sys.argv],cwd=str(Path.cwd()),profile_sha256=sha(args.profile)))96    archive_root=R/'docs/performance_archive' if args.report_dir==(R/'docs/reports/comparison').resolve() else args.report_dir/'performance_archive'97    archive_dir=archive_root/('before_'+STAMP);archive_dir.mkdir(parents=True)98    for name in [f'performance_{cache}m_'+('callback' if cache==256 else 'zero')+('_session' if mode=='session' else '')+'.html' for cache in args.cache for mode in write_modes]:99        if (args.report_dir/name).exists(): shutil.copy2(args.report_dir/name,archive_dir/name)100rows=[json.loads(line) for line in (ROOT/'raw.jsonl').read_text(encoding='utf-8').splitlines()] if args.resume and (ROOT/'raw.jsonl').exists() else []101started=time.monotonic();log=(ROOT/'raw.jsonl').open('a',encoding='utf-8');cmdlog=(ROOT/'commands.jsonl').open('a',encoding='utf-8')102counts={};checksums={}103for row in rows:104    r=row['result'];c=r['config'];cache=row['cache_mib']105    key=(cache,c['payload_bytes'],c['callback_iterations'],r['effective_warmup_messages'])106    if key in checksums and checksums[key]!=r['callback_work_checksum']: raise ValueError('Callback checksum differs')107    checksums[key]=r['callback_work_checksum'];counts[cache]=counts.get(cache,0)+1108def run(cache,label,c,mode,trial,stage):109 c=dict(c,mode=mode,messages=args.messages,random_seed=42+trial,verify_payload=stage=='validation')110 for prefix in ('source','output'):assert c[prefix+'_block_size']*c[prefix+'_pool_min']==c[prefix+'_block_size']*c[prefix+'_pool_max']==cache*1048576111 existing=next((x for x in rows if (x["cache_mib"],x["label"],x["stage"],x["trial"],x["result"]["config"]["mode"])==(cache,label,stage,trial,mode)),None)112 if existing:113  if existing["result"]["config"]!=c: raise ValueError("resume configuration mismatch")114  return existing["result"]115 command=[str(B)]116 for k,v in c.items():command += ['--'+k,str(v).lower() if isinstance(v,bool) else str(v)]117 cmdlog.write(json.dumps(dict(cache_mib=cache,label=label,stage=stage,trial=trial,command=command))+'\n');cmdlog.flush()118 proc=subprocess.run(command,capture_output=True,text=True,timeout=c['timeout_seconds']+15)119 if proc.returncode:120  save(ROOT/'failure.json',dict(label=label,command=command,exit_code=proc.returncode,stdout=proc.stdout,stderr=proc.stderr));raise RuntimeError(proc.stderr)121 r=json.loads(proc.stdout)122 assert r['config']==c,(label,set(r['config'])^set(c))123 C=cache*1048576; n=c['payload_bytes']124 if c['transport']=='two_buf_queue':125  retained=2*C126  assert r['nominal_buffer_bytes']==retained127 elif c['transport']=='readerwriterqueue':128  assert r['actual_spsc_capacity']%511==0129  retained=(r['actual_spsc_capacity']//511)*512*n130 elif c['transport']=='rigtorp': retained=(r['actual_spsc_capacity']+1+2*((64+n-1)//n))*n131 else: retained=r['actual_spsc_capacity']*n132 assert retained<=(2*C if c['transport']=='two_buf_queue' else C)133 assert r['consumed']==r['timed_callback_messages']==c['messages']==args.messages134 assert r['timing_scope']=='producer_rounds' and r['seconds']==r['producer_seconds']135 if c['producer_qos']=='user_interactive': assert r['producer_qos_applied']136 if c['producer_cpu']>=0: assert r['producer_hard_affinity']137 else: assert not r['producer_hard_affinity']138 if mode=='write_latency':assert r['latency_samples']*c['latency_batch_size']+r['latency_tail_writes']==args.messages and not r['migrated_samples']139 if c['use_write_session']:assert r['session_block_exhaustions']>=0140 else:assert r['session_block_exhaustions']==0141 key=(cache,c['payload_bytes'],c['callback_iterations'],r['effective_warmup_messages'])142 if key in checksums:assert checksums[key]==r['callback_work_checksum']143 checksums[key]=r['callback_work_checksum']144 row=dict(cache_mib=cache,label=label,stage=stage,trial=trial,command=command,retained_payload_storage_bytes=retained,result=r);rows.append(row);log.write(json.dumps(row)+'\n');log.flush()145 counts[cache]=counts.get(cache,0)+1146 if stage=='callback_calibration' or counts[cache]%10==0:147  print(f'{cache} MiB: {counts[cache]} runs; {stage} {label} {mode}: {r["m_messages_per_second"]:.1f} M/s, ratio={r["callback_to_producer_ratio"]:.2f}, elapsed={time.monotonic()-started:.0f}s',flush=True)148 return r149peers=[dict(file='performance_256m_callback.html',label='256M 普通'),dict(file='performance_256m_callback_session.html',label='256M Session'),dict(file='performance_8m_zero.html',label='8M 普通'),dict(file='performance_8m_zero_session.html',label='8M Session')]150for cache in args.cache:151 policy='zero' if cache==8 else ('fixed' if fixed_loops is not None else 'calibrated')152 specs={label:dict(c) for label,c in settings['caches'][str(cache)].items() if c['payload_bytes'] in args.payload and (c['transport']!='two_buf_queue' or ('session' if c['use_write_session'] else 'normal') in write_modes)}153 matched=args.write_mode=='both' and all({k:v for k,v in specs[f'two_buf_queue_{n}'].items() if k!='use_write_session'}=={k:v for k,v in specs[f'session_{n}'].items() if k!='use_write_session'} for n in args.payload)154 for c in specs.values():155  c.update(messages=args.messages,producer_cpu=args.producer_cpu,producer_qos=args.producer_qos,timer=args.timer,callback_iterations=0)156  assert c['prepared_input'] and c['variable_payload']157  if c['use_write_session']: assert not c['is_scan_current'] and c['publish_count']==4294967295158 calibrations=[]159 for n in args.payload:160  iterations=fixed_loops[n] if policy=='fixed' else 0161  if policy=='calibrated':162   observations={};reference=dict(settings['caches'][str(cache)][f'two_buf_queue_{n}'],messages=args.messages,producer_cpu=args.producer_cpu,producer_qos=args.producer_qos,timer=args.timer,callback_iterations=0);cal_settings=dict(target_ratio=2.5,min_ratio=2,max_ratio=3,repetitions=args.repetitions,messages=args.messages,min_rounds=0,max_iterations=1024,max_candidates=8,reference_publish_count=reference['publish_count'])163   def probe(it):164    vv=[run(cache,f'calibration_{n}_{it}',dict(reference,callback_iterations=it),'throughput',t,'callback_calibration') for t in range(args.repetitions)]165    ratio=statistics.fmean(x['callback_to_producer_ratio'] for x in vv)166    observations[it]=dict(iterations=it,ratio=ratio,ratios=[x['callback_to_producer_ratio'] for x in vv]);return ratio167   iterations=tune.choose_callback_iterations(probe,cal_settings)168   ratio=observations[iterations]['ratio']169   calibrations.append(dict(payload_bytes=n,reference_transport='two_buf_queue',reference_parameters=dict(reference),selected_iterations=iterations,selected_ratio=ratio,target_met=2<=ratio<=3,observations=list(observations.values()),settings=cal_settings))170   print(f'CALIBRATED {n} B: loops={iterations}, ordinary ratio={ratio:.2f}',flush=True)171  for c in specs.values():172   if c['payload_bytes']==n:c['callback_iterations']=iterations173 save(ROOT/f'configs_{cache}m.json',specs);save(ROOT/f'calibration_{cache}m.json',calibrations)174 for label,c in specs.items():run(cache,label,c,'throughput',0,'validation')175 for trial in range(args.repetitions):176  jobs=[(label,mode) for label in specs for mode in ('throughput','write_latency')];random.Random(1701+trial+cache).shuffle(jobs)177  for label,mode in jobs:run(cache,label,specs[label],mode,trial,'final')178 def group(label):179  cc=specs[label];tt=[x['result'] for x in rows if x['cache_mib']==cache and x['label']==label and x['stage']=='final' and x['result']['config']['mode']=='throughput'];ll=[x['result'] for x in rows if x['cache_mib']==cache and x['label']==label and x['stage']=='final' and x['result']['config']['mode']=='write_latency']180  assert len(tt)==len(ll)==args.repetitions181  avg=lambda k:statistics.fmean(x[k] for x in tt)182  g=dict(payload_bytes=cc['payload_bytes'],transport=cc['transport'],callback_iterations=cc['callback_iterations'],parameters=cc,throughput_mps=avg('m_messages_per_second'),throughput_runs_mps=[x['m_messages_per_second'] for x in tt],payload_gb_per_second=avg('payload_gb_per_second'))183  for q in ('p50','p99','p999'):g[q+'_ns']=statistics.fmean(x['batch_average_write_ns'][q] for x in ll);g[q+'_runs_ns']=[x['batch_average_write_ns'][q] for x in ll]184  g['latency_runs']=[x['batch_average_write_ns'] for x in ll]185  for k in ('callback_to_producer_ratio','producer_seconds','callback_processing_seconds','wall_seconds','drain_seconds','failed_write_calls','backpressured_messages','producer_retry_seconds','cpu_cores','input_preparation_seconds','session_block_exhaustions'):g[k]=avg(k)186  for k in ('actual_spsc_capacity','completed_rounds','effective_warmup_messages','prepared_input_bytes'):g[k]=sorted({x[k] for x in tt+ll})187  for k in ('latency_samples','latency_tail_writes','timer'):g[k]=sorted({x[k] for x in ll})188  for k in ('writer_pressure_events','copy_memory_full_events'):g[k]=statistics.fmean((x['library_backpressure'] or {}).get(k,0) for x in tt)189  g['internal_backpressure']=dict(writer_pressure_events=g['writer_pressure_events'],copy_memory_full_events=g['copy_memory_full_events']) if all(x['library_backpressure'] is not None for x in tt) else None190  g['nominal_buffer_mib']=sorted({x['nominal_buffer_bytes']/1048576 for x in tt+ll});g['peak_rss_mib']=max(x['peak_rss_mib'] for x in tt+ll);g['latency_quantum_ns']=max(x['latency_quantum_ns'] for x in ll);g['resolution_limited']=any(x['resolution_limited'] for x in ll);g['p999_sample_count_warning']=any(x['p999_sample_count_warning'] for x in ll);g['library_pressure_stats_available']=all(x['library_backpressure'] is not None for x in tt)191  return g192 for mode in write_modes:193  directory=ROOT/f'{cache}m_{mode}';directory.mkdir(exist_ok=True)194  selected=[(f'session_{n}' if mode=='session' else f'two_buf_queue_{n}') if t=='two_buf_queue' else f'{t}_{n}' for n in args.payload for t in ('two_buf_queue','readerwriterqueue','rigtorp','atomic_queue')]195  rr=[x for x in rows if x['cache_mib']==cache and x['label'] in selected and x['stage'] in ('validation','final')]196  stem='performance_256m_callback' if cache==256 else 'performance_8m_zero';html_name=stem+('_session' if mode=='session' else '')+'.html'197  tbq_configs=[specs[f'session_{n}' if mode=='session' else f'two_buf_queue_{n}'] for n in args.payload]198  publish_counts=sorted({c['publish_count'] for c in tbq_configs})199  profile=dict(cache_mib=cache,callback_policy=policy,payload_bytes=args.payload,messages=args.messages,repetitions=args.repetitions,publish_count=publish_counts[0] if len(publish_counts)==1 else publish_counts,write_mode=mode,report_peers=peers,block_note='按长度保留冻结的块大小和每层容量。普通 write 与 Session 的读端及发布配置各自独立，完整参数如下。')200  notes=['本次没有重新寻优，吞吐和各延时指标使用同一构造配置。'+('普通 write 与 Session 使用相同参数。' if matched else '普通 write 与 Session 的读端/发布配置和测量轮次可能不同，不能当作同参数对照。'),('同缓存的两份报告共享本次实测的开源对照数据；没有使用历史成绩。' if args.write_mode=='both' else '只重新测量 '+mode+' 与三个开源对照；另一写模式的报告保持原样，没有把它的历史数据混入本轮主表。'),'每层数据池严格为 '+str(cache)+' MiB；原生队列的保留槽与 rigtorp 首尾 padding 纳入这个 payload 存储预算，队列元数据和输入另计。','Session 正常遇到块满返回 insufficient_space，次数记入 session_block_exhaustions；这不是背压或丢消息。池/队列受压通过 Writer 内部压力统计报告。']201  notes+=['rigtorp_write_prefetch 在完整参数中记录；开关在测试开始前选择模板，普通复制和带提示构造都只复制一次完整负载，不额外写零。atomic_queue 保持原有调用。']202  notes+=['ReaderWriterQueue 16/32 B 使用 copy_after_zero：队列槽位默认清零后 memcpy 一次完整负载；8/64 B 保留成员复制。清零和复制均在写入计时内。各实现的调用端优化在行标签和完整参数中公开，不能把所有差异归因于队列算法。']203  if mode=='session':notes+=['get_current_session → 内层逐条写 → 空间不足退出 → 外层 get_next_session。获取、等待、换块与 Session 状态归还包含在写入计时中。分组延时跨块继续计时，没有在块边界提前结束计时组。','不主动 flush；flush 只发布位置，不切换 block。每轮尾部在计时外 finish_block，最后一轮 finish，关闭 current 扫描也能收齐尾部。']204  if policy=='calibrated':notes+=['回调循环以普通 write 校准后固定，各实现共享同一长度的循环次数；实际回调/写入时间比可以不同。']205  if policy=='fixed':notes+=['256 MiB 回调循环按 8/16/32/64 B 固定为 '+ '/'.join(str(fixed_loops[n]) for n in (8,16,32,64))+' 次，本次未重新校准。用于控制读端/发布参数变化；实际时间比保留在表格中，不保证仍在 2～3 倍范围。']206  for c in tbq_configs:207   notes+=['TwoBufQueue '+str(c['payload_bytes'])+' B：is_scan_current='+str(c['is_scan_current']).lower()+', publish_count='+str(c['publish_count'])+', 初始/最大扫描间隔='+str(c['rescan_interval_ns'])+'/'+str(c['rescan_max_interval_ns'])+' ns，最小等待='+str(c['work_wait_min_ms'])+' ms。']208  summary=dict(schema_version=2,profile=profile,environment=env,groups=[group(x) for x in selected],calibration=calibrations,search=None,formal_runs=sum(x['stage']=='final' for x in rr),validation_runs=sum(x['stage']=='validation' for x in rr),calibration_runs=sum(x['cache_mib']==cache and x['stage']=='callback_calibration' for x in rows),measurement_kind='matched_parameters' if matched else 'fixed_parameters_independent_write_mode',audit_notes=notes)209  summary['copy_sensitivity']=dict(copy_sensitivity,source_prefix=os.path.relpath(args.copy_sensitivity.resolve().parent,args.report_dir).replace(os.sep,"/")) if copy_sensitivity else None210  save(directory/'results.json',summary);save(directory/'profile.json',profile);save(directory/'environment.json',env);save(directory/'dependencies.json',deps);211  save(directory/'reproduction.json',dict(source_archive='../source.tar.gz',invocation='../invocation.json',commands='../commands.jsonl',compile_commands='../compile_commands.json')); save(directory/'callback-calibration.json',calibrations)212  (directory/'raw.jsonl').write_text(''.join(json.dumps(x)+'\n' for x in rr))213  save(directory/'audit.json',dict(messages=args.messages,repetitions=args.repetitions,full_payload_fifo_validated=True,successful_timed_writes=args.messages,block_get_and_switch_timed=True,session_block_exhaustions_are_not_backpressure=True,native_controls_shared=args.write_mode=='both',binary_sha256=sha(B),source_sha256=sources,parameters_matched=matched,runtime_cache_bounds_verified=True))214  write_html_report(summary,args.report_dir/html_name,directory)215  print('REPORT',html_name,flush=True)216  for g in summary['groups']:217   if g['transport']=='two_buf_queue':print(mode,g['payload_bytes'],*(round(g[k],3) for k in ('throughput_mps','p50_ns','p99_ns','p999_ns')),flush=True)218 log.flush()219(ROOT/'power-after.txt').write_text(power());save(ROOT/'manifest.json',dict(timestamp=STAMP,binary_sha256=sha(B),source_sha256=sources,environment=env,elapsed_seconds_this_invocation=time.monotonic()-started,resumed=args.resume,runs=len(rows),counts=counts,policy='All trials retained; write modes configured independently; original upstream headers'))220log.close();cmdlog.close();print('COMPLETE',ROOT,flush=True)