Spaces:

varshithreddy1807
/

aria

Configuration error

File size: 9,847 Bytes

import os
import shutil
import time
import sys
from collections import defaultdict

# Add project root to python path so backend/core/services can be imported
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

from core.analysis_registry import AnalysisRegistry
from core.build_pipeline import BuildPipeline
from services.symbol_service import SymbolService
from services.architecture_service import ArchitectureService
from services.call_graph_service import CallGraphService
from services.api_surface_service import APISurfaceService
from backend.dependencies import snapshot_store, analysis_cache


def create_mock_repo(repo_dir, num_files=50):
    os.makedirs(repo_dir, exist_ok=True)
    for i in range(1, num_files + 1):
        content = f"""# Mock file {i} for benchmarking
def function_{i}(x):
    return x + {i}

class Class_{i}:
    def method_{i}(self):
        return function_{i}(10)
"""
        with open(os.path.join(repo_dir, f"file_{i}.py"), "w") as f:
            f.write(content)


def modify_files(repo_dir, file_indices):
    for i in file_indices:
        content = f"""# Mock file {i} modified
def function_{i}(x):
    # Added comment or change
    return x + {i} + 100

class Class_{i}:
    def method_{i}(self):
        return function_{i}(20)
"""
        with open(os.path.join(repo_dir, f"file_{i}.py"), "w") as f:
            f.write(content)


def main():
    repo_name = "benchmark/repo"
    repo_dir = os.path.abspath(
        os.path.join(os.path.dirname(__file__), "..", "benchmark_temp_repo")
    )

    # Setup custom registry and pipeline
    benchmark_registry = AnalysisRegistry()
    benchmark_registry.register(
        "Symbol Index", SymbolService, dependencies=[], outputs=["symbols"]
    )
    benchmark_registry.register(
        "Dependency Graph",
        ArchitectureService,
        dependencies=["Symbol Index"],
        outputs=["graphs/dependency"],
    )
    benchmark_registry.register(
        "Call Graph",
        CallGraphService,
        dependencies=["Symbol Index", "Dependency Graph"],
        outputs=["graphs/call", "call_graphs"],
    )
    benchmark_registry.register(
        "API Surface",
        APISurfaceService,
        dependencies=["Symbol Index", "Dependency Graph"],
        outputs=["api_surface"],
    )

    pipeline = BuildPipeline(benchmark_registry)

    def run_build(max_workers: int, force_rebuild=False):
        # Track CPU time
        cpu_start = time.process_time()
        start_time = time.time()

        events = list(
            pipeline.build(
                repo_name,
                repo_path=repo_dir,
                force_rebuild=force_rebuild,
                max_workers=max_workers,
            )
        )

        wall_duration = (time.time() - start_time) * 1000
        cpu_duration = (time.process_time() - cpu_start) * 1000

        node_durations = {}
        tasks_executed = 0
        tasks_skipped = 0

        for e in events:
            if e.get("event") == "ERROR":
                print(f"Error executing build: {e.get('message')}")
                return None
            elif e.get("event") == "TASK_STARTED":
                tasks_executed += 1
            elif e.get("event") == "TASK_SKIPPED":
                tasks_skipped += 1
            elif e.get("event") == "BUILD TIME":
                node_durations[e["node"]] = e["duration_ms"]

        return {
            "wall_ms": wall_duration,
            "cpu_ms": cpu_duration,
            "node_durations": node_durations,
            "executed": tasks_executed,
            "skipped": tasks_skipped,
        }

    def clear_repo_snapshots():
        try:
            snapshot_store.delete(repo_name, "build_manifest")
            snapshot_store.delete(repo_name, "symbols")
            snapshot_store.delete(repo_name, "graphs/dependency")
            snapshot_store.delete(repo_name, "graphs/call")
            snapshot_store.delete(repo_name, "call_graphs")
            snapshot_store.delete(repo_name, "api_surface")
            analysis_cache.clear()
        except Exception:
            pass

    # Ensure clean state
    if os.path.exists(repo_dir):
        shutil.rmtree(repo_dir)
    clear_repo_snapshots()

    try:
        # --- Warmup Run ---
        print("Warming up parser libraries and cache...")
        create_mock_repo(repo_dir, 10)
        run_build(max_workers=1)
        modify_files(repo_dir, [1])
        run_build(max_workers=2)
        shutil.rmtree(repo_dir)
        clear_repo_snapshots()

        num_runs = 3
        # Results structure: mode (sequential/parallel) -> scenario -> run metrics
        metrics = {"sequential": defaultdict(list), "parallel": defaultdict(list)}

        print(
            f"\nRunning {num_runs} iterations of sequential (1 worker) vs parallel (4 workers) builds..."
        )

        scenarios = [
            ("Full Build", lambda dir: None),  # No modifications, runs on fresh repo
            ("No-change Rebuild", lambda dir: None),  # Second run, nothing modified
            ("Single-file Change", lambda dir: modify_files(dir, [1])),
            ("5-file Change", lambda dir: modify_files(dir, [2, 3, 4, 5, 6])),
            ("50-file Change (All)", lambda dir: modify_files(dir, list(range(1, 51)))),
        ]

        for run in range(1, num_runs + 1):
            print(f"Iteration {run}/{num_runs}...")

            # --- Test Sequential Mode ---
            if os.path.exists(repo_dir):
                shutil.rmtree(repo_dir)
            clear_repo_snapshots()
            create_mock_repo(repo_dir, 50)

            for idx, (name, mod_fn) in enumerate(scenarios):
                if idx > 0:
                    mod_fn(repo_dir)
                res = run_build(max_workers=1)
                if res:
                    metrics["sequential"][name].append(res)

            # --- Test Parallel Mode ---
            if os.path.exists(repo_dir):
                shutil.rmtree(repo_dir)
            clear_repo_snapshots()
            create_mock_repo(repo_dir, 50)

            for idx, (name, mod_fn) in enumerate(scenarios):
                if idx > 0:
                    mod_fn(repo_dir)
                res = run_build(max_workers=4)
                if res:
                    metrics["parallel"][name].append(res)

        # Compute Averages
        avg_metrics = defaultdict(dict)
        for mode in ["sequential", "parallel"]:
            for name, _ in scenarios:
                run_list = metrics[mode][name]
                if not run_list:
                    continue

                avg_wall = sum(r["wall_ms"] for r in run_list) / len(run_list)
                avg_cpu = sum(r["cpu_ms"] for r in run_list) / len(run_list)
                avg_exec = sum(r["executed"] for r in run_list) / len(run_list)
                avg_skip = sum(r["skipped"] for r in run_list) / len(run_list)

                node_avg = defaultdict(float)
                for r in run_list:
                    for n, d in r["node_durations"].items():
                        node_avg[n] += d
                for n in node_avg:
                    node_avg[n] /= len(run_list)

                avg_metrics[mode][name] = {
                    "wall": avg_wall,
                    "cpu": avg_cpu,
                    "executed": avg_exec,
                    "skipped": avg_skip,
                    "node_durations": dict(node_avg),
                }

        # Print detailed report
        report = []
        report.append("# Concurrency Performance Benchmark Report")
        report.append("\n## Average Scenario Performance Summary\n")
        report.append(
            "| Scenario | Sequential Wall-clock (ms) | Parallel Wall-clock (ms) | Speedup Factor | CPU Time (ms) | Executed Tasks | Skipped Tasks | Cache Hit Rate |"
        )
        report.append("| --- | --- | --- | --- | --- | --- | --- | --- |")

        for name, _ in scenarios:
            seq = avg_metrics["sequential"][name]
            par = avg_metrics["parallel"][name]

            speedup = seq["wall"] / max(par["wall"], 0.001)
            total_tasks = seq["executed"] + seq["skipped"]
            hit_rate = (seq["skipped"] / total_tasks) * 100 if total_tasks > 0 else 0.0

            report.append(
                f"| **{name}** | {seq['wall']:.2f} ms | {par['wall']:.2f} ms | {speedup:.2f}x | {par['cpu']:.2f} ms | {par['executed']:.1f} | {par['skipped']:.1f} | {hit_rate:.1f}% |"
            )

        report.append("\n## Concurrency Utilization Metrics\n")
        report.append(
            "| Scenario | Parallel Duration (ms) | Total Task Time (ms) | Worker Utilization | Parallel Efficiency |"
        )
        report.append("| --- | --- | --- | --- | --- |")

        for name, _ in scenarios:
            seq = avg_metrics["sequential"][name]
            par = avg_metrics["parallel"][name]

            # Sum of task durations inside parallel threads
            total_task_time = sum(par["node_durations"].values())
            # Utilization = total task time / (wall clock time * workers)
            utilization = (
                total_task_time / (par["wall"] * 4) if par["wall"] > 0 else 0.0
            )
            # Efficiency = seq total task time / (par total time * workers)
            efficiency = seq["wall"] / (par["wall"] * 4) if par["wall"] > 0 else 0.0

            report.append(
                f"| **{name}** | {par['wall']:.2f} ms | {total_task_time:.2f} ms | {utilization:.2%} | {efficiency:.2%} |"
            )

        report_text = "\n".join(report)
        print("\n=== BENCHMARK REPORT ===")
        print(report_text)

    finally:
        # Cleanup
        print(f"\nCleaning up temporary repository at {repo_dir}...")
        if os.path.exists(repo_dir):
            shutil.rmtree(repo_dir)


if __name__ == "__main__":
    main()