llm-benchmark.py

#!/usr/bin/env python3
"""
LLM Inference Benchmark - 3D Performance Surface Measurement
=============================================================

PURPOSE:
    Measures LLM inference throughput across a matrix of operating conditions
    to characterize the full performance surface of a deployed model.

    Works with any OpenAI-compatible API endpoint:
    - vLLM
    - TensorRT-LLM (trtllm-serve)
    - SGLang
    - Ollama (with OpenAI compatibility layer)

MODES:
    --fast:  6 test points, ~5-10 min  (ctx: 512, 32K | par: 1, 64 | out: 256, 4K)
    --quick: 12 test points, ~15 min   (ctx: 512, 8K, 32K | par: 1, 16, 64)
    --rag:   RAG-focused configs       (ctx: 4K, 8K | par: 1, 4, 16, 64)
    default: 48 test points, ~45 min   (full matrix)

USAGE:
    python llm-benchmark.py <deployment_directory> [--fast|--quick|--rag]

    Example:
    python llm-benchmark.py DgxSpark/MoE/trt-llm.nvidia.Qwen3-30B-A3B-FP4.benchmarked --fast

    The script will:
    1. Read spec.yaml from the deployment directory to get endpoint URL
    2. Read model.context.configured_length to filter test matrix (avoids OOM)
    3. Verify the model server is responding
    4. Run the benchmark matrix
    5. Save results to YYYY-MM-DD-benchmark-{mode}.csv
    6. Update spec.yaml status if successful

REQUIREMENTS:
    - Python 3.8+
    - aiohttp
    - pyyaml
    - Running inference server at the endpoint specified in spec.yaml

OUTPUT FORMAT:
    CSV with columns: gpu, input_ctx, output_max, parallelism, tokens_generated,
                      successful_requests, time_seconds, throughput_tok_s, latency_per_req_s

COMPATIBILITY:
    - Any OpenAI-compatible /v1/completions endpoint
    - Tested: vLLM, TensorRT-LLM, SGLang
    - Tested GPUs: RTX 4090, RTX Pro 6000, DGX Spark (GB10)

Author: Claude Code
Date: 2026-01-06
Updated: 2026-01-26 (renamed from vllm-matrix-benchmark.py, platform-agnostic)
"""

import asyncio
import aiohttp
import argparse
import csv
import os
import sys
import time
import yaml
from datetime import datetime
from pathlib import Path

# =============================================================================
# BENCHMARK CONFIGURATION
# =============================================================================

# Full benchmark matrix (48 points)
INPUT_CONTEXTS_FULL = [512, 2048, 8192, 32768]
OUTPUT_LENGTHS_FULL = [256, 1024, 4096]
PARALLELISM_LEVELS_FULL = [1, 4, 16, 64]

# Quick diagnostic matrix (tests extremes only, 8 points, ~19 min)
# Tests: min/max context, min/max output, min/max parallelism
INPUT_CONTEXTS_QUICK = [512, 32768]
OUTPUT_LENGTHS_QUICK = [256, 4096]
PARALLELISM_LEVELS_QUICK = [1, 64]

# Fast diagnostic matrix (skip expensive corners, 6 points, ~3 min)
# Same as quick but skips par=64 × out=4096 combinations (16 min savings)
INPUT_CONTEXTS_FAST = [512, 32768]
OUTPUT_LENGTHS_FAST = [256, 4096]
PARALLELISM_LEVELS_FAST = [1, 64]
FAST_SKIP_COMBINATIONS = [(64, 4096)]  # (parallelism, output_len) pairs to skip

# RAG test matrix (small outputs, large contexts, 45 points)
# Simulates retrieval-augmented generation: large context in, short answer out
INPUT_CONTEXTS_RAG = [4096, 8192, 16384]
OUTPUT_LENGTHS_RAG = [64, 128, 256]
PARALLELISM_LEVELS_RAG = [1, 8, 16, 32, 64]

# Defaults (will be set by main based on flags)
INPUT_CONTEXTS = INPUT_CONTEXTS_FULL
OUTPUT_LENGTHS = OUTPUT_LENGTHS_FULL
PARALLELISM_LEVELS = PARALLELISM_LEVELS_FULL
SKIP_COMBINATIONS = []  # (parallelism, output_len) pairs to skip

# Prompt template that encourages long output
# The {padding} will be filled with context to reach target input length
PROMPT_TEMPLATE = """You are a helpful assistant. Please write an extremely detailed,
comprehensive response. Do not stop early. Continue writing until you reach the maximum
length allowed. Be thorough and verbose.

Context information:
{padding}

Task: Write a very long, detailed essay about the nature of intelligence, consciousness,
and the future of AI. Include multiple sections, examples, and thorough explanations.
Do not summarize or be brief - expand on every point extensively."""

# Padding text to reach target context length (repeated as needed)
PADDING_TEXT = """Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do
eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam,
quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. """

# =============================================================================
# HELPER FUNCTIONS
# =============================================================================

def load_spec(deployment_dir: Path) -> dict:
    """Load and parse the spec.yaml from deployment directory."""
    spec_path = deployment_dir / "spec.yaml"
    if not spec_path.exists():
        raise FileNotFoundError(f"spec.yaml not found in {deployment_dir}")

    with open(spec_path) as f:
        return yaml.safe_load(f)


def get_endpoint_from_spec(spec: dict) -> tuple[str, str, int]:
    """
    Extract API endpoint URL, model name, and max context from spec.
    Returns (endpoint, model, max_context_length).
    """
    endpoint = None
    model = None
    max_context = None

    # Look for endpoint URL
    if "deployment" in spec:
        dep = spec["deployment"]
        if "endpoint" in dep:
            endpoint = dep["endpoint"].get("url")
        if "vllm_args" in dep:
            # Extract port from vllm args if present
            args = dep["vllm_args"]
            if "--port" in args:
                port_idx = args.index("--port") + 1
                if port_idx < len(args):
                    endpoint = f"http://localhost:{args[port_idx]}/v1"

    # Look for model name
    if "sources" in spec:
        sources = spec["sources"]
        if isinstance(sources, dict) and "model" in sources:
            model = sources["model"].get("repo")

    if "model" in spec:
        m = spec["model"]
        if isinstance(m, dict):
            model = m.get("model") or m.get("name") or model
            # Get configured context length (max_model_len)
            if "context" in m:
                max_context = m["context"].get("configured_length")

    # Default endpoint if not found
    if not endpoint:
        endpoint = "http://localhost:8000/v1"

    return endpoint, model, max_context


def generate_prompt(target_tokens: int) -> str:
    """Generate a prompt with approximately target_tokens of context."""
    # Rough estimate: 4 chars per token
    target_chars = target_tokens * 4
    base_len = len(PROMPT_TEMPLATE.format(padding=""))
    padding_needed = max(0, target_chars - base_len)

    # Repeat padding text to reach target
    padding = (PADDING_TEXT * (padding_needed // len(PADDING_TEXT) + 1))[:padding_needed]

    return PROMPT_TEMPLATE.format(padding=padding)


async def check_server(endpoint: str, model: str) -> tuple[bool, str]:
    """
    Check if the vLLM server is responding.
    Returns (success, actual_model_name).
    If server has a different model, returns the server's model name.
    """
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{endpoint}/models", timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    models = [m["id"] for m in data.get("data", [])]
                    if model and model in models:
                        return True, model
                    elif models:
                        actual_model = models[0]  # Use first available model
                        print(f"Note: Using model '{actual_model}' from server (spec had '{model}')")
                        return True, actual_model
                return False, model
    except Exception as e:
        return False, model


async def generate_completion(
    session: aiohttp.ClientSession,
    endpoint: str,
    model: str,
    prompt: str,
    max_tokens: int,
    timeout: int = 300
) -> tuple[int, float]:
    """
    Send a completion request and return (tokens_generated, elapsed_time).
    Returns (0, 0) on error.
    """
    payload = {
        "model": model,
        "prompt": prompt,
        "max_tokens": max_tokens,
        "temperature": 0.7,
    }

    try:
        start = time.perf_counter()
        async with session.post(
            f"{endpoint}/completions",
            json=payload,
            timeout=aiohttp.ClientTimeout(total=timeout)
        ) as resp:
            result = await resp.json()
            elapsed = time.perf_counter() - start

            if "usage" in result:
                tokens = result["usage"].get("completion_tokens", 0)
                return tokens, elapsed
            elif "error" in result:
                print(f"  API error: {result['error']}")
                return 0, 0
            else:
                return 0, 0
    except asyncio.TimeoutError:
        print(f"  Request timed out after {timeout}s")
        return 0, 0
    except Exception as e:
        print(f"  Request error: {e}")
        return 0, 0


async def benchmark_point(
    endpoint: str,
    model: str,
    input_ctx: int,
    output_max: int,
    parallelism: int,
    gpu: str = None
) -> dict:
    """
    Benchmark a single point on the performance surface.
    Returns dict with metrics.
    """
    prompt = generate_prompt(input_ctx)

    # Calculate appropriate timeout based on expected generation time
    # Assume minimum 10 tok/s, add buffer
    timeout = max(60, (output_max / 10) * parallelism * 2)

    connector = aiohttp.TCPConnector(limit=parallelism + 10)
    async with aiohttp.ClientSession(connector=connector) as session:
        start = time.perf_counter()

        tasks = [
            generate_completion(session, endpoint, model, prompt, output_max, timeout)
            for _ in range(parallelism)
        ]
        results = await asyncio.gather(*tasks)

        elapsed = time.perf_counter() - start

        total_tokens = sum(r[0] for r in results)
        successful = sum(1 for r in results if r[0] > 0)

        if elapsed > 0 and total_tokens > 0:
            throughput = total_tokens / elapsed
            latency_per_req = elapsed / parallelism
        else:
            throughput = 0
            latency_per_req = 0

        return {
            "gpu": gpu or "",
            "input_ctx": input_ctx,
            "output_max": output_max,
            "parallelism": parallelism,
            "tokens_generated": total_tokens,
            "successful_requests": successful,
            "time_seconds": round(elapsed, 2),
            "throughput_tok_s": round(throughput, 1),
            "latency_per_req_s": round(latency_per_req, 2)
        }


async def run_benchmark(
    endpoint: str,
    model: str,
    csv_path: Path,
    completed: set[tuple[int, int, int]],
    gpu: str = None
) -> list[dict]:
    """Run the benchmark matrix with incremental writes, skipping completed tests."""
    results = []

    # Build list of tests to run
    all_tests = []
    for input_ctx in INPUT_CONTEXTS:
        for output_max in OUTPUT_LENGTHS:
            for parallelism in PARALLELISM_LEVELS:
                if (parallelism, output_max) in SKIP_COMBINATIONS:
                    continue
                all_tests.append((input_ctx, output_max, parallelism))

    # Filter out completed tests
    remaining_tests = [t for t in all_tests if t not in completed]
    total_points = len(all_tests)
    already_done = len(completed)

    if already_done > 0:
        print(f"\nResuming: {already_done}/{total_points} tests already complete")

    print(f"\nRunning {len(remaining_tests)} benchmark points...")
    if gpu:
        print(f"GPU: {gpu}")
    print(f"Input contexts: {INPUT_CONTEXTS}")
    print(f"Output lengths: {OUTPUT_LENGTHS}")
    print(f"Parallelism: {PARALLELISM_LEVELS}")
    if SKIP_COMBINATIONS:
        print(f"Skipping: {SKIP_COMBINATIONS} (expensive corners)")
    print("-" * 70)

    # Warmup
    print("Warming up...", end=" ", flush=True)
    connector = aiohttp.TCPConnector(limit=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        await generate_completion(session, endpoint, model, "Hello", 10, 30)
    print("done")

    current = already_done
    for input_ctx, output_max, parallelism in remaining_tests:
        current += 1
        print(f"[{current}/{total_points}] ctx={input_ctx}, out={output_max}, "
              f"par={parallelism}...", end=" ", flush=True)

        result = await benchmark_point(
            endpoint, model, input_ctx, output_max, parallelism, gpu
        )
        results.append(result)

        # Write incrementally
        append_result(csv_path, result)

        print(f"{result['throughput_tok_s']} tok/s "
              f"({result['successful_requests']}/{parallelism} ok)")

    return results


CSV_FIELDNAMES = [
    "gpu", "input_ctx", "output_max", "parallelism", "tokens_generated",
    "successful_requests", "time_seconds", "throughput_tok_s", "latency_per_req_s"
]


def get_benchmark_paths(deployment_dir: Path, mode: str, gpu: str = None) -> tuple[Path, Path, Path]:
    """Get benchmark file paths: (benchmarks_dir, incomplete_path, complete_path)."""
    benchmarks_dir = deployment_dir / "benchmarks"
    date_str = datetime.now().strftime("%Y-%m-%d")

    mode_suffix = f"-{mode}" if mode else ""
    gpu_suffix = f"-{gpu}" if gpu else ""
    incomplete_path = benchmarks_dir / f"{date_str}-benchmark{mode_suffix}{gpu_suffix}-incomplete.csv"
    complete_path = benchmarks_dir / f"{date_str}-benchmark{mode_suffix}{gpu_suffix}.csv"

    return benchmarks_dir, incomplete_path, complete_path


def find_next_suffix(benchmarks_dir: Path, base_name: str) -> str:
    """Find next available suffix (b, c, d...) if base file exists."""
    if not (benchmarks_dir / f"{base_name}.csv").exists():
        return ""

    for suffix in "bcdefghijklmnopqrstuvwxyz":
        # Insert suffix after date: 2026-01-07b-benchmark-fast.csv
        parts = base_name.split("-benchmark", 1)
        suffixed_name = f"{parts[0]}{suffix}-benchmark{parts[1] if len(parts) > 1 else ''}"
        if not (benchmarks_dir / f"{suffixed_name}.csv").exists():
            return suffix

    raise RuntimeError("Too many benchmark runs today (exhausted a-z)")


def load_completed_tests(csv_path: Path) -> set[tuple[int, int, int]]:
    """Load completed (input_ctx, output_max, parallelism) tuples from CSV."""
    completed = set()
    if csv_path.exists():
        with open(csv_path, newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                completed.add((
                    int(row["input_ctx"]),
                    int(row["output_max"]),
                    int(row["parallelism"])
                ))
    return completed


def init_benchmark_file(csv_path: Path) -> None:
    """Initialize benchmark CSV with header."""
    csv_path.parent.mkdir(parents=True, exist_ok=True)
    with open(csv_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_FIELDNAMES)
        writer.writeheader()


def append_result(csv_path: Path, result: dict) -> None:
    """Append a single result to CSV."""
    with open(csv_path, "a", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_FIELDNAMES)
        writer.writerow(result)


def finalize_benchmark(incomplete_path: Path, complete_path: Path) -> Path:
    """Rename incomplete file to complete."""
    incomplete_path.rename(complete_path)
    return complete_path


def update_spec_benchmarked(deployment_dir: Path):
    """Update spec.yaml to set benchmarked: true."""
    spec_path = deployment_dir / "spec.yaml"

    with open(spec_path) as f:
        content = f.read()

    # Simple text replacement - look for benchmarked: false and replace
    if "benchmarked: false" in content:
        content = content.replace("benchmarked: false", "benchmarked: true")
        with open(spec_path, "w") as f:
            f.write(content)
        print("Updated spec.yaml: benchmarked: true")
    elif "benchmarked: true" in content:
        print("spec.yaml already has benchmarked: true")
    else:
        print("Note: Could not find benchmarked field in spec.yaml")


def print_summary(results: list[dict]):
    """Print a summary table of results."""
    print("\n" + "=" * 70)
    print("BENCHMARK RESULTS SUMMARY")
    print("=" * 70)

    # Group by input context
    for input_ctx in INPUT_CONTEXTS:
        print(f"\nInput Context: {input_ctx} tokens")
        print("-" * 50)

        # Header
        print(f"{'Parallel':>10}", end="")
        for out_len in OUTPUT_LENGTHS:
            print(f"{out_len:>12}", end="")
        print()

        # Data rows
        for parallelism in PARALLELISM_LEVELS:
            print(f"{parallelism:>10}", end="")
            for out_len in OUTPUT_LENGTHS:
                point = next(
                    (r for r in results
                     if r["input_ctx"] == input_ctx
                     and r["output_max"] == out_len
                     and r["parallelism"] == parallelism),
                    None
                )
                if point:
                    print(f"{point['throughput_tok_s']:>12.1f}", end="")
                else:
                    print(f"{'N/A':>12}", end="")
            print()

    # Find peak
    peak = max(results, key=lambda x: x["throughput_tok_s"])
    print(f"\nPEAK THROUGHPUT: {peak['throughput_tok_s']} tok/s")
    print(f"  at input_ctx={peak['input_ctx']}, output_max={peak['output_max']}, "
          f"parallelism={peak['parallelism']}")


async def main():
    global INPUT_CONTEXTS, OUTPUT_LENGTHS, PARALLELISM_LEVELS

    parser = argparse.ArgumentParser(
        description="vLLM Matrix Benchmark - 3D Performance Surface Measurement"
    )
    parser.add_argument(
        "deployment_dir",
        help="Path to deployment directory (contains spec.yaml)"
    )
    parser.add_argument(
        "--endpoint",
        help="Override endpoint URL (default: read from spec.yaml)"
    )
    parser.add_argument(
        "--model",
        help="Override model name (default: read from spec.yaml)"
    )
    parser.add_argument(
        "--quick",
        action="store_true",
        help="Quick diagnostic mode: test extremes only (8 points, ~19 min)"
    )
    parser.add_argument(
        "--fast",
        action="store_true",
        help="Fast diagnostic mode: skip expensive corners (6 points, ~3 min)"
    )
    parser.add_argument(
        "--rag",
        action="store_true",
        help="RAG test mode: small outputs (64-256), large context (4K-16K), par 1-64 (45 points)"
    )
    parser.add_argument(
        "--gpu",
        help="GPU tag for benchmark identification (e.g., rtx4090, rtx6000, dgxspark)"
    )

    args = parser.parse_args()

    # Set benchmark matrix based on flags
    global SKIP_COMBINATIONS
    if args.rag:
        INPUT_CONTEXTS = INPUT_CONTEXTS_RAG.copy()
        OUTPUT_LENGTHS = OUTPUT_LENGTHS_RAG.copy()
        PARALLELISM_LEVELS = PARALLELISM_LEVELS_RAG.copy()
        SKIP_COMBINATIONS = []
        print("RAG test mode: small outputs (64-256), large context (4K-16K) (45 points)")
    elif args.fast:
        INPUT_CONTEXTS = INPUT_CONTEXTS_FAST.copy()
        OUTPUT_LENGTHS = OUTPUT_LENGTHS_FAST.copy()
        PARALLELISM_LEVELS = PARALLELISM_LEVELS_FAST.copy()
        SKIP_COMBINATIONS = FAST_SKIP_COMBINATIONS.copy()
        print("Fast diagnostic mode: skipping expensive par×output corners (~3 min)")
    elif args.quick:
        INPUT_CONTEXTS = INPUT_CONTEXTS_QUICK.copy()
        OUTPUT_LENGTHS = OUTPUT_LENGTHS_QUICK.copy()
        PARALLELISM_LEVELS = PARALLELISM_LEVELS_QUICK.copy()
        SKIP_COMBINATIONS = []
        print("Quick diagnostic mode: testing extremes only (~19 min)")
    else:
        INPUT_CONTEXTS = INPUT_CONTEXTS_FULL.copy()
        OUTPUT_LENGTHS = OUTPUT_LENGTHS_FULL.copy()
        PARALLELISM_LEVELS = PARALLELISM_LEVELS_FULL.copy()
        SKIP_COMBINATIONS = []

    # Resolve deployment directory
    deployment_dir = Path(args.deployment_dir)
    if not deployment_dir.is_absolute():
        # Try relative to current dir, then relative to llm repo root
        if not deployment_dir.exists():
            llm_root = Path(__file__).parent.parent
            deployment_dir = llm_root / args.deployment_dir

    if not deployment_dir.exists():
        print(f"Error: Deployment directory not found: {args.deployment_dir}")
        sys.exit(1)

    print(f"Deployment directory: {deployment_dir}")

    # Load spec
    try:
        spec = load_spec(deployment_dir)
    except FileNotFoundError as e:
        print(f"Error: {e}")
        sys.exit(1)

    # Get endpoint, model, and max context from spec
    endpoint, model, max_context = get_endpoint_from_spec(spec)
    if args.endpoint:
        endpoint = args.endpoint
    if args.model:
        model = args.model

    print(f"Endpoint: {endpoint}")
    print(f"Model: {model}")

    # Filter input contexts based on configured_length from spec
    if max_context:
        original_contexts = INPUT_CONTEXTS.copy()
        INPUT_CONTEXTS = [ctx for ctx in INPUT_CONTEXTS if ctx <= max_context]
        filtered = [ctx for ctx in original_contexts if ctx > max_context]
        if filtered:
            print(f"Max context (from spec): {max_context} tokens")
            print(f"Skipping context sizes > {max_context}: {filtered}")
    else:
        print("Note: No configured_length in spec.yaml - testing all context sizes")

    # Check server is responding
    print("\nChecking server connectivity...", end=" ", flush=True)
    server_ok, actual_model = await check_server(endpoint, model)
    if not server_ok:
        print("FAILED")
        print(f"""
Error: Cannot connect to vLLM server at {endpoint}

To run this benchmark, the model must be served. Example:

    python -m vllm.entrypoints.openai.api_server \\
        --model {model or 'YOUR_MODEL'} \\
        --port 8000 \\
        --trust-remote-code

Then re-run this benchmark.
""")
        sys.exit(1)
    print("OK")

    # Use the actual model name from server
    model = actual_model
    print(f"Using model: {model}")

    # Determine mode string for filenames
    mode = "rag" if args.rag else ("fast" if args.fast else ("quick" if args.quick else ""))

    # Set up benchmark paths
    benchmarks_dir, incomplete_path, complete_path = get_benchmark_paths(deployment_dir, mode, args.gpu)
    benchmarks_dir.mkdir(parents=True, exist_ok=True)

    # Check for resume or new run
    completed = set()
    csv_path = incomplete_path

    if incomplete_path.exists():
        # Resume from incomplete file
        completed = load_completed_tests(incomplete_path)
        print(f"Found incomplete benchmark: {incomplete_path.name}")
    elif complete_path.exists():
        # Previous run exists, need new suffix
        date_str = datetime.now().strftime("%Y-%m-%d")
        base_name = f"{date_str}-benchmark" + (f"-{mode}" if mode else "")
        suffix = find_next_suffix(benchmarks_dir, base_name)
        parts = base_name.split("-benchmark", 1)
        new_base = f"{parts[0]}{suffix}-benchmark{parts[1] if len(parts) > 1 else ''}"
        csv_path = benchmarks_dir / f"{new_base}-incomplete.csv"
        complete_path = benchmarks_dir / f"{new_base}.csv"
        print(f"Starting new run: {csv_path.name}")
        init_benchmark_file(csv_path)
    else:
        # Fresh start
        print(f"Output: {incomplete_path.name}")
        init_benchmark_file(csv_path)

    # Run benchmark with incremental writes
    new_results = await run_benchmark(endpoint, model, csv_path, completed, args.gpu)

    # Load all results (including previously completed) for summary
    all_results = []
    with open(csv_path, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            all_results.append({
                "gpu": row.get("gpu", ""),
                "input_ctx": int(row["input_ctx"]),
                "output_max": int(row["output_max"]),
                "parallelism": int(row["parallelism"]),
                "tokens_generated": int(row["tokens_generated"]),
                "successful_requests": int(row["successful_requests"]),
                "time_seconds": float(row["time_seconds"]),
                "throughput_tok_s": float(row["throughput_tok_s"]),
                "latency_per_req_s": float(row["latency_per_req_s"])
            })

    # Check for failures
    failed_points = [r for r in all_results if r["throughput_tok_s"] == 0]
    if failed_points:
        print(f"\nWarning: {len(failed_points)} test points failed")

    # Finalize: rename incomplete to complete
    final_path = finalize_benchmark(csv_path, complete_path)
    print(f"\nResults saved to: {final_path}")

    # Print summary
    print_summary(all_results)

    # Update spec if all points succeeded
    if not failed_points:
        update_spec_benchmarked(deployment_dir)
        print("\nBenchmark completed successfully!")
    else:
        print(f"\nBenchmark completed with {len(failed_points)} failures.")
        print("spec.yaml NOT updated due to failures.")


if __name__ == "__main__":
    asyncio.run(main())