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..d2c8018c01f8 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 @@ -24,12 +24,11 @@ import java.security.KeyManagementException; import java.security.NoSuchAlgorithmException; import java.security.SecureRandom; -import java.util.Collections; +import java.util.Arrays; import java.util.concurrent.CompletableFuture; /** * The Base Performance Test class for API based Perf Tests. - * * @param the performance test options to use while running the test. */ public abstract class ApiPerfTestBase extends PerfTestBase { @@ -47,7 +46,6 @@ public abstract class ApiPerfTestBase extend /** * Creates an instance of the Http Based Performance test. - * * @param options the performance test options to use while running the test. * @throws IllegalStateException if an errors is encountered with building ssl context. */ @@ -60,7 +58,7 @@ public ApiPerfTestBase(TOptions options) { recordPlaybackHttpClient = createRecordPlaybackClient(options); testProxy = options.getTestProxies().get(parallelIndex % options.getTestProxies().size()); testProxyPolicy = new TestProxyPolicy(testProxy); - policies = Collections.singletonList(testProxyPolicy); + policies = Arrays.asList(testProxyPolicy); } else { recordPlaybackHttpClient = null; testProxy = null; @@ -81,7 +79,7 @@ private static HttpClient createHttpClient(PerfStressOptions options) { reactor.netty.http.client.HttpClient nettyHttpClient = reactor.netty.http.client.HttpClient.create() - .secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)); + .secure(sslContextSpec -> sslContextSpec.sslContext(sslContext)); return new NettyAsyncHttpClientBuilder(nettyHttpClient).build(); } catch (SSLException e) { @@ -129,20 +127,21 @@ private static reactor.netty.http.client.HttpClient createRecordPlaybackClient(P } /** - * Attempts to configure a ClientBuilder using reflection. If a ClientBuilder does not follow the standard - * convention, it can be configured manually using the "httpClient" and "policies" fields. - * + * Attempts to configure a ClientBuilder using reflection. If a ClientBuilder does not follow the standard convention, + * it can be configured manually using the "httpClient" and "policies" fields. * @param clientBuilder The client builder. * @throws IllegalStateException If reflective access to get httpClient or addPolicy methods fail. */ protected void configureClientBuilder(HttpTrait clientBuilder) { - if (httpClient != null) { - clientBuilder.httpClient(httpClient); - } + if (httpClient != null || policies != null) { + if (httpClient != null) { + clientBuilder.httpClient(httpClient); + } - if (policies != null) { - for (HttpPipelinePolicy policy : policies) { - clientBuilder.addPolicy(policy); + if (policies != null) { + for (HttpPipelinePolicy policy : policies) { + clientBuilder.addPolicy(policy); + } } } } @@ -164,37 +163,28 @@ public Mono runAllAsync(long endNanoTime) { lastCompletionNanoTime = 0; long startNanoTime = System.nanoTime(); - return Flux.generate(sink -> { - if (System.nanoTime() < endNanoTime) { - sink.next(1); - } else { - sink.complete(); - } - }) - .flatMap(ignored -> { - if (System.nanoTime() < endNanoTime) { - return runTestAsync(); - } else { - return Mono.just(0); - } - }, 1) - .doOnNext(result -> { - completedOperations += result; + return Flux.just(1) + .repeat() + .flatMap(i -> runTestAsync(), 1) + .doOnNext(v -> { + completedOperations += v; lastCompletionNanoTime = System.nanoTime() - startNanoTime; }) + .takeWhile(i -> System.nanoTime() < endNanoTime) .then(); } /** - * Indicates how many operations were completed in a single run of the test. Good to be used for batch operations. + * Indicates how many operations were completed in a single run of the test. + * Good to be used for batch operations. * * @return the number of successful operations completed. */ abstract int runTest(); /** - * Indicates how many operations were completed in a single run of the async test. Good to be used for batch - * operations. + * Indicates how many operations were completed in a single run of the async test. + * Good to be used for batch operations. * * @return the number of successful operations completed. */ @@ -202,7 +192,6 @@ public Mono runAllAsync(long endNanoTime) { /** * Stops playback tests. - * * @return An empty {@link Mono}. */ public Mono stopPlaybackAsync() { @@ -256,7 +245,6 @@ private Mono startPlaybackAsync() { /** * Records responses and starts tests in playback mode. - * * @return */ @Override @@ -265,16 +253,16 @@ Mono postSetupAsync() { // Make one call to Run() before starting recording, to avoid capturing one-time setup like authorization requests. return runSyncOrAsync() - .then(startRecordingAsync()) - .then(Mono.defer(() -> { + .then(startRecordingAsync()) + .then(Mono.defer(() -> { testProxyPolicy.setRecordingId(recordingId); testProxyPolicy.setMode("record"); return Mono.empty(); })) - .then(runSyncOrAsync()) - .then(stopRecordingAsync()) - .then(startPlaybackAsync()) - .then(Mono.defer(() -> { + .then(runSyncOrAsync()) + .then(stopRecordingAsync()) + .then(startPlaybackAsync()) + .then(Mono.defer(() -> { testProxyPolicy.setRecordingId(recordingId); testProxyPolicy.setMode("playback"); return Mono.empty(); 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..8ca329511da8 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 @@ -18,12 +18,11 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; -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; +import java.util.stream.Stream; /** * Represents the main program class which reflectively runs and manages the performance tests. @@ -31,25 +30,14 @@ public class PerfStressProgram { private static final int NANOSECONDS_PER_SECOND = 1_000_000_000; - private static long getCompletedOperations(PerfTestBase[] tests) { - long completedOperations = 0; - for (PerfTestBase test : tests) { - completedOperations += test.getCompletedOperations(); - } - - return completedOperations; + private static int getCompletedOperations(PerfTestBase[] tests) { + return Stream.of(tests).mapToInt(perfStressTest -> Long.valueOf(perfStressTest.getCompletedOperations()).intValue()).sum(); } private static double getOperationsPerSecond(PerfTestBase[] tests) { - double operationsPerSecond = 0.0D; - for (PerfTestBase test : tests) { - double temp = test.getCompletedOperations() / (((double) test.lastCompletionNanoTime) / NANOSECONDS_PER_SECOND); - if (!Double.isNaN(temp)) { - operationsPerSecond += temp; - } - } - - return operationsPerSecond; + return IntStream.range(0, tests.length) + .mapToDouble(i -> tests[i].getCompletedOperations() / (((double) tests[i].lastCompletionNanoTime) / NANOSECONDS_PER_SECOND)) + .sum(); } /** @@ -199,7 +187,7 @@ public static void run(Class testClass, PerfStressOptions options) { if (!options.isNoCleanup()) { cleanupStatus = printStatus("=== Cleanup ===", () -> ".", false, false); - Flux.just(tests).flatMap(PerfTestBase::cleanupAsync).blockLast(); + Flux.just(tests).flatMap(t -> t.cleanupAsync()).blockLast(); } } } @@ -234,11 +222,11 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, long endNanoTime = System.nanoTime() + ((long) durationSeconds * 1000000000); - long[] lastCompleted = new long[]{0}; + int[] lastCompleted = new int[]{0}; Disposable progressStatus = printStatus( "=== " + title + " ===" + System.lineSeparator() + "Current\t\tTotal\t\tAverage", () -> { - long totalCompleted = getCompletedOperations(tests); - long currentCompleted = totalCompleted - lastCompleted[0]; + int totalCompleted = getCompletedOperations(tests); + int currentCompleted = totalCompleted - lastCompleted[0]; double averageCompleted = getOperationsPerSecond(tests); lastCompleted[0] = totalCompleted; @@ -248,16 +236,10 @@ 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.submit(() -> { + IntStream.range(0, parallel).parallel().forEach(i -> tests[i].runAll(endNanoTime)); + }).get(); - forkJoinPool.invokeAll(operations, (durationSeconds * 1000L) + 100L, TimeUnit.MILLISECONDS); - forkJoinPool.shutdown(); } 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,13 +251,13 @@ 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) + .parallel() + .runOn(Schedulers.boundedElastic()) + .flatMap(i -> tests[i].runAllAsync(endNanoTime)) .then() .block(); } - } catch (InterruptedException e) { + } 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); @@ -288,7 +270,7 @@ public static void runTests(PerfTestBase[] tests, boolean sync, int parallel, System.out.println("=== Results ==="); - long totalOperations = getCompletedOperations(tests); + int totalOperations = getCompletedOperations(tests); if (totalOperations == 0) { throw new IllegalStateException("Zero operations has been completed"); } diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/AsyncRestProxy.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/AsyncRestProxy.java index 7415f79571d3..7348c3cf7df3 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/AsyncRestProxy.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/AsyncRestProxy.java @@ -22,6 +22,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Signal; +import reactor.core.scheduler.Schedulers; import reactor.util.context.ContextView; import java.io.IOException; @@ -72,6 +73,7 @@ public Object invoke(Object proxy, Method method, RequestOptions options, EnumSe Context finalContext = context; final Mono asyncResponse = RestProxyUtils.validateLengthAsync(request) + .publishOn(Schedulers.boundedElastic()) .flatMap(r -> send(r, finalContext)); Mono asyncDecodedResponse = this.decoder 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..f30eb998da0b 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,8 +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; @@ -18,19 +16,10 @@ public ListBlobsTest(PerfStressOptions options) { } public Mono globalSetupAsync() { - // Perform blob uploading in parallel. - // - // This not only results in faster setup it also 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, options.getCount()) - .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) + .map(i -> "getblobstest-" + UUID.randomUUID()) + .flatMap(b -> blobContainerAsyncClient.getBlobAsyncClient(b).upload(Flux.empty(), null)) .then()); } 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..54894fa8db56 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,9 +10,6 @@ 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 { @@ -39,19 +36,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()); - } }