test(pubsub): add throughput benchmark (#5238)

Add a new throughput benchmark for measuring the performance of publish
and subscribe operations. The benchmark is implemented as a single
binary with subcommands for the publisher and subscriber, allowing for
unified metric formatting and simplified execution.

A version of this was used to benchmark our libraries to validate
acceptable performance.

For #3840

Author: suzmue

Cargo.lock

@@ -287,6 +287,7 @@ members = [
   "src/generated/type",
   "src/lro",
   "src/pubsub",
+  "src/pubsub/benchmarks/throughput",
   "src/pubsub/examples",
   "src/spanner",
   "src/spanner/grpc-mock",

Cargo.toml

@@ -0,0 +1,38 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[package]
+name    = "pubsub-throughput"
+version = "0.0.0"
+publish = false
+# Inherit other attributes from the workspace.
+edition.workspace    = true
+authors.workspace    = true
+license.workspace    = true
+repository.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+anyhow.workspace    = true
+bytes.workspace     = true
+clap                = { workspace = true, features = ["derive", "env", "help", "std"] }
+google-cloud-pubsub = { workspace = true, features = ["default-rustls-provider"] }
+futures.workspace   = true
+humantime.workspace = true
+rand.workspace      = true
+tokio               = { workspace = true, features = ["full"] }
+tracing.workspace   = true
+tracing-subscriber  = { workspace = true, features = ["fmt", "std"] }

src/pubsub/benchmarks/throughput/Cargo.toml

@@ -0,0 +1,66 @@
+# Pub/Sub Throughput Benchmark
+
+A throughput benchmark for the Cloud Pub/Sub Rust client library.
+
+This tool measures the performance of publishing messages to a Google Cloud
+Pub/Sub topic, or receiving messages from a subscription. It reports operation
+rates in messages per second and megabytes per second.
+
+## Usage
+
+```bash
+cargo run --release -p pubsub-throughput -- [COMMAND] [OPTIONS]
+```
+
+To see the commands and options use:
+
+```bash
+cargo run -p pubsub-throughput -- --help
+```
+
+## Output Format
+
+The benchmark outputs data in CSV format with the following columns:
+
+- `timestamp`: The Unix timestamp in milliseconds.
+- `elapsed(s)`: The elapsed time for the operation in seconds.
+- `op`: The operation being measured (`Pub`, `Ack`, or `Recv`).
+- `iteration`: The current iteration number.
+- `count`: The number of messages processed in the operation.
+- `msgs/s`: The number of messages per second.
+- `bytes`: The total number of bytes processed.
+- `MB/s`: The throughput in megabytes per second.
+- `errors`: The number of errors encountered (if any).
+- `errors/s`: The number of errors per second (if any).
+
+## Examples
+
+### Setup
+
+Create a topic and subscription:
+
+```bash
+gcloud pubsub topics create my-topic
+gcloud pubsub subscriptions create my-subscription --topic=my-topic
+```
+
+### Publisher Run
+
+```bash
+cargo run --release -p pubsub-throughput -- publisher \
+    --project my-gcp-project \
+    --topic-id my-topic \
+    --payload-size 2048 \
+    --report-interval 10s \
+    --duration 1m
+```
+
+### Subscriber Run
+
+```bash
+cargo run --release -p pubsub-throughput -- subscriber \
+    --project my-gcp-project \
+    --subscription-id my-subscription \
+    --report-interval 10s \
+    --duration 1m
+```

src/pubsub/benchmarks/throughput/README.md

@@ -0,0 +1,86 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use clap::{Args, Parser, Subcommand};
+use humantime::parse_duration;
+
+#[derive(Parser, Debug)]
+#[command(author, version, about = "Cloud Pub/Sub Throughput Benchmark", long_about = None)]
+pub struct Config {
+    #[command(subcommand)]
+    pub command: Commands,
+}
+
+#[derive(Subcommand, Debug)]
+pub enum Commands {
+    /// Run the publisher benchmark
+    Publisher(PublisherArgs),
+    /// Run the subscriber benchmark
+    Subscriber(SubscriberArgs),
+}
+
+#[derive(Args, Debug, Clone)]
+pub struct CommonArgs {
+    #[arg(long, default_value = "", env = "GOOGLE_CLOUD_PROJECT")]
+    pub project: String,
+
+    #[arg(long, value_parser = parse_duration, default_value = "5s")]
+    pub report_interval: std::time::Duration,
+
+    #[arg(long, value_parser = parse_duration, default_value = "5m")]
+    pub duration: std::time::Duration,
+
+    #[arg(long, default_value_t = 1)]
+    pub grpc_channels: usize,
+
+    #[arg(long, default_value_t = 100000)]
+    pub max_outstanding_messages: usize,
+}
+
+#[derive(Args, Debug, Clone)]
+pub struct PublisherArgs {
+    #[command(flatten)]
+    pub common: CommonArgs,
+
+    #[arg(long)]
+    pub topic_id: String,
+
+    #[arg(long, default_value_t = 1024)]
+    pub payload_size: i64,
+
+    #[arg(long, default_value_t = 1000)]
+    pub batch_size: u32,
+
+    #[arg(long, default_value_t = 10 * 1024 * 1024)] // 10 MB
+    pub batch_bytes: u32,
+
+    #[arg(long, value_parser = parse_duration, default_value = "100ms")]
+    pub batch_delay: std::time::Duration,
+}
+
+#[derive(Args, Debug, Clone)]
+pub struct SubscriberArgs {
+    #[command(flatten)]
+    pub common: CommonArgs,
+
+    #[arg(long)]
+    pub subscription_id: String,
+
+    #[arg(long, default_value_t = 1)]
+    pub streams: usize,
+}
+
+pub fn parse_args() -> Config {
+    Config::parse()
+}

src/pubsub/benchmarks/throughput/src/args.rs

@@ -0,0 +1,86 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+mod args;
+mod publisher;
+mod subscriber;
+
+#[tokio::main]
+async fn main() -> Result<(), anyhow::Error> {
+    let config = crate::args::parse_args();
+    /// The CSV header used for all throughput benchmark results.
+    const CSV_HEADER: &str =
+        "timestamp,elapsed(s),op,iteration,count,msgs/s,bytes,MB/s,errors,errors/s";
+
+    match config.command {
+        crate::args::Commands::Publisher(args) => {
+            println!("# Running publish benchmark with config: {:?}", args);
+            println!("{}", CSV_HEADER);
+            publisher::run(args).await?;
+        }
+        crate::args::Commands::Subscriber(args) => {
+            println!("# Running subscribe benchmark with config: {:?}", args);
+            println!("{}", CSV_HEADER);
+            subscriber::run(args).await?;
+        }
+    }
+
+    Ok(())
+}
+
+/// Returns true if the benchmark has exceeded its maximum allowed runtime.
+pub(crate) fn done(maximum_runtime: Duration, start: Instant) -> bool {
+    start.elapsed() >= maximum_runtime
+}
+
+/// Returns the current Unix timestamp in milliseconds.
+pub(crate) fn timestamp() -> u128 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_millis()
+}
+
+/// Formats and prints a single measurement row to stdout in CSV format.
+///
+/// The output includes throughput metrics (msgs/s and MB/s) calculated based
+/// on the elapsed time and the amount of data processed.
+pub(crate) fn print_result(
+    operation: &str,
+    iteration: i64,
+    count: i64,
+    bytes: i64,
+    errors: i64,
+    elapsed: Duration,
+) {
+    let elapsed_s = elapsed.as_secs_f64();
+    let mbs = (bytes as f64) / elapsed_s / 1_000_000.0;
+    let msgs = (count as f64) / elapsed_s;
+    let errs = (errors as f64) / elapsed_s;
+    println!(
+        "{},{},{},{},{},{:.2},{},{:.2},{},{:.2}",
+        timestamp(),
+        elapsed_s,
+        operation,
+        iteration,
+        count,
+        msgs,
+        bytes,
+        mbs,
+        errors,
+        errs
+    );
+}