Add Try/catch around runAsync() for tests so we always print results and dispose of progress (Azure#22352)

conniey · web-flow · commit c2c5265081f0 · 2021-06-17T20:08:43.000Z
* Add Try/catch around runAsync() for tests so we always print results.

* Always dispose of progress.
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
@@ -3,6 +3,15 @@
 
 package com.azure.perf.test.core;
 
+import com.beust.jcommander.JCommander;
+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;
@@ -14,16 +23,6 @@
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
 
-import com.beust.jcommander.JCommander;
-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;
-
 /**
  * Represents the main program class which reflectively runs and manages the performance tests.
  */
@@ -46,9 +45,10 @@ private static double getOperationsPerSecond() {
     /**
      * Runs the performance tests passed to be executed.
      *
-     * @throws RuntimeException if the execution fails.
      * @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) {
         List<Class<?>> classList = new ArrayList<>(Arrays.asList(classes));
@@ -62,13 +62,13 @@ public static void run(Class<?>[] classes, String[] args) {
         }
 
         String[] commands = classList.stream().map(c -> getCommandName(c.getSimpleName()))
-                .toArray(i -> new String[i]);
+            .toArray(i -> new String[i]);
 
         PerfStressOptions[] options = classList.stream().map(c -> {
             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]);
@@ -99,9 +99,10 @@ private static String getCommandName(String testName) {
     /**
      * Run the performance test passed to be executed.
      *
-     * @throws RuntimeException if the execution fails.
      * @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) {
         System.out.println("=== Options ===");
@@ -125,7 +126,7 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
             try {
                 tests[i] = (PerfStressTest<?>) testClass.getConstructor(options.getClass()).newInstance(options);
             } catch (InstantiationException | IllegalAccessException | IllegalArgumentException
-                    | InvocationTargetException | SecurityException | NoSuchMethodException e) {
+                | InvocationTargetException | SecurityException | NoSuchMethodException e) {
                 throw new RuntimeException(e);
             }
         }
@@ -149,9 +150,7 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
                 }
             } finally {
                 if (!options.isNoCleanup()) {
-                    if (cleanupStatus == null) {
-                        cleanupStatus = printStatus("=== Cleanup ===", () -> ".", false, false);
-                    }
+                    cleanupStatus = printStatus("=== Cleanup ===", () -> ".", false, false);
 
                     Flux.just(tests).flatMap(t -> t.cleanupAsync()).blockLast();
                 }
@@ -174,12 +173,13 @@ public static void run(Class<?> testClass, PerfStressOptions options) {
     /**
      * Runs the performance tests passed to be executed.
      *
-     * @throws RuntimeException if the execution fails.
      * @param tests the performance tests to be executed.
      * @param sync indicate if synchronous test should be run.
      * @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.
      */
     public static void runTests(PerfStressTest<?>[] tests, boolean sync, int parallel, int durationSeconds, String title) {
@@ -188,9 +188,9 @@ public static void runTests(PerfStressTest<?>[] tests, boolean sync, int paralle
 
         long endNanoTime = System.nanoTime() + ((long) durationSeconds * 1000000000);
 
-        int[] lastCompleted = new int[] { 0 };
+        int[] lastCompleted = new int[]{0};
         Disposable progressStatus = printStatus(
-                "=== " + title + " ===" + System.lineSeparator() + "Current\t\tTotal\t\tAverage", () -> {
+            "=== " + title + " ===" + System.lineSeparator() + "Current\t\tTotal\t\tAverage", () -> {
                 int totalCompleted = getCompletedOperations();
                 int currentCompleted = totalCompleted - lastCompleted[0];
                 double averageCompleted = getOperationsPerSecond();
@@ -199,35 +199,39 @@ public static void runTests(PerfStressTest<?>[] tests, boolean sync, int paralle
                 return String.format("%d\t\t%d\t\t%.2f", currentCompleted, totalCompleted, averageCompleted);
             }, true, true);
 
-        if (sync) {
-            ForkJoinPool forkJoinPool = new ForkJoinPool(parallel);
-            try {
+        try {
+            if (sync) {
+                ForkJoinPool forkJoinPool = new ForkJoinPool(parallel);
                 forkJoinPool.submit(() -> {
                     IntStream.range(0, parallel).parallel().forEach(i -> runLoop(tests[i], i, endNanoTime));
                 }).get();
-            } catch (InterruptedException | ExecutionException e) {
-                throw new RuntimeException(e);
+
+            } 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
+                // Schedulers.onHandleError.  This handler ensures the perf framework will fail fast on any such exceptions.
+                Schedulers.onHandleError((t, e) -> {
+                    System.err.print(t + " threw exception: ");
+                    e.printStackTrace();
+                    System.exit(1);
+                });
+
+                Flux.range(0, parallel)
+                    .parallel()
+                    .runOn(Schedulers.boundedElastic())
+                    .flatMap(i -> runLoopAsync(tests[i], i, endNanoTime))
+                    .then()
+                    .block();
             }
-        } 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
-            // Schedulers.onHandleError.  This handler ensures the perf framework will fail fast on any such exceptions.
-            Schedulers.onHandleError((t, e) -> {
-                System.err.print(t + " threw exception: ");
-                e.printStackTrace();
-                System.exit(1);
-            });
-
-            Flux.range(0, parallel)
-                .parallel()
-                .runOn(Schedulers.boundedElastic())
-                .flatMap(i -> runLoopAsync(tests[i], i, endNanoTime))
-                .then()
-                .block();
+        } catch (InterruptedException | ExecutionException e) {
+            System.err.println("Error occurred when submitting jobs to ForkJoinPool. " + System.lineSeparator() + e);
+            throw new RuntimeException(e);
+        } catch (Exception e) {
+            System.err.println("Error occurred running tests: " + System.lineSeparator() + e);
+        } finally {
+            progressStatus.dispose();
         }
 
-        progressStatus.dispose();
-
         System.out.println("=== Results ===");
 
         int totalOperations = getCompletedOperations();
@@ -239,7 +243,7 @@ public static void runTests(PerfStressTest<?>[] tests, boolean sync, int paralle
         double weightedAverageSeconds = totalOperations / operationsPerSecond;
 
         System.out.printf("Completed %,d operations in a weighted-average of %,.2fs (%,.2f ops/s, %,.3f s/op)%n",
-                totalOperations, weightedAverageSeconds, operationsPerSecond, secondsPerOperation);
+            totalOperations, weightedAverageSeconds, operationsPerSecond, secondsPerOperation);
         System.out.println();
     }
 
@@ -269,7 +273,7 @@ private static Mono<Void> runLoopAsync(PerfStressTest<?> test, int index, long e
     private static Disposable printStatus(String header, Supplier<Object> status, boolean newLine, boolean printFinalStatus) {
         System.out.println(header);
 
-        boolean[] needsExtraNewline = new boolean[] { false };
+        boolean[] needsExtraNewline = new boolean[]{false};
 
         return Flux.interval(Duration.ofSeconds(1)).doFinally(s -> {
             if (printFinalStatus) {
