From c6e57149a1fbdb2b40e4fee5a9062327a8466460 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:11:54 +0300 Subject: [PATCH 1/5] Add SDK-backed performance driver --- .github/workflows/ci.yml | 4 +- docs/performance-testing.md | 109 +++--- scripts/performance_runner.py | 153 +++----- tests/CMakeLists.txt | 16 + tests/performance/a2a_performance_driver.cpp | 364 +++++++++++++++++++ tests/scripts/performance_runner_test.py | 1 + 6 files changed, 494 insertions(+), 153 deletions(-) create mode 100644 tests/performance/a2a_performance_driver.cpp diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8cdadf3..f3f916c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,8 +111,8 @@ jobs: - name: Run report-only performance smoke suite env: A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json - A2A_PERF_STORE_BACKENDS: inmemory,postgres - A2A_PERF_REQUESTS: 100000 + A2A_PERF_STORE_BACKENDS: inmemory + 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..66af6b9 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,69 @@ 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. The runner +and report schema keep the `postgres` selection, but the checked-in driver fails +explicitly if PostgreSQL is requested from a build that does not provide a +PostgreSQL-backed performance driver. It must not be interpreted as silently +using in-memory storage for PostgreSQL rows. PostgreSQL benchmark runs should be +executed from a PostgreSQL-enabled build and configured with the same local DSN +style used by the repository store tests (`A2A_TEST_POSTGRES_DSN`). + +## 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_TRANSPORTS=grpc,jsonrpc,http_json \ A2A_PERF_STORE_BACKENDS=inmemory \ -A2A_PERF_REQUESTS=100 \ -A2A_PERF_CONCURRENCY=1,4 \ -A2A_PERF_WARMUP_SECONDS=0 \ +A2A_PERF_REQUESTS=10000 \ +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..040d136 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 @@ -47,10 +42,11 @@ ) DEFAULT_REQUESTS = 1_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,51 @@ 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() -> 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"] + 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() + 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 +157,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 +274,7 @@ 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)] 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..2d2c58f 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -777,3 +777,19 @@ 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 +) diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp new file mode 100644 index 0000000..0a71e82 --- /dev/null +++ b/tests/performance/a2a_performance_driver.cpp @@ -0,0 +1,364 @@ +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "a2a/server/push_notification_delivery.h" +#include "a2a/server/request_context.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 int kDefaultRequests = 1000; +constexpr int kDefaultConcurrency = 1; +constexpr int kListPageSize = 10; +constexpr int kPushConfigFanout = 8; + +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 = 200, .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; +}; + +std::string JsonEscape(std::string_view value) { + std::ostringstream escaped; + for (const char character : value) { + switch (character) { + case '"': + escaped << "\\\""; + break; + case '\\': + escaped << "\\\\"; + break; + case '\n': + escaped << "\\n"; + break; + default: + escaped << character; + break; + } + } + return escaped.str(); +} + +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: + ScenarioHarness() { + options_.push_delivery = &delivery_; + executor_ = std::make_unique(std::move(options_)); + existing_task_id_ = SeedTask("existing-seed"); + subscribe_task_id_ = SeedTask("subscribe-seed"); + } + + bool Execute(std::string_view scenario, int index) { + std::lock_guard lock(mutex_); + 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" || scenario == "ListTasks_WithPagination") { + a2a::server::ListTasksRequest request; + if (scenario == "ListTasks_WithPagination") request.page_size = kListPageSize; + return executor_->ListTasks(request, context).ok(); + } + 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(); + } + if (scenario == "SendStreamingMessage_FiniteStream") { + auto stream = executor_->SendStreamingMessage(MakeSendRequest(BuildId("stream", index)), context); + return stream.ok() && DrainStream(stream.value().get()); + } + if (scenario == "SubscribeToTask_FirstEventLatency") return SubscribeOnce(context); + if (scenario == "SubscribeToTask_MultiSubscriber") + return SubscribeOnce(context) && SubscribeOnce(context) && SubscribeOnce(context); + if (scenario == "SubscribeToTask_TerminalCompletionLatency") { + auto stream = executor_->SendStreamingMessage(MakeSendRequest(BuildId("terminal", index)), context); + return stream.ok() && DrainStream(stream.value().get()); + } + if (scenario == "SubscribeToTask_DisconnectOneSubscriber") return SubscribeOnce(context) && SubscribeOnce(context); + if (scenario == "PushConfig_Create") + return executor_ + ->CreateTaskPushNotificationConfig(MakePushConfig(existing_task_id_, BuildId("cfg-create", index)), context) + .ok(); + if (scenario == "PushConfig_Get") { + 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(); + } + if (scenario == "PushConfig_List") { + SeedPushConfig(existing_task_id_, "cfg-list"); + lf::a2a::v1::ListTaskPushNotificationConfigsRequest request; + request.set_task_id(existing_task_id_); + return executor_->ListTaskPushNotificationConfigs(request, context).ok(); + } + if (scenario == "PushConfig_Delete") { + const std::string config_id = BuildId("cfg-delete", index); + SeedPushConfig(existing_task_id_, config_id); + lf::a2a::v1::DeleteTaskPushNotificationConfigRequest request; + request.set_task_id(existing_task_id_); + request.set_id(config_id); + return executor_->DeleteTaskPushNotificationConfig(request, context).ok(); + } + if (scenario == "PushNotify_ManyConfigsOneTaskUpdate" || scenario == "PushDelivery_CallbackLatency") { + 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(); + } + return false; + } + + private: + static std::string BuildId(std::string_view prefix, int index) { + std::string value; + value.reserve(prefix.size() + 16U); + 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::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; + 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 (static_cast(futures.size()) >= 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, 0.000001); + std::sort(result.latencies.begin(), result.latencies.end()); + 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 WriteResultJson(const Options& options, const ScenarioResult& result, bool first) { + const double p50 = Percentile(result.latencies, 50.0); + const double p90 = Percentile(result.latencies, 90.0); + const double p95 = Percentile(result.latencies, 95.0); + const double p99 = Percentile(result.latencies, 99.0); + const double max = result.latencies.empty() ? 0.0 : result.latencies.back(); + if (!first) std::cout << ",\n"; + std::cout << " {\"scenario\":\"" << JsonEscape(result.scenario) << "\",\"transport\":\"" + << JsonEscape(options.transport) << "\",\"store_backend\":\"" << JsonEscape(options.store_backend) + << "\",\"concurrency\":" << options.concurrency << ",\"operations\":" << result.operations + << ",\"success\":" << result.success << ",\"errors\":" << result.errors + << ",\"throughput_ops_per_sec\":" << std::fixed << std::setprecision(6) << result.throughput + << ",\"warmup_seconds\":" << options.warmup_seconds << ",\"duration_seconds\":" << options.duration_seconds + << ",\"driver_type\":\"" << kDriverType << "\",\"transport_path\":\"sdk_" << JsonEscape(options.transport) + << "_server_dispatch\",\"latency_ms\":{\"p50\":" << p50 << ",\"p90\":" << p90 << ",\"p95\":" << p95 + << ",\"p99\":" << p99 << ",\"max\":" << max << "}}"; +} +} // namespace + +int main(int argc, char** argv) { + Options options; + if (!ParseArgs(argc, argv, &options)) return 2; + if (options.store_backend == kPostgresStore) { + std::cerr << "postgres performance backend requires a PostgreSQL-enabled driver build and A2A_TEST_POSTGRES_DSN; " + "this binary was built without that backend\n"; + return 3; + } + if (options.store_backend != kInMemoryStore || + (options.transport != kGrpcTransport && options.transport != kJsonRpcTransport && + options.transport != kHttpJsonTransport)) { + std::cerr << "unsupported transport or store backend\n"; + return 2; + } + 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..08e9fba 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -24,6 +24,7 @@ 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"]) 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 c907152d9461f6921179aeea0ddc5e5e4c4e513f Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Wed, 8 Jul 2026 22:24:14 +0300 Subject: [PATCH 2/5] Add SDK-backed performance scenarios and C++ performance driver ### 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. ### Testing - Built the performance driver with `cmake -S . -B build/perf-test -DA2A_ENABLE_TESTING=ON` and `cmake --build build/perf-test --target a2a_performance_driver`, and the driver linked successfully. - Ran a local smoke invocation `A2A_PERF_BUILD_DIR=build/perf-test ./scripts/run_performance_tests.sh --transports grpc,jsonrpc,http_json --store-backends inmemory --requests 1 --concurrency 1 --warmup-seconds 0 --report-dir /tmp/perf-all` which produced `results.json`, `results.csv`, and `summary.md` successfully. - Executed the Python runner script unit test with `A2A_PERF_BUILD_DIR=build/perf-test python3 tests/scripts/performance_runner_test.py`, and the test passed. - Ran the repository validation `./scripts/verify_changes.sh` (format, build, `ctest`, and `clang-tidy` profile steps used by CI) and it completed successfully on the smoke change set. --- .github/workflows/ci.yml | 4 +- tests/performance/a2a_performance_driver.cpp | 234 +++++++++++++------ 2 files changed, 169 insertions(+), 69 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f3f916c..5dd4359 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -108,7 +108,9 @@ jobs: runs-on: ubuntu-24.04 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 diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index 0a71e82..b80550d 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -8,7 +8,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -29,10 +31,18 @@ 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 int kUnavailableExitCode = 3; +constexpr int kOutputPrecision = 6; +constexpr std::size_t kIdReserveSlack = 16U; const std::vector& Scenarios() { static const std::vector scenarios = {"SendMessage_CreateTask", @@ -62,7 +72,7 @@ class RecordingPushDelivery final : public a2a::server::PushNotificationDelivery (void)request; std::lock_guard lock(mutex_); ++deliveries_; - return a2a::server::PushDeliveryResult{.http_status = 200, .error_message = {}}; + return a2a::server::PushDeliveryResult{.http_status = kHttpStatusOk, .error_message = {}}; } private: @@ -110,7 +120,9 @@ std::string JsonEscape(std::string_view value) { } double Percentile(const std::vector& sorted_values, double percentile) { - if (sorted_values.empty()) return 0.0; + 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)]; @@ -119,7 +131,9 @@ double Percentile(const std::vector& sorted_values, double percentile) { 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)); + if (!task_id.empty()) { + request.mutable_message()->set_task_id(std::string(task_id)); + } request.mutable_message()->add_parts()->set_text("hello"); return request; } @@ -144,8 +158,23 @@ class ScenarioHarness final { bool Execute(std::string_view scenario, int index) { std::lock_guard lock(mutex_); a2a::server::RequestContext context; - if (scenario == "SendMessage_CreateTask") + 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: + 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_); @@ -157,67 +186,122 @@ class ScenarioHarness final { request.set_id(task_id); return executor_->CancelTask(request, context).ok(); } - if (scenario == "ListTasks_NoPagination" || scenario == "ListTasks_WithPagination") { - a2a::server::ListTasksRequest request; - if (scenario == "ListTasks_WithPagination") request.page_size = kListPageSize; - return executor_->ListTasks(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") + 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") { - auto stream = executor_->SendStreamingMessage(MakeSendRequest(BuildId("stream", index)), context); - return stream.ok() && DrainStream(stream.value().get()); + return SendAndDrainStream(BuildId("stream", index), context); + } + if (scenario == "SubscribeToTask_FirstEventLatency") { + return SubscribeOnce(context); + } + if (scenario == "SubscribeToTask_MultiSubscriber") { + return SubscribeMany(kMultiSubscriberCount, context); } - if (scenario == "SubscribeToTask_FirstEventLatency") return SubscribeOnce(context); - if (scenario == "SubscribeToTask_MultiSubscriber") - return SubscribeOnce(context) && SubscribeOnce(context) && SubscribeOnce(context); if (scenario == "SubscribeToTask_TerminalCompletionLatency") { - auto stream = executor_->SendStreamingMessage(MakeSendRequest(BuildId("terminal", index)), context); - return stream.ok() && DrainStream(stream.value().get()); + return SendAndDrainStream(BuildId("terminal", index), context); + } + if (scenario == "SubscribeToTask_DisconnectOneSubscriber") { + return SubscribeMany(kDisconnectSubscriberCount, context); } - if (scenario == "SubscribeToTask_DisconnectOneSubscriber") return SubscribeOnce(context) && SubscribeOnce(context); - if (scenario == "PushConfig_Create") + 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") { - 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(); + return GetPushConfig(context); } if (scenario == "PushConfig_List") { - SeedPushConfig(existing_task_id_, "cfg-list"); - lf::a2a::v1::ListTaskPushNotificationConfigsRequest request; - request.set_task_id(existing_task_id_); - return executor_->ListTaskPushNotificationConfigs(request, context).ok(); + return ListPushConfigs(context); } if (scenario == "PushConfig_Delete") { - const std::string config_id = BuildId("cfg-delete", index); - SeedPushConfig(existing_task_id_, config_id); - lf::a2a::v1::DeleteTaskPushNotificationConfigRequest request; - request.set_task_id(existing_task_id_); - request.set_id(config_id); - return executor_->DeleteTaskPushNotificationConfig(request, context).ok(); + return DeletePushConfig(BuildId("cfg-delete", index), context); } - if (scenario == "PushNotify_ManyConfigsOneTaskUpdate" || scenario == "PushDelivery_CallbackLatency") { - 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(); + if (scenario == "PushNotify_ManyConfigsOneTaskUpdate") { + return NotifyPushConfigs(index, context); } - return false; + 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(); } - private: static std::string BuildId(std::string_view prefix, int index) { std::string value; - value.reserve(prefix.size() + 16U); + value.reserve(prefix.size() + kIdReserveSlack); value.append(prefix); value.push_back('-'); value.append(std::to_string(index)); @@ -226,7 +310,9 @@ class ScenarioHarness final { 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(); + 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) { @@ -242,8 +328,12 @@ class ScenarioHarness final { static bool DrainStream(a2a::server::ServerStreamSession* stream) { for (;;) { auto next = stream->Next(); - if (!next.ok()) return false; - if (!next.value().has_value()) return true; + if (!next.ok()) { + return false; + } + if (!next.value().has_value()) { + return true; + } } } @@ -259,7 +349,9 @@ ScenarioResult RunScenario(const Options& options, const std::string& scenario) ScenarioHarness harness; 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++); + while (std::chrono::steady_clock::now() < warmup_end) { + (void)harness.Execute(scenario, warmup_index++); + } ScenarioResult result; result.scenario = scenario; result.operations = options.requests; @@ -285,34 +377,36 @@ ScenarioResult RunScenario(const Options& options, const std::string& scenario) 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 (static_cast(futures.size()) >= options.concurrency) { + if (futures.size() >= static_cast(options.concurrency)) { futures.front().get(); futures.erase(futures.begin()); } } - for (auto& future : futures) future.get(); + 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, 0.000001); - std::sort(result.latencies.begin(), result.latencies.end()); + 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) + if (arg == "--transport" && index + 1 < argc) { options->transport = argv[++index]; - else if (arg == "--store-backend" && index + 1 < argc) + } else if (arg == "--store-backend" && index + 1 < argc) { options->store_backend = argv[++index]; - else if (arg == "--requests" && index + 1 < argc) + } else if (arg == "--requests" && index + 1 < argc) { options->requests = std::atoi(argv[++index]); - else if (arg == "--concurrency" && index + 1 < argc) + } else if (arg == "--concurrency" && index + 1 < argc) { options->concurrency = std::atoi(argv[++index]); - else if (arg == "--warmup-seconds" && index + 1 < argc) + } else if (arg == "--warmup-seconds" && index + 1 < argc) { options->warmup_seconds = std::atof(argv[++index]); - else if (arg == "--duration-seconds" && index + 1 < argc) + } else if (arg == "--duration-seconds" && index + 1 < argc) { options->duration_seconds = std::atof(argv[++index]); - else { + } else { std::cerr << "unknown or incomplete argument: " << arg << '\n'; return false; } @@ -326,32 +420,36 @@ void WriteResultJson(const Options& options, const ScenarioResult& result, bool const double p95 = Percentile(result.latencies, 95.0); const double p99 = Percentile(result.latencies, 99.0); const double max = result.latencies.empty() ? 0.0 : result.latencies.back(); - if (!first) std::cout << ",\n"; - std::cout << " {\"scenario\":\"" << JsonEscape(result.scenario) << "\",\"transport\":\"" - << JsonEscape(options.transport) << "\",\"store_backend\":\"" << JsonEscape(options.store_backend) - << "\",\"concurrency\":" << options.concurrency << ",\"operations\":" << result.operations - << ",\"success\":" << result.success << ",\"errors\":" << result.errors - << ",\"throughput_ops_per_sec\":" << std::fixed << std::setprecision(6) << result.throughput - << ",\"warmup_seconds\":" << options.warmup_seconds << ",\"duration_seconds\":" << options.duration_seconds - << ",\"driver_type\":\"" << kDriverType << "\",\"transport_path\":\"sdk_" << JsonEscape(options.transport) - << "_server_dispatch\",\"latency_ms\":{\"p50\":" << p50 << ",\"p90\":" << p90 << ",\"p95\":" << p95 - << ",\"p99\":" << p99 << ",\"max\":" << max << "}}"; + if (!first) { + std::cout << ",\n"; + } + std::cout << R"( {"scenario":")" << JsonEscape(result.scenario) << R"(","transport":")" + << JsonEscape(options.transport) << R"(","store_backend":")" << JsonEscape(options.store_backend) + << R"(","concurrency":)" << options.concurrency << R"(,"operations":)" << result.operations + << R"(,"success":)" << result.success << R"(,"errors":)" << result.errors << R"(,"throughput_ops_per_sec":)" + << std::fixed << std::setprecision(kOutputPrecision) << result.throughput << R"(,"warmup_seconds":)" + << options.warmup_seconds << R"(,"duration_seconds":)" << options.duration_seconds << R"(,"driver_type":")" + << kDriverType << R"(","transport_path":"sdk_)" << JsonEscape(options.transport) + << R"(_server_dispatch","latency_ms":{"p50":)" << p50 << R"(,"p90":)" << p90 << R"(,"p95":)" << p95 + << R"(,"p99":)" << p99 << R"(,"max":)" << max << "}}"; } } // namespace int main(int argc, char** argv) { Options options; - if (!ParseArgs(argc, argv, &options)) return 2; + if (!ParseArgs(argc, argv, &options)) { + return kUsageExitCode; + } if (options.store_backend == kPostgresStore) { std::cerr << "postgres performance backend requires a PostgreSQL-enabled driver build and A2A_TEST_POSTGRES_DSN; " "this binary was built without that backend\n"; - return 3; + return kUnavailableExitCode; } if (options.store_backend != kInMemoryStore || (options.transport != kGrpcTransport && options.transport != kJsonRpcTransport && options.transport != kHttpJsonTransport)) { std::cerr << "unsupported transport or store backend\n"; - return 2; + return kUsageExitCode; } std::cout << "[\n"; bool first = true; From fdbdaade3f386cfedee4fc9382f8ad4fd855c403 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:18:58 +0300 Subject: [PATCH 3/5] Add C++ SDK-backed performance driver; use it from runner, docs, and CI ### Motivation - Move measured operations out of a pure-Python harness and into a C++ SDK-backed in-process driver to exercise real SDK code paths and produce stable, machine-readable performance artifacts. - Provide a fall-back auto-build and explicit driver selection so CI and local runs can reuse an existing build or build the driver on demand. - Keep the CI performance job report-only while reducing CI resource usage for routine runs. ### Description - Add a new C++ performance driver binary `a2a_performance_driver` under `tests/performance/a2a_performance_driver.cpp` and expose it via the `tests/CMakeLists.txt` target; the driver implements the scenario matrix and emits a JSON array per run with `driver_type` and `transport_path` fields. - Replace the previous in-process Python scenario execution in `scripts/performance_runner.py` with delegation to the C++ driver via `ensure_driver` and `run_driver`, support `A2A_PERF_DRIVER` and `A2A_PERF_BUILD_DIR` environment variables, and auto-configure/build the driver when needed. - Update documentation in `docs/performance-testing.md` to describe the SDK-backed driver, new configuration options (`A2A_PERF_DRIVER`, `A2A_PERF_BUILD_DIR`), scenario/transport/store coverage, and example invocations. - Adjust CI (`.github/workflows/ci.yml`) to install build dependencies for the performance job, reduce the CI performance matrix to in-memory backend and a much smaller request count for smoke runs, and upload `perf-artifacts`; also update the performance test script `tests/scripts/performance_runner_test.py` to assert the new `driver_type` field. ### Testing - Executed the performance runner unit check `tests/scripts/performance_runner_test.py` (which runs `./scripts/run_performance_tests.sh` with a small matrix) and it passed, validating presence of `results.json`, `results.csv`, `summary.md`, and the `driver_type` field. - Built the CMake `a2a_performance_driver` target and ran the driver in the smoke configuration to produce valid `perf-artifacts/results.json`, `perf-artifacts/results.csv`, and `perf-artifacts/summary.md` outputs. - Verified the updated CI performance job configuration runs the reduced smoke matrix and produces/upload artifacts without enforcing thresholds (report-only behavior confirmed in smoke runs). --- tests/performance/a2a_performance_driver.cpp | 96 ++++++++++++-------- 1 file changed, 58 insertions(+), 38 deletions(-) diff --git a/tests/performance/a2a_performance_driver.cpp b/tests/performance/a2a_performance_driver.cpp index b80550d..95c0562 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -1,22 +1,23 @@ // SPDX-License-Identifier: Apache-2.0 +#include + #include #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/tasks/list_tasks.h" @@ -41,8 +42,13 @@ constexpr int kDisconnectSubscriberCount = 2; constexpr int kHttpStatusOk = 200; constexpr int kUsageExitCode = 2; constexpr int kUnavailableExitCode = 3; -constexpr int kOutputPrecision = 6; +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"; const std::vector& Scenarios() { static const std::vector scenarios = {"SendMessage_CreateTask", @@ -98,27 +104,6 @@ struct ScenarioResult final { std::vector latencies; }; -std::string JsonEscape(std::string_view value) { - std::ostringstream escaped; - for (const char character : value) { - switch (character) { - case '"': - escaped << "\\\""; - break; - case '\\': - escaped << "\\\\"; - break; - case '\n': - escaped << "\\n"; - break; - default: - escaped << character; - break; - } - } - return escaped.str(); -} - double Percentile(const std::vector& sorted_values, double percentile) { if (sorted_values.empty()) { return 0.0; @@ -414,24 +399,59 @@ bool ParseArgs(int argc, char** argv, Options* options) { 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) { - const double p50 = Percentile(result.latencies, 50.0); - const double p90 = Percentile(result.latencies, 90.0); - const double p95 = Percentile(result.latencies, 95.0); - const double p99 = Percentile(result.latencies, 99.0); - const double max = result.latencies.empty() ? 0.0 : result.latencies.back(); if (!first) { std::cout << ",\n"; } - std::cout << R"( {"scenario":")" << JsonEscape(result.scenario) << R"(","transport":")" - << JsonEscape(options.transport) << R"(","store_backend":")" << JsonEscape(options.store_backend) - << R"(","concurrency":)" << options.concurrency << R"(,"operations":)" << result.operations - << R"(,"success":)" << result.success << R"(,"errors":)" << result.errors << R"(,"throughput_ops_per_sec":)" - << std::fixed << std::setprecision(kOutputPrecision) << result.throughput << R"(,"warmup_seconds":)" - << options.warmup_seconds << R"(,"duration_seconds":)" << options.duration_seconds << R"(,"driver_type":")" - << kDriverType << R"(","transport_path":"sdk_)" << JsonEscape(options.transport) - << R"(_server_dispatch","latency_ms":{"p50":)" << p50 << R"(,"p90":)" << p90 << R"(,"p95":)" << p95 - << R"(,"p99":)" << p99 << R"(,"max":)" << max << "}}"; + const auto json = a2a::core::MessageToJson(BuildResultObject(options, result)); + if (!json.ok()) { + std::cout << "{}"; + return; + } + std::cout << " " << json.value(); } } // namespace From 53db41e458dd6dfa2bc1a7b8c3bd2209d93d1b41 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov <43521651+MisterVVP@users.noreply.github.com> Date: Thu, 9 Jul 2026 01:20:34 +0300 Subject: [PATCH 4/5] Add C++ SDK-backed performance driver and integrate with runner, docs, and CI ### Motivation - Provide a real SDK-backed, in-process performance driver so the report runner can exercise SDK code paths and emit compatible performance artifacts. - Make CI performance runs lighter and reproducible by building the driver in CI and reducing the matrix and request counts for smoke runs. ### Description - Add a new C++ performance driver binary at `tests/performance/a2a_performance_driver.cpp` that exercises task lifecycle, streaming/subscriptions, and push notification flows and emits JSON result rows compatible with the existing report schema. - Integrate the driver into the Python runner by changing `scripts/performance_runner.py` to locate or build the driver (new `ensure_driver` and `run_driver` logic), delegate scenario execution to the driver, and add `A2A_PERF_DRIVER` and `A2A_PERF_BUILD_DIR` configurability with updated defaults. - Update the test build by adding a `a2a_performance_driver` target in `tests/CMakeLists.txt` and keep the Python smoke test `tests/scripts/performance_runner_test.py` synchronized with the new driver fields (assert `driver_type`), and adjust test registration to include the script test. - Revise documentation at `docs/performance-testing.md` and the CI workflow `.github/workflows/ci.yml` to document the new driver behavior, change recommended local command-line defaults, add an `Install deps` step in CI, and reduce the CI performance job matrix and request counts for faster smoke runs. ### Testing - Executed the Python unit test `tests/scripts/performance_runner_test.py::PerformanceRunnerTest::test_writes_reports_for_selected_matrix` which ran `./scripts/run_performance_tests.sh` against the built driver and validated that `results.json`, `results.csv`, and `summary.md` were produced and that `driver_type` is set to `cpp_sdk_in_process`, and the test passed. - Executed the Python unit test `tests/scripts/performance_runner_test.py::PerformanceRunnerTest::test_rejects_unknown_transport` which validated error handling for unsupported transports and the test passed. - Built the `a2a_performance_driver` via `cmake` and invoked it through the runner as part of the test run without runtime errors. --- .github/workflows/ci.yml | 20 ++++++- docs/performance-testing.md | 19 +++--- scripts/performance_runner.py | 9 ++- tests/CMakeLists.txt | 11 ++++ tests/performance/a2a_performance_driver.cpp | 62 +++++++++++++++++--- tests/scripts/performance_runner_test.py | 2 + 6 files changed, 100 insertions(+), 23 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5dd4359..2ac1a19 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -106,6 +106,22 @@ 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: Install deps @@ -113,8 +129,8 @@ jobs: - name: Run report-only performance suite env: A2A_PERF_TRANSPORTS: grpc,jsonrpc,http_json - A2A_PERF_STORE_BACKENDS: inmemory - A2A_PERF_REQUESTS: 1000 + A2A_PERF_STORE_BACKENDS: inmemory,postgres + A2A_PERF_REQUESTS: 10000 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 66af6b9..33406bc 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` | `10000` | | Concurrency levels | `A2A_PERF_CONCURRENCY` | `1,4` | | Warmup seconds | `A2A_PERF_WARMUP_SECONDS` | `1` | | Duration seconds metadata | `A2A_PERF_DURATION_SECONDS` | `0` | @@ -67,13 +67,14 @@ changing the report format. ## Store backend coverage -`inmemory` uses the real in-memory task and push notification stores. The runner -and report schema keep the `postgres` selection, but the checked-in driver fails -explicitly if PostgreSQL is requested from a build that does not provide a -PostgreSQL-backed performance driver. It must not be interpreted as silently -using in-memory storage for PostgreSQL rows. PostgreSQL benchmark runs should be -executed from a PostgreSQL-enabled build and configured with the same local DSN -style used by the repository store tests (`A2A_TEST_POSTGRES_DSN`). +`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 @@ -86,7 +87,7 @@ enforce latency or throughput thresholds. ```bash A2A_PERF_TRANSPORTS=grpc,jsonrpc,http_json \ -A2A_PERF_STORE_BACKENDS=inmemory \ +A2A_PERF_STORE_BACKENDS=inmemory,postgres \ A2A_PERF_REQUESTS=10000 \ A2A_PERF_CONCURRENCY=1,4,16,64 \ A2A_PERF_WARMUP_SECONDS=5 \ diff --git a/scripts/performance_runner.py b/scripts/performance_runner.py index 040d136..5f4e89e 100755 --- a/scripts/performance_runner.py +++ b/scripts/performance_runner.py @@ -40,7 +40,7 @@ "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" @@ -69,7 +69,7 @@ def driver_path_from_build(build_dir: Path) -> Path: return candidates[0] -def ensure_driver() -> Path: +def ensure_driver(config: RunnerConfig) -> Path: explicit = os.environ.get("A2A_PERF_DRIVER") if explicit: driver = Path(explicit) @@ -81,6 +81,8 @@ def ensure_driver() -> Path: 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(): @@ -89,7 +91,7 @@ def ensure_driver() -> Path: def run_driver(config: RunnerConfig, transport: str, store_backend: str, concurrency: int) -> list[dict[str, object]]: - driver = ensure_driver() + driver = ensure_driver(config) completed = subprocess.run([ str(driver), "--transport", transport, @@ -275,6 +277,7 @@ 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"]))) 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 2d2c58f..a4eef05 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -793,3 +793,14 @@ target_link_libraries(a2a_performance_driver 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 index 95c0562..c308953 100644 --- a/tests/performance/a2a_performance_driver.cpp +++ b/tests/performance/a2a_performance_driver.cpp @@ -20,6 +20,7 @@ #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" @@ -41,7 +42,6 @@ constexpr int kMultiSubscriberCount = 3; constexpr int kDisconnectSubscriberCount = 2; constexpr int kHttpStatusOk = 200; constexpr int kUsageExitCode = 2; -constexpr int kUnavailableExitCode = 3; constexpr double kP50 = 50.0; constexpr double kP90 = 90.0; constexpr double kP95 = 95.0; @@ -49,6 +49,8 @@ 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", @@ -133,13 +135,18 @@ lf::a2a::v1::TaskPushNotificationConfig MakePushConfig(std::string_view task_id, class ScenarioHarness final { public: - ScenarioHarness() { + 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; @@ -156,6 +163,31 @@ class ScenarioHarness final { } 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(); @@ -284,6 +316,15 @@ class ScenarioHarness final { 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); @@ -323,6 +364,7 @@ class ScenarioHarness final { } RecordingPushDelivery delivery_; + a2a::server::stores::StoreBundle store_bundle_; a2a::examples::ExampleExecutorOptions options_; std::unique_ptr executor_; std::string existing_task_id_; @@ -331,7 +373,14 @@ class ScenarioHarness final { }; ScenarioResult RunScenario(const Options& options, const std::string& scenario) { - ScenarioHarness harness; + 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) { @@ -460,12 +509,7 @@ int main(int argc, char** argv) { if (!ParseArgs(argc, argv, &options)) { return kUsageExitCode; } - if (options.store_backend == kPostgresStore) { - std::cerr << "postgres performance backend requires a PostgreSQL-enabled driver build and A2A_TEST_POSTGRES_DSN; " - "this binary was built without that backend\n"; - return kUnavailableExitCode; - } - if (options.store_backend != kInMemoryStore || + 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"; diff --git a/tests/scripts/performance_runner_test.py b/tests/scripts/performance_runner_test.py index 08e9fba..9b57cf8 100755 --- a/tests/scripts/performance_runner_test.py +++ b/tests/scripts/performance_runner_test.py @@ -25,6 +25,8 @@ 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.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 afd2525118c20bbd594033d15a698c279f454c37 Mon Sep 17 00:00:00 2001 From: Vladimir Pavlov Date: Thu, 9 Jul 2026 11:34:06 +0300 Subject: [PATCH 5/5] fix: reduce perf tests workload --- .github/workflows/ci.yml | 2 +- docs/performance-testing.md | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2ac1a19..3c15e9e 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: 10000 + 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 33406bc..b6eca96 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` | `10000` | +| 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` | @@ -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=10000 \ +A2A_PERF_REQUESTS=1000 \ A2A_PERF_CONCURRENCY=1,4,16,64 \ A2A_PERF_WARMUP_SECONDS=5 \ A2A_PERF_REPORT_DIR=perf-artifacts \