core_bench.py (1892B)
1 #!/usr/bin/env python3 2 """Fresh-process C ABI linkage smoke with timing observations.""" 3 import argparse 4 import json 5 from pathlib import Path 6 import statistics 7 import subprocess 8 9 parser = argparse.ArgumentParser(description=__doc__) 10 parser.add_argument('--dynamic', required=True) 11 parser.add_argument('--archive', required=True) 12 parser.add_argument('--output', required=True, type=Path) 13 args = parser.parse_args() 14 rows = [] 15 summary = [] 16 for index, (kind, size) in enumerate((kind, size) for kind in ('constant', 'distribution') 17 for size in (4096, 65536)): 18 roles = ('dynamic', 'archive') if index % 2 == 0 else ('archive', 'dynamic') 19 for role in roles: 20 result = subprocess.run([getattr(args, role), kind, str(size)], check=True, 21 text=True, capture_output=True, timeout=180) 22 assert not result.stderr, result.stderr 23 observed = [json.loads(line) for line in result.stdout.splitlines()] 24 assert len(observed) == 7 25 assert [row['round'] for row in observed] == list(range(7)) 26 assert all(row['kind'] == kind and row['size'] == size for row in observed) 27 rows.extend(dict(role=role, **row) for row in observed) 28 summary.append(dict(role=role, kind=kind, size=size, 29 build_median_seconds=statistics.median(row['build_seconds'] for row in observed), 30 reduce_median_seconds=statistics.median(row['reduce_seconds'] for row in observed), 31 peak_rss_kib=max(row['peak_rss_kib'] for row in observed))) 32 args.output.write_text(json.dumps(dict( 33 schema='tricu.embedding-linkage-smoke.v1', 34 scope='Same public C API, fresh graphs, separate graph-build and WHNF-demand timings; no speed threshold', 35 samples=rows, summary=summary), indent=2) + '\n') 36 print(json.dumps(summary, indent=2))