Use parallel flux to ensure all cores are used (#6)

jianghaolu · Nov 21, 2019 · a546101 · a546101
1 parent d00e451
commit a546101
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/java/azure-perfstress/src/main/java/com/azure/perfstress/PerfStressProgram.java b/java/azure-perfstress/src/main/java/com/azure/perfstress/PerfStressProgram.java
@@ -20,6 +20,7 @@
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 public class PerfStressProgram {
     private static int[] _completedOperations;
@@ -165,7 +166,12 @@ public static void RunTests(PerfStressTest<?>[] tests, boolean sync, int paralle
             }
         }
         else {
-            Flux.range(0, parallel).flatMap(i -> RunLoopAsync(tests[i], i, endNanoTime)).blockLast();
+            Flux.range(0, parallel)
+                .parallel()
+                .runOn(Schedulers.boundedElastic())
+                .flatMap(i -> RunLoopAsync(tests[i], i, endNanoTime))
+                .then()
+                .block();
         }
 
         progressStatus.dispose();
@@ -174,7 +180,7 @@ public static void RunTests(PerfStressTest<?>[] tests, boolean sync, int paralle
 
         int totalOperations = IntStream.of(_completedOperations).sum();
         double operationsPerSecond = IntStream.range(0, parallel)
-                .mapToDouble(i -> ((double)_completedOperations[i]) / (_lastCompletionNanoTimes[i] / 1000000000))
+                .mapToDouble(i -> _completedOperations[i] / (((double)_lastCompletionNanoTimes[i]) / 1000000000))
                 .sum();
         double secondsPerOperation = 1 / operationsPerSecond;
         double weightedAverageSeconds = totalOperations / operationsPerSecond;