S3 Benchmarks - support java-based multipart client

test/s3-benchmarks/.scripts/benchmark

@@ -64,7 +64,7 @@ if [ ! -d result ]; then
 fi
 
 sizes_str="1B 8MB+1 8MB-1 128MB 4GB 30GB"
-versions_str="v1 v2 CRT"
+versions_str="v1 v2 CRT java"
 sizes=( $sizes_str )
 versions=( $versions_str )
 

test/s3-benchmarks/.scripts/create_benchmark_files

@@ -1,14 +1,13 @@
-head -c 1B </dev/urandom >/dev/shm/1B
-head -c 8388607B </dev/urandom >/dev/shm/8MB-1
-head -c 8388609B </dev/urandom >/dev/shm/8MB+1
-head -c 128M </dev/urandom >/dev/shm/128MB
-head -c 4B </dev/urandom >/dev/shm/4GB
-head -c 30GB </dev/urandom >/dev/shm/30GB
-
-head -c 1B </dev/urandom >/1B
-head -c 8388607B </dev/urandom >/8MB-1
-head -c 8388609B </dev/urandom >/8MB+1
-head -c 128M </dev/urandom >/128MB
-head -c 4B </dev/urandom >/4GB
-head -c 30GB </dev/urandom >/30GB
+head -c 1 </dev/urandom >/dev/shm/1B
+head -c $((8*1024*1024-1)) </dev/urandom >/dev/shm/8MB-1
+head -c $((8*1024*1024+1)) </dev/urandom >/dev/shm/8MB+1
+head -c $((128*1024*1024)) </dev/urandom >/dev/shm/128MB
+head -c $((4*1024*1024*1024)) </dev/urandom >/dev/shm/4GB
+head -c $((30*1024*1024*1024)) </dev/urandom >/dev/shm/30GB
 
+head -c 1 </dev/urandom >/1B
+head -c $((8*1024*1024-1)) </dev/urandom >/8MB-1
+head -c $((8*1024*1024+1)) </dev/urandom >/8MB+1
+head -c $((128*1024*1024)) </dev/urandom >/128MB
+head -c $((4*1024*1024*1024)) </dev/urandom >/4GB
+head -c $((30*1024*1024*1024)) </dev/urandom >/30GB

test/s3-benchmarks/README.md

@@ -1,7 +1,6 @@
 # S3 Benchmark Harness
 
-
-This module contains performance tests for `S3AsyncClient` and 
+This module contains performance tests for `S3AsyncClient` and
 `S3TransferManager`
 
 ## How to run
@@ -17,6 +16,31 @@ java -jar s3-benchmarks.jar --bucket=bucket --key=key -file=/path/to/destionfile
 java -jar s3-benchmarks.jar --bucket=bucket --key=key -file=/path/to/sourcefile/ --operation=upload --partSizeInMB=20 --maxThroughput=100.0
 ```
 
+## Command line arguments
+
+### Benchmark version
+
+The `--version` command line option is used to determine which component is under test:
+
+- `--version=crt` : Indicate to run the benchmark for the CRT's S3Client
+- `--version=java` : Indicate to run the benchmark for the java based S3 Async Client (`MultipartS3AsyncClient` class)
+- `--version=v2`: SDK v2 transfer manager (using `S3CrtAsyncClient` to delegate requests)
+- `--version=v1`: SDK v1 transfer manager (using `AmazonS3Client` to delegate requests)
+
+### Operation
+
+The `--operation` command line argument determine which transfer operation is used
+
+|operation|supported version|
+|---|-------|
+|download | v1 v2 java crt |
+|upload | v1 v2 java crt |
+|download_directory | v1 v2 |
+|upload_directory | v1 v2 |
+|copy | v1 v2 java |
+
+> All command line argument can be found in the `BenchmarkRunner` class.
+
 # Benchmark scripts Automation
 From the `.script` folder, use one of the `benchamrk` scripts to run a test suite.
 

test/s3-benchmarks/pom.xml

@@ -87,6 +87,16 @@
             <artifactId>log4j-slf4j-impl</artifactId>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>netty-nio-client</artifactId>
+            <version>${awsjavasdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>aws-crt-client</artifactId>
+            <version>${awsjavasdk.version}</version>
+        </dependency>
     </dependencies>
 
     <build>

test/s3-benchmarks/src/main/java/software/amazon/awssdk/s3benchmarks/BaseJavaS3ClientBenchmark.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file 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 software.amazon.awssdk.s3benchmarks;
+
+import static software.amazon.awssdk.s3benchmarks.BenchmarkUtils.BENCHMARK_ITERATIONS;
+import static software.amazon.awssdk.s3benchmarks.BenchmarkUtils.DEFAULT_TIMEOUT;
+import static software.amazon.awssdk.s3benchmarks.BenchmarkUtils.printOutResult;
+import static software.amazon.awssdk.transfer.s3.SizeConstant.MB;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.http.crt.AwsCrtAsyncHttpClient;
+import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
+import software.amazon.awssdk.utils.Logger;
+import software.amazon.awssdk.utils.Validate;
+
+public abstract class BaseJavaS3ClientBenchmark implements TransferManagerBenchmark {
+    private static final Logger logger = Logger.loggerFor(BaseJavaS3ClientBenchmark.class);
+
+    protected final S3Client s3Client;
+
+    protected final S3AsyncClient s3AsyncClient;
+    protected final String bucket;
+    protected final String key;
+    protected final Duration timeout;
+    private final ChecksumAlgorithm checksumAlgorithm;
+    private final int iteration;
+
+    protected BaseJavaS3ClientBenchmark(TransferManagerBenchmarkConfig config) {
+        this.bucket = Validate.paramNotNull(config.bucket(), "bucket");
+        this.key = Validate.paramNotNull(config.key(), "key");
+        this.timeout = Validate.getOrDefault(config.timeout(), () -> DEFAULT_TIMEOUT);
+        this.iteration = Validate.getOrDefault(config.iteration(), () -> BENCHMARK_ITERATIONS);
+        this.checksumAlgorithm = config.checksumAlgorithm();
+
+        this.s3Client = S3Client.create();
+
+        long partSizeInMb = Validate.paramNotNull(config.partSizeInMb(), "partSize");
+        long readBufferInMb = Validate.paramNotNull(config.readBufferSizeInMb(), "readBufferSizeInMb");
+        Validate.mutuallyExclusive("cannot use forceCrtHttpClient and connectionAcquisitionTimeoutInSec",
+                                   config.forceCrtHttpClient(), config.connectionAcquisitionTimeoutInSec());
+        this.s3AsyncClient = S3AsyncClient.builder()
+                                          .multipartEnabled(true)
+                                          .multipartConfiguration(c -> c.minimumPartSizeInBytes(partSizeInMb * MB)
+                                                                        .thresholdInBytes(partSizeInMb * 2 * MB)
+                                                                        .apiCallBufferSizeInBytes(readBufferInMb * MB))
+                                          .httpClientBuilder(httpClient(config))
+                                          .build();
+    }
+
+    private SdkAsyncHttpClient.Builder httpClient(TransferManagerBenchmarkConfig config) {
+        if (config.forceCrtHttpClient()) {
+            logger.info(() -> "Using CRT HTTP client");
+            AwsCrtAsyncHttpClient.Builder builder = AwsCrtAsyncHttpClient.builder();
+            if (config.readBufferSizeInMb() != null) {
+                builder.readBufferSizeInBytes(config.readBufferSizeInMb() * MB);
+            }
+            if (config.maxConcurrency() != null) {
+                builder.maxConcurrency(config.maxConcurrency());
+            }
+            return builder;
+        }
+        NettyNioAsyncHttpClient.Builder builder = NettyNioAsyncHttpClient.builder();
+        if (config.connectionAcquisitionTimeoutInSec() != null) {
+            Duration connAcqTimeout = Duration.ofSeconds(config.connectionAcquisitionTimeoutInSec());
+            builder.connectionAcquisitionTimeout(connAcqTimeout);
+        }
+        if (config.maxConcurrency() != null) {
+            builder.maxConcurrency(config.maxConcurrency());
+        }
+        return builder;
+    }
+
+    protected abstract void sendOneRequest(List<Double> latencies) throws Exception;
+
+    protected abstract long contentLength() throws Exception;
+
+    @Override
+    public void run() {
+        try {
+            warmUp();
+            doRunBenchmark();
+        } catch (Exception e) {
+            logger.error(() -> "Exception occurred", e);
+        } finally {
+            cleanup();
+        }
+    }
+
+    private void cleanup() {
+        s3Client.close();
+        s3AsyncClient.close();
+    }
+
+    private void warmUp() throws Exception {
+        logger.info(() -> "Starting to warm up");
+        for (int i = 0; i < 3; i++) {
+            sendOneRequest(new ArrayList<>());
+            Thread.sleep(500);
+        }
+        logger.info(() -> "Ending warm up");
+    }
+
+    private void doRunBenchmark() throws Exception {
+        List<Double> metrics = new ArrayList<>();
+        for (int i = 0; i < iteration; i++) {
+            sendOneRequest(metrics);
+        }
+        printOutResult(metrics, "S3 Async client", contentLength());
+    }
+
+}

test/s3-benchmarks/src/main/java/software/amazon/awssdk/s3benchmarks/BaseTransferManagerBenchmark.java

@@ -34,6 +34,7 @@
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3CrtAsyncClientBuilder;
 import software.amazon.awssdk.services.s3.internal.crt.S3CrtAsyncClient;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
@@ -61,15 +62,18 @@ public abstract class BaseTransferManagerBenchmark implements TransferManagerBen
         logger.info(() -> "Benchmark config: " + config);
         Long partSizeInMb = config.partSizeInMb() == null ? null : config.partSizeInMb() * MB;
         Long readBufferSizeInMb = config.readBufferSizeInMb() == null ? null : config.readBufferSizeInMb() * MB;
-        s3 = S3CrtAsyncClient.builder()
-                             .targetThroughputInGbps(config.targetThroughput())
-                             .minimumPartSizeInBytes(partSizeInMb)
-                             .initialReadBufferSizeInBytes(readBufferSizeInMb)
-                             .targetThroughputInGbps(config.targetThroughput() == null ?
-                                                     Double.valueOf(100.0) : config.targetThroughput())
-                             .build();
-        s3Sync = S3Client.builder()
-                         .build();
+        S3CrtAsyncClientBuilder builder = S3CrtAsyncClient.builder()
+                                                          .targetThroughputInGbps(config.targetThroughput())
+                                                          .minimumPartSizeInBytes(partSizeInMb)
+                                                          .initialReadBufferSizeInBytes(readBufferSizeInMb)
+                                                          .targetThroughputInGbps(config.targetThroughput() == null ?
+                                                                                  Double.valueOf(100.0) :
+                                                                                  config.targetThroughput());
+        if (config.maxConcurrency() != null) {
+            builder.maxConcurrency(config.maxConcurrency());
+        }
+        s3 = builder.build();
+        s3Sync = S3Client.builder().build();
         transferManager = S3TransferManager.builder()
                                            .s3Client(s3)
                                            .build();

test/s3-benchmarks/src/main/java/software/amazon/awssdk/s3benchmarks/BenchmarkRunner.java

@@ -45,6 +45,11 @@ public final class BenchmarkRunner {
 
     private static final String TIMEOUT = "timeoutInMin";
 
+    private static final String CONN_ACQ_TIMEOUT_IN_SEC = "connAcqTimeoutInSec";
+
+    private static final String FORCE_CRT_HTTP_CLIENT = "crtHttp";
+    private static final String MAX_CONCURRENCY = "maxConcurrency";
+
     private static final Map<TransferManagerOperation, Function<TransferManagerBenchmarkConfig, TransferManagerBenchmark>>
         OPERATION_TO_BENCHMARK_V1 = new EnumMap<>(TransferManagerOperation.class);
     private static final Map<TransferManagerOperation, Function<TransferManagerBenchmarkConfig, TransferManagerBenchmark>>
@@ -83,8 +88,8 @@ public static void main(String... args) throws org.apache.commons.cli.ParseExcep
         options.addOption(null, CHECKSUM_ALGORITHM, true, "The checksum algorithm to use");
         options.addOption(null, ITERATION, true, "The number of iterations");
         options.addOption(null, READ_BUFFER_IN_MB, true, "Read buffer size in MB");
-        options.addOption(null, VERSION, true, "The major version of the transfer manager to run test: v1 | v2 | crt, default: "
-                                               + "v2");
+        options.addOption(null, VERSION, true, "The major version of the transfer manager to run test: "
+                                               + "v1 | v2 | crt | java, default: v2");
         options.addOption(null, PREFIX, true, "S3 Prefix used in downloadDirectory and uploadDirectory");
 
         options.addOption(null, CONTENT_LENGTH, true, "Content length to upload from memory. Used only in the "
@@ -93,6 +98,12 @@ public static void main(String... args) throws org.apache.commons.cli.ParseExcep
 
         options.addOption(null, TIMEOUT, true, "Amount of minute to wait before a single operation "
                                                + "times out and is cancelled. Optional, defaults to 10 minutes if no specified");
+        options.addOption(null, CONN_ACQ_TIMEOUT_IN_SEC, true, "Timeout for acquiring an already-established"
+                                                               + " connection from a connection pool to a remote service.");
+        options.addOption(null, FORCE_CRT_HTTP_CLIENT, true,
+                          "Force the CRT http client to be used in JavaBased benchmarks");
+        options.addOption(null, MAX_CONCURRENCY, true,
+                          "The Maximum number of allowed concurrent requests. For HTTP/1.1 this is the same as max connections.");
 
         CommandLine cmd = parser.parse(options, args);
         TransferManagerBenchmarkConfig config = parseConfig(cmd);
@@ -114,11 +125,22 @@ public static void main(String... args) throws org.apache.commons.cli.ParseExcep
                 if (operation == TransferManagerOperation.DOWNLOAD) {
                     benchmark = new CrtS3ClientDownloadBenchmark(config);
                     break;
-                } else if (operation == TransferManagerOperation.UPLOAD) {
+                }
+                if (operation == TransferManagerOperation.UPLOAD) {
                     benchmark = new CrtS3ClientUploadBenchmark(config);
                     break;
                 }
                 throw new UnsupportedOperationException();
+            case JAVA:
+                if (operation == TransferManagerOperation.UPLOAD) {
+                    benchmark = new JavaS3ClientUploadBenchmark(config);
+                    break;
+                }
+                if (operation == TransferManagerOperation.COPY) {
+                    benchmark = new JavaS3ClientCopyBenchmark(config);
+                    break;
+                }
+                throw new UnsupportedOperationException("Java based s3 client benchmark only support upload and copy");
             default:
                 throw new UnsupportedOperationException();
         }
@@ -158,6 +180,15 @@ private static TransferManagerBenchmarkConfig parseConfig(CommandLine cmd) {
         Duration timeout = cmd.getOptionValue(TIMEOUT) == null ? null :
                            Duration.ofMinutes(Long.parseLong(cmd.getOptionValue(TIMEOUT)));
 
+        Long connAcqTimeoutInSec = cmd.getOptionValue(CONN_ACQ_TIMEOUT_IN_SEC) == null ? null :
+                                   Long.parseLong(cmd.getOptionValue(CONN_ACQ_TIMEOUT_IN_SEC));
+
+        Boolean forceCrtHttpClient = cmd.getOptionValue(FORCE_CRT_HTTP_CLIENT) != null
+                                     && Boolean.parseBoolean(cmd.getOptionValue(FORCE_CRT_HTTP_CLIENT));
+
+        Integer maxConcurrency = cmd.getOptionValue(MAX_CONCURRENCY) == null ? null :
+                                 Integer.parseInt(cmd.getOptionValue(MAX_CONCURRENCY));
+
         return TransferManagerBenchmarkConfig.builder()
                                              .key(key)
                                              .bucket(bucket)
@@ -171,6 +202,9 @@ private static TransferManagerBenchmarkConfig parseConfig(CommandLine cmd) {
                                              .prefix(prefix)
                                              .contentLengthInMb(contentLengthInMb)
                                              .timeout(timeout)
+                                             .connectionAcquisitionTimeoutInSec(connAcqTimeoutInSec)
+                                             .forceCrtHttpClient(forceCrtHttpClient)
+                                             .maxConcurrency(maxConcurrency)
                                              .build();
     }
 
@@ -185,6 +219,7 @@ public enum TransferManagerOperation {
     private enum SdkVersion {
         V1,
         V2,
-        CRT
+        CRT,
+        JAVA
     }
 }