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)