Skip to content

[Perf] Revert "Change how List Blobs Perf Creates Blobs and minor changes to Perf Common (#31817)" #31990

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mikeharder wants to merge 1 commit into from
Closed

[Perf] Revert "Change how List Blobs Perf Creates Blobs and minor changes to Perf Common (#31817)" #31990

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 29 additions & 41 deletions common/perf-test-core/src/main/java/com/azure/perf/test/core/ApiPerfTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 <TOptions> the performance test options to use while running the test.
*/
public abstract class ApiPerfTestBase<TOptions extends PerfStressOptions> extends PerfTestBase<TOptions> {
Expand All @@ -47,7 +46,6 @@ public abstract class ApiPerfTestBase<TOptions extends PerfStressOptions> 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.
*/
Expand All @@ -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;
Expand All @@ -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) {
Expand Down Expand Up @@ -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);
}
}
}
}
Expand All @@ -164,45 +163,35 @@ public Mono<Void> 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.
*/
abstract Mono<Integer> runTestAsync();

/**
* Stops playback tests.
*
* @return An empty {@link Mono}.
*/
public Mono<Void> stopPlaybackAsync() {
Expand Down Expand Up @@ -256,7 +245,6 @@ private Mono<Void> startPlaybackAsync() {

/**
* Records responses and starts tests in playback mode.
*
* @return
*/
@Override
Expand All @@ -265,16 +253,16 @@ Mono<Void> 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();
Expand Down
54 changes: 18 additions & 36 deletions common/perf-test-core/src/main/java/com/azure/perf/test/core/PerfStressProgram.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,38 +18,26 @@
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.
*/
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();
}

/**
Expand Down Expand Up @@ -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();
}
}
}
Expand Down Expand Up @@ -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;
Expand All @@ -248,16 +236,10 @@ public static void runTests(PerfTestBase<?>[] tests, boolean sync, int parallel,
try {
if (sync) {
ForkJoinPool forkJoinPool = new ForkJoinPool(parallel);
List<Callable<Integer>> 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
Expand All @@ -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);
Expand All @@ -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");
}
Expand Down
2 changes: 2 additions & 0 deletions ...core/azure-core/src/main/java/com/azure/core/implementation/http/rest/AsyncRestProxy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -72,6 +73,7 @@ public Object invoke(Object proxy, Method method, RequestOptions options, EnumSe

Context finalContext = context;
final Mono<HttpResponse> asyncResponse = RestProxyUtils.validateLengthAsync(request)
.publishOn(Schedulers.boundedElastic())
.flatMap(r -> send(r, finalContext));

Mono<HttpResponseDecoder.HttpDecodedResponse> asyncDecodedResponse = this.decoder
Expand Down
15 changes: 2 additions & 13 deletions sdk/storage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/ListBlobsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -18,19 +16,10 @@ public ListBlobsTest(PerfStressOptions options) {
}

public Mono<Void> 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());
}

Expand Down
18 changes: 0 additions & 18 deletions ...torage/azure-storage-perf/src/main/java/com/azure/storage/blob/perf/core/ServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<TOptions extends PerfStressOptions> extends PerfStressTest<TOptions> {

Expand All @@ -39,19 +36,4 @@ public ServiceTest(TOptions options) {
blobServiceClient = builder.buildClient();
blobServiceAsyncClient = builder.buildAsyncClient();
}

@Override
public Mono<Void> 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());
}
}