From 40ebb7030546717e6b10d3fc0151b8bc5bb48c7b Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Tue, 8 Nov 2022 16:55:55 -0500 Subject: [PATCH 01/12] Make some changes --- .../perf/test/core/PerfStressProgram.java | 84 +++++++++---------- .../storage/blob/perf/ListBlobsTest.java | 4 +- .../storage/blob/perf/core/BlobTestBase.java | 31 +------ .../storage/blob/perf/core/ContainerTest.java | 2 +- .../storage/blob/perf/core/ServiceTest.java | 21 +---- 5 files changed, 46 insertions(+), 96 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 1b7e36351258..1e3460a5fd29 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -7,23 +7,21 @@ 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 +55,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) { @@ -81,7 +78,7 @@ public static void run(Class[] classes, String[] args) { try { return c.getConstructors()[0].getParameterTypes()[0].getConstructors()[0].newInstance(); } catch (InstantiationException | IllegalAccessException | IllegalArgumentException - | InvocationTargetException | SecurityException e) { + | InvocationTargetException | SecurityException e) { throw new RuntimeException(e); } }).toArray(i -> new PerfStressOptions[i]); @@ -92,7 +89,6 @@ public static void run(Class[] classes, String[] args) { jc.addCommand(commands[i], options[i]); } - jc.parse(args); String parsedCommand = jc.getParsedCommand(); @@ -114,7 +110,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 +125,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 +146,21 @@ 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); + + Flux.range(0, tests.length) + .parallel(tests.length) + .runOn(Schedulers.boundedElastic()) + .flatMap(i -> tests[i].postSetupAsync(), false, 1, 1) + .sequential() + .then() + .block(); startedPlayback = true; - recordStatus.dispose(); + recordStatus.cancel(); } if (options.getWarmup() > 0) { @@ -185,7 +177,7 @@ 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(); @@ -193,7 +185,7 @@ public static void run(Class testClass, PerfStressOptions options) { return Mono.error(new IllegalStateException("Test Proxy not supported.")); } }).blockLast(); - playbackStatus.dispose(); + playbackStatus.cancel(); } } finally { if (!options.isNoCleanup()) { @@ -214,7 +206,7 @@ public static void run(Class testClass, PerfStressOptions options) { } if (cleanupStatus != null) { - cleanupStatus.dispose(); + cleanupStatus.cancel(); } } @@ -226,7 +218,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 +226,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]; @@ -270,8 +261,9 @@ 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) + .runOn(Schedulers.boundedElastic()) + .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, 1, 1) + .sequential() .then() .block(); } @@ -283,7 +275,7 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, System.err.println("Error occurred running tests: " + System.lineSeparator() + e); e.printStackTrace(System.err); } finally { - progressStatus.dispose(); + progressStatus.cancel(); } System.out.println("=== Results ==="); @@ -304,23 +296,27 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, System.out.println(); } - private static Disposable printStatus(String header, Supplier status, boolean newLine, boolean printFinalStatus) { + private static Timer printStatus(String header, Supplier status, boolean newLine, boolean printFinalStatus) { System.out.println(header); boolean[] needsExtraNewline = new boolean[]{false}; - return Flux.interval(Duration.ofSeconds(1)).doFinally(s -> { - if (printFinalStatus) { - printStatusHelper(status, newLine, needsExtraNewline); - } + Timer timer = new Timer(true); + timer.scheduleAtFixedRate(new TimerTask() { + @Override + public void run() { + if (printFinalStatus) { + printStatusHelper(status, newLine, needsExtraNewline); + } - if (needsExtraNewline[0]) { + if (needsExtraNewline[0]) { + System.out.println(); + } System.out.println(); } - System.out.println(); - }).subscribe(i -> { - printStatusHelper(status, newLine, needsExtraNewline); - }); + }, 1000, 1000); + + return timer; } private static void printStatusHelper(Supplier status, boolean newLine, boolean[] needsExtraNewline) { diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index 71a3365de858..fec55b89dfae 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/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; @@ -30,7 +29,8 @@ public Mono globalSetupAsync() { .parallel(options.getParallel()) .runOn(Schedulers.boundedElastic()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) - .upload(Flux.empty(), null), false, Math.min(options.getParallel(), 1000 / options.getParallel()), 1) + .upload(Flux.empty(), null), false, 1, 1) + .sequential() .then()); } diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/BlobTestBase.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/BlobTestBase.java index 16d170c0b0ff..7eeaa84ebc45 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/BlobTestBase.java +++ b/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 extends ContainerTest { @@ -63,27 +59,4 @@ public BlobTestBase(TOptions options, String blobName) { blockBlobClient = blobContainerClient.getBlobClient(blobName).getBlockBlobClient(); blockBlobAsyncClient = blobContainerAsyncClient.getBlobAsyncClient(blobName).getBlockBlobAsyncClient(); } - - @Override - public Mono globalSetupAsync() { - return super.globalSetupAsync() - .then(); - } - - @Override - public Mono 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; - } } diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ContainerTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ContainerTest.java index 34dddf9b1a03..37d4ce002e83 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ContainerTest.java +++ b/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 extends ServiceTest { - 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; diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ServiceTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ServiceTest.java index be3f013c92c6..a3a6e1f9a519 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ServiceTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ServiceTest.java @@ -10,20 +10,16 @@ import com.azure.storage.blob.BlobServiceAsyncClient; import com.azure.storage.blob.BlobServiceClient; import com.azure.storage.blob.BlobServiceClientBuilder; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.scheduler.Schedulers; public abstract class ServiceTest extends PerfStressTest { protected final BlobServiceClient blobServiceClient; protected final BlobServiceAsyncClient blobServiceAsyncClient; - private final Configuration configuration; protected String connectionString; public ServiceTest(TOptions options) { super(options); - configuration = Configuration.getGlobalConfiguration().clone(); + Configuration configuration = Configuration.getGlobalConfiguration().clone(); connectionString = configuration.get("STORAGE_CONNECTION_STRING"); if (CoreUtils.isNullOrEmpty(connectionString)) { @@ -39,19 +35,4 @@ public ServiceTest(TOptions options) { blobServiceClient = builder.buildClient(); blobServiceAsyncClient = builder.buildAsyncClient(); } - - @Override - public Mono globalSetupAsync() { - // Arbitrarily run 1000 service get properties calls to warm up the connection pool used by the HttpClient. - // This helps guard against an edge case seen in Reactor Netty where only one IO thread could end up owning all - // connections in the connection pool. This results in 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. - return super.globalSetupAsync().then(Flux.range(0, 1000) - .parallel(options.getParallel()) - .runOn(Schedulers.boundedElastic()) - .flatMap(ignored -> blobServiceAsyncClient.getProperties(), false, - Math.min(options.getParallel(), 1000 / options.getParallel()), 1) - .then()); - } } From 4250b1b6c20e07232bd7a43edb73a17ea44ee4f9 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Tue, 8 Nov 2022 17:18:11 -0500 Subject: [PATCH 02/12] Fix logging --- .../java/com/azure/perf/test/core/PerfStressProgram.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index f0b09545ef76..05a24c8ea8d9 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -312,6 +312,11 @@ private static Timer printStatus(String header, Supplier status, boolean timer.scheduleAtFixedRate(new TimerTask() { @Override public void run() { + printStatusHelper(status, newLine, needsExtraNewline); + } + + @Override + public boolean cancel() { if (printFinalStatus) { printStatusHelper(status, newLine, needsExtraNewline); } @@ -320,6 +325,7 @@ public void run() { System.out.println(); } System.out.println(); + return super.cancel(); } }, 1000, 1000); From 5da0fa10ac7f595d390ac02b78daff8d96e406f3 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Tue, 8 Nov 2022 17:35:03 -0500 Subject: [PATCH 03/12] A few more changes --- .../com/azure/perf/test/core/PerfStressProgram.java | 12 +++++------- .../com/azure/storage/blob/perf/ListBlobsTest.java | 2 +- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 05a24c8ea8d9..5e61b47aa326 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -12,7 +12,6 @@ import reactor.core.scheduler.Schedulers; import java.io.IOException; -import java.lang.reflect.InvocationTargetException; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -77,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]); @@ -154,7 +152,7 @@ public static void run(Class testClass, PerfStressOptions options) { Flux.range(0, tests.length) .parallel(tests.length) .runOn(Schedulers.boundedElastic()) - .flatMap(i -> tests[i].postSetupAsync(), false, 1, 1) + .flatMap(i -> tests[i].postSetupAsync(), false, Schedulers.DEFAULT_POOL_SIZE, 1) .sequential() .then() .block(); @@ -254,8 +252,8 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, forkJoinPool.shutdown(); - // Wait 10 seconds for operations to shut down. - forkJoinPool.awaitTermination(10, TimeUnit.SECONDS); + // Wait 1 second for operations to shut down. + forkJoinPool.awaitTermination(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 @@ -269,7 +267,7 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, Flux.range(0, parallel) .parallel(parallel) .runOn(Schedulers.boundedElastic()) - .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, 1, 1) + .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, Schedulers.DEFAULT_POOL_SIZE, 1) .sequential() .then() .block(); diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index fec55b89dfae..436c1d6ba1b3 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java @@ -29,7 +29,7 @@ public Mono globalSetupAsync() { .parallel(options.getParallel()) .runOn(Schedulers.boundedElastic()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) - .upload(Flux.empty(), null), false, 1, 1) + .upload(Flux.empty(), null), false, Schedulers.DEFAULT_POOL_SIZE, 1) .sequential() .then()); } From 61e50303552eccbb13cc51cbb39dc0798c414c50 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Wed, 9 Nov 2022 07:27:38 -0500 Subject: [PATCH 04/12] Revert part of last change --- .../main/java/com/azure/perf/test/core/PerfStressProgram.java | 4 ++-- .../main/java/com/azure/storage/blob/perf/ListBlobsTest.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 5e61b47aa326..50eda4491ad8 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -152,7 +152,7 @@ public static void run(Class testClass, PerfStressOptions options) { Flux.range(0, tests.length) .parallel(tests.length) .runOn(Schedulers.boundedElastic()) - .flatMap(i -> tests[i].postSetupAsync(), false, Schedulers.DEFAULT_POOL_SIZE, 1) + .flatMap(i -> tests[i].postSetupAsync(), false, 1, 1) .sequential() .then() .block(); @@ -267,7 +267,7 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, Flux.range(0, parallel) .parallel(parallel) .runOn(Schedulers.boundedElastic()) - .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, Schedulers.DEFAULT_POOL_SIZE, 1) + .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, 1, 1) .sequential() .then() .block(); diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index 436c1d6ba1b3..fec55b89dfae 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java @@ -29,7 +29,7 @@ public Mono globalSetupAsync() { .parallel(options.getParallel()) .runOn(Schedulers.boundedElastic()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) - .upload(Flux.empty(), null), false, Schedulers.DEFAULT_POOL_SIZE, 1) + .upload(Flux.empty(), null), false, 1, 1) .sequential() .then()); } From d925279a4c54a4a267538ea6d366a1850b81691f Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Wed, 9 Nov 2022 12:26:20 -0500 Subject: [PATCH 05/12] Change Scheduler --- .../main/java/com/azure/perf/test/core/PerfStressProgram.java | 4 ++-- .../main/java/com/azure/storage/blob/perf/ListBlobsTest.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 50eda4491ad8..3d5ed3d5bfd8 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -151,7 +151,7 @@ public static void run(Class testClass, PerfStressOptions options) { Flux.range(0, tests.length) .parallel(tests.length) - .runOn(Schedulers.boundedElastic()) + .runOn(Schedulers.parallel()) .flatMap(i -> tests[i].postSetupAsync(), false, 1, 1) .sequential() .then() @@ -266,7 +266,7 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, Flux.range(0, parallel) .parallel(parallel) - .runOn(Schedulers.boundedElastic()) + .runOn(Schedulers.parallel()) .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, 1, 1) .sequential() .then() diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index fec55b89dfae..15ec3600c7b5 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java @@ -27,7 +27,7 @@ public Mono globalSetupAsync() { return super.globalSetupAsync().then( Flux.range(0, options.getCount()) .parallel(options.getParallel()) - .runOn(Schedulers.boundedElastic()) + .runOn(Schedulers.parallel()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) .upload(Flux.empty(), null), false, 1, 1) .sequential() From 577bc625129fd134721f0c4aa9fef185452bb0d5 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Wed, 9 Nov 2022 15:21:29 -0500 Subject: [PATCH 06/12] Another minor change to parallelism --- .../java/com/azure/perf/test/core/PerfStressProgram.java | 9 +++++---- .../java/com/azure/storage/blob/perf/ListBlobsTest.java | 5 +++-- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 3d5ed3d5bfd8..2274d8f78c77 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -149,10 +149,11 @@ public static void run(Class testClass, PerfStressOptions options) { if (options.getTestProxies() != null && !options.getTestProxies().isEmpty()) { Timer recordStatus = printStatus("=== Record and Start Playback ===", () -> ".", false, false); - Flux.range(0, tests.length) - .parallel(tests.length) + int parallel = tests.length; + Flux.range(0, parallel) + .parallel(parallel) .runOn(Schedulers.parallel()) - .flatMap(i -> tests[i].postSetupAsync(), false, 1, 1) + .flatMap(i -> tests[i].postSetupAsync(), false, Math.min(10000 / parallel, parallel), 1) .sequential() .then() .block(); @@ -267,7 +268,7 @@ 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, 1, 1) + .flatMap(i -> tests[i].runAllAsync(endNanoTime), false, Math.min(10000 / parallel, parallel), 1) .sequential() .then() .block(); diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index 15ec3600c7b5..4d3d432f9ca3 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java @@ -24,12 +24,13 @@ public Mono 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()) + .parallel(parallel) .runOn(Schedulers.parallel()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) - .upload(Flux.empty(), null), false, 1, 1) + .upload(Flux.empty(), null), false, Math.min(1000 / parallel, parallel), 1) .sequential() .then()); } From ab7a11b3342c001cffeaebc48ef71be9c63a2a7e Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Thu, 10 Nov 2022 11:20:43 -0500 Subject: [PATCH 07/12] Attempt to resolve thread pinning in download --- .../java/com/azure/perf/test/core/PerfStressProgram.java | 4 ++-- .../azure/storage/blob/perf/core/AbstractDownloadTest.java | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 2274d8f78c77..9fb25ee107b1 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -153,7 +153,7 @@ public static void run(Class testClass, PerfStressOptions options) { Flux.range(0, parallel) .parallel(parallel) .runOn(Schedulers.parallel()) - .flatMap(i -> tests[i].postSetupAsync(), false, Math.min(10000 / parallel, parallel), 1) + .flatMap(i -> tests[i].postSetupAsync()) .sequential() .then() .block(); @@ -268,7 +268,7 @@ 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(10000 / parallel, parallel), 1) + .flatMap(i -> tests[i].runAllAsync(endNanoTime)) .sequential() .then() .block(); diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java index 99af714190c8..435d5a4faf4c 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java @@ -3,12 +3,14 @@ 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 extends BlobTestBase { + private static final long GB = 1024 * 1024 * 1024; public AbstractDownloadTest(TOptions options) { super(options, BLOB_NAME_PREFIX); @@ -17,7 +19,8 @@ public AbstractDownloadTest(TOptions options) { // Upload one blob for the whole test run. All tests can download the same blob public Mono globalSetupAsync() { return super.globalSetupAsync() - .then(blobAsyncClient.upload(createRandomByteBufferFlux(options.getSize()), null)) + .then(blobAsyncClient.upload(createRandomByteBufferFlux(options.getSize()), new ParallelTransferOptions() + .setMaxSingleUploadSizeLong(GB).setBlockSizeLong(GB))) .then(); } } From 9e2b8b1da3c32defd475e2a25bb64493422b6f39 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Thu, 10 Nov 2022 12:28:45 -0500 Subject: [PATCH 08/12] Minor cleanup for single thread tests --- .../perf/test/core/PerfStressProgram.java | 52 +++++++++---------- .../storage/blob/perf/ListBlobsTest.java | 3 +- 2 files changed, 27 insertions(+), 28 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index 9fb25ee107b1..cd6c1e3e999b 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -237,24 +237,22 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, try { if (sync) { - ForkJoinPool forkJoinPool = new ForkJoinPool(parallel); - List> operations = new ArrayList<>(parallel); - for (PerfTestBase test : tests) { - operations.add(() -> { - test.runAll(endNanoTime); - return 1; - }); - } - - forkJoinPool.invokeAll(operations); - - // Sleep until the tests complete. - Thread.sleep(durationSeconds * 1000L); + if (parallel > 1) { + ForkJoinPool forkJoinPool = new ForkJoinPool(parallel); + List> operations = new ArrayList<>(parallel); + for (PerfTestBase test : tests) { + operations.add(() -> { + test.runAll(endNanoTime); + return 1; + }); + } - forkJoinPool.shutdown(); + forkJoinPool.invokeAll(operations); - // Wait 1 second for operations to shut down. - forkJoinPool.awaitTermination(1, TimeUnit.SECONDS); + forkJoinPool.awaitQuiescence(durationSeconds + 1, TimeUnit.SECONDS); + } else { + tests[0].runAll(endNanoTime); + } } 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 @@ -265,18 +263,18 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, System.exit(1); }); - Flux.range(0, parallel) - .parallel(parallel) - .runOn(Schedulers.parallel()) - .flatMap(i -> tests[i].runAllAsync(endNanoTime)) - .sequential() - .then() - .block(); + if (parallel > 1) { + Flux.range(0, parallel) + .parallel(parallel) + .runOn(Schedulers.parallel()) + .flatMap(i -> tests[i].runAllAsync(endNanoTime)) + .sequential() + .then() + .block(); + } else { + tests[0].runAllAsync(endNanoTime).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); diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index 4d3d432f9ca3..64ea33ce9da0 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java @@ -30,7 +30,8 @@ public Mono globalSetupAsync() { .parallel(parallel) .runOn(Schedulers.parallel()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) - .upload(Flux.empty(), null), false, Math.min(1000 / parallel, parallel), 1) + .getBlockBlobAsyncClient() + .upload(Flux.empty(), 0L), false, Math.min(1000 / parallel, parallel), 1) .sequential() .then()); } From 56640c4907bfb281e7a5e67678a5e738506ed666 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Thu, 10 Nov 2022 17:13:28 -0500 Subject: [PATCH 09/12] Remove specialization for parallel=1 --- .../perf/test/core/PerfStressProgram.java | 42 ++++++++----------- 1 file changed, 17 insertions(+), 25 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java index cd6c1e3e999b..19ec7a45e031 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java @@ -237,22 +237,18 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, try { if (sync) { - if (parallel > 1) { - ForkJoinPool forkJoinPool = new ForkJoinPool(parallel); - List> operations = new ArrayList<>(parallel); - for (PerfTestBase test : tests) { - operations.add(() -> { - test.runAll(endNanoTime); - return 1; - }); - } + ForkJoinPool forkJoinPool = new ForkJoinPool(parallel); + List> operations = new ArrayList<>(parallel); + for (PerfTestBase test : tests) { + operations.add(() -> { + test.runAll(endNanoTime); + return 1; + }); + } - forkJoinPool.invokeAll(operations); + forkJoinPool.invokeAll(operations); - forkJoinPool.awaitQuiescence(durationSeconds + 1, TimeUnit.SECONDS); - } else { - tests[0].runAll(endNanoTime); - } + 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 @@ -263,17 +259,13 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, System.exit(1); }); - if (parallel > 1) { - Flux.range(0, parallel) - .parallel(parallel) - .runOn(Schedulers.parallel()) - .flatMap(i -> tests[i].runAllAsync(endNanoTime)) - .sequential() - .then() - .block(); - } else { - tests[0].runAllAsync(endNanoTime).block(); - } + Flux.range(0, parallel) + .parallel(parallel) + .runOn(Schedulers.parallel()) + .flatMap(i -> tests[i].runAllAsync(endNanoTime)) + .sequential() + .then() + .block(); } } catch (Exception e) { System.err.println("Error occurred running tests: " + System.lineSeparator() + e); From 432246e2f3ea9ffd85a9d4c6db95aa92899cefe2 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Thu, 10 Nov 2022 17:18:26 -0500 Subject: [PATCH 10/12] Always run final operations --- .../java/com/azure/perf/test/core/ApiPerfTestBase.java | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/common/perf-test-core/src/main/java/com/azure/perf/test/core/ApiPerfTestBase.java b/common/perf-test-core/src/main/java/com/azure/perf/test/core/ApiPerfTestBase.java index 595cdb54d372..6e44c3438768 100644 --- a/common/perf-test-core/src/main/java/com/azure/perf/test/core/ApiPerfTestBase.java +++ b/common/perf-test-core/src/main/java/com/azure/perf/test/core/ApiPerfTestBase.java @@ -171,13 +171,7 @@ public Mono 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; From ddfbdaaa12dfe793ccdbac85dec7d1b745c193e1 Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Fri, 11 Nov 2022 14:26:40 -0500 Subject: [PATCH 11/12] Comment explaining change --- .../blob/perf/core/AbstractDownloadTest.java | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java index 435d5a4faf4c..29097d7f3a1d 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/AbstractDownloadTest.java @@ -18,6 +18,20 @@ public AbstractDownloadTest(TOptions options) { // Upload one blob for the whole test run. All tests can download the same blob public Mono 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()), new ParallelTransferOptions() .setMaxSingleUploadSizeLong(GB).setBlockSizeLong(GB))) From 8a17289e70556944082cce0fdd5d81494570640c Mon Sep 17 00:00:00 2001 From: alzimmermsft <48699787+alzimmermsft@users.noreply.github.com> Date: Fri, 11 Nov 2022 14:34:06 -0500 Subject: [PATCH 12/12] Change ListBlobTests parallel --- .../main/java/com/azure/storage/blob/perf/ListBlobsTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java index 64ea33ce9da0..37aa64cb2b10 100644 --- a/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java +++ b/sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java @@ -31,7 +31,7 @@ public Mono globalSetupAsync() { .runOn(Schedulers.parallel()) .flatMap(iteration -> blobContainerAsyncClient.getBlobAsyncClient("getblobstest-" + UUID.randomUUID()) .getBlockBlobAsyncClient() - .upload(Flux.empty(), 0L), false, Math.min(1000 / parallel, parallel), 1) + .upload(Flux.empty(), 0L), false, parallel, 1) .sequential() .then()); }