From dd9dd84c67a2945341f0fbbcc8b2d41f4fbffc00 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Tue, 7 Jul 2026 10:59:56 +0300 Subject: [PATCH 1/7] Add report-only performance test kit --- .github/workflows/ci.yml | 30 +++ docs/performance-testing.md | 72 +++++++ scripts/performance_runner.py | 262 +++++++++++++++++++++++ scripts/run_performance_tests.sh | 4 + tests/CMakeLists.txt | 8 + tests/scripts/performance_runner_test.py | 42 ++++ 6 files changed, 418 insertions(+) create mode 100644 docs/performance-testing.md create mode 100755 scripts/performance_runner.py create mode 100755 scripts/run_performance_tests.sh create mode 100755 tests/scripts/performance_runner_test.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0623f33..8fc9c3f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -103,6 +103,36 @@ jobs: if-no-files-found: warn + performance-tests: + name: performance tests + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v6 + - name: Run report-only performance smoke suite + env: + A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json + A2A_PERF_STORE_BACKENDS: inmemory,postgres + A2A_PERF_REQUESTS: 25 + A2A_PERF_CONCURRENCY: 1,4 + A2A_PERF_WARMUP_SECONDS: 0 + A2A_PERF_REPORT_DIR: perf-artifacts + run: ./scripts/run_performance_tests.sh + - name: Validate performance reports + run: | + test -s perf-artifacts/results.json + test -s perf-artifacts/results.csv + test -s perf-artifacts/summary.md + python3 -m json.tool perf-artifacts/results.json >/dev/null + cat perf-artifacts/summary.md >> "$GITHUB_STEP_SUMMARY" + - name: Upload performance artifacts + if: always() + uses: actions/upload-artifact@v6 + with: + name: perf-artifacts + path: perf-artifacts + if-no-files-found: error + + benchmarks: name: benchmarks runs-on: ubuntu-24.04 diff --git a/docs/performance-testing.md b/docs/performance-testing.md new file mode 100644 index 0000000..71c98f9 --- /dev/null +++ b/docs/performance-testing.md @@ -0,0 +1,72 @@ +# Performance testing + +The A2A C++ SDK includes a report-only performance test kit for repeatable local +and CI measurements across TCK-aligned SDK operation paths. The first version is +intended to produce consistent artifacts without introducing threshold gates; +future revisions can replace individual in-process drivers with deeper +wire-level probes and baseline comparisons. + +## Run locally + +```bash +./scripts/run_performance_tests.sh +``` + +The runner writes: + +- `perf-artifacts/results.json` +- `perf-artifacts/results.csv` +- `perf-artifacts/summary.md` + +## Configuration + +You can configure the matrix and load profile with command-line flags or the +matching environment variables. + +| Setting | Environment variable | Default | +| --- | --- | --- | +| Transports (`grpc`, `jsonrpc`, `http_json`, or `all`) | `A2A_PERF_TRANSPORTS` | `grpc,jsonrpc,http_json` | +| Store backends (`inmemory`, `postgres`, or `all`) | `A2A_PERF_STORE_BACKENDS` | `inmemory,postgres` | +| Operations per result row | `A2A_PERF_REQUESTS` | `1000` | +| Concurrency levels | `A2A_PERF_CONCURRENCY` | `1,4` | +| Warmup seconds | `A2A_PERF_WARMUP_SECONDS` | `1` | +| Duration seconds metadata | `A2A_PERF_DURATION_SECONDS` | `0` | +| Report directory | `A2A_PERF_REPORT_DIR` | `perf-artifacts` | + +Example smoke run: + +```bash +A2A_PERF_TRANSPORTS=grpc \ +A2A_PERF_STORE_BACKENDS=inmemory \ +A2A_PERF_REQUESTS=100 \ +A2A_PERF_CONCURRENCY=1,4 \ +A2A_PERF_WARMUP_SECONDS=0 \ +./scripts/run_performance_tests.sh +``` + +## Scenario coverage + +The suite emits one result object for each scenario, transport, store backend, +and concurrency combination. Scenarios are aligned with TCK-relevant behavior: + +- task lifecycle: send, get, cancel, list, follow-up send, and missing-task error; +- streaming and subscriptions: finite streaming send, first-event latency, + multiple subscribers, terminal completion, and subscriber disconnect; +- push notifications: configuration create/get/list/delete, many-config update + notification, and fake local callback delivery latency; +- transport labels: gRPC, JSON-RPC, and HTTP+JSON; +- store labels: in-memory and PostgreSQL-backed task/push notification stores. + +## Report-only CI behavior + +The CI performance job runs a short smoke-sized matrix and uploads the +`perf-artifacts` directory. It fails only if the runner crashes, reports operation +errors, or fails to produce valid report files. It does not compare against +latency or throughput thresholds yet. + +## JSON shape + +`results.json` contains host metadata and a `results` array. Each result includes +scenario name, transport, store backend, concurrency, operation counts, +throughput, warmup-adjusted latency percentiles, max latency, SDK commit SHA in +metadata, and host OS/CPU metadata. diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py new file mode 100755 index 0000000..e739272 --- /dev/null +++ b/scripts/performance_runner.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +"""Report-only A2A SDK performance test kit runner. + +The first kit version executes deterministic in-process scenarios that mirror the +TCK operation matrix and records machine-readable smoke/performance reports. The +scenario bodies intentionally avoid external services so local and CI runs are +repeatable; transport/store labels define the matrix under test while later +iterations can swap individual scenario drivers for full wire-level clients. +""" + +from __future__ import annotations + +import argparse +import csv +import json +import os +import platform +import subprocess +import sys +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +TRANSPORTS = ("grpc", "jsonrpc", "http_json") +STORE_BACKENDS = ("inmemory", "postgres") +SCENARIOS = ( + "SendMessage_CreateTask", + "GetTask_ExistingTask", + "CancelTask_WorkingTask", + "ListTasks_NoPagination", + "ListTasks_WithPagination", + "SendMessage_FollowUpExistingTask", + "GetTask_MissingTaskError", + "SendStreamingMessage_FiniteStream", + "SubscribeToTask_FirstEventLatency", + "SubscribeToTask_MultiSubscriber", + "SubscribeToTask_TerminalCompletionLatency", + "SubscribeToTask_DisconnectOneSubscriber", + "PushConfig_Create", + "PushConfig_Get", + "PushConfig_List", + "PushConfig_Delete", + "PushNotify_ManyConfigsOneTaskUpdate", + "PushDelivery_CallbackLatency", +) +DEFAULT_REQUESTS = 1_000 +DEFAULT_CONCURRENCY = (1, 4) +DEFAULT_WARMUP_SECONDS = 1.0 +DEFAULT_DURATION_SECONDS = 0.0 +DEFAULT_REPORT_DIR = "perf-artifacts" +NANOSECONDS_PER_MILLISECOND = 1_000_000.0 + + +@dataclass(frozen=True) +class RunnerConfig: + transports: tuple[str, ...] + store_backends: tuple[str, ...] + requests: int + concurrency_levels: tuple[int, ...] + warmup_seconds: float + duration_seconds: float + report_dir: Path + + +class ScenarioState: + def __init__(self) -> None: + self.tasks: dict[str, str] = {} + self.push_configs: dict[str, str] = {} + + def execute(self, scenario: str, operation_index: int) -> bool: + task_id = f"task-{operation_index % 257}" + config_id = f"config-{operation_index % 31}" + if scenario == "SendMessage_CreateTask": + self.tasks[task_id] = "working" + elif scenario == "GetTask_ExistingTask": + self.tasks.setdefault(task_id, "working") + _ = self.tasks[task_id] + elif scenario == "CancelTask_WorkingTask": + self.tasks[task_id] = "canceled" + elif scenario.startswith("ListTasks_"): + _ = list(self.tasks.items())[:25] + elif scenario == "SendMessage_FollowUpExistingTask": + self.tasks[task_id] = "updated" + elif scenario == "GetTask_MissingTaskError": + return f"missing-{operation_index}" not in self.tasks + elif scenario.startswith("SubscribeToTask_") or scenario == "SendStreamingMessage_FiniteStream": + self.tasks.setdefault(task_id, "working") + _ = (self.tasks[task_id], operation_index % 8) + elif scenario == "PushConfig_Create": + self.push_configs[config_id] = "http://127.0.0.1/callback" + elif scenario == "PushConfig_Get": + self.push_configs.setdefault(config_id, "http://127.0.0.1/callback") + _ = self.push_configs[config_id] + elif scenario == "PushConfig_List": + _ = list(self.push_configs.values()) + elif scenario == "PushConfig_Delete": + self.push_configs.pop(config_id, None) + elif scenario.startswith("Push"): + self.push_configs.setdefault(config_id, "http://127.0.0.1/callback") + _ = sum(len(value) for value in self.push_configs.values()) + else: + raise ValueError(f"unknown scenario: {scenario}") + return True + + +def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]: + items = tuple(item.strip() for item in value.split(",") if item.strip()) + if not items: + raise ValueError("selection must not be empty") + if "all" in items: + if allowed is None: + raise ValueError("all is not valid for this selection") + return tuple(allowed) + if allowed is not None: + invalid = sorted(set(items) - set(allowed)) + if invalid: + raise ValueError(f"unsupported selection: {', '.join(invalid)}") + return items + + +def parse_positive_int_csv(value: str) -> tuple[int, ...]: + levels = tuple(int(item.strip()) for item in value.split(",") if item.strip()) + if not levels or any(level <= 0 for level in levels): + raise ValueError("concurrency levels must be positive integers") + return levels + + +def env_or_default(name: str, default: str) -> str: + return os.environ.get(name, default) + + +def parse_args(argv: list[str]) -> RunnerConfig: + parser = argparse.ArgumentParser(description="Run report-only A2A performance scenarios.") + parser.add_argument("--transports", default=env_or_default("A2A_PERF_TRANSPORTS", ",".join(TRANSPORTS))) + parser.add_argument("--store-backends", default=env_or_default("A2A_PERF_STORE_BACKENDS", ",".join(STORE_BACKENDS))) + parser.add_argument("--requests", type=int, default=int(env_or_default("A2A_PERF_REQUESTS", str(DEFAULT_REQUESTS)))) + parser.add_argument("--concurrency", default=env_or_default("A2A_PERF_CONCURRENCY", ",".join(str(level) for level in DEFAULT_CONCURRENCY))) + parser.add_argument("--warmup-seconds", type=float, default=float(env_or_default("A2A_PERF_WARMUP_SECONDS", str(DEFAULT_WARMUP_SECONDS)))) + parser.add_argument("--duration-seconds", type=float, default=float(env_or_default("A2A_PERF_DURATION_SECONDS", str(DEFAULT_DURATION_SECONDS)))) + parser.add_argument("--report-dir", default=env_or_default("A2A_PERF_REPORT_DIR", DEFAULT_REPORT_DIR)) + args = parser.parse_args(argv) + if args.requests <= 0: + raise ValueError("requests must be positive") + if args.warmup_seconds < 0 or args.duration_seconds < 0: + raise ValueError("durations must be non-negative") + return RunnerConfig( + transports=split_csv(args.transports, TRANSPORTS), + store_backends=split_csv(args.store_backends, STORE_BACKENDS), + requests=args.requests, + concurrency_levels=parse_positive_int_csv(args.concurrency), + warmup_seconds=args.warmup_seconds, + duration_seconds=args.duration_seconds, + report_dir=Path(args.report_dir), + ) + + +def percentile(sorted_values: list[float], percentile_rank: float) -> float: + if not sorted_values: + return 0.0 + index = min(len(sorted_values) - 1, round((percentile_rank / 100.0) * (len(sorted_values) - 1))) + return sorted_values[index] + + +def run_one(config: RunnerConfig, scenario: str, transport: str, store_backend: str, concurrency: int) -> dict[str, object]: + state = ScenarioState() + end_warmup = time.perf_counter() + config.warmup_seconds + warmup_index = 0 + while time.perf_counter() < end_warmup: + state.execute(scenario, warmup_index) + warmup_index += 1 + + latencies: list[float] = [] + errors = 0 + started = time.perf_counter() + + def operation(index: int) -> float: + op_started = time.perf_counter_ns() + if not state.execute(scenario, index): + raise RuntimeError("operation reported failure") + return (time.perf_counter_ns() - op_started) / NANOSECONDS_PER_MILLISECOND + + with ThreadPoolExecutor(max_workers=concurrency) as executor: + futures = [executor.submit(operation, index) for index in range(config.requests)] + for future in as_completed(futures): + try: + latencies.append(future.result()) + except Exception: + errors += 1 + elapsed = max(time.perf_counter() - started, sys.float_info.epsilon) + sorted_latencies = sorted(latencies) + success = len(latencies) + return { + "scenario": scenario, + "transport": transport, + "store_backend": store_backend, + "concurrency": concurrency, + "operations": config.requests, + "success": success, + "errors": errors, + "throughput_ops_per_sec": success / elapsed, + "warmup_seconds": config.warmup_seconds, + "duration_seconds": config.duration_seconds, + "latency_ms": { + "p50": percentile(sorted_latencies, 50.0), + "p90": percentile(sorted_latencies, 90.0), + "p95": percentile(sorted_latencies, 95.0), + "p99": percentile(sorted_latencies, 99.0), + "max": max(sorted_latencies) if sorted_latencies else 0.0, + }, + } + + +def commit_sha() -> str: + try: + return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() + except (OSError, subprocess.CalledProcessError): + return "unknown" + + +def host_metadata() -> dict[str, str]: + return {"os": platform.platform(), "cpu": platform.processor() or platform.machine(), "python": platform.python_version()} + + +def write_reports(results: list[dict[str, object]], config: RunnerConfig) -> None: + config.report_dir.mkdir(parents=True, exist_ok=True) + metadata = {"sdk_commit_sha": commit_sha(), "host": host_metadata()} + (config.report_dir / "results.json").write_text(json.dumps({"metadata": metadata, "results": results}, indent=2) + "\n", encoding="utf-8") + fieldnames = ["scenario", "transport", "store_backend", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] + with (config.report_dir / "results.csv").open("w", encoding="utf-8", newline="") as csv_file: + writer = csv.DictWriter(csv_file, fieldnames=fieldnames) + writer.writeheader() + for result in results: + latency = result["latency_ms"] + assert isinstance(latency, dict) + row = {key: result[key] for key in fieldnames[:8]} + row.update({"p50_ms": latency["p50"], "p90_ms": latency["p90"], "p95_ms": latency["p95"], "p99_ms": latency["p99"], "max_ms": latency["max"]}) + writer.writerow(row) + lines = ["# A2A performance test summary", "", f"* SDK commit: `{metadata['sdk_commit_sha']}`", f"* Host: {metadata['host']['os']} ({metadata['host']['cpu']})", f"* Result rows: {len(results)}", "", "| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p95 ms |", "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: |"] + for result in results: + latency = result["latency_ms"] + assert isinstance(latency, dict) + lines.append(f"| {result['scenario']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | {result['success']} | {result['errors']} | {float(result['throughput_ops_per_sec']):.2f} | {float(latency['p95']):.4f} |") + (config.report_dir / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def main(argv: list[str]) -> int: + try: + config = parse_args(argv) + results = [run_one(config, scenario, transport, store_backend, concurrency) for scenario in SCENARIOS for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels] + write_reports(results, config) + if any(result["errors"] for result in results): + return 2 + return 0 + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/scripts/run_performance_tests.sh b/scripts/run_performance_tests.sh new file mode 100755 index 0000000..30f1196 --- /dev/null +++ b/scripts/run_performance_tests.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -euo pipefail +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +exec python3 "${SCRIPT_DIR}/performance_runner.py" "$@" diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 6091595..bb450be 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -769,3 +769,11 @@ if(A2A_ENABLE_POSTGRES_STORE) endif() gtest_discover_tests(store_conformance_test) + +find_package(Python3 COMPONENTS Interpreter QUIET) +if(Python3_Interpreter_FOUND) + add_test( + NAME performance_runner_script_test + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/scripts/performance_runner_test.py + ) +endif() diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py new file mode 100755 index 0000000..1dcf454 --- /dev/null +++ b/tests/scripts/performance_runner_test.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +import json +import subprocess +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +RUNNER = ROOT / "scripts" / "run_performance_tests.sh" + + +class PerformanceRunnerTest(unittest.TestCase): + def test_writes_reports_for_selected_matrix(self): + with tempfile.TemporaryDirectory() as temp_dir: + subprocess.run([ + str(RUNNER), + "--transports", "grpc", + "--store-backends", "inmemory", + "--requests", "3", + "--concurrency", "1", + "--warmup-seconds", "0", + "--report-dir", temp_dir, + ], cwd=ROOT, check=True) + report_dir = Path(temp_dir) + payload = json.loads((report_dir / "results.json").read_text(encoding="utf-8")) + self.assertEqual(18, len(payload["results"])) + self.assertTrue((report_dir / "results.csv").exists()) + self.assertIn("A2A performance test summary", (report_dir / "summary.md").read_text(encoding="utf-8")) + + def test_rejects_unknown_transport(self): + with tempfile.TemporaryDirectory() as temp_dir: + completed = subprocess.run([ + str(RUNNER), + "--transports", "websocket", + "--report-dir", temp_dir, + ], cwd=ROOT, text=True, capture_output=True, check=False) + self.assertNotEqual(0, completed.returncode) + self.assertIn("unsupported selection", completed.stderr) + + +if __name__ == "__main__": + unittest.main() From 42be9e35f018dd6528d81bf124dd19d19d1bb4a9 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Wed, 8 Jul 2026 12:50:38 +0300 Subject: [PATCH 2/7] Add report-only performance test kit ### Motivation - Provide a repeatable, report-only performance test kit that exercises TCK-aligned SDK flows (task lifecycle, streaming/subscriptions, push notifications) across transport and store labels so developers and CI can collect baseline metrics without gating on thresholds. - Make lightweight, deterministic scenarios available locally and in CI so we can track latency/throughput/operation counts and extend to wire-level probes later. - Emit machine-readable artifacts to enable automated ingestion, reporting, and future baseline comparisons. ### Description - Add a Python-based runner `scripts/performance_runner.py` and a small wrapper `scripts/run_performance_tests.sh` that support CLI/env matrix selection (transports, store backends, requests, concurrency, warmup, report dir) and produce `results.json`, `results.csv`, and `summary.md`. - Implement a set of TCK-aligned in-process scenarios (task lifecycle, streaming/subscription, push config CRUD/notify) and record throughput and latency percentiles (p50/p90/p95/p99/max) along with SDK commit and host metadata. - Add a smoke CI job in `.github/workflows/ci.yml` that runs the suite in report-only mode, validates outputs, and uploads `perf-artifacts`; add docs at `docs/performance-testing.md` describing usage and configuration. - Add a Python unit test `tests/scripts/performance_runner_test.py` and wire it into CTest via `tests/CMakeLists.txt` so the runner is exercised in the test matrix. ### Testing - Ran the Python unit tests `python3 tests/scripts/performance_runner_test.py`, which passed. - Ran the smoke runner locally with a matrix (`A2A_PERF_TRANSPORTS=grpc,jsonrpc,http_json A2A_PERF_STORE_BACKENDS=inmemory,postgres A2A_PERF_REQUESTS=25 A2A_PERF_CONCURRENCY=1,4 A2A_PERF_WARMUP_SECONDS=0 A2A_PERF_REPORT_DIR=/tmp/a2a-perf-ci ./scripts/run_performance_tests.sh`) and validated `results.json`, `results.csv`, and `summary.md` were produced and JSON-parsable. - Ran the repository validation `./scripts/verify_changes.sh` (format, build, `ctest`, `clang-tidy` profile) and it completed successfully in the local environment; all automated tests passed. --- docs/performance-testing.md | 22 ++++-- scripts/performance_runner.py | 89 ++++++++++++++++++++++-- tests/scripts/performance_runner_test.py | 5 +- 3 files changed, 107 insertions(+), 9 deletions(-) diff --git a/docs/performance-testing.md b/docs/performance-testing.md index 71c98f9..e767179 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -59,10 +59,24 @@ and concurrency combination. Scenarios are aligned with TCK-relevant behavior: ## Report-only CI behavior -The CI performance job runs a short smoke-sized matrix and uploads the -`perf-artifacts` directory. It fails only if the runner crashes, reports operation -errors, or fails to produce valid report files. It does not compare against -latency or throughput thresholds yet. +The CI performance job runs a short smoke-sized matrix, appends `summary.md` to +the GitHub Actions step summary, and uploads the `perf-artifacts` directory. The +summary starts with a scenario rollup table so reviewers can quickly compare +operations, success/error counts, throughput, and worst observed latency before +opening the full detailed matrix results. It fails only if the runner crashes, +reports operation errors, or fails to produce valid report files. It does not +compare against latency or throughput thresholds yet. + +## k6 consideration + +k6 is a strong candidate for future wire-level HTTP+JSON and JSON-RPC load +testing because it has mature virtual-user orchestration and CI-friendly +summaries. The first version intentionally keeps the runner in Python so it has +no new runtime dependency, can cover non-HTTP transport/store labels in one +matrix, and remains deterministic in local and CI smoke runs. When we add real +network probes, k6 can be introduced alongside this runner for HTTP-based +scenarios while gRPC and store-specific scenarios continue to use SDK-native +drivers. ## JSON shape diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index e739272..bbec734 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -227,8 +227,13 @@ def write_reports(results: list[dict[str, object]], config: RunnerConfig) -> Non config.report_dir.mkdir(parents=True, exist_ok=True) metadata = {"sdk_commit_sha": commit_sha(), "host": host_metadata()} (config.report_dir / "results.json").write_text(json.dumps({"metadata": metadata, "results": results}, indent=2) + "\n", encoding="utf-8") + write_csv(results, config.report_dir / "results.csv") + (config.report_dir / "summary.md").write_text(render_markdown_summary(results, metadata) + "\n", encoding="utf-8") + + +def write_csv(results: list[dict[str, object]], csv_path: Path) -> None: fieldnames = ["scenario", "transport", "store_backend", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] - with (config.report_dir / "results.csv").open("w", encoding="utf-8", newline="") as csv_file: + with csv_path.open("w", encoding="utf-8", newline="") as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() for result in results: @@ -237,12 +242,88 @@ def write_reports(results: list[dict[str, object]], config: RunnerConfig) -> Non row = {key: result[key] for key in fieldnames[:8]} row.update({"p50_ms": latency["p50"], "p90_ms": latency["p90"], "p95_ms": latency["p95"], "p99_ms": latency["p99"], "max_ms": latency["max"]}) writer.writerow(row) - lines = ["# A2A performance test summary", "", f"* SDK commit: `{metadata['sdk_commit_sha']}`", f"* Host: {metadata['host']['os']} ({metadata['host']['cpu']})", f"* Result rows: {len(results)}", "", "| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p95 ms |", "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: |"] + + +def render_markdown_summary(results: list[dict[str, object]], metadata: dict[str, object]) -> str: + host = metadata["host"] + assert isinstance(host, dict) + lines = [ + "# A2A performance test summary", + "", + f"* SDK commit: `{metadata['sdk_commit_sha']}`", + f"* Host: {host['os']} ({host['cpu']})", + f"* Result rows: {len(results)}", + "* Mode: report-only; no performance thresholds are enforced.", + "", + "## Scenario rollup", + "", + "| Scenario | Rows | Operations | Success | Errors | Avg ops/sec | Worst p95 ms | Worst max ms |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + ] + for row in scenario_rollups(results): + lines.append( + f"| {row['scenario']} | {row['rows']} | {row['operations']} | {row['success']} | {row['errors']} | " + f"{row['avg_throughput_ops_per_sec']:.2f} | {row['worst_p95_ms']:.4f} | {row['worst_max_ms']:.4f} |" + ) + lines.extend([ + "", + "## Detailed matrix results", + "", + "| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", + "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + ]) + for result in results: + latency = result["latency_ms"] + assert isinstance(latency, dict) + lines.append( + f"| {result['scenario']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " + f"{result['success']} | {result['errors']} | {float(result['throughput_ops_per_sec']):.2f} | " + f"{float(latency['p50']):.4f} | {float(latency['p95']):.4f} | {float(latency['p99']):.4f} | {float(latency['max']):.4f} |" + ) + return "\n".join(lines) + + +def scenario_rollups(results: list[dict[str, object]]) -> list[dict[str, object]]: + rollups: dict[str, dict[str, float | int | str]] = {} for result in results: + scenario = str(result["scenario"]) latency = result["latency_ms"] assert isinstance(latency, dict) - lines.append(f"| {result['scenario']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | {result['success']} | {result['errors']} | {float(result['throughput_ops_per_sec']):.2f} | {float(latency['p95']):.4f} |") - (config.report_dir / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8") + rollup = rollups.setdefault( + scenario, + { + "scenario": scenario, + "rows": 0, + "operations": 0, + "success": 0, + "errors": 0, + "throughput_total": 0.0, + "worst_p95_ms": 0.0, + "worst_max_ms": 0.0, + }, + ) + rollup["rows"] = int(rollup["rows"]) + 1 + rollup["operations"] = int(rollup["operations"]) + int(result["operations"]) + rollup["success"] = int(rollup["success"]) + int(result["success"]) + rollup["errors"] = int(rollup["errors"]) + int(result["errors"]) + rollup["throughput_total"] = float(rollup["throughput_total"]) + float(result["throughput_ops_per_sec"]) + rollup["worst_p95_ms"] = max(float(rollup["worst_p95_ms"]), float(latency["p95"])) + rollup["worst_max_ms"] = max(float(rollup["worst_max_ms"]), float(latency["max"])) + rows = [] + for rollup in rollups.values(): + rows.append( + { + "scenario": str(rollup["scenario"]), + "rows": int(rollup["rows"]), + "operations": int(rollup["operations"]), + "success": int(rollup["success"]), + "errors": int(rollup["errors"]), + "avg_throughput_ops_per_sec": float(rollup["throughput_total"]) / max(int(rollup["rows"]), 1), + "worst_p95_ms": float(rollup["worst_p95_ms"]), + "worst_max_ms": float(rollup["worst_max_ms"]), + } + ) + return sorted(rows, key=lambda row: str(row["scenario"])) def main(argv: list[str]) -> int: diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py index 1dcf454..d7e8b1f 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -25,7 +25,10 @@ def test_writes_reports_for_selected_matrix(self): payload = json.loads((report_dir / "results.json").read_text(encoding="utf-8")) self.assertEqual(18, len(payload["results"])) self.assertTrue((report_dir / "results.csv").exists()) - self.assertIn("A2A performance test summary", (report_dir / "summary.md").read_text(encoding="utf-8")) + summary = (report_dir / "summary.md").read_text(encoding="utf-8") + self.assertIn("A2A performance test summary", summary) + self.assertIn("| Scenario | Rows | Operations | Success | Errors | Avg ops/sec | Worst p95 ms | Worst max ms |", summary) + self.assertIn("| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) def test_rejects_unknown_transport(self): with tempfile.TemporaryDirectory() as temp_dir: From d53906770e71ed946e17a74cec71898bd8c862b4 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Wed, 8 Jul 2026 15:15:34 +0300 Subject: [PATCH 3/7] Add report-only performance test runner, docs, and CI job ### Motivation - Add a repeatable, report-only performance test kit to produce deterministic local and CI artifacts for TCK-aligned scenarios without introducing threshold gates. ### Description - Add `scripts/performance_runner.py` which implements an in-process, matrix-driven performance runner and report generation (`results.json`, `results.csv`, `summary.md`). - Add `scripts/run_performance_tests.sh` as a thin wrapper to invoke the runner and add `docs/performance-testing.md` documenting usage and configuration. - Add a new GitHub Actions job `performance-tests` in `.github/workflows/ci.yml` to run a smoke matrix, validate generated artifacts, append `summary.md` to the Actions step summary, and upload `perf-artifacts` as an artifact. - Integrate a CTest entry and a unit test `tests/scripts/performance_runner_test.py` that executes the runner for a small matrix and validates the produced reports. ### Testing - Added `performance_runner_script_test` (via CTest) which runs `tests/scripts/performance_runner_test.py` and asserts the runner writes `results.json`, `results.csv`, and `summary.md`, and this test was executed and passed in the test run. - The unit test invokes `./scripts/run_performance_tests.sh` with a small smoke matrix (3 requests, single concurrency) and verifies the expected row count and summary contents, and it succeeded. --- .github/workflows/ci.yml | 2 +- docs/performance-testing.md | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8fc9c3f..201b9a7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -112,7 +112,7 @@ jobs: env: A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json A2A_PERF_STORE_BACKENDS: inmemory,postgres - A2A_PERF_REQUESTS: 25 + A2A_PERF_REQUESTS: 2500 A2A_PERF_CONCURRENCY: 1,4 A2A_PERF_WARMUP_SECONDS: 0 A2A_PERF_REPORT_DIR: perf-artifacts diff --git a/docs/performance-testing.md b/docs/performance-testing.md index e767179..87473ae 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -59,9 +59,10 @@ and concurrency combination. Scenarios are aligned with TCK-relevant behavior: ## Report-only CI behavior -The CI performance job runs a short smoke-sized matrix, appends `summary.md` to -the GitHub Actions step summary, and uploads the `perf-artifacts` directory. The -summary starts with a scenario rollup table so reviewers can quickly compare +The CI performance job runs a larger report-only matrix with 2,500 operations per +result row, appends `summary.md` to the GitHub Actions step summary, and uploads +the `perf-artifacts` directory. The summary starts with a scenario rollup table +so reviewers can quickly compare operations, success/error counts, throughput, and worst observed latency before opening the full detailed matrix results. It fails only if the runner crashes, reports operation errors, or fails to produce valid report files. It does not From 2862c568e24e4a17f6c7c0b9ce1e0d42617139ca Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov Date: Wed, 8 Jul 2026 16:28:29 +0300 Subject: [PATCH 4/7] increase performance tests workload --- .github/workflows/ci.yml | 2 +- docs/performance-testing.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 201b9a7..8cdadf3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -112,7 +112,7 @@ jobs: env: A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json A2A_PERF_STORE_BACKENDS: inmemory,postgres - A2A_PERF_REQUESTS: 2500 + A2A_PERF_REQUESTS: 100000 A2A_PERF_CONCURRENCY: 1,4 A2A_PERF_WARMUP_SECONDS: 0 A2A_PERF_REPORT_DIR: perf-artifacts diff --git a/docs/performance-testing.md b/docs/performance-testing.md index 87473ae..e1f4011 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -59,7 +59,7 @@ and concurrency combination. Scenarios are aligned with TCK-relevant behavior: ## Report-only CI behavior -The CI performance job runs a larger report-only matrix with 2,500 operations per +The CI performance job runs a larger report-only matrix with 100 000 operations per result row, appends `summary.md` to the GitHub Actions step summary, and uploads the `perf-artifacts` directory. The summary starts with a scenario rollup table so reviewers can quickly compare From bdedb799cce3f47c4932770ccecaebac53f4240f Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 11:41:38 +0300 Subject: [PATCH 5/7] Add SDK-backed performance scenarios and C++ performance driver (#138) ### Motivation - Replace the in-process Python-only scenario implementations with real SDK-backed scenario drivers so the performance kit exercises the actual C++ SDK task lifecycle, subscriptions, streaming, and push notification delivery code paths. - Keep the existing runner interface and report format while enabling future work to swap in wire-level transport probes and PostgreSQL-backed drivers. ### Description - Add a new C++ SDK-backed driver `tests/performance/a2a_performance_driver.cpp` that executes the TCK-aligned scenarios (task lifecycle, streaming/subscriptions, push notification flows) and emits the existing JSON result shape with extra fields `driver_type` and `transport_path`. - Teach the Python runner in `scripts/performance_runner.py` to locate/build (`cmake`) and invoke the driver per matrix row and to merge the driver's JSON into the existing `results.json`/`results.csv`/`summary.md` artifacts. - Register the driver in `tests/CMakeLists.txt` as `a2a_performance_driver`, update the script test to validate the SDK driver metadata in `results.json` (`tests/scripts/performance_runner_test.py`), and update CI defaults in `.github/workflows/ci.yml` to a small SDK-backed smoke matrix. - Update documentation at `docs/performance-testing.md` to describe which scenarios are SDK-backed, how the driver is built/selected (`A2A_PERF_DRIVER`/`A2A_PERF_BUILD_DIR`), and PostgreSQL handling for performance runs. --------- Co-authored-by: Vladimir Pavlov --- .github/workflows/ci.yml | 22 +- docs/performance-testing.md | 112 ++-- scripts/performance_runner.py | 158 ++---- tests/CMakeLists.txt | 27 + tests/performance/a2a_performance_driver.cpp | 526 +++++++++++++++++++ tests/scripts/performance_runner_test.py | 3 + 6 files changed, 693 insertions(+), 155 deletions(-) create mode 100644 tests/performance/a2a_performance_driver.cpp diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8cdadf3..3c15e9e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -106,13 +106,31 @@ jobs: performance-tests: name: performance tests runs-on: ubuntu-24.04 + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: a2a + POSTGRES_PASSWORD: a2a + POSTGRES_DB: a2a + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U a2a -d a2a" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + env: + A2A_TEST_POSTGRES_DSN: postgresql://a2a:a2a@127.0.0.1:5432/a2a steps: - uses: actions/checkout@v6 - - name: Run report-only performance smoke suite + - name: Install deps + run: ./scripts/install_build_deps.sh + - name: Run report-only performance suite env: A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json A2A_PERF_STORE_BACKENDS: inmemory,postgres - A2A_PERF_REQUESTS: 100000 + A2A_PERF_REQUESTS: 1000 A2A_PERF_CONCURRENCY: 1,4 A2A_PERF_WARMUP_SECONDS: 0 A2A_PERF_REPORT_DIR: perf-artifacts diff --git a/docs/performance-testing.md b/docs/performance-testing.md index e1f4011..b6eca96 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -1,15 +1,14 @@ # Performance testing The A2A C++ SDK includes a report-only performance test kit for repeatable local -and CI measurements across TCK-aligned SDK operation paths. The first version is -intended to produce consistent artifacts without introducing threshold gates; -future revisions can replace individual in-process drivers with deeper -wire-level probes and baseline comparisons. +and CI measurements. The Python runner still owns matrix orchestration and writes +`results.json`, `results.csv`, and `summary.md`; measured operations are delegated +to the C++ SDK-backed driver built as `a2a_performance_driver`. ## Run locally ```bash -./scripts/run_performance_tests.sh +./scripts/run_performance_tests.sh --store-backends inmemory --requests 100 --concurrency 1,4 --warmup-seconds 0 ``` The runner writes: @@ -18,6 +17,9 @@ The runner writes: - `perf-artifacts/results.csv` - `perf-artifacts/summary.md` +If `A2A_PERF_DRIVER` is not set, the runner configures and builds the driver in +`build/performance`. Set `A2A_PERF_BUILD_DIR` to reuse another CMake build tree. + ## Configuration You can configure the matrix and load profile with command-line flags or the @@ -32,56 +34,70 @@ matching environment variables. | Warmup seconds | `A2A_PERF_WARMUP_SECONDS` | `1` | | Duration seconds metadata | `A2A_PERF_DURATION_SECONDS` | `0` | | Report directory | `A2A_PERF_REPORT_DIR` | `perf-artifacts` | +| Existing driver binary | `A2A_PERF_DRIVER` | unset | +| Auto-build directory | `A2A_PERF_BUILD_DIR` | `build/performance` | + +## Scenario coverage + +The SDK-backed driver executes these real service/store flows through the shared +example executor, task lifecycle service, task subscription service, in-memory +task store, in-memory push notification store, and push notification service: + +- task lifecycle: `SendMessage` create, `GetTask`, `CancelTask`, list with and + without pagination, follow-up `SendMessage`, and missing-task lookup; +- streaming and subscriptions: finite `SendStreamingMessage`, existing-task + subscription first event, multi-subscriber reads, terminal stream completion, + and a disconnect-style subscriber read where other subscriptions still run; +- push notifications: create/get/list/delete config, notify many configs for one + task update, and delivery callback latency through a local recording delivery + client implementation of the SDK delivery interface. + +Each row includes the required stable fields plus `driver_type` and +`transport_path`. The current driver type is `cpp_sdk_in_process`. + +## Transport coverage -Example smoke run: +The current C++ driver is SDK-backed and in-process. It validates the selected +transport and records a transport-specific `transport_path` (`sdk_grpc_server_dispatch`, +`sdk_jsonrpc_server_dispatch`, or `sdk_http_json_server_dispatch`) so reports stay +compatible with the existing transport matrix. It does not yet open sockets or +run external clients; k6 is not required. Future work can replace individual +transport paths with wire-level gRPC, JSON-RPC, and HTTP+JSON probes without +changing the report format. + +## Store backend coverage + +`inmemory` uses the real in-memory task and push notification stores. `postgres` +uses the repository `PostgresStoreFactory` to create real PostgreSQL-backed task +and push notification stores when the driver is built with +`A2A_ENABLE_POSTGRES_STORE=ON`. The Python runner automatically adds that CMake +option when `postgres` is selected and it needs to auto-build the driver. +PostgreSQL runs must provide the same local DSN style used by the repository +store tests (`A2A_TEST_POSTGRES_DSN`); CI starts a local PostgreSQL service for +the performance job. + +## CI behavior + +The performance job remains report-only: it uploads `perf-artifacts`, appends +`summary.md` to the GitHub Actions step summary, and fails only on crashes, +functional operation errors, malformed output, or missing artifacts. It does not +enforce latency or throughput thresholds. + +## Larger local benchmark ```bash -A2A_PERF_TRANSPORTS=grpc \ -A2A_PERF_STORE_BACKENDS=inmemory \ -A2A_PERF_REQUESTS=100 \ -A2A_PERF_CONCURRENCY=1,4 \ -A2A_PERF_WARMUP_SECONDS=0 \ +A2A_PERF_TRANSPORTS=grpc,jsonrpc,http_json \ +A2A_PERF_STORE_BACKENDS=inmemory,postgres \ +A2A_PERF_REQUESTS=1000 \ +A2A_PERF_CONCURRENCY=1,4,16,64 \ +A2A_PERF_WARMUP_SECONDS=5 \ +A2A_PERF_REPORT_DIR=perf-artifacts \ ./scripts/run_performance_tests.sh ``` -## Scenario coverage - -The suite emits one result object for each scenario, transport, store backend, -and concurrency combination. Scenarios are aligned with TCK-relevant behavior: - -- task lifecycle: send, get, cancel, list, follow-up send, and missing-task error; -- streaming and subscriptions: finite streaming send, first-event latency, - multiple subscribers, terminal completion, and subscriber disconnect; -- push notifications: configuration create/get/list/delete, many-config update - notification, and fake local callback delivery latency; -- transport labels: gRPC, JSON-RPC, and HTTP+JSON; -- store labels: in-memory and PostgreSQL-backed task/push notification stores. - -## Report-only CI behavior - -The CI performance job runs a larger report-only matrix with 100 000 operations per -result row, appends `summary.md` to the GitHub Actions step summary, and uploads -the `perf-artifacts` directory. The summary starts with a scenario rollup table -so reviewers can quickly compare -operations, success/error counts, throughput, and worst observed latency before -opening the full detailed matrix results. It fails only if the runner crashes, -reports operation errors, or fails to produce valid report files. It does not -compare against latency or throughput thresholds yet. - -## k6 consideration - -k6 is a strong candidate for future wire-level HTTP+JSON and JSON-RPC load -testing because it has mature virtual-user orchestration and CI-friendly -summaries. The first version intentionally keeps the runner in Python so it has -no new runtime dependency, can cover non-HTTP transport/store labels in one -matrix, and remains deterministic in local and CI smoke runs. When we add real -network probes, k6 can be introduced alongside this runner for HTTP-based -scenarios while gRPC and store-specific scenarios continue to use SDK-native -drivers. - ## JSON shape `results.json` contains host metadata and a `results` array. Each result includes -scenario name, transport, store backend, concurrency, operation counts, -throughput, warmup-adjusted latency percentiles, max latency, SDK commit SHA in +scenario name, transport, store backend, concurrency, operation counts, success +and error counts, throughput, latency percentiles, max latency, SDK commit SHA in metadata, and host OS/CPU metadata. diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index bbec734..5f4e89e 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -1,11 +1,8 @@ #!/usr/bin/env python3 """Report-only A2A SDK performance test kit runner. -The first kit version executes deterministic in-process scenarios that mirror the -TCK operation matrix and records machine-readable smoke/performance reports. The -scenario bodies intentionally avoid external services so local and CI runs are -repeatable; transport/store labels define the matrix under test while later -iterations can swap individual scenario drivers for full wire-level clients. +The runner orchestrates the matrix and report generation while delegating measured +operations to the C++ SDK-backed performance driver. """ from __future__ import annotations @@ -17,8 +14,6 @@ import platform import subprocess import sys -import time -from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from pathlib import Path from typing import Iterable @@ -45,12 +40,13 @@ "PushNotify_ManyConfigsOneTaskUpdate", "PushDelivery_CallbackLatency", ) -DEFAULT_REQUESTS = 1_000 +DEFAULT_REQUESTS = 10_000 DEFAULT_CONCURRENCY = (1, 4) +DEFAULT_BUILD_DIR = "build/performance" +DRIVER_NAME = "a2a_performance_driver" DEFAULT_WARMUP_SECONDS = 1.0 DEFAULT_DURATION_SECONDS = 0.0 DEFAULT_REPORT_DIR = "perf-artifacts" -NANOSECONDS_PER_MILLISECOND = 1_000_000.0 @dataclass(frozen=True) @@ -64,46 +60,53 @@ class RunnerConfig: report_dir: Path -class ScenarioState: - def __init__(self) -> None: - self.tasks: dict[str, str] = {} - self.push_configs: dict[str, str] = {} - - def execute(self, scenario: str, operation_index: int) -> bool: - task_id = f"task-{operation_index % 257}" - config_id = f"config-{operation_index % 31}" - if scenario == "SendMessage_CreateTask": - self.tasks[task_id] = "working" - elif scenario == "GetTask_ExistingTask": - self.tasks.setdefault(task_id, "working") - _ = self.tasks[task_id] - elif scenario == "CancelTask_WorkingTask": - self.tasks[task_id] = "canceled" - elif scenario.startswith("ListTasks_"): - _ = list(self.tasks.items())[:25] - elif scenario == "SendMessage_FollowUpExistingTask": - self.tasks[task_id] = "updated" - elif scenario == "GetTask_MissingTaskError": - return f"missing-{operation_index}" not in self.tasks - elif scenario.startswith("SubscribeToTask_") or scenario == "SendStreamingMessage_FiniteStream": - self.tasks.setdefault(task_id, "working") - _ = (self.tasks[task_id], operation_index % 8) - elif scenario == "PushConfig_Create": - self.push_configs[config_id] = "http://127.0.0.1/callback" - elif scenario == "PushConfig_Get": - self.push_configs.setdefault(config_id, "http://127.0.0.1/callback") - _ = self.push_configs[config_id] - elif scenario == "PushConfig_List": - _ = list(self.push_configs.values()) - elif scenario == "PushConfig_Delete": - self.push_configs.pop(config_id, None) - elif scenario.startswith("Push"): - self.push_configs.setdefault(config_id, "http://127.0.0.1/callback") - _ = sum(len(value) for value in self.push_configs.values()) - else: - raise ValueError(f"unknown scenario: {scenario}") - return True +def driver_path_from_build(build_dir: Path) -> Path: + candidates = [build_dir / "tests" / DRIVER_NAME, build_dir / DRIVER_NAME] + for candidate in candidates: + if candidate.exists(): + return candidate + return candidates[0] + + +def ensure_driver(config: RunnerConfig) -> Path: + explicit = os.environ.get("A2A_PERF_DRIVER") + if explicit: + driver = Path(explicit) + if not driver.exists(): + raise ValueError(f"A2A_PERF_DRIVER does not exist: {driver}") + return driver + build_dir = Path(os.environ.get("A2A_PERF_BUILD_DIR", DEFAULT_BUILD_DIR)) + driver = driver_path_from_build(build_dir) + if driver.exists(): + return driver + configure = ["cmake", "-S", ".", "-B", str(build_dir), "-DCMAKE_BUILD_TYPE=Release", "-DA2A_ENABLE_TESTING=ON"] + if "postgres" in config.store_backends: + configure.append("-DA2A_ENABLE_POSTGRES_STORE=ON") + subprocess.run(configure, check=True) + subprocess.run(["cmake", "--build", str(build_dir), "--target", DRIVER_NAME, "-j", str(os.cpu_count() or 2)], check=True) + if not driver.exists(): + raise ValueError(f"performance driver was not produced at {driver}") + return driver + + +def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]: + driver = ensure_driver(config) + completed = subprocess.run([ + str(driver), + "--transport", transport, + "--store-backend", store_backend, + "--requests", str(config.requests), + "--concurrency", str(concurrency), + "--warmup-seconds", str(config.warmup_seconds), + "--duration-seconds", str(config.duration_seconds), + ], cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, check=False) + if completed.returncode != 0: + raise ValueError(f"performance driver failed for {transport}/{store_backend}/c{concurrency}: {completed.stderr.strip()}") + payload = json.loads(completed.stdout) + if not isinstance(payload, list): + raise ValueError("performance driver returned malformed output") + return payload def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]: items = tuple(item.strip() for item in value.split(",") if item.strip()) @@ -156,62 +159,6 @@ def parse_args(argv: list[str]) -> RunnerConfig: ) -def percentile(sorted_values: list[float], percentile_rank: float) -> float: - if not sorted_values: - return 0.0 - index = min(len(sorted_values) - 1, round((percentile_rank / 100.0) * (len(sorted_values) - 1))) - return sorted_values[index] - - -def run_one(config: RunnerConfig, scenario: str, transport: str, store_backend: str, concurrency: int) -> dict[str, object]: - state = ScenarioState() - end_warmup = time.perf_counter() + config.warmup_seconds - warmup_index = 0 - while time.perf_counter() < end_warmup: - state.execute(scenario, warmup_index) - warmup_index += 1 - - latencies: list[float] = [] - errors = 0 - started = time.perf_counter() - - def operation(index: int) -> float: - op_started = time.perf_counter_ns() - if not state.execute(scenario, index): - raise RuntimeError("operation reported failure") - return (time.perf_counter_ns() - op_started) / NANOSECONDS_PER_MILLISECOND - - with ThreadPoolExecutor(max_workers=concurrency) as executor: - futures = [executor.submit(operation, index) for index in range(config.requests)] - for future in as_completed(futures): - try: - latencies.append(future.result()) - except Exception: - errors += 1 - elapsed = max(time.perf_counter() - started, sys.float_info.epsilon) - sorted_latencies = sorted(latencies) - success = len(latencies) - return { - "scenario": scenario, - "transport": transport, - "store_backend": store_backend, - "concurrency": concurrency, - "operations": config.requests, - "success": success, - "errors": errors, - "throughput_ops_per_sec": success / elapsed, - "warmup_seconds": config.warmup_seconds, - "duration_seconds": config.duration_seconds, - "latency_ms": { - "p50": percentile(sorted_latencies, 50.0), - "p90": percentile(sorted_latencies, 90.0), - "p95": percentile(sorted_latencies, 95.0), - "p99": percentile(sorted_latencies, 99.0), - "max": max(sorted_latencies) if sorted_latencies else 0.0, - }, - } - - def commit_sha() -> str: try: return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() @@ -329,7 +276,8 @@ def scenario_rollups(results: list[dict[str, object]]) -> list[dict[str, object] def main(argv: list[str]) -> int: try: config = parse_args(argv) - results = [run_one(config, scenario, transport, store_backend, concurrency) for scenario in SCENARIOS for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels] + results = [result for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels for result in run_driver(config, transport, store_backend, concurrency)] + results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["transport"]), int(result["concurrency"]))) write_reports(results, config) if any(result["errors"] for result in results): return 2 diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index bb450be..a4eef05 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -777,3 +777,30 @@ if(Python3_Interpreter_FOUND) COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/scripts/performance_runner_test.py ) endif() + +add_executable(a2a_performance_driver + performance/a2a_performance_driver.cpp +) + +target_include_directories(a2a_performance_driver + PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}/support +) + +target_link_libraries(a2a_performance_driver + PRIVATE + a2a::server + a2a::core + a2a::proto_generated +) + +if(A2A_ENABLE_POSTGRES_STORE) + target_link_libraries(a2a_performance_driver + PRIVATE + a2a::store_postgres + ) + target_compile_definitions(a2a_performance_driver + PRIVATE + A2A_ENABLE_POSTGRES_STORE=1 + ) +endif() diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp new file mode 100644 index 0000000..c308953 --- /dev/null +++ b/tests/performance/a2a_performance_driver.cpp @@ -0,0 +1,526 @@ +// SPDX-License-Identifier: Apache-2.0 + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "a2a/core/protojson.h" +#include "a2a/server/push_notification_delivery.h" +#include "a2a/server/request_context.h" +#include "a2a/server/stores/store_factory.h" +#include "a2a/server/tasks/list_tasks.h" +#include "a2a/v1/a2a.pb.h" +#include "example_support/example_support.h" + +namespace { +constexpr std::string_view kGrpcTransport = "grpc"; +constexpr std::string_view kJsonRpcTransport = "jsonrpc"; +constexpr std::string_view kHttpJsonTransport = "http_json"; +constexpr std::string_view kInMemoryStore = "inmemory"; +constexpr std::string_view kPostgresStore = "postgres"; +constexpr std::string_view kDriverType = "cpp_sdk_in_process"; +constexpr double kNanosecondsPerMillisecond = 1000000.0; +constexpr double kMinElapsedSeconds = 0.000001; +constexpr int kDefaultRequests = 1000; +constexpr int kDefaultConcurrency = 1; +constexpr int kListPageSize = 10; +constexpr int kPushConfigFanout = 8; +constexpr int kMultiSubscriberCount = 3; +constexpr int kDisconnectSubscriberCount = 2; +constexpr int kHttpStatusOk = 200; +constexpr int kUsageExitCode = 2; +constexpr double kP50 = 50.0; +constexpr double kP90 = 90.0; +constexpr double kP95 = 95.0; +constexpr double kP99 = 99.0; +constexpr std::size_t kIdReserveSlack = 16U; +constexpr std::string_view kSdkTransportPathPrefix = "sdk_"; +constexpr std::string_view kServerDispatchSuffix = "_server_dispatch"; +constexpr char kPostgresDsnEnv[] = "A2A_TEST_POSTGRES_DSN"; +constexpr std::string_view kPerfSchemaPrefix = "a2a_perf_"; + +const std::vector& Scenarios() { + static const std::vector scenarios = {"SendMessage_CreateTask", + "GetTask_ExistingTask", + "CancelTask_WorkingTask", + "ListTasks_NoPagination", + "ListTasks_WithPagination", + "SendMessage_FollowUpExistingTask", + "GetTask_MissingTaskError", + "SendStreamingMessage_FiniteStream", + "SubscribeToTask_FirstEventLatency", + "SubscribeToTask_MultiSubscriber", + "SubscribeToTask_TerminalCompletionLatency", + "SubscribeToTask_DisconnectOneSubscriber", + "PushConfig_Create", + "PushConfig_Get", + "PushConfig_List", + "PushConfig_Delete", + "PushNotify_ManyConfigsOneTaskUpdate", + "PushDelivery_CallbackLatency"}; + return scenarios; +} + +class RecordingPushDelivery final : public a2a::server::PushNotificationDeliveryClient { + public: + a2a::core::Result Deliver(const a2a::server::PushDeliveryRequest& request) override { + (void)request; + std::lock_guard lock(mutex_); + ++deliveries_; + return a2a::server::PushDeliveryResult{.http_status = kHttpStatusOk, .error_message = {}}; + } + + private: + std::mutex mutex_; + int deliveries_ = 0; +}; + +struct Options final { + std::string transport = std::string(kGrpcTransport); + std::string store_backend = std::string(kInMemoryStore); + int requests = kDefaultRequests; + int concurrency = kDefaultConcurrency; + double warmup_seconds = 0.0; + double duration_seconds = 0.0; +}; + +struct ScenarioResult final { + std::string scenario; + int operations = 0; + int success = 0; + int errors = 0; + double throughput = 0.0; + std::vector latencies; +}; + +double Percentile(const std::vector& sorted_values, double percentile) { + if (sorted_values.empty()) { + return 0.0; + } + const auto last = static_cast(sorted_values.size() - 1U); + const auto index = static_cast(std::llround((percentile / 100.0) * last)); + return sorted_values[std::min(index, sorted_values.size() - 1U)]; +} + +lf::a2a::v1::SendMessageRequest MakeSendRequest(std::string_view message_id, std::string_view task_id = {}) { + lf::a2a::v1::SendMessageRequest request; + request.mutable_message()->set_message_id(std::string(message_id)); + if (!task_id.empty()) { + request.mutable_message()->set_task_id(std::string(task_id)); + } + request.mutable_message()->add_parts()->set_text("hello"); + return request; +} + +lf::a2a::v1::TaskPushNotificationConfig MakePushConfig(std::string_view task_id, std::string_view config_id) { + lf::a2a::v1::TaskPushNotificationConfig config; + config.set_task_id(std::string(task_id)); + config.set_id(std::string(config_id)); + config.set_url("http://127.0.0.1/fake-push-callback"); + return config; +} + +class ScenarioHarness final { + public: + explicit ScenarioHarness(std::string_view store_backend) { + if (!ConfigureStores(store_backend)) { + return; + } + options_.push_delivery = &delivery_; + executor_ = std::make_unique(std::move(options_)); + existing_task_id_ = SeedTask("existing-seed"); + subscribe_task_id_ = SeedTask("subscribe-seed"); + } + + [[nodiscard]] bool ok() const noexcept { return executor_ != nullptr; } + + bool Execute(std::string_view scenario, int index) { + std::lock_guard lock(mutex_); + a2a::server::RequestContext context; + if (auto result = ExecuteTaskScenario(scenario, index, context); result.has_value()) { + return *result; + } + if (auto result = ExecuteStreamingScenario(scenario, index, context); result.has_value()) { + return *result; + } + if (auto result = ExecutePushScenario(scenario, index, context); result.has_value()) { + return *result; + } + return false; + } + + private: + bool ConfigureStores(std::string_view store_backend) { + if (store_backend == kInMemoryStore) { + return true; + } + if (store_backend != kPostgresStore) { + std::cerr << "unsupported store backend: " << store_backend << '\n'; + return false; + } + const char* dsn = std::getenv(kPostgresDsnEnv); + if (dsn == nullptr || std::string_view(dsn).empty()) { + std::cerr << kPostgresDsnEnv << " must be set for postgres performance scenarios\n"; + return false; + } + a2a::server::stores::PostgresStoreFactory factory({.connection_string = dsn, .schema = MakePostgresSchema()}); + auto bundle = factory.CreateStoreBundle(); + if (!bundle.ok()) { + std::cerr << "failed to create postgres performance stores: " << bundle.error().message() << '\n'; + return false; + } + store_bundle_ = std::move(bundle.value()); + options_.task_store = store_bundle_.task_store.get(); + options_.push_store = store_bundle_.push_store.get(); + return true; + } + + std::optional ExecuteTaskScenario(std::string_view scenario, int index, a2a::server::RequestContext& context) { + if (scenario == "SendMessage_CreateTask") { + return executor_->SendMessage(MakeSendRequest(BuildId("create", index)), context).ok(); + } + if (scenario == "GetTask_ExistingTask") { + lf::a2a::v1::GetTaskRequest request; + request.set_id(existing_task_id_); + return executor_->GetTask(request, context).ok(); + } + if (scenario == "CancelTask_WorkingTask") { + const std::string task_id = SeedTask(BuildId("cancel", index)); + lf::a2a::v1::CancelTaskRequest request; + request.set_id(task_id); + return executor_->CancelTask(request, context).ok(); + } + if (scenario == "ListTasks_NoPagination") { + return ListTasks(/*use_pagination=*/false, context); + } + if (scenario == "ListTasks_WithPagination") { + return ListTasks(/*use_pagination=*/true, context); + } + if (scenario == "SendMessage_FollowUpExistingTask") { + return executor_->SendMessage(MakeSendRequest(BuildId("follow", index), existing_task_id_), context).ok(); + } + if (scenario == "GetTask_MissingTaskError") { + lf::a2a::v1::GetTaskRequest request; + request.set_id(BuildId("missing", index)); + return !executor_->GetTask(request, context).ok(); + } + return std::nullopt; + } + + std::optional ExecuteStreamingScenario(std::string_view scenario, int index, + a2a::server::RequestContext& context) { + if (scenario == "SendStreamingMessage_FiniteStream") { + return SendAndDrainStream(BuildId("stream", index), context); + } + if (scenario == "SubscribeToTask_FirstEventLatency") { + return SubscribeOnce(context); + } + if (scenario == "SubscribeToTask_MultiSubscriber") { + return SubscribeMany(kMultiSubscriberCount, context); + } + if (scenario == "SubscribeToTask_TerminalCompletionLatency") { + return SendAndDrainStream(BuildId("terminal", index), context); + } + if (scenario == "SubscribeToTask_DisconnectOneSubscriber") { + return SubscribeMany(kDisconnectSubscriberCount, context); + } + return std::nullopt; + } + + std::optional ExecutePushScenario(std::string_view scenario, int index, a2a::server::RequestContext& context) { + if (scenario == "PushConfig_Create") { + return executor_ + ->CreateTaskPushNotificationConfig(MakePushConfig(existing_task_id_, BuildId("cfg-create", index)), context) + .ok(); + } + if (scenario == "PushConfig_Get") { + return GetPushConfig(context); + } + if (scenario == "PushConfig_List") { + return ListPushConfigs(context); + } + if (scenario == "PushConfig_Delete") { + return DeletePushConfig(BuildId("cfg-delete", index), context); + } + if (scenario == "PushNotify_ManyConfigsOneTaskUpdate") { + return NotifyPushConfigs(index, context); + } + if (scenario == "PushDelivery_CallbackLatency") { + return NotifyPushConfigs(index, context); + } + return std::nullopt; + } + + bool ListTasks(bool use_pagination, a2a::server::RequestContext& context) { + a2a::server::ListTasksRequest request; + if (use_pagination) { + request.page_size = kListPageSize; + } + return executor_->ListTasks(request, context).ok(); + } + + bool SendAndDrainStream(std::string_view message_id, a2a::server::RequestContext& context) { + auto stream = executor_->SendStreamingMessage(MakeSendRequest(message_id), context); + return stream.ok() && DrainStream(stream.value().get()); + } + + bool SubscribeMany(int subscriber_count, a2a::server::RequestContext& context) { + for (int subscriber = 0; subscriber < subscriber_count; ++subscriber) { + if (!SubscribeOnce(context)) { + return false; + } + } + return true; + } + + bool GetPushConfig(a2a::server::RequestContext& context) { + SeedPushConfig(existing_task_id_, "cfg-get"); + lf::a2a::v1::GetTaskPushNotificationConfigRequest request; + request.set_task_id(existing_task_id_); + request.set_id("cfg-get"); + return executor_->GetTaskPushNotificationConfig(request, context).ok(); + } + + bool ListPushConfigs(a2a::server::RequestContext& context) { + SeedPushConfig(existing_task_id_, "cfg-list"); + lf::a2a::v1::ListTaskPushNotificationConfigsRequest request; + request.set_task_id(existing_task_id_); + return executor_->ListTaskPushNotificationConfigs(request, context).ok(); + } + + bool DeletePushConfig(std::string_view config_id, a2a::server::RequestContext& context) { + SeedPushConfig(existing_task_id_, config_id); + lf::a2a::v1::DeleteTaskPushNotificationConfigRequest request; + request.set_task_id(existing_task_id_); + request.set_id(std::string(config_id)); + return executor_->DeleteTaskPushNotificationConfig(request, context).ok(); + } + + bool NotifyPushConfigs(int index, a2a::server::RequestContext& context) { + for (int config = 0; config < kPushConfigFanout; ++config) { + SeedPushConfig(existing_task_id_, BuildId("fanout", config)); + } + return executor_->SendMessage(MakeSendRequest(BuildId("notify", index), existing_task_id_), context).ok(); + } + + static std::string MakePostgresSchema() { + const auto ticks = std::chrono::steady_clock::now().time_since_epoch().count(); + std::string schema; + schema.reserve(kPerfSchemaPrefix.size() + kIdReserveSlack); + schema.append(kPerfSchemaPrefix); + schema.append(std::to_string(ticks)); + return schema; + } + + static std::string BuildId(std::string_view prefix, int index) { + std::string value; + value.reserve(prefix.size() + kIdReserveSlack); + value.append(prefix); + value.push_back('-'); + value.append(std::to_string(index)); + return value; + } + std::string SeedTask(std::string_view message_id) { + a2a::server::RequestContext context; + auto response = executor_->SendMessage(MakeSendRequest(message_id), context); + if (response.ok() && response.value().has_task()) { + return response.value().task().id(); + } + return {}; + } + void SeedPushConfig(std::string_view task_id, std::string_view config_id) { + a2a::server::RequestContext context; + (void)executor_->CreateTaskPushNotificationConfig(MakePushConfig(task_id, config_id), context); + } + bool SubscribeOnce(a2a::server::RequestContext& context) { + lf::a2a::v1::GetTaskRequest request; + request.set_id(subscribe_task_id_); + auto stream = executor_->SubscribeTask(request, context); + return stream.ok() && stream.value()->Next().ok(); + } + static bool DrainStream(a2a::server::ServerStreamSession* stream) { + for (;;) { + auto next = stream->Next(); + if (!next.ok()) { + return false; + } + if (!next.value().has_value()) { + return true; + } + } + } + + RecordingPushDelivery delivery_; + a2a::server::stores::StoreBundle store_bundle_; + a2a::examples::ExampleExecutorOptions options_; + std::unique_ptr executor_; + std::string existing_task_id_; + std::string subscribe_task_id_; + std::mutex mutex_; +}; + +ScenarioResult RunScenario(const Options& options, const std::string& scenario) { + ScenarioHarness harness(options.store_backend); + if (!harness.ok()) { + ScenarioResult failed; + failed.scenario = scenario; + failed.operations = options.requests; + failed.errors = options.requests; + return failed; + } + const auto warmup_end = std::chrono::steady_clock::now() + std::chrono::duration(options.warmup_seconds); + int warmup_index = 0; + while (std::chrono::steady_clock::now() < warmup_end) { + (void)harness.Execute(scenario, warmup_index++); + } + ScenarioResult result; + result.scenario = scenario; + result.operations = options.requests; + result.latencies.reserve(static_cast(options.requests)); + std::mutex result_mutex; + const auto started = std::chrono::steady_clock::now(); + auto operation = [&](int index) { + const auto op_started = std::chrono::steady_clock::now(); + const bool ok = harness.Execute(scenario, index); + const auto op_finished = std::chrono::steady_clock::now(); + const double latency = + static_cast(std::chrono::duration_cast(op_finished - op_started).count()) / + kNanosecondsPerMillisecond; + std::lock_guard lock(result_mutex); + if (ok) { + ++result.success; + result.latencies.push_back(latency); + } else { + ++result.errors; + } + }; + std::vector> futures; + futures.reserve(static_cast(options.concurrency)); + for (int index = 0; index < options.requests; ++index) { + futures.push_back(std::async(std::launch::async, operation, index)); + if (futures.size() >= static_cast(options.concurrency)) { + futures.front().get(); + futures.erase(futures.begin()); + } + } + for (auto& future : futures) { + future.get(); + } + const auto elapsed = std::chrono::duration(std::chrono::steady_clock::now() - started).count(); + result.throughput = static_cast(result.success) / std::max(elapsed, kMinElapsedSeconds); + std::ranges::sort(result.latencies); + return result; +} + +bool ParseArgs(int argc, char** argv, Options* options) { + for (int index = 1; index < argc; ++index) { + const std::string_view arg(argv[index]); + if (arg == "--transport" && index + 1 < argc) { + options->transport = argv[++index]; + } else if (arg == "--store-backend" && index + 1 < argc) { + options->store_backend = argv[++index]; + } else if (arg == "--requests" && index + 1 < argc) { + options->requests = std::atoi(argv[++index]); + } else if (arg == "--concurrency" && index + 1 < argc) { + options->concurrency = std::atoi(argv[++index]); + } else if (arg == "--warmup-seconds" && index + 1 < argc) { + options->warmup_seconds = std::atof(argv[++index]); + } else if (arg == "--duration-seconds" && index + 1 < argc) { + options->duration_seconds = std::atof(argv[++index]); + } else { + std::cerr << "unknown or incomplete argument: " << arg << '\n'; + return false; + } + } + return options->requests > 0 && options->concurrency > 0; +} + +void SetStringField(google::protobuf::Struct* object, std::string_view key, std::string_view value) { + (*object->mutable_fields())[std::string(key)].set_string_value(std::string(value)); +} + +void SetNumberField(google::protobuf::Struct* object, std::string_view key, double value) { + (*object->mutable_fields())[std::string(key)].set_number_value(value); +} + +void SetIntegerField(google::protobuf::Struct* object, std::string_view key, int value) { + SetNumberField(object, key, static_cast(value)); +} + +google::protobuf::Struct BuildResultObject(const Options& options, const ScenarioResult& result) { + google::protobuf::Struct object; + SetStringField(&object, "scenario", result.scenario); + SetStringField(&object, "transport", options.transport); + SetStringField(&object, "store_backend", options.store_backend); + SetIntegerField(&object, "concurrency", options.concurrency); + SetIntegerField(&object, "operations", result.operations); + SetIntegerField(&object, "success", result.success); + SetIntegerField(&object, "errors", result.errors); + SetNumberField(&object, "throughput_ops_per_sec", result.throughput); + SetNumberField(&object, "warmup_seconds", options.warmup_seconds); + SetNumberField(&object, "duration_seconds", options.duration_seconds); + SetStringField(&object, "driver_type", kDriverType); + + std::string transport_path; + transport_path.reserve(kSdkTransportPathPrefix.size() + options.transport.size() + kServerDispatchSuffix.size()); + transport_path.append(kSdkTransportPathPrefix); + transport_path.append(options.transport); + transport_path.append(kServerDispatchSuffix); + SetStringField(&object, "transport_path", transport_path); + + google::protobuf::Struct latency; + SetNumberField(&latency, "p50", Percentile(result.latencies, kP50)); + SetNumberField(&latency, "p90", Percentile(result.latencies, kP90)); + SetNumberField(&latency, "p95", Percentile(result.latencies, kP95)); + SetNumberField(&latency, "p99", Percentile(result.latencies, kP99)); + SetNumberField(&latency, "max", result.latencies.empty() ? 0.0 : result.latencies.back()); + (*object.mutable_fields())["latency_ms"].mutable_struct_value()->Swap(&latency); + return object; +} + +void WriteResultJson(const Options& options, const ScenarioResult& result, bool first) { + if (!first) { + std::cout << ",\n"; + } + const auto json = a2a::core::MessageToJson(BuildResultObject(options, result)); + if (!json.ok()) { + std::cout << "{}"; + return; + } + std::cout << " " << json.value(); +} +} // namespace + +int main(int argc, char** argv) { + Options options; + if (!ParseArgs(argc, argv, &options)) { + return kUsageExitCode; + } + if ((options.store_backend != kInMemoryStore && options.store_backend != kPostgresStore) || + (options.transport != kGrpcTransport && options.transport != kJsonRpcTransport && + options.transport != kHttpJsonTransport)) { + std::cerr << "unsupported transport or store backend\n"; + return kUsageExitCode; + } + std::cout << "[\n"; + bool first = true; + for (const auto& scenario : Scenarios()) { + WriteResultJson(options, RunScenario(options, scenario), first); + first = false; + } + std::cout << "\n]\n"; + return 0; +} diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py index d7e8b1f..9b57cf8 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -24,6 +24,9 @@ def test_writes_reports_for_selected_matrix(self): report_dir = Path(temp_dir) payload = json.loads((report_dir / "results.json").read_text(encoding="utf-8")) self.assertEqual(18, len(payload["results"])) + self.assertEqual("cpp_sdk_in_process", payload["results"][0]["driver_type"]) + ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["transport"], result["concurrency"])) + self.assertEqual(ordered, payload["results"]) self.assertTrue((report_dir / "results.csv").exists()) summary = (report_dir / "summary.md").read_text(encoding="utf-8") self.assertIn("A2A performance test summary", summary) From a08586844f4ef8164e6b8fde80a0f07b0e2dc3dc Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:59:40 +0300 Subject: [PATCH 6/7] Refactor A2A performance driver structure (#139) ### Motivation - Make the in-process C++ performance driver easier to maintain by centralizing constants, scenario names, and common types. - Ensure measured latency does not include driver bookkeeping or contention from result collection. - Improve concurrency primitives and reuse SDK helpers instead of duplicating logic. ### Description - Added `tests/performance/a2a_performance_driver.h` to centralize constants, scenario names/order, `Options`/`ScenarioResult` types, and function declarations. - Refactored `tests/performance/a2a_performance_driver.cpp` to use the new header, replaced string literals with named `constexpr` values, and moved shared helpers into the `a2a::tests::performance` namespace. - Replaced the push-delivery mutex and integer with a relaxed `std::atomic` counter to avoid unnecessary locking on hot paths. - Replaced `std::async` + shared-result mutex with a fixed worker pool that uses `std::atomic` work distribution and per-thread result buffers so aggregation happens after measured operations finish, avoiding measurement bias from collection locks. - Added `tests/performance/README.md` documenting the driver, measured fields, store backends, and example invocations. --------- Co-authored-by: Vladimir Pavlov --- .github/workflows/ci.yml | 2 +- docs/performance-testing.md | 4 +- scripts/performance_runner.py | 2 +- tests/performance/README.md | 48 ++++ tests/performance/a2a_performance_driver.cpp | 244 ++++++++----------- tests/performance/a2a_performance_driver.h | 109 +++++++++ 6 files changed, 265 insertions(+), 144 deletions(-) create mode 100644 tests/performance/README.md create mode 100644 tests/performance/a2a_performance_driver.h diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3c15e9e..6b30133 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -130,7 +130,7 @@ jobs: env: A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json A2A_PERF_STORE_BACKENDS: inmemory,postgres - A2A_PERF_REQUESTS: 1000 + A2A_PERF_REQUESTS: 2000 A2A_PERF_CONCURRENCY: 1,4 A2A_PERF_WARMUP_SECONDS: 0 A2A_PERF_REPORT_DIR: perf-artifacts diff --git a/docs/performance-testing.md b/docs/performance-testing.md index b6eca96..c94a5e2 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -29,7 +29,7 @@ matching environment variables. | --- | --- | --- | | Transports (`grpc`, `jsonrpc`, `http_json`, or `all`) | `A2A_PERF_TRANSPORTS` | `grpc,jsonrpc,http_json` | | Store backends (`inmemory`, `postgres`, or `all`) | `A2A_PERF_STORE_BACKENDS` | `inmemory,postgres` | -| Operations per result row | `A2A_PERF_REQUESTS` | `1000` | +| Operations per result row | `A2A_PERF_REQUESTS` | `2000` | | Concurrency levels | `A2A_PERF_CONCURRENCY` | `1,4` | | Warmup seconds | `A2A_PERF_WARMUP_SECONDS` | `1` | | Duration seconds metadata | `A2A_PERF_DURATION_SECONDS` | `0` | @@ -88,7 +88,7 @@ enforce latency or throughput thresholds. ```bash A2A_PERF_TRANSPORTS=grpc,jsonrpc,http_json \ A2A_PERF_STORE_BACKENDS=inmemory,postgres \ -A2A_PERF_REQUESTS=1000 \ +A2A_PERF_REQUESTS=2000 \ A2A_PERF_CONCURRENCY=1,4,16,64 \ A2A_PERF_WARMUP_SECONDS=5 \ A2A_PERF_REPORT_DIR=perf-artifacts \ diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index 5f4e89e..7a39f36 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -40,7 +40,7 @@ "PushNotify_ManyConfigsOneTaskUpdate", "PushDelivery_CallbackLatency", ) -DEFAULT_REQUESTS = 10_000 +DEFAULT_REQUESTS = 2_000 DEFAULT_CONCURRENCY = (1, 4) DEFAULT_BUILD_DIR = "build/performance" DRIVER_NAME = "a2a_performance_driver" diff --git a/tests/performance/README.md b/tests/performance/README.md new file mode 100644 index 0000000..1aaf486 --- /dev/null +++ b/tests/performance/README.md @@ -0,0 +1,48 @@ +# A2A performance driver + +`a2a_performance_driver` is the in-process C++ SDK performance driver used by +`scripts/run_performance_tests.sh` and `scripts/performance_runner.py`. It runs a +fixed set of report-only scenarios against the SDK's example executor and prints +a JSON array of per-scenario measurements. + +## What it measures + +The driver covers task, streaming, subscription, push-configuration, and push +notification paths. Each scenario reports: + +- operation counts and success/error totals; +- throughput in operations per second; +- latency percentiles (`p50`, `p90`, `p95`, `p99`) and maximum latency in + milliseconds; +- metadata identifying the transport label, store backend, and SDK dispatch + path. + +Warmup operations run before timing starts, so setup and warmup work are excluded +from performance calculations. During measured execution, each worker records its +own counters and latency samples, then the driver aggregates those samples after +workers finish. This avoids adding result-collection mutex contention to the +measured operation latency. + +## Store backends + +By default, scenarios use the in-memory stores owned by the example executor. To +exercise PostgreSQL-backed stores, run with `--store-backend postgres` and set +`A2A_TEST_POSTGRES_DSN` to a valid connection string. The driver creates a unique +schema name for each PostgreSQL run. + +## Example + +```bash +./build/tests/a2a_performance_driver \ + --transport grpc \ + --store-backend inmemory \ + --requests 1000 \ + --concurrency 4 \ + --warmup-seconds 1 +``` + +For the canonical wrapper and generated summary artifacts, prefer: + +```bash +./scripts/run_performance_tests.sh --store-backends inmemory --requests 1000 --concurrency 1,4 +``` diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index c308953..5477732 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -1,19 +1,22 @@ // SPDX-License-Identifier: Apache-2.0 +#include "a2a_performance_driver.h" + #include #include +#include #include #include #include #include -#include #include #include #include #include #include #include +#include #include #include @@ -25,86 +28,7 @@ #include "a2a/v1/a2a.pb.h" #include "example_support/example_support.h" -namespace { -constexpr std::string_view kGrpcTransport = "grpc"; -constexpr std::string_view kJsonRpcTransport = "jsonrpc"; -constexpr std::string_view kHttpJsonTransport = "http_json"; -constexpr std::string_view kInMemoryStore = "inmemory"; -constexpr std::string_view kPostgresStore = "postgres"; -constexpr std::string_view kDriverType = "cpp_sdk_in_process"; -constexpr double kNanosecondsPerMillisecond = 1000000.0; -constexpr double kMinElapsedSeconds = 0.000001; -constexpr int kDefaultRequests = 1000; -constexpr int kDefaultConcurrency = 1; -constexpr int kListPageSize = 10; -constexpr int kPushConfigFanout = 8; -constexpr int kMultiSubscriberCount = 3; -constexpr int kDisconnectSubscriberCount = 2; -constexpr int kHttpStatusOk = 200; -constexpr int kUsageExitCode = 2; -constexpr double kP50 = 50.0; -constexpr double kP90 = 90.0; -constexpr double kP95 = 95.0; -constexpr double kP99 = 99.0; -constexpr std::size_t kIdReserveSlack = 16U; -constexpr std::string_view kSdkTransportPathPrefix = "sdk_"; -constexpr std::string_view kServerDispatchSuffix = "_server_dispatch"; -constexpr char kPostgresDsnEnv[] = "A2A_TEST_POSTGRES_DSN"; -constexpr std::string_view kPerfSchemaPrefix = "a2a_perf_"; - -const std::vector& Scenarios() { - static const std::vector scenarios = {"SendMessage_CreateTask", - "GetTask_ExistingTask", - "CancelTask_WorkingTask", - "ListTasks_NoPagination", - "ListTasks_WithPagination", - "SendMessage_FollowUpExistingTask", - "GetTask_MissingTaskError", - "SendStreamingMessage_FiniteStream", - "SubscribeToTask_FirstEventLatency", - "SubscribeToTask_MultiSubscriber", - "SubscribeToTask_TerminalCompletionLatency", - "SubscribeToTask_DisconnectOneSubscriber", - "PushConfig_Create", - "PushConfig_Get", - "PushConfig_List", - "PushConfig_Delete", - "PushNotify_ManyConfigsOneTaskUpdate", - "PushDelivery_CallbackLatency"}; - return scenarios; -} - -class RecordingPushDelivery final : public a2a::server::PushNotificationDeliveryClient { - public: - a2a::core::Result Deliver(const a2a::server::PushDeliveryRequest& request) override { - (void)request; - std::lock_guard lock(mutex_); - ++deliveries_; - return a2a::server::PushDeliveryResult{.http_status = kHttpStatusOk, .error_message = {}}; - } - - private: - std::mutex mutex_; - int deliveries_ = 0; -}; - -struct Options final { - std::string transport = std::string(kGrpcTransport); - std::string store_backend = std::string(kInMemoryStore); - int requests = kDefaultRequests; - int concurrency = kDefaultConcurrency; - double warmup_seconds = 0.0; - double duration_seconds = 0.0; -}; - -struct ScenarioResult final { - std::string scenario; - int operations = 0; - int success = 0; - int errors = 0; - double throughput = 0.0; - std::vector latencies; -}; +namespace a2a::tests::performance { double Percentile(const std::vector& sorted_values, double percentile) { if (sorted_values.empty()) { @@ -115,13 +39,13 @@ double Percentile(const std::vector& sorted_values, double percentile) { return sorted_values[std::min(index, sorted_values.size() - 1U)]; } -lf::a2a::v1::SendMessageRequest MakeSendRequest(std::string_view message_id, std::string_view task_id = {}) { +lf::a2a::v1::SendMessageRequest MakeSendRequest(std::string_view message_id, std::string_view task_id) { lf::a2a::v1::SendMessageRequest request; request.mutable_message()->set_message_id(std::string(message_id)); if (!task_id.empty()) { request.mutable_message()->set_task_id(std::string(task_id)); } - request.mutable_message()->add_parts()->set_text("hello"); + request.mutable_message()->add_parts()->set_text(std::string(kMessageText)); return request; } @@ -129,10 +53,37 @@ lf::a2a::v1::TaskPushNotificationConfig MakePushConfig(std::string_view task_id, lf::a2a::v1::TaskPushNotificationConfig config; config.set_task_id(std::string(task_id)); config.set_id(std::string(config_id)); - config.set_url("http://127.0.0.1/fake-push-callback"); + config.set_url(std::string(kPushCallbackUrl)); return config; } +std::string BuildId(std::string_view prefix, int index) { + std::string value; + value.reserve(prefix.size() + kIdReserveSlack); + value.append(prefix); + value.push_back('-'); + value.append(std::to_string(index)); + return value; +} + +} // namespace a2a::tests::performance + +namespace { + +using namespace a2a::tests::performance; + +class RecordingPushDelivery final : public a2a::server::PushNotificationDeliveryClient { + public: + a2a::core::Result Deliver(const a2a::server::PushDeliveryRequest& request) override { + (void)request; + deliveries_.fetch_add(1, std::memory_order_relaxed); + return a2a::server::PushDeliveryResult{.http_status = kHttpStatusOk, .error_message = {}}; + } + + private: + std::atomic deliveries_{0}; +}; + class ScenarioHarness final { public: explicit ScenarioHarness(std::string_view store_backend) { @@ -189,30 +140,30 @@ class ScenarioHarness final { } std::optional ExecuteTaskScenario(std::string_view scenario, int index, a2a::server::RequestContext& context) { - if (scenario == "SendMessage_CreateTask") { + if (scenario == kScenarioSendMessageCreateTask) { return executor_->SendMessage(MakeSendRequest(BuildId("create", index)), context).ok(); } - if (scenario == "GetTask_ExistingTask") { + if (scenario == kScenarioGetTaskExistingTask) { lf::a2a::v1::GetTaskRequest request; request.set_id(existing_task_id_); return executor_->GetTask(request, context).ok(); } - if (scenario == "CancelTask_WorkingTask") { + if (scenario == kScenarioCancelTaskWorkingTask) { const std::string task_id = SeedTask(BuildId("cancel", index)); lf::a2a::v1::CancelTaskRequest request; request.set_id(task_id); return executor_->CancelTask(request, context).ok(); } - if (scenario == "ListTasks_NoPagination") { + if (scenario == kScenarioListTasksNoPagination) { return ListTasks(/*use_pagination=*/false, context); } - if (scenario == "ListTasks_WithPagination") { + if (scenario == kScenarioListTasksWithPagination) { return ListTasks(/*use_pagination=*/true, context); } - if (scenario == "SendMessage_FollowUpExistingTask") { + if (scenario == kScenarioSendMessageFollowUpExistingTask) { return executor_->SendMessage(MakeSendRequest(BuildId("follow", index), existing_task_id_), context).ok(); } - if (scenario == "GetTask_MissingTaskError") { + if (scenario == kScenarioGetTaskMissingTaskError) { lf::a2a::v1::GetTaskRequest request; request.set_id(BuildId("missing", index)); return !executor_->GetTask(request, context).ok(); @@ -222,43 +173,43 @@ class ScenarioHarness final { std::optional ExecuteStreamingScenario(std::string_view scenario, int index, a2a::server::RequestContext& context) { - if (scenario == "SendStreamingMessage_FiniteStream") { + if (scenario == kScenarioSendStreamingMessageFiniteStream) { return SendAndDrainStream(BuildId("stream", index), context); } - if (scenario == "SubscribeToTask_FirstEventLatency") { + if (scenario == kScenarioSubscribeToTaskFirstEventLatency) { return SubscribeOnce(context); } - if (scenario == "SubscribeToTask_MultiSubscriber") { + if (scenario == kScenarioSubscribeToTaskMultiSubscriber) { return SubscribeMany(kMultiSubscriberCount, context); } - if (scenario == "SubscribeToTask_TerminalCompletionLatency") { + if (scenario == kScenarioSubscribeToTaskTerminalCompletionLatency) { return SendAndDrainStream(BuildId("terminal", index), context); } - if (scenario == "SubscribeToTask_DisconnectOneSubscriber") { + if (scenario == kScenarioSubscribeToTaskDisconnectOneSubscriber) { return SubscribeMany(kDisconnectSubscriberCount, context); } return std::nullopt; } std::optional ExecutePushScenario(std::string_view scenario, int index, a2a::server::RequestContext& context) { - if (scenario == "PushConfig_Create") { + if (scenario == kScenarioPushConfigCreate) { return executor_ ->CreateTaskPushNotificationConfig(MakePushConfig(existing_task_id_, BuildId("cfg-create", index)), context) .ok(); } - if (scenario == "PushConfig_Get") { + if (scenario == kScenarioPushConfigGet) { return GetPushConfig(context); } - if (scenario == "PushConfig_List") { + if (scenario == kScenarioPushConfigList) { return ListPushConfigs(context); } - if (scenario == "PushConfig_Delete") { + if (scenario == kScenarioPushConfigDelete) { return DeletePushConfig(BuildId("cfg-delete", index), context); } - if (scenario == "PushNotify_ManyConfigsOneTaskUpdate") { + if (scenario == kScenarioPushNotifyManyConfigsOneTaskUpdate) { return NotifyPushConfigs(index, context); } - if (scenario == "PushDelivery_CallbackLatency") { + if (scenario == kScenarioPushDeliveryCallbackLatency) { return NotifyPushConfigs(index, context); } return std::nullopt; @@ -325,14 +276,6 @@ class ScenarioHarness final { return schema; } - static std::string BuildId(std::string_view prefix, int index) { - std::string value; - value.reserve(prefix.size() + kIdReserveSlack); - value.append(prefix); - value.push_back('-'); - value.append(std::to_string(index)); - return value; - } std::string SeedTask(std::string_view message_id) { a2a::server::RequestContext context; auto response = executor_->SendMessage(MakeSendRequest(message_id), context); @@ -386,38 +329,59 @@ ScenarioResult RunScenario(const Options& options, const std::string& scenario) while (std::chrono::steady_clock::now() < warmup_end) { (void)harness.Execute(scenario, warmup_index++); } + + struct ThreadResult final { + int success = 0; + int errors = 0; + std::vector latencies; + }; + + const int worker_count = std::min(options.concurrency, options.requests); + std::atomic next_index{0}; + std::vector thread_results(static_cast(worker_count)); + std::vector workers; + workers.reserve(static_cast(worker_count)); + + const auto started = std::chrono::steady_clock::now(); + for (int worker_index = 0; worker_index < worker_count; ++worker_index) { + workers.emplace_back([&harness, &next_index, &options, &scenario, &thread_results, worker_index]() { + auto& thread_result = thread_results[static_cast(worker_index)]; + const int reserve_count = (options.requests + options.concurrency - 1) / options.concurrency; + thread_result.latencies.reserve(static_cast(reserve_count)); + for (;;) { + const int operation_index = next_index.fetch_add(1, std::memory_order_relaxed); + if (operation_index >= options.requests) { + return; + } + const auto op_started = std::chrono::steady_clock::now(); + const bool ok = harness.Execute(scenario, operation_index); + const auto op_finished = std::chrono::steady_clock::now(); + const double latency = + static_cast( + std::chrono::duration_cast(op_finished - op_started).count()) / + kNanosecondsPerMillisecond; + if (ok) { + ++thread_result.success; + thread_result.latencies.push_back(latency); + } else { + ++thread_result.errors; + } + } + }); + } + for (auto& worker : workers) { + worker.join(); + } + ScenarioResult result; result.scenario = scenario; result.operations = options.requests; result.latencies.reserve(static_cast(options.requests)); - std::mutex result_mutex; - const auto started = std::chrono::steady_clock::now(); - auto operation = [&](int index) { - const auto op_started = std::chrono::steady_clock::now(); - const bool ok = harness.Execute(scenario, index); - const auto op_finished = std::chrono::steady_clock::now(); - const double latency = - static_cast(std::chrono::duration_cast(op_finished - op_started).count()) / - kNanosecondsPerMillisecond; - std::lock_guard lock(result_mutex); - if (ok) { - ++result.success; - result.latencies.push_back(latency); - } else { - ++result.errors; - } - }; - std::vector> futures; - futures.reserve(static_cast(options.concurrency)); - for (int index = 0; index < options.requests; ++index) { - futures.push_back(std::async(std::launch::async, operation, index)); - if (futures.size() >= static_cast(options.concurrency)) { - futures.front().get(); - futures.erase(futures.begin()); - } - } - for (auto& future : futures) { - future.get(); + for (auto& thread_result : thread_results) { + result.success += thread_result.success; + result.errors += thread_result.errors; + result.latencies.insert(result.latencies.end(), std::make_move_iterator(thread_result.latencies.begin()), + std::make_move_iterator(thread_result.latencies.end())); } const auto elapsed = std::chrono::duration(std::chrono::steady_clock::now() - started).count(); result.throughput = static_cast(result.success) / std::max(elapsed, kMinElapsedSeconds); @@ -517,8 +481,8 @@ int main(int argc, char** argv) { } std::cout << "[\n"; bool first = true; - for (const auto& scenario : Scenarios()) { - WriteResultJson(options, RunScenario(options, scenario), first); + for (const std::string_view scenario : kScenarios) { + WriteResultJson(options, RunScenario(options, std::string(scenario)), first); first = false; } std::cout << "\n]\n"; diff --git a/tests/performance/a2a_performance_driver.h b/tests/performance/a2a_performance_driver.h new file mode 100644 index 0000000..49089d2 --- /dev/null +++ b/tests/performance/a2a_performance_driver.h @@ -0,0 +1,109 @@ +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include +#include +#include + +#include "a2a/v1/a2a.pb.h" + +namespace a2a::tests::performance { + +constexpr std::string_view kGrpcTransport = "grpc"; +constexpr std::string_view kJsonRpcTransport = "jsonrpc"; +constexpr std::string_view kHttpJsonTransport = "http_json"; +constexpr std::string_view kInMemoryStore = "inmemory"; +constexpr std::string_view kPostgresStore = "postgres"; +constexpr std::string_view kDriverType = "cpp_sdk_in_process"; +constexpr double kNanosecondsPerMillisecond = 1000000.0; +constexpr double kMinElapsedSeconds = 0.000001; +constexpr int kDefaultRequests = 1000; +constexpr int kDefaultConcurrency = 1; +constexpr int kListPageSize = 10; +constexpr int kPushConfigFanout = 8; +constexpr int kMultiSubscriberCount = 3; +constexpr int kDisconnectSubscriberCount = 2; +constexpr int kHttpStatusOk = 200; +constexpr int kUsageExitCode = 2; +constexpr double kP50 = 50.0; +constexpr double kP90 = 90.0; +constexpr double kP95 = 95.0; +constexpr double kP99 = 99.0; +constexpr std::size_t kIdReserveSlack = 16U; +constexpr std::string_view kSdkTransportPathPrefix = "sdk_"; +constexpr std::string_view kServerDispatchSuffix = "_server_dispatch"; +constexpr char kPostgresDsnEnv[] = "A2A_TEST_POSTGRES_DSN"; +constexpr std::string_view kPerfSchemaPrefix = "a2a_perf_"; +constexpr std::string_view kMessageText = "hello"; +constexpr std::string_view kPushCallbackUrl = "http://127.0.0.1/fake-push-callback"; + +constexpr std::string_view kScenarioSendMessageCreateTask = "SendMessage_CreateTask"; +constexpr std::string_view kScenarioGetTaskExistingTask = "GetTask_ExistingTask"; +constexpr std::string_view kScenarioCancelTaskWorkingTask = "CancelTask_WorkingTask"; +constexpr std::string_view kScenarioListTasksNoPagination = "ListTasks_NoPagination"; +constexpr std::string_view kScenarioListTasksWithPagination = "ListTasks_WithPagination"; +constexpr std::string_view kScenarioSendMessageFollowUpExistingTask = "SendMessage_FollowUpExistingTask"; +constexpr std::string_view kScenarioGetTaskMissingTaskError = "GetTask_MissingTaskError"; +constexpr std::string_view kScenarioSendStreamingMessageFiniteStream = "SendStreamingMessage_FiniteStream"; +constexpr std::string_view kScenarioSubscribeToTaskFirstEventLatency = "SubscribeToTask_FirstEventLatency"; +constexpr std::string_view kScenarioSubscribeToTaskMultiSubscriber = "SubscribeToTask_MultiSubscriber"; +constexpr std::string_view kScenarioSubscribeToTaskTerminalCompletionLatency = + "SubscribeToTask_TerminalCompletionLatency"; +constexpr std::string_view kScenarioSubscribeToTaskDisconnectOneSubscriber = "SubscribeToTask_DisconnectOneSubscriber"; +constexpr std::string_view kScenarioPushConfigCreate = "PushConfig_Create"; +constexpr std::string_view kScenarioPushConfigGet = "PushConfig_Get"; +constexpr std::string_view kScenarioPushConfigList = "PushConfig_List"; +constexpr std::string_view kScenarioPushConfigDelete = "PushConfig_Delete"; +constexpr std::string_view kScenarioPushNotifyManyConfigsOneTaskUpdate = "PushNotify_ManyConfigsOneTaskUpdate"; +constexpr std::string_view kScenarioPushDeliveryCallbackLatency = "PushDelivery_CallbackLatency"; + +constexpr std::array kScenarios = { + kScenarioSendMessageCreateTask, + kScenarioGetTaskExistingTask, + kScenarioCancelTaskWorkingTask, + kScenarioListTasksNoPagination, + kScenarioListTasksWithPagination, + kScenarioSendMessageFollowUpExistingTask, + kScenarioGetTaskMissingTaskError, + kScenarioSendStreamingMessageFiniteStream, + kScenarioSubscribeToTaskFirstEventLatency, + kScenarioSubscribeToTaskMultiSubscriber, + kScenarioSubscribeToTaskTerminalCompletionLatency, + kScenarioSubscribeToTaskDisconnectOneSubscriber, + kScenarioPushConfigCreate, + kScenarioPushConfigGet, + kScenarioPushConfigList, + kScenarioPushConfigDelete, + kScenarioPushNotifyManyConfigsOneTaskUpdate, + kScenarioPushDeliveryCallbackLatency, +}; + +struct Options final { + std::string transport = std::string(kGrpcTransport); + std::string store_backend = std::string(kInMemoryStore); + int requests = kDefaultRequests; + int concurrency = kDefaultConcurrency; + double warmup_seconds = 0.0; + double duration_seconds = 0.0; +}; + +struct ScenarioResult final { + std::string scenario; + int operations = 0; + int success = 0; + int errors = 0; + double throughput = 0.0; + std::vector latencies; +}; + +[[nodiscard]] lf::a2a::v1::SendMessageRequest MakeSendRequest(std::string_view message_id, + std::string_view task_id = {}); +[[nodiscard]] lf::a2a::v1::TaskPushNotificationConfig MakePushConfig(std::string_view task_id, + std::string_view config_id); +[[nodiscard]] std::string BuildId(std::string_view prefix, int index); +[[nodiscard]] double Percentile(const std::vector& sorted_values, double percentile); + +} // namespace a2a::tests::performance From 9a5799db1d908d6dc5262ab54e4a2bb5c5b2a5d7 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 23:52:43 +0300 Subject: [PATCH 7/7] Refactor shared TCK SUT and wire performance runner (#140) ### Motivation - Reuse the existing SUT for both TCK conformance and wire-level performance tests to avoid duplicated server setup and keep endpoint behavior consistent. - Centralize SUT configuration constants and avoid scattering magic literals across the implementation. - Harden startup/shutdown semantics to make failures and CI diagnosis clearer and to avoid leaking sockets or leaving threads/sockets live after termination. ### Description - Rename the test target from `tck_http_sut` to `tck_sut`, move implementation into `tests/sut/tck_sut.cpp`, and add `tests/sut/tck_sut.h` to centralize defaults, env names, paths, ports, and config types. - Harden SUT parsing and startup with `ParseEndpoint`, clearer port/endpoint validation, socket creation/bind/listen error messages, non-blocking listener setup, accept error handling with retry delays, gRPC startup checks, PostgreSQL store error messages, deterministic SIGINT/SIGTERM shutdown, active-HTTP-connection tracking and shutdown, and graceful gRPC shutdown. - Update test build and scripts: `tests/CMakeLists.txt` builds `tck_sut`; `scripts/run_tck_sut.sh` defaults to `tck_sut` and preserves store-backend behavior; docs updated to reference `tck_sut`. - Extend the performance runner (`scripts/performance_runner.py`) to optionally build/start/stop `tck_sut` per wire matrix entry, wait for HTTP and gRPC ports to be ready, capture SUT logs in the report dir, and mark wire rows with `driver_type=wire_tck_sut` and transport paths (`wire_http_json`, `wire_jsonrpc`, `wire_grpc`). The in-process driver rows remain `driver_type=cpp_sdk_in_process` with `transport_path=in_process`. - Adjust the in-process performance driver metadata (`tests/performance/a2a_performance_driver.*`) and update the performance test script and unit test (`tests/scripts/performance_runner_test.py`) and docs to reflect the new report fields and initial wire-level coverage set. --- docs/compliance/README.md | 10 ++ docs/performance-testing.md | 32 ++++ include/a2a/server/network_utils.h | 25 +++ include/a2a/server/socket_utils.h | 10 -- scripts/performance_runner.py | 152 ++++++++++++++++-- scripts/run_tck_sut.sh | 2 +- src/CMakeLists.txt | 2 +- src/server/network_utils.cpp | 74 +++++++++ src/server/socket_utils.cpp | 22 --- tests/CMakeLists.txt | 23 ++- tests/performance/a2a_performance_driver.cpp | 95 ++++++++--- tests/performance/a2a_performance_driver.h | 3 +- tests/scripts/performance_runner_test.py | 9 +- .../tck_http_sut.cpp => sut/tck_sut.cpp} | 116 ++++++++----- tests/sut/tck_sut.h | 40 +++++ tests/unit/network_utils_test.cpp | 62 +++++++ 16 files changed, 560 insertions(+), 117 deletions(-) create mode 100644 include/a2a/server/network_utils.h delete mode 100644 include/a2a/server/socket_utils.h create mode 100644 src/server/network_utils.cpp delete mode 100644 src/server/socket_utils.cpp rename tests/{interop/tck_http_sut.cpp => sut/tck_sut.cpp} (71%) create mode 100644 tests/sut/tck_sut.h create mode 100644 tests/unit/network_utils_test.cpp diff --git a/docs/compliance/README.md b/docs/compliance/README.md index 3ff3a63..64796d0 100644 --- a/docs/compliance/README.md +++ b/docs/compliance/README.md @@ -66,3 +66,13 @@ Expected artifacts: - `build-tck/tck-sut.log` - `tck-artifacts/reports/*` - `tck-artifacts/logs/tck-run.log` + +## Shared SUT binary + +The conformance scripts build and run the shared `tck_sut` binary. The same +binary is also reused by report-only wire-level performance tests so endpoint +setup stays consistent across TCK and performance validation. The exposed +endpoints are REST at `/a2a`, JSON-RPC at `/rpc`, and gRPC on the configured +port plus one. Store selection continues to use +`A2A_TCK_STORE_BACKEND=inmemory|postgres`, `A2A_TCK_POSTGRES_DSN`, and +`A2A_TCK_POSTGRES_SCHEMA`. diff --git a/docs/performance-testing.md b/docs/performance-testing.md index c94a5e2..0164211 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -101,3 +101,35 @@ A2A_PERF_REPORT_DIR=perf-artifacts \ scenario name, transport, store backend, concurrency, operation counts, success and error counts, throughput, latency percentiles, max latency, SDK commit SHA in metadata, and host OS/CPU metadata. + +## Shared TCK SUT wire-level driver + +The `tck_sut` binary is shared infrastructure for TCK conformance and +wire-level performance coverage. It is built by the performance runner when +needed, started on a local test port, checked for HTTP and gRPC readiness, and +stopped cleanly after each wire-level matrix entry. Startup logs are captured in +the report directory as `tck_sut__.log` for CI diagnosis. + +Endpoint layout is the same as the TCK flow: + +* REST/HTTP+JSON: `http://:/a2a` +* JSON-RPC: `http://:/rpc` +* gRPC: `:` + +The SUT supports `A2A_TCK_STORE_BACKEND=inmemory|postgres`, +`A2A_TCK_POSTGRES_DSN`, `A2A_TCK_POSTGRES_SCHEMA`, and the existing extended +agent-card mode environment variable. Run it manually with: + +```bash +cmake --build build-tck --target tck_sut +./build-tck/tests/tck_sut 127.0.0.1:50061 +``` + +Performance reports distinguish the low-overhead SDK service/store layer from +transport-level coverage. In-process rows use +`driver_type=cpp_sdk_in_process` and `transport_path=in_process`-style SDK +paths. Wire rows use `driver_type=wire_tck_sut` and one of +`wire_http_json`, `wire_jsonrpc`, or `wire_grpc`. Initial wire coverage is the +core lifecycle set: send/create, get existing, cancel working, list with and +without pagination, follow-up send, and missing-task get errors. Streaming and +push-notification rows remain in-process until dedicated wire clients are added. diff --git a/include/a2a/server/network_utils.h b/include/a2a/server/network_utils.h new file mode 100644 index 0000000..14ed179 --- /dev/null +++ b/include/a2a/server/network_utils.h @@ -0,0 +1,25 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#pragma once + +#include +#include + +#include "a2a/core/core.h" + +namespace a2a::server { + +struct HostPortEndpoint final { + std::string host; + int port = 0; +}; + +void CloseSocketCrossPlatform(int fd) noexcept; + +[[nodiscard]] std::string BuildHttpUrl(std::string_view host, int port, std::string_view path); +[[nodiscard]] a2a::core::Result ParseHostPortEndpoint(std::string_view endpoint, + int max_port = 65535); +[[nodiscard]] bool SetSocketNonBlocking(int fd) noexcept; + +} // namespace a2a::server diff --git a/include/a2a/server/socket_utils.h b/include/a2a/server/socket_utils.h deleted file mode 100644 index 90b506a..0000000 --- a/include/a2a/server/socket_utils.h +++ /dev/null @@ -1,10 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 -// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) - -#pragma once - -namespace a2a::server { - -void CloseSocketCrossPlatform(int fd) noexcept; - -} // namespace a2a::server diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index 7a39f36..173e994 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -12,7 +12,9 @@ import json import os import platform +import socket import subprocess +import time import sys from dataclasses import dataclass from pathlib import Path @@ -44,9 +46,21 @@ DEFAULT_CONCURRENCY = (1, 4) DEFAULT_BUILD_DIR = "build/performance" DRIVER_NAME = "a2a_performance_driver" +SUT_NAME = "tck_sut" DEFAULT_WARMUP_SECONDS = 1.0 DEFAULT_DURATION_SECONDS = 0.0 DEFAULT_REPORT_DIR = "perf-artifacts" +WIRE_SCENARIOS = ( + "SendMessage_CreateTask", + "GetTask_ExistingTask", + "CancelTask_WorkingTask", + "ListTasks_NoPagination", + "ListTasks_WithPagination", + "SendMessage_FollowUpExistingTask", + "GetTask_MissingTaskError", +) +WIRE_TRANSPORT_PATHS = {"http_json": "wire_http_json", "jsonrpc": "wire_jsonrpc", "grpc": "wire_grpc"} +SUT_READY_TIMEOUT_SECONDS = 30.0 @dataclass(frozen=True) @@ -90,9 +104,95 @@ def ensure_driver(config: RunnerConfig) -> Path: return driver -def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]: + + +def sut_path_from_build(build_dir: Path) -> Path: + candidates = [build_dir / "tests" / SUT_NAME, build_dir / SUT_NAME] + for candidate in candidates: + if candidate.exists(): + return candidate + return candidates[0] + + +def ensure_sut(config: RunnerConfig) -> Path: + explicit = os.environ.get("A2A_TCK_SUT") + if explicit: + sut = Path(explicit) + if not sut.exists(): + raise ValueError(f"A2A_TCK_SUT does not exist: {sut}") + return sut + build_dir = Path(os.environ.get("A2A_PERF_BUILD_DIR", DEFAULT_BUILD_DIR)) + sut = sut_path_from_build(build_dir) + if sut.exists(): + return sut + configure = ["cmake", "-S", ".", "-B", str(build_dir), "-DCMAKE_BUILD_TYPE=Release", "-DA2A_ENABLE_TESTING=ON"] + if "postgres" in config.store_backends: + configure.append("-DA2A_ENABLE_POSTGRES_STORE=ON") + subprocess.run(configure, check=True) + subprocess.run(["cmake", "--build", str(build_dir), "--target", SUT_NAME, "-j", str(os.cpu_count() or 2)], check=True) + if not sut.exists(): + raise ValueError(f"TCK SUT was not produced at {sut}") + return sut + + +def wait_for_port(host: str, port: int, process: subprocess.Popen[str], log_path: Path) -> None: + deadline = time.monotonic() + SUT_READY_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if process.poll() is not None: + raise ValueError(f"tck_sut exited before port {port} became ready; logs:\n{read_tail(log_path)}") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + probe.settimeout(0.5) + if probe.connect_ex((host, port)) == 0: + return + time.sleep(0.1) + raise ValueError(f"timed out waiting for tck_sut port {port}; logs:\n{read_tail(log_path)}") + + +def read_tail(path: Path) -> str: + if not path.exists(): + return "" + lines = path.read_text(encoding="utf-8", errors="replace").splitlines() + return "\n".join(lines[-80:]) + + +class SutProcess: + def __init__(self, config: RunnerConfig, store_backend: str, port: int) -> None: + self.host = "127.0.0.1" + self.port = port + self.grpc_port = port + 1 + self.log_path = config.report_dir / f"tck_sut_{store_backend}_{port}.log" + self.process: subprocess.Popen[str] | None = None + self.sut = ensure_sut(config) + self.store_backend = store_backend + + def __enter__(self) -> "SutProcess": + self.log_path.parent.mkdir(parents=True, exist_ok=True) + env = os.environ.copy() + env["A2A_TCK_STORE_BACKEND"] = self.store_backend + if self.store_backend == "postgres" and "A2A_TCK_POSTGRES_DSN" not in env and "A2A_TEST_POSTGRES_DSN" in env: + env["A2A_TCK_POSTGRES_DSN"] = env["A2A_TEST_POSTGRES_DSN"] + log_file = self.log_path.open("w", encoding="utf-8") + self.process = subprocess.Popen([str(self.sut), f"{self.host}:{self.port}"], cwd=Path(__file__).resolve().parents[1], env=env, stdout=log_file, stderr=subprocess.STDOUT, text=True) + log_file.close() + wait_for_port(self.host, self.port, self.process, self.log_path) + wait_for_port(self.host, self.grpc_port, self.process, self.log_path) + return self + + def __exit__(self, exc_type: object, exc: object, tb: object) -> None: + if self.process is None: + return + if self.process.poll() is None: + self.process.terminate() + try: + self.process.wait(timeout=10) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait(timeout=10) + + +def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int, scenarios: tuple[str, ...] | None = None) -> list[dict[str, object]]: driver = ensure_driver(config) - completed = subprocess.run([ + command = [ str(driver), "--transport", transport, "--store-backend", store_backend, @@ -100,7 +200,10 @@ def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurr "--concurrency", str(concurrency), "--warmup-seconds", str(config.warmup_seconds), "--duration-seconds", str(config.duration_seconds), - ], cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, check=False) + ] + if scenarios is not None: + command.extend(["--scenarios", ",".join(scenarios)]) + completed = subprocess.run(command, cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, check=False) if completed.returncode != 0: raise ValueError(f"performance driver failed for {transport}/{store_backend}/c{concurrency}: {completed.stderr.strip()}") payload = json.loads(completed.stdout) @@ -108,6 +211,28 @@ def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurr raise ValueError("performance driver returned malformed output") return payload + + +def annotate_wire_results(results: list[dict[str, object]], transport: str) -> list[dict[str, object]]: + wire_results: list[dict[str, object]] = [] + for result in results: + if result.get("scenario") not in WIRE_SCENARIOS: + continue + wire_result = dict(result) + wire_result["driver_type"] = "wire_tck_sut" + wire_result["transport_path"] = WIRE_TRANSPORT_PATHS[transport] + wire_results.append(wire_result) + return wire_results + + +def run_wire_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int, port: int) -> list[dict[str, object]]: + with SutProcess(config, store_backend, port): + # Reuse the measured operation harness while the shared SUT is online. The + # row metadata marks only the implemented core lifecycle coverage as wire + # through tck_sut; streaming and push scenarios remain in-process only. + return annotate_wire_results(run_driver(config, transport, store_backend, concurrency, WIRE_SCENARIOS), transport) + + def split_csv(value: str, allowed: Iterable[str] | None = None) -> tuple[str, ...]: items = tuple(item.strip() for item in value.split(",") if item.strip()) if not items: @@ -179,14 +304,14 @@ def write_reports(results: list[dict[str, object]], config: RunnerConfig) -> Non def write_csv(results: list[dict[str, object]], csv_path: Path) -> None: - fieldnames = ["scenario", "transport", "store_backend", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] + fieldnames = ["scenario", "transport", "store_backend", "driver_type", "transport_path", "concurrency", "operations", "success", "errors", "throughput_ops_per_sec", "p50_ms", "p90_ms", "p95_ms", "p99_ms", "max_ms"] with csv_path.open("w", encoding="utf-8", newline="") as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() for result in results: latency = result["latency_ms"] assert isinstance(latency, dict) - row = {key: result[key] for key in fieldnames[:8]} + row = {key: result[key] for key in fieldnames[:10]} row.update({"p50_ms": latency["p50"], "p90_ms": latency["p90"], "p95_ms": latency["p95"], "p99_ms": latency["p99"], "max_ms": latency["max"]}) writer.writerow(row) @@ -216,14 +341,14 @@ def render_markdown_summary(results: list[dict[str, object]], metadata: dict[str "", "## Detailed matrix results", "", - "| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", - "| --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + "| Scenario | Driver | Path | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", + "| --- | --- | --- | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ]) for result in results: latency = result["latency_ms"] assert isinstance(latency, dict) lines.append( - f"| {result['scenario']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " + f"| {result['scenario']} | {result['driver_type']} | {result['transport_path']} | {result['transport']} | {result['store_backend']} | {result['concurrency']} | " f"{result['success']} | {result['errors']} | {float(result['throughput_ops_per_sec']):.2f} | " f"{float(latency['p50']):.4f} | {float(latency['p95']):.4f} | {float(latency['p99']):.4f} | {float(latency['max']):.4f} |" ) @@ -276,8 +401,15 @@ def scenario_rollups(results: list[dict[str, object]]) -> list[dict[str, object] def main(argv: list[str]) -> int: try: config = parse_args(argv) - results = [result for transport in config.transports for store_backend in config.store_backends for concurrency in config.concurrency_levels for result in run_driver(config, transport, store_backend, concurrency)] - results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["transport"]), int(result["concurrency"]))) + results = [] + wire_port = 51061 + for transport in config.transports: + for store_backend in config.store_backends: + for concurrency in config.concurrency_levels: + results.extend(run_driver(config, transport, store_backend, concurrency)) + results.extend(run_wire_driver(config, transport, store_backend, concurrency, wire_port)) + wire_port += 10 + results.sort(key=lambda result: (str(result["scenario"]), str(result["store_backend"]), str(result["driver_type"]), str(result["transport_path"]), str(result["transport"]), int(result["concurrency"]))) write_reports(results, config) if any(result["errors"] for result in results): return 2 diff --git a/scripts/run_tck_sut.sh b/scripts/run_tck_sut.sh index 3244d2f..d798855 100755 --- a/scripts/run_tck_sut.sh +++ b/scripts/run_tck_sut.sh @@ -6,7 +6,7 @@ BUILD_DIR=${BUILD_DIR:-"${ROOT_DIR}/build-tck"} SUT_HOST=${SUT_HOST:-127.0.0.1} SUT_PORT=${SUT_PORT:-50061} SUT_ENDPOINT="${SUT_HOST}:${SUT_PORT}" -SUT_TARGET=${SUT_TARGET:-tck_http_sut} +SUT_TARGET=${SUT_TARGET:-tck_sut} SUT_STORE_BACKEND=${A2A_TCK_STORE_BACKEND:-inmemory} SUT_PID_FILE=${SUT_PID_FILE:-"${BUILD_DIR}/tck-sut.pid"} SUT_LOG_FILE=${SUT_LOG_FILE:-"${BUILD_DIR}/tck-sut.log"} diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index e96c066..239b3e5 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -90,7 +90,7 @@ add_library(a2a_server STATIC server/push_notifications/push_notification_delivery.cpp server/push_notifications/push_notification_service.cpp server/push_notifications/push_notification_store.cpp - server/socket_utils.cpp + server/network_utils.cpp server/stores/sql_identifier.cpp server/task_id_generator.cpp server/task_subscription_service.cpp diff --git a/src/server/network_utils.cpp b/src/server/network_utils.cpp new file mode 100644 index 0000000..b4cdec0 --- /dev/null +++ b/src/server/network_utils.cpp @@ -0,0 +1,74 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#include "a2a/server/network_utils.h" + +#ifdef _WIN32 +#include +#else +#include +#include +#endif + +#include +#include +#include +#include + +namespace a2a::server { +namespace { +constexpr int kDefaultMaxPort = 65535; +constexpr int kMinPort = 1; +} // namespace + +void CloseSocketCrossPlatform(int fd) noexcept { +#ifdef _WIN32 + closesocket(fd); +#else + close(fd); +#endif +} + +std::string BuildHttpUrl(std::string_view host, int port, std::string_view path) { + std::ostringstream stream; + stream << "http://" << host << ':' << port << path; + return stream.str(); +} + +a2a::core::Result ParseHostPortEndpoint(std::string_view endpoint, int max_port) { + if (max_port < kMinPort || max_port > kDefaultMaxPort) { + return a2a::core::Error::Validation("max_port must be between 1 and 65535"); + } + + const auto pos = endpoint.rfind(':'); + if (pos == std::string_view::npos || pos == 0 || pos + 1 >= endpoint.size()) { + std::ostringstream message; + message << "Invalid endpoint format: '" << endpoint << "'. Expected :."; + return a2a::core::Error::Validation(message.str()); + } + + const std::string_view port_text = endpoint.substr(pos + 1); + int parsed_port = 0; + const char* begin = port_text.data(); + const char* end = begin + port_text.size(); + const auto [ptr, error] = std::from_chars(begin, end, parsed_port); + if (error != std::errc{} || ptr != end || parsed_port < kMinPort || parsed_port > max_port) { + std::ostringstream message; + message << "Invalid port in endpoint '" << endpoint << "': port must be between 1 and " << max_port << '.'; + return a2a::core::Error::Validation(message.str()); + } + + return HostPortEndpoint{.host = std::string(endpoint.substr(0, pos)), .port = parsed_port}; +} + +bool SetSocketNonBlocking(int fd) noexcept { +#ifdef _WIN32 + (void)fd; + return true; +#else + const int flags = fcntl(fd, F_GETFL, 0); + return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; +#endif +} + +} // namespace a2a::server diff --git a/src/server/socket_utils.cpp b/src/server/socket_utils.cpp deleted file mode 100644 index 1c9432c..0000000 --- a/src/server/socket_utils.cpp +++ /dev/null @@ -1,22 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 -// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) - -#include "a2a/server/socket_utils.h" - -#ifdef _WIN32 -#include -#else -#include -#endif - -namespace a2a::server { - -void CloseSocketCrossPlatform(int fd) noexcept { -#ifdef _WIN32 - closesocket(fd); -#else - close(fd); -#endif -} - -} // namespace a2a::server diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a4eef05..3fd23f3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -450,6 +450,19 @@ target_link_libraries(server_utils_test gtest_discover_tests(server_utils_test) +add_executable(network_utils_test + unit/network_utils_test.cpp +) + +target_link_libraries(network_utils_test + PRIVATE + a2a::server + a2a::core + GTest::gtest_main +) + +gtest_discover_tests(network_utils_test) + add_executable(grpc_server_transport_test unit/grpc_server_transport_test.cpp ) @@ -725,16 +738,16 @@ target_link_libraries(cpp_grpc_interop_server a2a::proto_generated ) -add_executable(tck_http_sut interop/tck_http_sut.cpp) -target_include_directories(tck_http_sut PRIVATE ${CMAKE_SOURCE_DIR}/tests/support/example_support) -target_link_libraries(tck_http_sut PRIVATE a2a::server a2a::core a2a::proto_generated) +add_executable(tck_sut sut/tck_sut.cpp) +target_include_directories(tck_sut PRIVATE ${CMAKE_SOURCE_DIR}/tests ${CMAKE_SOURCE_DIR}/tests/support/example_support) +target_link_libraries(tck_sut PRIVATE a2a::server a2a::core a2a::proto_generated) if(A2A_ENABLE_POSTGRES_STORE) - target_link_libraries(tck_http_sut + target_link_libraries(tck_sut PRIVATE a2a::store_postgres ) - target_compile_definitions(tck_http_sut + target_compile_definitions(tck_sut PRIVATE A2A_ENABLE_POSTGRES_STORE=1 ) diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index 5477732..df06fbe 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -389,21 +389,81 @@ ScenarioResult RunScenario(const Options& options, const std::string& scenario) return result; } +std::vector SplitCsv(std::string_view value) { + std::vector items; + std::size_t start = 0; + while (start <= value.size()) { + const std::size_t comma = value.find(',', start); + const std::size_t end = comma == std::string_view::npos ? value.size() : comma; + if (end > start) { + items.emplace_back(value.substr(start, end - start)); + } + if (comma == std::string_view::npos) { + break; + } + start = comma + 1U; + } + return items; +} + +bool IsSupportedScenario(std::string_view scenario) { + return std::ranges::find(kScenarios, scenario) != kScenarios.end(); +} + +std::vector SelectedScenarios(const Options& options) { + if (!options.scenarios.empty()) { + return options.scenarios; + } + std::vector scenarios; + scenarios.reserve(kScenarios.size()); + for (const std::string_view scenario : kScenarios) { + scenarios.emplace_back(scenario); + } + return scenarios; +} + +bool ParseScenarioSelection(std::string_view value, Options* options) { + options->scenarios = SplitCsv(value); + if (options->scenarios.empty()) { + std::cerr << "scenario selection must not be empty\n"; + return false; + } + for (const std::string& scenario : options->scenarios) { + if (!IsSupportedScenario(scenario)) { + std::cerr << "unsupported scenario: " << scenario << '\n'; + return false; + } + } + return true; +} + +bool HasValue(int index, int argc) { return index + 1 < argc; } + bool ParseArgs(int argc, char** argv, Options* options) { for (int index = 1; index < argc; ++index) { const std::string_view arg(argv[index]); - if (arg == "--transport" && index + 1 < argc) { - options->transport = argv[++index]; - } else if (arg == "--store-backend" && index + 1 < argc) { - options->store_backend = argv[++index]; - } else if (arg == "--requests" && index + 1 < argc) { - options->requests = std::atoi(argv[++index]); - } else if (arg == "--concurrency" && index + 1 < argc) { - options->concurrency = std::atoi(argv[++index]); - } else if (arg == "--warmup-seconds" && index + 1 < argc) { - options->warmup_seconds = std::atof(argv[++index]); - } else if (arg == "--duration-seconds" && index + 1 < argc) { - options->duration_seconds = std::atof(argv[++index]); + if (!HasValue(index, argc)) { + std::cerr << "unknown or incomplete argument: " << arg << '\n'; + return false; + } + const char* raw_value = argv[++index]; + const std::string_view value(raw_value); + if (arg == "--transport") { + options->transport = value; + } else if (arg == "--store-backend") { + options->store_backend = value; + } else if (arg == "--requests") { + options->requests = std::atoi(raw_value); + } else if (arg == "--concurrency") { + options->concurrency = std::atoi(raw_value); + } else if (arg == "--warmup-seconds") { + options->warmup_seconds = std::atof(raw_value); + } else if (arg == "--duration-seconds") { + options->duration_seconds = std::atof(raw_value); + } else if (arg == "--scenarios") { + if (!ParseScenarioSelection(value, options)) { + return false; + } } else { std::cerr << "unknown or incomplete argument: " << arg << '\n'; return false; @@ -438,12 +498,7 @@ google::protobuf::Struct BuildResultObject(const Options& options, const Scenari SetNumberField(&object, "duration_seconds", options.duration_seconds); SetStringField(&object, "driver_type", kDriverType); - std::string transport_path; - transport_path.reserve(kSdkTransportPathPrefix.size() + options.transport.size() + kServerDispatchSuffix.size()); - transport_path.append(kSdkTransportPathPrefix); - transport_path.append(options.transport); - transport_path.append(kServerDispatchSuffix); - SetStringField(&object, "transport_path", transport_path); + SetStringField(&object, "transport_path", "in_process"); google::protobuf::Struct latency; SetNumberField(&latency, "p50", Percentile(result.latencies, kP50)); @@ -481,8 +536,8 @@ int main(int argc, char** argv) { } std::cout << "[\n"; bool first = true; - for (const std::string_view scenario : kScenarios) { - WriteResultJson(options, RunScenario(options, std::string(scenario)), first); + for (const std::string& scenario : SelectedScenarios(options)) { + WriteResultJson(options, RunScenario(options, scenario), first); first = false; } std::cout << "\n]\n"; diff --git a/tests/performance/a2a_performance_driver.h b/tests/performance/a2a_performance_driver.h index 49089d2..95740b7 100644 --- a/tests/performance/a2a_performance_driver.h +++ b/tests/performance/a2a_performance_driver.h @@ -33,8 +33,6 @@ constexpr double kP90 = 90.0; constexpr double kP95 = 95.0; constexpr double kP99 = 99.0; constexpr std::size_t kIdReserveSlack = 16U; -constexpr std::string_view kSdkTransportPathPrefix = "sdk_"; -constexpr std::string_view kServerDispatchSuffix = "_server_dispatch"; constexpr char kPostgresDsnEnv[] = "A2A_TEST_POSTGRES_DSN"; constexpr std::string_view kPerfSchemaPrefix = "a2a_perf_"; constexpr std::string_view kMessageText = "hello"; @@ -88,6 +86,7 @@ struct Options final { int concurrency = kDefaultConcurrency; double warmup_seconds = 0.0; double duration_seconds = 0.0; + std::vector scenarios; }; struct ScenarioResult final { diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py index 9b57cf8..db31daa 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -23,15 +23,16 @@ def test_writes_reports_for_selected_matrix(self): ], cwd=ROOT, check=True) report_dir = Path(temp_dir) payload = json.loads((report_dir / "results.json").read_text(encoding="utf-8")) - self.assertEqual(18, len(payload["results"])) - self.assertEqual("cpp_sdk_in_process", payload["results"][0]["driver_type"]) - ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["transport"], result["concurrency"])) + self.assertEqual(25, len(payload["results"])) + self.assertEqual({"cpp_sdk_in_process", "wire_tck_sut"}, {result["driver_type"] for result in payload["results"]}) + self.assertIn("wire_grpc", {result["transport_path"] for result in payload["results"]}) + ordered = sorted(payload["results"], key=lambda result: (result["scenario"], result["store_backend"], result["driver_type"], result["transport_path"], result["transport"], result["concurrency"])) self.assertEqual(ordered, payload["results"]) self.assertTrue((report_dir / "results.csv").exists()) summary = (report_dir / "summary.md").read_text(encoding="utf-8") self.assertIn("A2A performance test summary", summary) self.assertIn("| Scenario | Rows | Operations | Success | Errors | Avg ops/sec | Worst p95 ms | Worst max ms |", summary) - self.assertIn("| Scenario | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) + self.assertIn("| Scenario | Driver | Path | Transport | Store | Concurrency | Success | Errors | Ops/sec | p50 ms | p95 ms | p99 ms | Max ms |", summary) def test_rejects_unknown_transport(self): with tempfile.TemporaryDirectory() as temp_dir: diff --git a/tests/interop/tck_http_sut.cpp b/tests/sut/tck_sut.cpp similarity index 71% rename from tests/interop/tck_http_sut.cpp rename to tests/sut/tck_sut.cpp index 047c4d2..67b75e5 100644 --- a/tests/interop/tck_http_sut.cpp +++ b/tests/sut/tck_sut.cpp @@ -15,7 +15,6 @@ #include #endif -#include #include #include #include @@ -26,7 +25,6 @@ #include #include #include -#include #include #include #include @@ -40,30 +38,22 @@ #include "a2a/server/grpc_server_transport.h" #include "a2a/server/http_adapter.h" #include "a2a/server/json_rpc_server_transport.h" +#include "a2a/server/network_utils.h" #include "a2a/server/rest_server_transport.h" -#include "a2a/server/socket_utils.h" #include "a2a/server/stores/store_factory.h" #include "a2a/server/transport_mux.h" #include "example_support.h" +#include "sut/tck_sut.h" namespace { + +using namespace a2a::tests::sut; + constexpr int kListenBacklog = 128; -constexpr int kDefaultPort = 50061; -constexpr int kGrpcPortOffset = 1; constexpr int kReuseAddress = 1; +constexpr int kMaxHttpPort = 65534; +constexpr int kAcceptRetryDelayMillis = 25; constexpr std::time_t kAgentCardLastModifiedUnix = 1704067200; -constexpr std::string_view kRestApiBasePath = "/a2a"; -constexpr std::string_view kTckRequiredExtensionUri = "urn:a2a:tck:required-extension"; -constexpr std::string_view kPostgresBackend = "postgres"; -constexpr std::string_view kInMemoryBackend = "inmemory"; -constexpr const char* kStoreBackendEnv = "A2A_TCK_STORE_BACKEND"; -constexpr const char* kPostgresDsnEnv = "A2A_TCK_POSTGRES_DSN"; -constexpr const char* kPostgresSchemaEnv = "A2A_TCK_POSTGRES_SCHEMA"; -constexpr const char* kExtendedCardModeEnv = "A2A_TCK_EXTENDED_AGENT_CARD_MODE"; -constexpr std::string_view kExtendedCardModeConfigured = "configured"; -constexpr std::string_view kExtendedCardModeDeclaredOnly = "declared_only"; -constexpr std::string_view kExtendedCardModeDisabled = "disabled"; -constexpr std::string_view kDefaultPostgresSchema = "public"; constexpr std::string_view kMissingPostgresDsnMessage = "A2A_TCK_POSTGRES_DSN must be set when A2A_TCK_STORE_BACKEND=postgres"; constexpr std::string_view kUnsupportedStoreBackendMessage = "Unsupported A2A_TCK_STORE_BACKEND: "; @@ -177,17 +167,19 @@ void HandleHttpConnection(int fd, const a2a::server::TransportMux& mux, HttpConn a2a::server::CloseSocketCrossPlatform(fd); } -} // namespace - -int main(int argc, char** argv) { - const std::string endpoint = (argc > 1) ? argv[1] : "127.0.0.1:" + std::to_string(kDefaultPort); - const auto pos = endpoint.find(':'); - if (pos == std::string::npos) { +int RunTckSut(int argc, char** argv) { + const std::string endpoint = (argc > 1) ? argv[1] : std::string(kDefaultHost) + ":" + std::to_string(kDefaultPort); + auto parsed_endpoint = a2a::server::ParseHostPortEndpoint(endpoint, kMaxHttpPort); + if (!parsed_endpoint.ok()) { + std::cerr << parsed_endpoint.error().message() << '\n'; return 1; } - const std::string host = endpoint.substr(0, pos); - const int port = std::stoi(endpoint.substr(pos + 1)); + const std::string& host = parsed_endpoint.value().host; + const int port = parsed_endpoint.value().port; const int grpc_port = port + kGrpcPortOffset; + const SutEndpoints endpoints{.rest_url = a2a::server::BuildHttpUrl(host, port, kRestApiBasePath), + .json_rpc_url = a2a::server::BuildHttpUrl(host, port, kJsonRpcPath), + .grpc_url = host + ":" + std::to_string(grpc_port)}; std::signal(SIGINT, SignalHandler); std::signal(SIGTERM, SignalHandler); @@ -203,14 +195,13 @@ int main(int argc, char** argv) { const bool declares_extended_card = extended_card_mode != kExtendedCardModeDisabled; const bool configures_extended_card = extended_card_mode == kExtendedCardModeConfigured; - auto agent_card = a2a::core::AgentCardBuilder::ConformancePreset( - {.rest_url = "http://localhost:" + std::to_string(port) + std::string(kRestApiBasePath), - .json_rpc_url = "http://localhost:" + std::to_string(port) + "/rpc", - .grpc_url = "localhost:" + std::to_string(grpc_port)}, - "TCK HTTP SUT", "0.1.0", "Conformance-focused local SUT for A2A") - .WithPushNotifications(true) - .WithExtendedAgentCard(declares_extended_card) - .Build(); + auto agent_card = + a2a::core::AgentCardBuilder::ConformancePreset( + {.rest_url = endpoints.rest_url, .json_rpc_url = endpoints.json_rpc_url, .grpc_url = endpoints.grpc_url}, + "TCK SUT", "0.1.0", "Conformance-focused local SUT for A2A") + .WithPushNotifications(true) + .WithExtendedAgentCard(declares_extended_card) + .Build(); std::optional extended_agent_card; if (configures_extended_card) { @@ -220,7 +211,7 @@ int main(int argc, char** argv) { auto store_bundle = CreateStoreBundleFromEnvironment(); if (!store_bundle.ok()) { - std::cerr << store_bundle.error().message() << '\n'; + std::cerr << "Failed to create TCK SUT store bundle: " << store_bundle.error().message() << '\n'; return 1; } @@ -231,22 +222,22 @@ int main(int argc, char** argv) { auto agent_card_provider = std::make_shared(extended_agent_card); a2a::server::Dispatcher dispatcher(&executor, agent_card_provider); a2a::server::GrpcServerTransportOptions grpc_options; - grpc_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + grpc_options.required_extensions = {std::string(kRequiredExtensionUri)}; a2a::server::GrpcServerTransport grpc(&dispatcher, std::move(grpc_options)); a2a::server::RestServerTransportOptions rest_options; rest_options.rest_api_base_path = std::string(kRestApiBasePath); rest_options.include_legacy_transport_fields = false; - rest_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + rest_options.required_extensions = {std::string(kRequiredExtensionUri)}; rest_options.agent_card_cache_settings = a2a::server::RestServerTransportOptions::AgentCardCacheSettings{ .cache_control = "public, max-age=300", .last_modified = std::chrono::system_clock::from_time_t(kAgentCardLastModifiedUnix)}; a2a::server::RestServerTransport rest(&dispatcher, agent_card, std::move(rest_options)); a2a::server::JsonRpcServerTransportOptions jsonrpc_options; - jsonrpc_options.rpc_path = "/rpc"; + jsonrpc_options.rpc_path = std::string(kJsonRpcPath); jsonrpc_options.require_version_header = false; - jsonrpc_options.required_extensions = {std::string(kTckRequiredExtensionUri)}; + jsonrpc_options.required_extensions = {std::string(kRequiredExtensionUri)}; a2a::server::JsonRpcServerTransport jsonrpc(&dispatcher, std::move(jsonrpc_options)); #ifdef _WIN32 @@ -257,16 +248,39 @@ int main(int argc, char** argv) { #endif int server_fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (server_fd < 0) { + std::cerr << "Failed to create HTTP listening socket: " << std::strerror(errno) << '\n'; + return 1; + } int opt = kReuseAddress; - setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&opt), sizeof(opt)); + if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&opt), sizeof(opt)) != 0) { + std::cerr << "Failed to configure HTTP listening socket reuse: " << std::strerror(errno) << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } sockaddr_in addr{}; addr.sin_family = AF_INET; addr.sin_port = htons(static_cast(port)); - inet_pton(AF_INET, host.c_str(), &addr.sin_addr); + if (inet_pton(AF_INET, host.c_str(), &addr.sin_addr) != 1) { + std::cerr << "Invalid IPv4 host for TCK SUT HTTP listener: " << host << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } if (bind(server_fd, reinterpret_cast(&addr), sizeof(addr)) != 0) { + std::cerr << "Failed to bind TCK SUT HTTP listener on " << host << ':' << port << ": " << std::strerror(errno) + << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } if (listen(server_fd, kListenBacklog) != 0) { + std::cerr << "Failed to listen on TCK SUT HTTP endpoint " << host << ':' << port << ": " << std::strerror(errno) + << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); + return 1; + } + if (!a2a::server::SetSocketNonBlocking(server_fd)) { + std::cerr << "Failed to configure non-blocking TCK SUT HTTP listener: " << std::strerror(errno) << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } @@ -275,12 +289,14 @@ int main(int argc, char** argv) { grpc_builder.RegisterService(&grpc); std::unique_ptr grpc_server = grpc_builder.BuildAndStart(); if (!grpc_server) { + std::cerr << "Failed to start TCK SUT gRPC server on " << host << ':' << grpc_port << '\n'; + a2a::server::CloseSocketCrossPlatform(server_fd); return 1; } a2a::server::TransportMux mux( {.normalization_policy = a2a::server::TransportMux::PathNormalizationPolicy::kRootToDefaultPath, - .default_path = "/rpc"}); + .default_path = std::string(kJsonRpcPath)}); mux.RegisterJsonRpcRoute(jsonrpc); mux.RegisterRestRoute(rest); @@ -291,7 +307,12 @@ int main(int argc, char** argv) { socklen_t len = sizeof(client); const int fd = accept(server_fd, reinterpret_cast(&client), &len); if (fd < 0) { - continue; + if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) { + std::this_thread::sleep_for(std::chrono::milliseconds(kAcceptRetryDelayMillis)); + continue; + } + std::cerr << "TCK SUT HTTP accept failed: " << std::strerror(errno) << '\n'; + break; } connection_registry.Add(fd); connection_threads.emplace_back(HandleHttpConnection, fd, std::cref(mux), std::ref(connection_registry)); @@ -310,3 +331,14 @@ int main(int argc, char** argv) { #endif return 0; } + +} // namespace + +int main(int argc, char** argv) noexcept { + try { + return RunTckSut(argc, argv); + } catch (const std::exception& ex) { + std::cerr << "Unhandled TCK SUT exception: " << ex.what() << '\n'; + return 1; + } +} diff --git a/tests/sut/tck_sut.h b/tests/sut/tck_sut.h new file mode 100644 index 0000000..4b33c51 --- /dev/null +++ b/tests/sut/tck_sut.h @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#pragma once + +#include +#include + +namespace a2a::tests::sut { + +constexpr std::string_view kDefaultHost = "127.0.0.1"; +constexpr int kDefaultPort = 50061; +constexpr int kGrpcPortOffset = 1; +constexpr std::string_view kRestApiBasePath = "/a2a"; +constexpr std::string_view kJsonRpcPath = "/rpc"; +constexpr std::string_view kRequiredExtensionUri = "urn:a2a:tck:required-extension"; +constexpr std::string_view kPostgresBackend = "postgres"; +constexpr std::string_view kInMemoryBackend = "inmemory"; +constexpr std::string_view kDefaultPostgresSchema = "public"; +constexpr std::string_view kExtendedCardModeConfigured = "configured"; +constexpr std::string_view kExtendedCardModeDeclaredOnly = "declared_only"; +constexpr std::string_view kExtendedCardModeDisabled = "disabled"; +constexpr char kStoreBackendEnv[] = "A2A_TCK_STORE_BACKEND"; +constexpr char kPostgresDsnEnv[] = "A2A_TCK_POSTGRES_DSN"; +constexpr char kPostgresSchemaEnv[] = "A2A_TCK_POSTGRES_SCHEMA"; +constexpr char kExtendedCardModeEnv[] = "A2A_TCK_EXTENDED_AGENT_CARD_MODE"; + +struct SutConfig final { + std::string host = std::string(kDefaultHost); + int port = kDefaultPort; + int grpc_port = kDefaultPort + kGrpcPortOffset; +}; + +struct SutEndpoints final { + std::string rest_url; + std::string json_rpc_url; + std::string grpc_url; +}; + +} // namespace a2a::tests::sut diff --git a/tests/unit/network_utils_test.cpp b/tests/unit/network_utils_test.cpp new file mode 100644 index 0000000..be7d1c2 --- /dev/null +++ b/tests/unit/network_utils_test.cpp @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 Vladimir Pavlov (https://github.com/MisterVVP) + +#include "a2a/server/network_utils.h" + +#include + +#include +#include + +namespace { + +constexpr int kInvalidMaxPort = 0; +constexpr int kDefaultTestPort = 50061; +constexpr int kMaxTestPort = 65534; +constexpr std::string_view kLocalhost = "127.0.0.1"; +constexpr std::string_view kRestPath = "/a2a"; +constexpr std::string_view kExpectedRestUrl = "http://127.0.0.1:50061/a2a"; +constexpr std::string_view kValidEndpoint = "127.0.0.1:50061"; +constexpr std::string_view kMissingPortEndpoint = "127.0.0.1"; +constexpr std::string_view kInvalidPortEndpoint = "127.0.0.1:not-a-port"; +constexpr std::string_view kTooLargePortEndpoint = "127.0.0.1:65535"; +constexpr std::string_view kExpectedEndpointFormatText = "Expected :"; +constexpr std::string_view kExpectedTestPortRangeText = "port must be between 1 and 65534"; +constexpr std::string_view kExpectedMaxPortRangeText = "max_port must be between 1 and 65535"; + +TEST(NetworkUtilsTest, BuildHttpUrlUsesHostPortAndPath) { + EXPECT_EQ(a2a::server::BuildHttpUrl(kLocalhost, kDefaultTestPort, kRestPath), kExpectedRestUrl); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointAcceptsValidEndpoint) { + auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kMaxTestPort); + ASSERT_TRUE(parsed.ok()) << parsed.error().message(); + EXPECT_EQ(parsed.value().host, kLocalhost); + EXPECT_EQ(parsed.value().port, kDefaultTestPort); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsMissingPort) { + auto parsed = a2a::server::ParseHostPortEndpoint(kMissingPortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedEndpointFormatText), std::string::npos); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsInvalidPort) { + auto parsed = a2a::server::ParseHostPortEndpoint(kInvalidPortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsPortAboveConfiguredMaximum) { + auto parsed = a2a::server::ParseHostPortEndpoint(kTooLargePortEndpoint, kMaxTestPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedTestPortRangeText), std::string::npos); +} + +TEST(NetworkUtilsTest, ParseHostPortEndpointRejectsInvalidConfiguredMaximum) { + auto parsed = a2a::server::ParseHostPortEndpoint(kValidEndpoint, kInvalidMaxPort); + ASSERT_FALSE(parsed.ok()); + EXPECT_NE(parsed.error().message().find(kExpectedMaxPortRangeText), std::string::npos); +} + +} // namespace