Performance Framework Investigation

common/perf-test-core/src/main/java/com/azure/perf/test/core/ApiPerfTestBase.java

@@ -171,13 +171,7 @@ public Mono<Void> runAllAsync(long endNanoTime) {
                     sink.complete();
                 }
             })
-            .flatMap(ignored -> {
-                if (System.nanoTime() < endNanoTime) {
-                    return runTestAsync();
-                } else {
-                    return Mono.just(0);
-                }
-            }, 1)
+            .flatMap(ignored -> runTestAsync(), 1)
             .doOnNext(result -> {
                 completedOperations += result;
                 lastCompletionNanoTime = System.nanoTime() - startNanoTime;

common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java

@@ -7,23 +7,20 @@
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
-import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
 import java.io.IOException;
-import java.lang.reflect.InvocationTargetException;
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
-import java.util.stream.IntStream;
 
 /**
  * Represents the main program class which reflectively runs and manages the performance tests.
@@ -57,7 +54,6 @@ private static double getOperationsPerSecond(PerfTestBase<?>[] tests) {
      *
      * @param classes the performance test classes to execute.
      * @param args the command line arguments ro run performance tests with.
-     *
      * @throws RuntimeException if the execution fails.
      */
     public static void run(Class<?>[] classes, String[] args) {
@@ -80,8 +76,7 @@ public static void run(Class<?>[] classes, String[] args) {
         PerfStressOptions[] options = classList.stream().map(c -> {
             try {
                 return c.getConstructors()[0].getParameterTypes()[0].getConstructors()[0].newInstance();
-            } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
-                | InvocationTargetException | SecurityException e) {
+            } catch (ReflectiveOperationException e) {
                 throw new RuntimeException(e);
             }
         }).toArray(i -> new PerfStressOptions[i]);
@@ -92,7 +87,6 @@ public static void run(Class<?>[] classes, String[] args) {
             jc.addCommand(commands[i], options[i]);
         }
 
-
         jc.parse(args);
 
         String parsedCommand = jc.getParsedCommand();
@@ -114,7 +108,6 @@ private static String getCommandName(String testName) {
      *
      * @param testClass the performance test class to execute.
      * @param options the configuration ro run performance test with.
-     *
      * @throws RuntimeException if the execution fails.
      */
     public static void run(Class<?> testClass, PerfStressOptions options) {
@@ -130,16 +123,16 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
 
         System.out.println();
         System.out.println();
-        Disposable setupStatus = printStatus("=== Setup ===", () -> ".", false, false);
-        Disposable cleanupStatus = null;
+
+        Timer setupStatus = printStatus("=== Setup ===", () -> ".", false, false);
+        Timer cleanupStatus = null;
 
         PerfTestBase<?>[] tests = new PerfTestBase<?>[options.getParallel()];
 
         for (int i = 0; i < options.getParallel(); i++) {
             try {
                 tests[i] = (PerfTestBase<?>) testClass.getConstructor(options.getClass()).newInstance(options);
-            } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
-                | InvocationTargetException | SecurityException | NoSuchMethodException e) {
+            } catch (ReflectiveOperationException e) {
                 throw new RuntimeException(e);
             }
         }
@@ -151,24 +144,22 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
 
             try {
                 Flux.just(tests).flatMap(PerfTestBase::setupAsync).blockLast();
-                setupStatus.dispose();
+                setupStatus.cancel();
 
                 if (options.getTestProxies() != null && !options.getTestProxies().isEmpty()) {
-                    Disposable recordStatus = printStatus("=== Record and Start Playback ===", () -> ".", false, false);
-
-                    try {
-                        ForkJoinPool forkJoinPool = new ForkJoinPool(tests.length);
-                        forkJoinPool.submit(() -> {
-                            IntStream.range(0, tests.length).parallel().forEach(i -> tests[i].postSetupAsync().block());
-                        }).get();
-                    } catch (InterruptedException | ExecutionException e) {
-                        System.err.println("Error occurred when submitting jobs to ForkJoinPool. " + System.lineSeparator() + e);
-                        e.printStackTrace(System.err);
-                        throw new RuntimeException(e);
-                    }
+                    Timer recordStatus = printStatus("=== Record and Start Playback ===", () -> ".", false, false);
+
+                    int parallel = tests.length;
+                    Flux.range(0, parallel)
+                        .parallel(parallel)
+                        .runOn(Schedulers.parallel())
+                        .flatMap(i -> tests[i].postSetupAsync())
+                        .sequential()
+                        .then()
+                        .block();
 
                     startedPlayback = true;
-                    recordStatus.dispose();
+                    recordStatus.cancel();
                 }
 
                 if (options.getWarmup() > 0) {
@@ -185,15 +176,15 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
             } finally {
                 try {
                     if (startedPlayback) {
-                        Disposable playbackStatus = printStatus("=== Stop Playback ===", () -> ".", false, false);
+                        Timer playbackStatus = printStatus("=== Stop Playback ===", () -> ".", false, false);
                         Flux.just(tests).flatMap(perfTestBase -> {
                             if (perfTestBase instanceof ApiPerfTestBase) {
                                 return ((ApiPerfTestBase<?>) perfTestBase).stopPlaybackAsync();
                             } else {
                                 return Mono.error(new IllegalStateException("Test Proxy not supported."));
                             }
                         }).blockLast();
-                        playbackStatus.dispose();
+                        playbackStatus.cancel();
                     }
                 } finally {
                     if (!options.isNoCleanup()) {
@@ -214,7 +205,7 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
         }
 
         if (cleanupStatus != null) {
-            cleanupStatus.dispose();
+            cleanupStatus.cancel();
         }
     }
 
@@ -226,7 +217,6 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
      * @param parallel the number of parallel threads to run the performance test on.
      * @param durationSeconds the duration for which performance test should be run on.
      * @param title the title of the performance tests.
-     *
      * @throws RuntimeException if the execution fails.
      * @throws IllegalStateException if zero operations completed of the performance test.
      */
@@ -235,7 +225,7 @@ public static void runTests(PerfTestBase<?>[] tests, boolean sync, int parallel,
         long endNanoTime = System.nanoTime() + ((long) durationSeconds * 1000000000);
 
         long[] lastCompleted = new long[]{0};
-        Disposable progressStatus = printStatus(
+        Timer progressStatus = printStatus(
             "=== " + title + " ===" + System.lineSeparator() + "Current\t\tTotal\t\tAverage", () -> {
                 long totalCompleted = getCompletedOperations(tests);
                 long currentCompleted = totalCompleted - lastCompleted[0];
@@ -258,13 +248,7 @@ public static void runTests(PerfTestBase<?>[] tests, boolean sync, int parallel,
 
                 forkJoinPool.invokeAll(operations);
 
-                // Sleep until the tests complete.
-                Thread.sleep(durationSeconds * 1000L);
-
-                forkJoinPool.shutdown();
-
-                // Wait 10 seconds for operations to shut down.
-                forkJoinPool.awaitTermination(10, TimeUnit.SECONDS);
+                forkJoinPool.awaitQuiescence(durationSeconds + 1, TimeUnit.SECONDS);
             } else {
                 // Exceptions like OutOfMemoryError are handled differently by the default Reactor schedulers. Instead of terminating the
                 // Flux, the Flux will hang and the exception is only sent to the thread's uncaughtExceptionHandler and the Reactor
@@ -278,19 +262,16 @@ public static void runTests(PerfTestBase<?>[] tests, boolean sync, int parallel,
                 Flux.range(0, parallel)
                     .parallel(parallel)
                     .runOn(Schedulers.parallel())
-                    .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, Math.min(parallel, 1000 / parallel), 1)
+                    .flatMap(i -> tests[i].runAllAsync(endNanoTime))
+                    .sequential()
                     .then()
                     .block();
             }
-        } catch (InterruptedException e) {
-            System.err.println("Error occurred when submitting jobs to ForkJoinPool. " + System.lineSeparator() + e);
-            e.printStackTrace(System.err);
-            throw new RuntimeException(e);
         } catch (Exception e) {
             System.err.println("Error occurred running tests: " + System.lineSeparator() + e);
             e.printStackTrace(System.err);
         } finally {
-            progressStatus.dispose();
+            progressStatus.cancel();
         }
 
         System.out.println("=== Results ===");
@@ -311,23 +292,33 @@ public static void runTests(PerfTestBase<?>[] tests, boolean sync, int parallel,
         System.out.println();
     }
 
-    private static Disposable printStatus(String header, Supplier<Object> status, boolean newLine, boolean printFinalStatus) {
+    private static Timer printStatus(String header, Supplier<Object> status, boolean newLine, boolean printFinalStatus) {
         System.out.println(header);
 
         boolean[] needsExtraNewline = new boolean[]{false};
 
-        return Flux.interval(Duration.ofSeconds(1)).doFinally(s -> {
-            if (printFinalStatus) {
+        Timer timer = new Timer(true);
+        timer.scheduleAtFixedRate(new TimerTask() {
+            @Override
+            public void run() {
                 printStatusHelper(status, newLine, needsExtraNewline);
             }
 
-            if (needsExtraNewline[0]) {
+            @Override
+            public boolean cancel() {
+                if (printFinalStatus) {
+                    printStatusHelper(status, newLine, needsExtraNewline);
+                }
+
+                if (needsExtraNewline[0]) {
+                    System.out.println();
+                }
                 System.out.println();
+                return super.cancel();
             }
-            System.out.println();
-        }).subscribe(i -> {
-            printStatusHelper(status, newLine, needsExtraNewline);
-        });
+        }, 1000, 1000);
+
+        return timer;
     }
 
     private static void printStatusHelper(Supplier<Object> status, boolean newLine, boolean[] needsExtraNewline) {

sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java

@@ -7,7 +7,6 @@
 import com.azure.storage.blob.perf.core.ContainerTest;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import java.util.UUID;
@@ -25,12 +24,15 @@ public Mono<Void> globalSetupAsync() {
         // drastically less CPU usage and throughput, there is ongoing discussions with Reactor Netty on what causes
         // this edge case, whether we had a design flaw in the performance tests, or if there is a configuration change
         // needed in Reactor Netty.
+        int parallel = options.getParallel();
         return super.globalSetupAsync().then(
             Flux.range(0, options.getCount())
-                .parallel(options.getParallel())
-                .runOn(Schedulers.boundedElastic())
+                .parallel(parallel)
+                .runOn(Schedulers.parallel())
                 .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID())
-                    .upload(Flux.empty(), null), false, Math.min(options.getParallel(), 1000 / options.getParallel()), 1)
+                    .getBlockBlobAsyncClient()
+                    .upload(Flux.empty(), 0L), false, parallel, 1)
+                .sequential()
                 .then());
     }
 

sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java

@@ -3,21 +3,38 @@
 
 package com.azure.storage.blob.perf.core;
 
+import com.azure.storage.blob.models.ParallelTransferOptions;
 import com.azure.storage.blob.perf.BlobPerfStressOptions;
 import reactor.core.publisher.Mono;
 
 import static com.azure.perf.test.core.TestDataCreationHelper.createRandomByteBufferFlux;
 
 public abstract class AbstractDownloadTest <TOptions extends BlobPerfStressOptions> extends BlobTestBase<TOptions> {
+    private static final long GB = 1024 * 1024 * 1024;
 
     public AbstractDownloadTest(TOptions options) {
         super(options, BLOB_NAME_PREFIX);
     }
 
     // Upload one blob for the whole test run. All tests can download the same blob
     public Mono<Void> globalSetupAsync() {
+        /*
+         * Uploading the blob is set to use a "single shot" and block size of 1GB to have the blob upload over a single
+         * connection. There was an investigation into an issue in the performance tests where all connections were
+         * being handled by a single IO thread, the root cause was found to be that when 1GB download set up resources
+         * here the sequential uploading of 4MB blocks resulted in Reactor Netty using the same thread to manage all
+         * upload connections (1GB / 4MB = 256). The test only used 8 threads to perform parallel 1GB download and since
+         * 8 connections already existed they were reused and managed by that single IO thread. So, changing upload to
+         * be done with a single connection fixes that, where when 8 threads begin performing download at the same time
+         * Reactor Netty has to even spread those requests over the available IO threads instead of being pinned to
+         * the one IO thread.
+         *
+         * In the future there will be work to separate the HttpClients used to perform resource preparation and running
+         * the performance test. As part of that work this can be reverted to using the default ParallelTransferOptions.
+         */
         return super.globalSetupAsync()
-            .then(blobAsyncClient.upload(createRandomByteBufferFlux(options.getSize()), null))
+            .then(blobAsyncClient.upload(createRandomByteBufferFlux(options.getSize()), new ParallelTransferOptions()
+                .setMaxSingleUploadSizeLong(GB).setBlockSizeLong(GB)))
             .then();
     }
 }

sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/BlobTestBase.java

@@ -3,11 +3,6 @@
 
 package com.azure.storage.blob.perf.core;
 
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Random;
-
 import com.azure.security.keyvault.keys.cryptography.models.KeyWrapAlgorithm;
 import com.azure.storage.blob.BlobAsyncClient;
 import com.azure.storage.blob.BlobClient;
@@ -16,7 +11,8 @@
 import com.azure.storage.blob.specialized.BlockBlobClient;
 import com.azure.storage.blob.specialized.cryptography.EncryptedBlobClientBuilder;
 import com.azure.storage.blob.specialized.cryptography.EncryptionVersion;
-import reactor.core.publisher.Mono;
+
+import java.util.Random;
 
 public abstract class BlobTestBase<TOptions extends BlobPerfStressOptions> extends ContainerTest<TOptions> {
 
@@ -63,27 +59,4 @@ public BlobTestBase(TOptions options, String blobName) {
         blockBlobClient = blobContainerClient.getBlobClient(blobName).getBlockBlobClient();
         blockBlobAsyncClient = blobContainerAsyncClient.getBlobAsyncClient(blobName).getBlockBlobAsyncClient();
     }
-
-    @Override
-    public Mono<Void> globalSetupAsync() {
-        return super.globalSetupAsync()
-            .then();
-    }
-
-    @Override
-    public Mono<Void> setupAsync() {
-        return super.setupAsync()
-            .then();
-    }
-
-    public long copyStream(InputStream input, OutputStream out) throws IOException {
-        long transferred = 0;
-        byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
-        int read;
-        while ((read = input.read(buffer, 0, DEFAULT_BUFFER_SIZE)) >= 0) {
-            out.write(buffer, 0, read);
-            transferred += read;
-        }
-        return transferred;
-    }
 }

sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ContainerTest.java

@@ -12,7 +12,7 @@
 import java.util.UUID;
 
 public abstract class ContainerTest<TOptions extends PerfStressOptions> extends ServiceTest<TOptions> {
-    protected static final String CONTAINER_NAME = "perfstress-" + UUID.randomUUID().toString();
+    protected static final String CONTAINER_NAME = "perfstress-" + UUID.randomUUID();
 
     protected final BlobContainerClient blobContainerClient;
     protected final BlobContainerAsyncClient blobContainerAsyncClient;