#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import json import math import subprocess import textwrap import time from dataclasses import dataclass from pathlib import Path from run_cache_analysis import ( CacheLevel, BenchmarkPoint, collect_system_info, draw_line_chart, format_bytes, latency_series, load_benchmark_points, map_points, query_harness_environment, resolve_executable, throughput_series, write_capabilities_report, write_counter_availability_report, write_run_metadata, write_summary_csv, write_system_info, ) SIZE_LADDER = [ "4K", "8K", "16K", "32K", "64K", "128K", "256K", "512K", "1M", "2M", "4M", "8M", "16M", "32M", "64M", ] PERF_EVENTS = [ "cycles", "instructions", "cache-references", "cache-misses", "branches", "branch-misses", "task-clock", "context-switches", "cpu-migrations", "page-faults", ] OPTIONAL_PERF_EVENTS = [ "L1-dcache-loads", "L1-dcache-load-misses", "dTLB-loads", "dTLB-load-misses", ] PALETTE = { "naive": "#6c757d", "cache_aware": "#0b6e4f", "simd": "#ff7b00", "cache_aware_simd": "#0f7173", "cache_marker": "#ef476f", } @dataclass(frozen=True) class Variant: key: str label: str benchmark: str @dataclass(frozen=True) class Question: slug: str title: str problem: str metric: str variants: tuple[Variant, Variant, Variant, Variant] QUESTIONS = [ Question( "stock_profit", "Best Time to Buy and Sell Stock", "Given daily prices, find the maximum profit from one buy followed by one sell.", "latency", ( Variant("naive", "naive", "dsa_scalar_stock_profit_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_stock_profit_soa"), Variant("simd", "SIMD", "dsa_simd_stock_profit_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_stock_profit_soa"), ), ), Question( "product_except_self", "Product of Array Except Self", "For every index, return the product of all other elements without division.", "latency", ( Variant("naive", "naive", "dsa_scalar_product_except_self_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_product_except_self_soa"), Variant("simd", "SIMD", "dsa_simd_product_except_self_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_product_except_self_soa"), ), ), Question( "max_subarray", "Maximum Subarray", "Find the contiguous subarray with the largest sum.", "latency", ( Variant("naive", "naive", "dsa_scalar_max_subarray_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_max_subarray_soa"), Variant("simd", "SIMD", "dsa_simd_max_subarray_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_max_subarray_soa"), ), ), Question( "trapping_rain", "Trapping Rain Water", "Given bar heights, compute how much water is trapped.", "latency", ( Variant("naive", "naive", "dsa_scalar_trapping_rain_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_trapping_rain_soa"), Variant("simd", "SIMD", "dsa_simd_trapping_rain_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_trapping_rain_soa"), ), ), Question( "sliding_window", "K-Sized Subarray Maximum", "For every fixed-width window, report the maximum value.", "latency", ( Variant("naive", "naive", "dsa_scalar_sliding_window_max_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_sliding_window_max_soa"), Variant("simd", "SIMD", "dsa_simd_sliding_window_max_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_sliding_window_max_soa"), ), ), Question( "rotated_search", "Search in Rotated Sorted Array", "Search target values in a sorted array that has been rotated.", "latency", ( Variant("naive", "naive", "dsa_scalar_rotated_search_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_rotated_search_soa"), Variant("simd", "SIMD", "dsa_simd_rotated_search_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_rotated_search_soa"), ), ), Question( "sort012", "Sort 0s, 1s, and 2s", "Sort an array whose values are only 0, 1, and 2.", "latency", ( Variant("naive", "naive", "dsa_scalar_sort012_aos"), Variant("cache_aware", "cache-aware", "dsa_scalar_sort012_soa"), Variant("simd", "SIMD", "dsa_simd_sort012_aos"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_sort012_soa"), ), ), Question( "bfs", "BFS of Graph", "Traverse graph vertices in breadth-first order.", "latency", ( Variant("naive", "naive", "dsa_scalar_bfs_adjlist"), Variant("cache_aware", "cache-aware", "dsa_scalar_bfs_csr"), Variant("simd", "SIMD", "dsa_simd_bfs_adjlist"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_bfs_csr"), ), ), Question( "transpose", "Transpose of Matrix", "Transpose a dense row-major matrix.", "throughput", ( Variant("naive", "naive", "dsa_scalar_transpose_matrix"), Variant("cache_aware", "cache-aware", "dsa_scalar_blocked_transpose_matrix"), Variant("simd", "SIMD", "dsa_simd_transpose_matrix"), Variant("cache_aware_simd", "cache-aware + SIMD", "dsa_simd_blocked_transpose_matrix"), ), ), Question( "matrix_multiply", "Multiply Two Matrices", "Multiply two dense square matrices.", "throughput", ( Variant("naive", "naive", "simd_scalar_matmul_naive"), Variant("cache_aware", "cache-aware", "simd_scalar_matmul_transposed"), Variant("simd", "SIMD", "simd_avx2_matmul_gather"), Variant("cache_aware_simd", "cache-aware + SIMD", "simd_avx2_matmul_blocked"), ), ), ] def main() -> int: parser = argparse.ArgumentParser(description="Run the DSA full-evidence suite.") parser.add_argument("--exe", default=None) parser.add_argument("--results-dir", default="results/dsa_haswell_full_evidence") parser.add_argument("--iterations", type=int, default=3) parser.add_argument("--threads", type=int, default=4) parser.add_argument("--sizes", default=",".join(SIZE_LADDER)) parser.add_argument("--taskset", default="", help="Optional CPU list passed to taskset -c.") parser.add_argument("--skip-timing", action="store_true") parser.add_argument("--skip-perf", action="store_true") parser.add_argument("--skip-run", action="store_true", help="Regenerate summaries/charts from existing files.") parser.add_argument("--perf-repeats", type=int, default=1) parser.add_argument("--timing-timeout", type=int, default=900) parser.add_argument("--matmul-timeout", type=int, default=1800) parser.add_argument("--collect-counters", default="off") args = parser.parse_args() root = Path(__file__).resolve().parent.parent results_dir = (root / args.results_dir).resolve() results_dir.mkdir(parents=True, exist_ok=True) (results_dir / "charts").mkdir(exist_ok=True) (results_dir / "perf_raw").mkdir(exist_ok=True) (results_dir / "perf_bench_csv").mkdir(exist_ok=True) exe = resolve_executable(root, args.exe) sizes = [size.strip() for size in args.sizes.split(",") if size.strip()] system_info = collect_system_info() harness_info = query_harness_environment(exe) write_system_info(results_dir / "system_info.txt", system_info) write_capabilities_report(results_dir / "capabilities.txt", system_info, harness_info) write_counter_availability_report(results_dir / "counter_availability.txt", system_info, args.collect_counters) write_run_metadata(results_dir / "run_metadata.json", exe, args, system_info, harness_info) if not args.skip_run and not args.skip_timing: run_timing(exe, results_dir, sizes, args) raw_csv = results_dir / "benchmark_results_raw.csv" summary_csv = results_dir / "benchmark_results_summary.csv" if raw_csv.exists(): points = load_benchmark_points(raw_csv) write_summary_csv(summary_csv, points) write_timing_charts(results_dir / "charts", points, system_info["cache_levels"]) else: points = [] availability = probe_perf_events(results_dir, args) if not args.skip_run else load_json(results_dir / "perf_event_availability.json") if not args.skip_run and not args.skip_perf: run_perf(exe, results_dir, sizes, args, availability) perf_rows = parse_perf_outputs(results_dir) write_perf_summaries(results_dir, perf_rows) write_perf_charts(results_dir / "charts", results_dir / "perf_derived_summary.csv", system_info["cache_levels"]) write_question_summary(results_dir, points) write_code_appendix(root, results_dir / "cache_simd_dsa_code_appendix.md") write_source_snapshots(root, results_dir / "source") print(results_dir) return 0 def base_command(exe: Path, benchmark: str, size: str, iterations: int, threads: int, args: argparse.Namespace) -> list[str]: command = [ str(exe), "--bench", benchmark, "--sizes", size, "--iterations", str(iterations), "--threads", str(threads), "--pin", "none", "--collect-counters", args.collect_counters, "--output", "csv", ] if args.taskset: return ["taskset", "-c", args.taskset, *command] return command def timeout_for(benchmark: str, size: str, args: argparse.Namespace) -> int: if benchmark.startswith("simd_") and "matmul" in benchmark: return args.matmul_timeout if parse_size(size) >= 32 * 1024 * 1024 and "matmul" in benchmark: return args.matmul_timeout return args.timing_timeout def run_timing(exe: Path, results_dir: Path, sizes: list[str], args: argparse.Namespace) -> None: raw_csv = results_dir / "benchmark_results_raw.csv" failures_csv = results_dir / "timing_failures.csv" header_written = False start = time.time() with raw_csv.open("w", encoding="utf-8", newline="") as raw_handle, failures_csv.open( "w", encoding="utf-8", newline="" ) as failure_handle: failure_writer = csv.writer(failure_handle) failure_writer.writerow(["question", "variant", "benchmark", "size", "returncode", "reason", "stderr"]) for question in QUESTIONS: for variant in question.variants: for size in sizes: command = base_command(exe, variant.benchmark, size, args.iterations, args.threads, args) try: completed = subprocess.run( command, capture_output=True, text=True, timeout=timeout_for(variant.benchmark, size, args), ) except subprocess.TimeoutExpired as exc: failure_writer.writerow( [question.slug, variant.key, variant.benchmark, size, "timeout", "timeout", str(exc)] ) continue if completed.returncode != 0: failure_writer.writerow( [ question.slug, variant.key, variant.benchmark, size, completed.returncode, "nonzero", completed.stderr.strip(), ] ) continue lines = [line for line in completed.stdout.splitlines() if line.strip()] if not lines: failure_writer.writerow([question.slug, variant.key, variant.benchmark, size, 0, "empty", ""]) continue if header_written: lines = lines[1:] else: header_written = True raw_handle.write("\n".join(lines) + "\n") raw_handle.flush() (results_dir / "timing_elapsed_seconds.txt").write_text(f"{time.time() - start:.2f}\n", encoding="utf-8") def probe_perf_events(results_dir: Path, args: argparse.Namespace) -> dict[str, object]: events = PERF_EVENTS + OPTIONAL_PERF_EVENTS availability: dict[str, object] = {} for event in events: command = ["perf", "stat", "-x,", "-e", event, "--", "true"] if args.taskset: command = ["taskset", "-c", args.taskset, *command] completed = subprocess.run(command, capture_output=True, text=True) counted = completed.returncode == 0 and "" not in completed.stderr availability[event] = { "available": counted, "returncode": completed.returncode, "stderr": completed.stderr.strip(), } (results_dir / "perf_event_availability.json").write_text( json.dumps(availability, indent=2) + "\n", encoding="utf-8" ) return availability def run_perf( exe: Path, results_dir: Path, sizes: list[str], args: argparse.Namespace, availability: dict[str, object], ) -> None: requested_events = PERF_EVENTS + OPTIONAL_PERF_EVENTS available_events = [ event for event in requested_events if isinstance(availability.get(event), dict) and availability[event].get("available") ] if not available_events: (results_dir / "perf_skipped.txt").write_text("No requested perf events were available.\n", encoding="utf-8") return failures_csv = results_dir / "perf_failures.csv" with failures_csv.open("w", encoding="utf-8", newline="") as failure_handle: failure_writer = csv.writer(failure_handle) failure_writer.writerow(["question", "variant", "benchmark", "size", "returncode", "reason"]) for question in QUESTIONS: for variant in question.variants: for size in sizes: stem = safe_name(f"{question.slug}--{variant.key}--{variant.benchmark}--{size}") perf_path = results_dir / "perf_raw" / f"{stem}.perf.csv" bench_path = results_dir / "perf_bench_csv" / f"{stem}.bench.csv" bench_command = base_command(exe, variant.benchmark, size, 1, args.threads, args) command = [ "perf", "stat", "-x,", "-r", str(args.perf_repeats), "-e", ",".join(available_events), "--", *bench_command, ] try: completed = subprocess.run( command, capture_output=True, text=True, timeout=timeout_for(variant.benchmark, size, args), ) except subprocess.TimeoutExpired: failure_writer.writerow([question.slug, variant.key, variant.benchmark, size, "timeout", "timeout"]) continue bench_path.write_text(completed.stdout, encoding="utf-8") perf_path.write_text(completed.stderr, encoding="utf-8") if completed.returncode != 0: failure_writer.writerow( [question.slug, variant.key, variant.benchmark, size, completed.returncode, "nonzero"] ) def parse_perf_outputs(results_dir: Path) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for path in sorted((results_dir / "perf_raw").glob("*.perf.csv")): meta = metadata_from_perf_name(path.stem) for row in csv.reader(path.read_text(encoding="utf-8", errors="ignore").splitlines()): if len(row) < 3: continue raw_value = row[0].strip() event = row[2].strip() if not event or raw_value.startswith("#"): continue counted = not raw_value.startswith("<") value = parse_float(raw_value) if counted else None rows.append( { **meta, "event": event, "value": value, "counted": counted, "unit": row[1].strip() if len(row) > 1 else "", "raw": ",".join(row), "file": str(path.name), } ) return rows def write_perf_summaries(results_dir: Path, rows: list[dict[str, object]]) -> None: raw_path = results_dir / "perf_results_summary.csv" with raw_path.open("w", encoding="utf-8", newline="") as handle: fieldnames = ["question", "variant", "benchmark", "size", "event", "value", "counted", "unit", "file", "raw"] writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({name: row.get(name, "") for name in fieldnames}) grouped: dict[tuple[str, str, str, str], dict[str, float]] = {} for row in rows: if not row["counted"] or row["value"] is None: continue key = (str(row["question"]), str(row["variant"]), str(row["benchmark"]), str(row["size"])) grouped.setdefault(key, {})[str(row["event"])] = float(row["value"]) derived_path = results_dir / "perf_derived_summary.csv" with derived_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.writer(handle) writer.writerow( [ "question", "variant", "benchmark", "size", "size_bytes", "ipc", "cache_miss_rate", "branch_miss_rate", "l1_load_miss_rate", "dtlb_load_miss_rate", "cycles", "instructions", "cache_references", "cache_misses", "branches", "branch_misses", "l1_loads", "l1_load_misses", "dtlb_loads", "dtlb_load_misses", ] ) for (question, variant, benchmark, size), values in sorted(grouped.items(), key=lambda item: (item[0][0], item[0][1], parse_size(item[0][3]))): cycles = values.get("cycles", 0.0) instructions = values.get("instructions", 0.0) cache_refs = values.get("cache-references", 0.0) cache_misses = values.get("cache-misses", 0.0) branches = values.get("branches", 0.0) branch_misses = values.get("branch-misses", 0.0) l1_loads = values.get("L1-dcache-loads", 0.0) l1_load_misses = values.get("L1-dcache-load-misses", 0.0) dtlb_loads = values.get("dTLB-loads", 0.0) dtlb_load_misses = values.get("dTLB-load-misses", 0.0) ipc = instructions / cycles if cycles > 0 else 0.0 cache_miss_rate = cache_misses / cache_refs if cache_refs > 0 else 0.0 branch_miss_rate = branch_misses / branches if branches > 0 else 0.0 l1_load_miss_rate = l1_load_misses / l1_loads if l1_loads > 0 else 0.0 dtlb_load_miss_rate = dtlb_load_misses / dtlb_loads if dtlb_loads > 0 else 0.0 writer.writerow( [ question, variant, benchmark, size, parse_size(size), f"{ipc:.6f}", f"{cache_miss_rate:.6f}", f"{branch_miss_rate:.6f}", f"{l1_load_miss_rate:.6f}", f"{dtlb_load_miss_rate:.6f}", f"{cycles:.0f}", f"{instructions:.0f}", f"{cache_refs:.0f}", f"{cache_misses:.0f}", f"{branches:.0f}", f"{branch_misses:.0f}", f"{l1_loads:.0f}", f"{l1_load_misses:.0f}", f"{dtlb_loads:.0f}", f"{dtlb_load_misses:.0f}", ] ) def write_timing_charts(charts_dir: Path, points: list[BenchmarkPoint], cache_levels: list[CacheLevel]) -> None: benchmark_map = map_points(points) verticals = cache_verticals(cache_levels) for question in QUESTIONS: series = [] for variant in question.variants: data_points = benchmark_map.get(variant.benchmark, []) data = latency_series(data_points) if question.metric == "latency" else throughput_series(data_points) series.append((variant.label, PALETTE[variant.key], data)) y_label = "Latency (ns per element/query)" if question.metric == "latency" else "Throughput (GiB/s)" draw_line_chart(charts_dir / f"dsa_{question.slug}_timing.svg", question.title, y_label, series, verticals) def write_perf_charts(charts_dir: Path, perf_csv: Path, cache_levels: list[CacheLevel]) -> None: if not perf_csv.exists(): return rows = list(csv.DictReader(perf_csv.open("r", encoding="utf-8", newline=""))) for question in QUESTIONS: question_rows = [row for row in rows if row["question"] == question.slug] if not question_rows: continue draw_perf_chart(charts_dir / f"dsa_{question.slug}_perf.svg", question.title, question_rows, cache_levels) def draw_perf_chart(path: Path, title: str, rows: list[dict[str, str]], cache_levels: list[CacheLevel]) -> None: width = 1100 height = 760 left = 100 right = 50 top = 80 gap = 70 panel_height = 250 bottom = 80 plot_width = width - left - right xs = sorted({int(row["size_bytes"]) for row in rows if row["size_bytes"]}) if not xs: return min_x = min(xs) max_x = max(xs) def x_pos(value: int) -> float: if min_x == max_x: return left + plot_width / 2 return left + (math.log2(value) - math.log2(min_x)) / (math.log2(max_x) - math.log2(min_x)) * plot_width def panel(row_key: str, y_top: int, label: str) -> list[str]: values = [float(row[row_key]) for row in rows if row.get(row_key)] max_y = max(values) if values else 1.0 max_y = max(max_y, 0.001) def y_pos(value: float) -> float: return y_top + panel_height - (value / max_y) * panel_height lines = [ f'{escape_xml(label)}', f'', f'', f'{max_y:.3f}', f'0', ] for cache in cache_levels: if cache.cache_type not in {"Data", "Unified"} or cache.level not in {1, 2, 3}: continue if cache.size_bytes < min_x or cache.size_bytes > max_x: continue x = x_pos(cache.size_bytes) lines.append( f'' ) for variant in ("naive", "cache_aware", "simd", "cache_aware_simd"): variant_rows = sorted( [row for row in rows if row["variant"] == variant], key=lambda row: int(row["size_bytes"]), ) points = [] for row in variant_rows: value = float(row[row_key]) points.append(f'{x_pos(int(row["size_bytes"])):.1f},{y_pos(value):.1f}') if points: lines.append( f'' ) return lines svg = [ f'', '', f'{escape_xml(title)} perf evidence', *panel("ipc", top, "Instructions per cycle"), *panel("cache_miss_rate", top + panel_height + gap, "Cache misses / cache references"), ] legend_x = left legend_y = height - bottom + 35 for variant in ("naive", "cache_aware", "simd", "cache_aware_simd"): svg.append(f'') svg.append(f'{escape_xml(variant.replace("_", " "))}') legend_x += 220 svg.append("") path.write_text("\n".join(svg) + "\n", encoding="utf-8") def write_question_summary(results_dir: Path, points: list[BenchmarkPoint]) -> None: benchmark_map = map_points(points) if points else {} perf_rows = list(csv.DictReader((results_dir / "perf_derived_summary.csv").open("r", encoding="utf-8", newline=""))) if (results_dir / "perf_derived_summary.csv").exists() else [] lines = ["# DSA Full Evidence Summary", ""] for question in QUESTIONS: lines.append(f"## {question.title}") lines.append("") lines.append(f"- Question: {question.problem}") variant_points = [] for variant in question.variants: data = benchmark_map.get(variant.benchmark, []) if not data: lines.append(f"- `{variant.label}`: no timing row completed.") continue row = max(data, key=lambda point: point.requested_size_bytes) score = row.ns_per_access if question.metric == "latency" else row.throughput_gib_per_s variant_points.append((variant, row, score)) unit = "ns/element-query" if question.metric == "latency" else "GiB/s" lines.append(f"- `{variant.label}` at largest completed size `{format_bytes(row.requested_size_bytes)}`: `{score:.4f}` {unit}.") if variant_points: best = min(variant_points, key=lambda item: item[2]) if question.metric == "latency" else max(variant_points, key=lambda item: item[2]) lines.append(f"- Evidence verdict: timing evidence favors `{best[0].label}` at the largest completed size.") question_perf = [row for row in perf_rows if row["question"] == question.slug] if question_perf: lines.append("- Perf evidence: generic counters were captured; use the perf chart and CSV for IPC/cache-miss-rate support.") else: lines.append("- Perf evidence: no counted perf rows were available for this question.") lines.append("") (results_dir / "SUMMARY.md").write_text("\n".join(lines), encoding="utf-8") def write_code_appendix(root: Path, path: Path) -> None: dsa = (root / "src/dsa_benchmarks.cpp").read_text(encoding="utf-8") simd = (root / "src/simd_benchmarks.cpp").read_text(encoding="utf-8") sections = [ ("Shared DSA support types", extract_range(dsa, "template \nstruct WideRecord", "struct GraphCsr")), ("Stock profit", extract_functions(dsa, ["scalar_stock_profit_aos", "scalar_stock_profit_soa", "simd_stock_profit_aos", "simd_stock_profit_soa"])), ("Maximum subarray", extract_functions(dsa, ["scalar_max_subarray_aos", "scalar_max_subarray_soa", "simd_max_subarray_aos", "simd_max_subarray_soa"])), ("Trapping rain water", extract_functions(dsa, ["scalar_trapping_rain_aos", "scalar_trapping_rain_soa", "simd_trapping_rain_aos", "simd_trapping_rain_soa"])), ("Sliding-window maximum", extract_functions(dsa, ["scalar_sliding_max_aos", "scalar_sliding_max_soa", "simd_sliding_max_aos", "simd_sliding_max_soa"])), ("Product except self", extract_functions(dsa, ["scalar_product_except_self_aos", "scalar_product_except_self_soa", "simd_product_except_self_aos", "simd_product_except_self_soa"])), ("Rotated search", extract_functions(dsa, ["scalar_rotated_search_aos_value", "scalar_rotated_search_soa_value", "scalar_rotated_search_aos", "scalar_rotated_search_soa", "simd_rotated_search_batch", "simd_rotated_search_aos", "simd_rotated_search_soa"])), ("Sort 0/1/2", extract_functions(dsa, ["sort012_scalar_aos", "sort012_scalar_soa", "sort012_simd_aos", "sort012_simd_soa"])), ("BFS", extract_functions(dsa, ["bfs_scalar_adjlist", "bfs_scalar_csr", "bfs_simd_adjlist", "bfs_simd_csr"])), ("Transpose", extract_functions(dsa, ["transpose_scalar_naive", "transpose_scalar_blocked", "transpose_simd_naive", "transpose_simd_blocked"])), ("Matrix multiply", extract_functions(simd, ["scalar_matmul_naive", "transpose_square_matrix", "scalar_matmul_transposed", "avx2_matmul_gather", "avx2_matmul_blocked"])), ] lines = [ "# DSA Full Code Appendix", "", "This appendix contains the measured core functions. The benchmark harness also includes dataset generation, warmup, checksums, CSV output, and CLI plumbing.", "", ] for title, code in sections: lines.extend([f"## {title}", "", "```cpp", code.strip(), "```", ""]) path.write_text("\n".join(lines), encoding="utf-8") def write_source_snapshots(root: Path, out_dir: Path) -> None: out_dir.mkdir(parents=True, exist_ok=True) for relative in [ "src/benchmark.hpp", "src/dsa_benchmarks.cpp", "src/simd_benchmarks.cpp", "src/main.cpp", "scripts/run_dsa_evidence.py", "scripts/run_dsa_analysis.py", ]: source = root / relative if source.exists(): target = out_dir / relative.replace("/", "__") target.write_text(source.read_text(encoding="utf-8"), encoding="utf-8") def extract_functions(text: str, names: list[str]) -> str: chunks = [] for name in names: index = text.find(name) if index == -1: continue start = text.rfind("\n", 0, index) start = 0 if start == -1 else start + 1 brace = text.find("{", index) if brace == -1: continue depth = 0 end = brace for pos in range(brace, len(text)): if text[pos] == "{": depth += 1 elif text[pos] == "}": depth -= 1 if depth == 0: end = pos + 1 break while end < len(text) and text[end] in " \t\r\n": end += 1 chunks.append(text[start:end]) return "\n\n".join(chunks) def extract_range(text: str, start_marker: str, end_marker: str) -> str: start = text.find(start_marker) end = text.find(end_marker, start) if start == -1 or end == -1: return "" brace = text.find("};", end) return text[start : brace + 2] def metadata_from_perf_name(stem: str) -> dict[str, str]: name = stem if name.endswith(".perf"): name = name[:-5] parts = name.split("--") if len(parts) != 4: return {"question": "", "variant": "", "benchmark": "", "size": ""} return { "question": parts[0], "variant": parts[1], "benchmark": parts[2], "size": parts[3], } def safe_name(text: str) -> str: return text.replace("/", "_").replace(":", "_").replace(",", "_") def parse_float(text: str) -> float | None: try: return float(text.replace(",", "")) except ValueError: return None def parse_size(label: str) -> int: raw = label.strip().upper() if raw.endswith("K"): return int(raw[:-1]) * 1024 if raw.endswith("M"): return int(raw[:-1]) * 1024 * 1024 if raw.endswith("G"): return int(raw[:-1]) * 1024 * 1024 * 1024 return int(raw) def cache_verticals(cache_levels: list[CacheLevel]) -> list[tuple[int, str, str]]: return [ (cache.size_bytes, f"L{cache.level} {cache.cache_type}", PALETTE["cache_marker"]) for cache in cache_levels if cache.cache_type in {"Data", "Unified"} and cache.level in {1, 2, 3} ] def escape_xml(text: str) -> str: return text.replace("&", "&").replace("<", "<").replace(">", ">") def load_json(path: Path) -> dict[str, object]: if not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8")) if __name__ == "__main__": raise SystemExit(main())