Refactor s3-benchmarks and include v1 transfermanager tests

pom.xml

@@ -112,7 +112,7 @@
         <rxjava.version>2.2.21</rxjava.version>
         <commons-codec.verion>1.10</commons-codec.verion>
         <jmh.version>1.29</jmh.version>
-        <awscrt.version>0.13.2</awscrt.version>
+        <awscrt.version>0.13.5</awscrt.version>
 
         <!--Test dependencies -->
         <junit.version>4.13.1</junit.version>

test/s3-benchmarks/README.md

@@ -1,7 +1,7 @@
 # S3 Benchmark Harness
 
 
-This module contains performance tests for `S3AsyncClient` and 
+This module contains performance tests for `S3AsyncClient` and
 `S3TransferManager`
 
 ## How to run
@@ -11,8 +11,8 @@ This module contains performance tests for `S3AsyncClient` and
 mvn clean install -pl :s3-benchmarks -P quick --am
 
 # download
-java -jar s3-benchmarks.jar --bucket=bucket --key=key -file=/path/to/destionfile/ --operation=download --partSizeInMB=20 --maxThroughput=100.0
+java -jar s3-benchmarks.jar --bucket=bucket --key=key -file=/path/to/destionfile/ --operation=download --partSizeInMB=20 --maxThroughput=100.0 --version=v2 --warmupConcurrency=300
 
 # upload
-java -jar s3-benchmarks.jar --bucket=bucket --key=key -file=/path/to/sourcefile/ --operation=upload --partSizeInMB=20 --maxThroughput=100.0
+java -jar s3-benchmarks.jar --bucket=bucket --key=key -file=/path/to/sourcefile/ --operation=upload --partSizeInMB=20 --maxThroughput=100.0 --version=v2 --warmupConcurrency=300
 ```

test/s3-benchmarks/pom.xml

@@ -30,6 +30,7 @@
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
         <awsjavasdk.version>${project.version}</awsjavasdk.version>
+        <sdk-v1.version>1.12.1</sdk-v1.version>
     </properties>
     <name>AWS Java SDK :: Test :: S3 Benchmarks</name>
     <description>Contains benchmark code for S3 and TransferManager</description>
@@ -55,6 +56,11 @@
             <artifactId>s3-transfer-manager</artifactId>
             <version>${awsjavasdk.version}-PREVIEW</version>
         </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-s3</artifactId>
+            <version>${sdk-v1.version}</version>
+        </dependency>
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>test-utils</artifactId>

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

@@ -15,6 +15,11 @@
 
 package software.amazon.awssdk.s3benchmarks;
 
+import static software.amazon.awssdk.s3benchmarks.BenchmarkUtils.DEFAULT_WARMUP_CONCURRENCY;
+import static software.amazon.awssdk.s3benchmarks.BenchmarkUtils.PRE_WARMUP_ITERATIONS;
+import static software.amazon.awssdk.s3benchmarks.BenchmarkUtils.PRE_WARMUP_RUNS;
+import static software.amazon.awssdk.transfer.s3.SizeConstant.KB;
+import static software.amazon.awssdk.transfer.s3.SizeConstant.MB;
 import static software.amazon.awssdk.utils.FunctionalUtils.runAndLogError;
 
 import java.io.File;
@@ -28,46 +33,39 @@
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
-import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.testutils.RandomTempFile;
-import software.amazon.awssdk.transfer.s3.S3TransferManager;
 import software.amazon.awssdk.transfer.s3.internal.S3CrtAsyncClient;
 import software.amazon.awssdk.utils.Logger;
 
 public abstract class BaseTransferManagerBenchmark implements TransferManagerBenchmark {
-    protected static final int WARMUP_ITERATIONS = 10;
-    protected static final int BENCHMARK_ITERATIONS = 10;
 
     private static final Logger logger = Logger.loggerFor("TransferManagerBenchmark");
     private static final String WARMUP_KEY = "warmupobject";
 
-    protected final S3TransferManager transferManager;
     protected final S3CrtAsyncClient s3;
     protected final S3Client s3Sync;
     protected final String bucket;
     protected final String key;
     protected final String path;
-    private final File file;
+    protected final int warmupConcurrency;
+    protected final File tmpFile;
 
     BaseTransferManagerBenchmark(TransferManagerBenchmarkConfig config) {
         logger.info(() -> "Benchmark config: " + config);
-        Long partSizeInMb = config.partSizeInMb() == null ? null : config.partSizeInMb() * 1024 * 1024L;
+        Long partSizeInMb = config.partSizeInMb() == null ? null : config.partSizeInMb() * MB;
         s3 = S3CrtAsyncClient.builder()
                              .targetThroughputInGbps(config.targetThroughput())
                              .minimumPartSizeInBytes(partSizeInMb)
                              .build();
         s3Sync = S3Client.builder()
                          .build();
-        transferManager = S3TransferManager.builder()
-                                           .s3ClientConfiguration(b -> b.targetThroughputInGbps(config.targetThroughput())
-                                           .minimumPartSizeInBytes(partSizeInMb))
-                                           .build();
         bucket = config.bucket();
         key = config.key();
         path = config.filePath();
+        warmupConcurrency = config.warmupConcurrency() == null ? DEFAULT_WARMUP_CONCURRENCY : config.warmupConcurrency();
         try {
-            file = new RandomTempFile(1024 * 1000L);
+            tmpFile = new RandomTempFile(1000 * KB);
         } catch (IOException e) {
             logger.error(() -> "Failed to create the file");
             throw new RuntimeException("Failed to create the temp file", e);
@@ -95,36 +93,15 @@ protected void additionalWarmup() {
 
     protected abstract void doRunBenchmark();
 
-    protected final void printOutResult(List<Double> metrics, String name) {
-        logger.info(() -> String.format("===============  %s Result ================", name));
-        logger.info(() -> "" + metrics);
-        double averageLatency = metrics.stream()
-                                       .mapToDouble(a -> a)
-                                       .average()
-                                       .orElse(0.0);
-
-        double lowestLatency = metrics.stream()
-                                      .mapToDouble(a -> a)
-                                      .min().orElse(0.0);
-
-        HeadObjectResponse headObjectResponse = s3Sync.headObject(b -> b.bucket(bucket).key(key));
-        double contentLengthInGigabit = (headObjectResponse.contentLength() / (1000 * 1000 * 1000.0)) * 8.0;
-        logger.info(() -> "Average latency (s): " + averageLatency);
-        logger.info(() -> "Object size (Gigabit): " + contentLengthInGigabit);
-        logger.info(() -> "Average throughput (Gbps): " + contentLengthInGigabit / averageLatency);
-        logger.info(() -> "Highest average throughput (Gbps): " + contentLengthInGigabit / lowestLatency);
-        logger.info(() -> "==========================================================");
-    }
-
     private void cleanup() {
         s3Sync.deleteObject(b -> b.bucket(bucket).key(WARMUP_KEY));
-        transferManager.close();
+        s3.close();
     }
 
     private void warmUp() throws InterruptedException {
         logger.info(() -> "Starting to warm up");
 
-        for (int i = 0; i < WARMUP_ITERATIONS; i++) {
+        for (int i = 0; i < PRE_WARMUP_ITERATIONS; i++) {
             warmUpUploadBatch();
             warmUpDownloadBatch();
 
@@ -136,7 +113,7 @@ private void warmUp() throws InterruptedException {
 
     private void warmUpDownloadBatch() {
         List<CompletableFuture<?>> futures = new ArrayList<>();
-        for (int i = 0; i < 20; i++) {
+        for (int i = 0; i < PRE_WARMUP_RUNS; i++) {
             Path tempFile = RandomTempFile.randomUncreatedFile().toPath();
             futures.add(s3.getObject(GetObjectRequest.builder().bucket(bucket).key(WARMUP_KEY).build(),
                                      AsyncResponseTransformer.toFile(tempFile)).whenComplete((r, t) -> runAndLogError(
@@ -148,9 +125,9 @@ private void warmUpDownloadBatch() {
 
     private void warmUpUploadBatch() {
         List<CompletableFuture<?>> futures = new ArrayList<>();
-        for (int i = 0; i < 20; i++) {
+        for (int i = 0; i < PRE_WARMUP_RUNS; i++) {
             futures.add(s3.putObject(PutObjectRequest.builder().bucket(bucket).key(WARMUP_KEY).build(),
-                                     AsyncRequestBody.fromFile(file)));
+                                     AsyncRequestBody.fromFile(tmpFile)));
         }
 
         CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])).join();

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

@@ -29,6 +29,8 @@ public class BenchmarkRunner {
     private static final String MAX_THROUGHPUT = "maxThroughput";
     private static final String KEY = "key";
     private static final String OPERATION = "operation";
+    private static final String WARMUP_CONCURRENCY = "warmupConcurrency";
+    private static final String VERSION = "version";
 
     private BenchmarkRunner() {
     }
@@ -42,20 +44,35 @@ public static void main(String... args) throws org.apache.commons.cli.ParseExcep
                                                     + "uploaded");
         options.addRequiredOption(null, BUCKET, true, "The s3 bucket");
         options.addRequiredOption(null, KEY, true, "The s3 key");
-        options.addRequiredOption(null, OPERATION, true, "The operation to benchmark against");
+        options.addRequiredOption(null, OPERATION, true, "The operation to run tests: download | upload");
         options.addOption(null, PART_SIZE_IN_MB, true, "Part size in MB");
         options.addOption(null, MAX_THROUGHPUT, true, "The max throughput");
+        options.addOption(null, WARMUP_CONCURRENCY, true, "The number of concurrent requests to send during warmup, default: "
+                                                          + "100");
+        options.addOption(null, VERSION, true, "The major version of the transfer manager to run test: v1 | v2, default: v2");
 
         CommandLine cmd = parser.parse(options, args);
         TransferManagerBenchmarkConfig config = parseConfig(cmd);
         TransferManagerOperation operation = TransferManagerOperation.valueOf(cmd.getOptionValue(OPERATION)
                                                                                  .toUpperCase(Locale.ENGLISH));
+
+        SdkVersion version = SdkVersion.valueOf(cmd.getOptionValue(VERSION, "V2")
+                                                   .toUpperCase(Locale.ENGLISH));
+
         switch (operation) {
             case DOWNLOAD:
-                TransferManagerBenchmark.download(config).run();
+                if (version.equals(SdkVersion.V2)) {
+                    TransferManagerBenchmark.download(config).run();
+                } else {
+                    TransferManagerBenchmark.v1Download(config).run();
+                }
                 break;
             case UPLOAD:
-                TransferManagerBenchmark.upload(config).run();
+                if (version.equals(SdkVersion.V2)) {
+                    TransferManagerBenchmark.upload(config).run();
+                } else {
+                    TransferManagerBenchmark.v1Upload(config).run();
+                }
                 break;
             default:
                 throw new UnsupportedOperationException();
@@ -72,9 +89,13 @@ private static TransferManagerBenchmarkConfig parseConfig(CommandLine cmd) {
         Double maxThroughput = cmd.getOptionValue(MAX_THROUGHPUT) == null ? null :
                                Double.parseDouble(cmd.getOptionValue(MAX_THROUGHPUT));
 
+        Integer warmupConcurrency = cmd.getOptionValue(WARMUP_CONCURRENCY) == null ? null :
+                                    Integer.parseInt(cmd.getOptionValue(WARMUP_CONCURRENCY));
+
         return TransferManagerBenchmarkConfig.builder()
                                              .key(key)
                                              .bucket(bucket)
+                                             .warmupConcurrency(warmupConcurrency)
                                              .partSizeInMb(partSize)
                                              .targetThroughput(maxThroughput)
                                              .filePath(filePath)
@@ -85,4 +106,9 @@ private enum TransferManagerOperation {
         DOWNLOAD,
         UPLOAD
     }
+
+    private enum SdkVersion {
+        V1,
+        V2
+    }
 }

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

@@ -0,0 +1,52 @@
+/*
+ * 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 java.util.List;
+import software.amazon.awssdk.utils.Logger;
+
+public class BenchmarkUtils {
+    protected static final int PRE_WARMUP_ITERATIONS = 10;
+    protected static final int PRE_WARMUP_RUNS = 20;
+    protected static final int BENCHMARK_ITERATIONS = 10;
+    protected static final int DEFAULT_WARMUP_CONCURRENCY = 100;
+    protected static final String WARMUP_KEY = "warmupobject";
+
+    private static final Logger logger = Logger.loggerFor("TransferManagerBenchmark");
+
+    private BenchmarkUtils() {
+    }
+
+    public static void printOutResult(List<Double> metrics, String name, long contentLengthInByte) {
+        logger.info(() -> String.format("===============  %s Result ================", name));
+        logger.info(() -> "" + metrics);
+        double averageLatency = metrics.stream()
+                                       .mapToDouble(a -> a)
+                                       .average()
+                                       .orElse(0.0);
+
+        double lowestLatency = metrics.stream()
+                                      .mapToDouble(a -> a)
+                                      .min().orElse(0.0);
+
+        double contentLengthInGigabit = (contentLengthInByte / (1000 * 1000 * 1000.0)) * 8.0;
+        logger.info(() -> "Average latency (s): " + averageLatency);
+        logger.info(() -> "Object size (Gigabit): " + contentLengthInGigabit);
+        logger.info(() -> "Average throughput (Gbps): " + contentLengthInGigabit / averageLatency);
+        logger.info(() -> "Highest average throughput (Gbps): " + contentLengthInGigabit / lowestLatency);
+        logger.info(() -> "==========================================================");
+    }
+}

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

@@ -34,4 +34,12 @@ static TransferManagerBenchmark upload(TransferManagerBenchmarkConfig config) {
         return new TransferManagerUploadBenchmark(config);
     }
 
+    static TransferManagerBenchmark v1Download(TransferManagerBenchmarkConfig config) {
+        return new V1TransferManagerDownloadBenchmark(config);
+    }
+
+    static TransferManagerBenchmark v1Upload(TransferManagerBenchmarkConfig config) {
+        return new V1TransferManagerUploadBenchmark(config);
+    }
+
 }

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

@@ -21,13 +21,15 @@ public class TransferManagerBenchmarkConfig {
     private final String key;
     private final Double targetThroughput;
     private final Long partSizeInMb;
+    private final Integer warmupConcurrency;
 
     private TransferManagerBenchmarkConfig(Builder builder) {
         this.filePath = builder.filePath;
         this.bucket = builder.bucket;
         this.key = builder.key;
         this.targetThroughput = builder.targetThroughput;
         this.partSizeInMb = builder.partSizeInMb;
+        this.warmupConcurrency = builder.warmupConcurrency;
     }
 
     public String filePath() {
@@ -50,6 +52,10 @@ public Long partSizeInMb() {
         return partSizeInMb;
     }
 
+    public Integer warmupConcurrency() {
+        return warmupConcurrency;
+    }
+
     public static Builder builder() {
         return new Builder();
     }
@@ -61,7 +67,8 @@ public String toString() {
                ", bucket: '" + bucket + '\'' +
                ", key: '" + key + '\'' +
                ", targetThroughput: " + targetThroughput +
-               ", partSizeInMB: " + partSizeInMb +
+               ", partSizeInMb: " + partSizeInMb +
+               ", warmupConcurrency: " + warmupConcurrency +
                '}';
     }
 
@@ -71,6 +78,7 @@ static final class Builder {
         private String key;
         private Double targetThroughput;
         private Long partSizeInMb;
+        private Integer warmupConcurrency;
 
         public Builder filePath(String filePath) {
             this.filePath = filePath;
@@ -97,6 +105,11 @@ public Builder partSizeInMb(Long partSizeInMb) {
             return this;
         }
 
+        public Builder warmupConcurrency(Integer warmupConcurrency) {
+            this.warmupConcurrency = warmupConcurrency;
+            return this;
+        }
+
         public TransferManagerBenchmarkConfig build() {
             return new TransferManagerBenchmarkConfig(this);
         }