Add end-to-end throughput benchmarks for S3 Express (#3446)

## Description
See
[README](https://github.com/smithy-lang/smithy-rs/blob/b172a1ea4c76cf8f820d8ef4ae827d02451bddcd/aws/sdk/benchmarks/s3-express/README.md)

## Testing
Here are performance numbers measured on `Amazon Linux 2 x86_64 5.10`
with `c5.4xlarge` in `us-west-2`, and benchmarks ran with
`NUMBER_OF_ITERATIONS=1` to allow for quick execution

```
For 64KB objects:
PUT: [11.750 ms 12.025 ms 12.291 ms] / [67.737 ms 76.367 ms 85.531 ms]
GET: [16.071 ms 16.270 ms 16.461 ms] / [19.941 ms 22.622 ms 26.167 ms]
PUT + DELETE: [21.492 ms 22.132 ms 22.755 ms] / [75.608 ms 86.056 ms 98.329 ms]

For 1MB objects:
PUT [37.400 ms 39.130 ms 40.769 ms] / [144.30 ms 160.93 ms 180.93 ms]
GET [14.968 ms 15.193 ms 15.408 ms] / [24.872 ms 28.417 ms 32.984 ms]
PUT + DELETE: [52.106 ms 54.875 ms 57.503 ms] / [172.38 ms 185.00 ms 200.28 ms]
``` 
In each row, a tuple on the left of `/` is for execution time for an S3
Express bucket and that on the right is for a regular S3 bucket. Each
tuple is a 99% confidence interval, containing `[<the lower end> <the
mean> <the higher end>]`.

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

---------

Co-authored-by: John DiSanti <[email protected]>
Co-authored-by: AWS SDK Rust Bot <[email protected]>
Co-authored-by: AWS SDK Rust Bot <[email protected]>
Co-authored-by: Zelda Hessler <[email protected]>
Co-authored-by: Russell Cohen <[email protected]>

aws/sdk/benchmarks/s3-express/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "s3-express"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+aws-config = { path = "../../build/aws-sdk/sdk/aws-config", features = ["behavior-version-latest"] }
+aws-sdk-s3 = { path = "../../build/aws-sdk/sdk/s3", features = ["behavior-version-latest"] }
+criterion = { version = "0.5", features = ["async_tokio"] }
+futures-util = { version = "0.3.29", default-features = false, features = ["alloc"] }
+tokio = { version = "1.23.1", features = ["macros", "test-util", "rt-multi-thread"] }
+
+[[bench]]
+name = "concurrent_put_get"
+harness = false
+
+[[bench]]
+name = "get_object"
+harness = false
+
+[[bench]]
+name = "put_get_delete"
+harness = false
+
+[[bench]]
+name = "put_object"
+harness = false
+
+[profile.bench]
+debug = true

aws/sdk/benchmarks/s3-express/README.md

@@ -0,0 +1,62 @@
+# S3 Express Benchmark
+
+This directory contains S3 Express One Zone benchmarks that measure end-to-end throughput when using the AWS Rust SDK to put, get, and delete objects to/from S3 Express One Zone buckets. We use [`Criterion`](https://github.com/bheisler/criterion.rs) for benchmarks. A sibling directory `s3-throughput` also measures throughput to put and get objects to/from S3 but currently does not support generating benchmark results with a given confidence interval, which is why we have this separate benchmark suite.
+
+Performance numbers will vary depending on the benchmarking environment, but relative performance should still be accurate (i.e. regular S3 bucket vs. S3 Express bucket or comparing with a previous release of the Rust SDK).
+
+## Benchmark targets
+- `get_object`: Invoke `GetObject` the specified number of times (20 by default) against a given set of buckets, using both 64KB and 1MB objects.
+- `put_object`: Invoke `PutObject` the specified number of times (20 by default) against a given set of buckets, using both 64KB and 1MB objects.
+- `put_get_delete`: `PutObject`, `GetObject`, and `DeleteObject` using sequential invocations (20 by default) of operations across different buckets, switching buckets on every request and using both 64KB and 1MB objects.
+- `concurrent_put_get`: Schedule the equal number of async tasks of `PutObject` (20 by default) to different buckets, wait for completion, then schedule the equal number of async tasks of `GetObject` to different buckets, and wait for completion, using the 64KB objects.
+
+## Running benchmarks
+Example of running the `put_object` benchmark in local dev environment:
+
+```bash
+export BUCKETS=test0--usw2-az1--x-s3,test1--usw2-az1--x-s3
+cargo bench --bench put_object
+```
+To configure how the benchmark is run, set the following environment variables:
+#### required
+- `BUCKETS`: a list of comma separated bucket names
+
+#### optional
+- `CONFIDENCE_LEVEL`: the confidence level for benchmarks in a group (0.99 by default)
+- `NUMBER_OF_ITERATIONS`: the number of times a set of operations runs for measurement (20 by default)
+- `SAMPLE_SIZE`: the size of the sample for benchmarks in a group (10 by default)
+
+### Flamegraph generation
+Use [`flamegraph`](https://github.com/flamegraph-rs/flamegraph) to generate one for a target bench, for instance:
+```bash
+export BUCKETS=test0--usw2-az1--x-s3,test1--usw2-az1--x-s3
+cargo flamegraph --bench put_get_delete -- --bench
+```
+
+The resulting flamegraph `flamegraph.svg` should be generated in the current directory.
+
+
+## Limitation
+Benchmarks currently measure end-to-end throughput of operations, including both the Rust SDK latency and the server side latency. To detect regressions in the Rust SDK reliably, we should only capture the time taken before sending a request and after receiving a response.
+
+## Baseline
+As of b172a1e, here are performance numbers for the targets `get_object` and `put_object` run against a single express bucket within the `us-west-2` region (showing additional outputs to display config parameters). The benchmarks are measured on Amazon Linux 2 x86_64 5.10 Kernel with a host type c5.4xlarge.
+```
+[src/lib.rs:30] sample_size = 10
+[src/lib.rs:14] confidence_level = 0.99
+[src/lib.rs:23] number_of_iterations = 20
+measuring 20 of GetObject against [
+    "s3express-rust-sdk-benchmark--usw2-az1--x-s3",
+], switching buckets on every operation if more than one bucket is specified
+get_object/size/65536   time:   [304.20 ms 311.62 ms 317.62 ms]
+get_object/size/1048576 time:   [283.94 ms 289.16 ms 293.42 ms]
+
+[src/lib.rs:30] sample_size = 10
+[src/lib.rs:14] confidence_level = 0.99
+[src/lib.rs:23] number_of_iterations = 20
+measuring 20 of PutObject against [
+    "s3express-rust-sdk-benchmark--usw2-az1--x-s3",
+], switching buckets on every operation if more than one bucket is specified
+put_object/size/65536   time:   [163.01 ms 172.76 ms 185.16 ms]
+put_object/size/1048576 time:   [356.49 ms 368.64 ms 383.51 ms]
+```

aws/sdk/benchmarks/s3-express/benches/concurrent_put_get.rs

@@ -0,0 +1,97 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+use aws_config::BehaviorVersion;
+use aws_sdk_s3::primitives::ByteStream;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use s3_express::{confidence_level, number_of_iterations, sample_size};
+use tokio::runtime::Runtime;
+use tokio::task;
+
+pub fn concurrent_put_get(c: &mut Criterion) {
+    let buckets = if let Ok(buckets) = std::env::var("BUCKETS") {
+        buckets.split(",").map(String::from).collect::<Vec<_>>()
+    } else {
+        panic!("required environment variable `BUCKETS` should be set: e.g. `BUCKETS=\"bucket1,bucket2\"`")
+    };
+
+    let number_of_iterations = number_of_iterations();
+
+    println!(
+        "measuring {number_of_iterations} concurrent PutObject followed by \
+            {number_of_iterations} concurrent GetObject against \
+            {buckets:#?}, with each bucket being assigned equal number of operations\n"
+    );
+
+    let client = Runtime::new().unwrap().block_on(async {
+        let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+        aws_sdk_s3::Client::new(&config)
+    });
+
+    const KB: usize = 1024;
+    const SIZE: usize = 64 * KB;
+    let object: Vec<u8> = vec![0; SIZE];
+    let mut group = c.benchmark_group("concurrent_put_delete");
+    group.bench_with_input(BenchmarkId::new("size", SIZE), &SIZE, |b, _| {
+        b.to_async(Runtime::new().unwrap()).iter(|| async {
+            let put_futures = (0..number_of_iterations)
+                .map({
+                    let client = client.clone();
+                    let buckets = buckets.clone();
+                    let object = object.clone();
+                    move |i| {
+                        task::spawn({
+                            let client = client.clone();
+                            let buckets = buckets.clone();
+                            let object = object.clone();
+                            async move {
+                                client
+                                    .put_object()
+                                    .bucket(&buckets[i % buckets.len()])
+                                    .key(&format!("test{i}"))
+                                    .body(ByteStream::from(object))
+                                    .send()
+                                    .await
+                                    .unwrap();
+                            }
+                        })
+                    }
+                })
+                .collect::<Vec<_>>();
+            ::futures_util::future::join_all(put_futures).await;
+
+            let get_futures = (0..number_of_iterations)
+                .map({
+                    let client = client.clone();
+                    let buckets = buckets.clone();
+                    move |i| {
+                        task::spawn({
+                            let client = client.clone();
+                            let buckets = buckets.clone();
+                            async move {
+                                client
+                                    .get_object()
+                                    .bucket(&buckets[i % buckets.len()])
+                                    .key(&format!("test{i}"))
+                                    .send()
+                                    .await
+                                    .unwrap();
+                            }
+                        })
+                    }
+                })
+                .collect::<Vec<_>>();
+            ::futures_util::future::join_all(get_futures).await;
+        });
+    });
+    group.finish();
+}
+
+criterion_group!(
+    name = benches;
+    config = Criterion::default().sample_size(sample_size()).confidence_level(confidence_level());
+    targets = concurrent_put_get
+);
+criterion_main!(benches);

aws/sdk/benchmarks/s3-express/benches/get_object.rs

@@ -0,0 +1,99 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+use aws_config::BehaviorVersion;
+use aws_sdk_s3::primitives::ByteStream;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use s3_express::{confidence_level, number_of_iterations, sample_size};
+use tokio::runtime::Runtime;
+
+async fn prepare_test_objects<T: AsRef<str>>(buckets: &[T], object_sizes: &[usize]) {
+    let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+    let client = aws_sdk_s3::Client::new(&config);
+
+    for size in object_sizes {
+        let object: Vec<u8> = vec![0; *size];
+        for bucket in buckets.as_ref() {
+            client
+                .put_object()
+                .bucket(bucket.as_ref())
+                .key(&format!("test-{size}"))
+                .body(ByteStream::from(object.clone()))
+                .send()
+                .await
+                .unwrap();
+        }
+    }
+}
+
+pub fn get_object(c: &mut Criterion) {
+    let buckets = if let Ok(buckets) = std::env::var("BUCKETS") {
+        buckets.split(",").map(String::from).collect::<Vec<_>>()
+    } else {
+        panic!("required environment variable `BUCKETS` should be set: e.g. `BUCKETS=\"bucket1,bucket2\"`")
+    };
+
+    let number_of_iterations = number_of_iterations();
+
+    println!(
+        "measuring {number_of_iterations} of GetObject against {buckets:#?}, \
+        switching buckets on every operation if more than one bucket is specified\n"
+    );
+
+    const KB: usize = 1024;
+    let sizes = [64 * KB, KB * KB];
+
+    let client = Runtime::new().unwrap().block_on(async {
+        prepare_test_objects(&buckets, &sizes).await;
+
+        // Return a new client that has an empty identity cache
+        let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+        aws_sdk_s3::Client::new(&config)
+    });
+
+    let mut group = c.benchmark_group("get_object");
+
+    for size in sizes.iter() {
+        let key = format!("test-{size}");
+        group.bench_with_input(BenchmarkId::new("size", size), size, |b, _| {
+            b.to_async(Runtime::new().unwrap()).iter(|| async {
+                for i in 0..number_of_iterations {
+                    let bucket = &buckets[i % buckets.len()];
+                    client
+                        .get_object()
+                        .bucket(bucket)
+                        .key(&key)
+                        .send()
+                        .await
+                        .unwrap();
+                }
+            });
+        });
+    }
+    group.finish();
+
+    // Clean up test objects
+    Runtime::new().unwrap().block_on(async {
+        for size in sizes {
+            let key = format!("test-{size}");
+            for bucket in &buckets {
+                client
+                    .delete_object()
+                    .bucket(bucket)
+                    .key(&key)
+                    .send()
+                    .await
+                    .unwrap();
+            }
+        }
+    });
+}
+
+criterion_group!(
+    name = benches;
+    config = Criterion::default().sample_size(sample_size()).confidence_level(confidence_level());
+    targets = get_object
+);
+criterion_main!(benches);

aws/sdk/benchmarks/s3-express/benches/put_get_delete.rs

@@ -0,0 +1,78 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+use aws_config::BehaviorVersion;
+use aws_sdk_s3::primitives::ByteStream;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use s3_express::{confidence_level, number_of_iterations, sample_size};
+use tokio::runtime::Runtime;
+
+pub fn put_get_delete(c: &mut Criterion) {
+    let buckets = if let Ok(buckets) = std::env::var("BUCKETS") {
+        buckets.split(",").map(String::from).collect::<Vec<_>>()
+    } else {
+        panic!("required environment variable `BUCKETS` should be set: e.g. `BUCKETS=\"bucket1,bucket2\"`")
+    };
+
+    let number_of_iterations = number_of_iterations();
+
+    println!(
+        "measuring {number_of_iterations} sequences of \
+            PutObject -> GetObject -> DeleteObject against \
+            {buckets:#?}, switching buckets on every sequence of operations if more than one bucket is specified\n"
+    );
+
+    let client = Runtime::new().unwrap().block_on(async {
+        let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+        aws_sdk_s3::Client::new(&config)
+    });
+
+    let mut group = c.benchmark_group("put_get_delete");
+
+    const KB: usize = 1024;
+    for size in [64 * KB, KB * KB].iter() {
+        let object: Vec<u8> = vec![0; *size];
+        group.bench_with_input(BenchmarkId::new("size", size), size, |b, _| {
+            b.to_async(Runtime::new().unwrap()).iter(|| async {
+                for i in 0..number_of_iterations {
+                    let bucket = &buckets[i % buckets.len()];
+                    let key = "test";
+                    client
+                        .put_object()
+                        .bucket(bucket)
+                        .key(key)
+                        .body(ByteStream::from(object.clone()))
+                        .send()
+                        .await
+                        .unwrap();
+
+                    client
+                        .get_object()
+                        .bucket(bucket)
+                        .key(key)
+                        .send()
+                        .await
+                        .unwrap();
+
+                    client
+                        .delete_object()
+                        .bucket(bucket)
+                        .key(key)
+                        .send()
+                        .await
+                        .unwrap();
+                }
+            });
+        });
+    }
+    group.finish();
+}
+
+criterion_group!(
+    name = benches;
+    config = Criterion::default().sample_size(sample_size()).confidence_level(confidence_level());
+    targets = put_get_delete
+);
+criterion_main!(benches);