Merge pull request #3354 from akarnokd/PerfAwaitFix2x

2.x: perf change wait to spin-loop for short async benchmarks

Author: akarnokd

src/perf/java/io/reactivex/OperatorFlatMapPerf.java

@@ -47,7 +47,11 @@ public void flatMapIntPassthruAsync(Input input) throws InterruptedException {
         input.observable.flatMap(i -> {
                 return Observable.just(i).subscribeOn(Schedulers.computation());
         }).subscribe(latchedObserver);
-        latchedObserver.latch.await();
+        if (input.size == 1) {
+            while (latchedObserver.latch.getCount() != 0);
+        } else {
+            latchedObserver.latch.await();
+        }
     }
 
     @Benchmark

src/perf/java/io/reactivex/OperatorMergePerf.java

@@ -31,7 +31,12 @@ public void oneStreamOfNthatMergesIn1(final InputMillion input) throws Interrupt
         Observable<Observable<Integer>> os = Observable.range(1, input.size).map(Observable::just);
         LatchedObserver<Integer> o = input.newLatchedObserver();
         Observable.merge(os).subscribe(o);
-        o.latch.await();
+
+        if (input.size == 1) {
+            while (o.latch.getCount() != 0);
+        } else {
+            o.latch.await();
+        }
     }
 
     // flatMap
@@ -42,7 +47,12 @@ public void merge1SyncStreamOfN(final InputMillion input) throws InterruptedExce
         });
         LatchedObserver<Integer> o = input.newLatchedObserver();
         Observable.merge(os).subscribe(o);
-        o.latch.await();
+        
+        if (input.size == 1) {
+            while (o.latch.getCount() != 0);
+        } else {
+            o.latch.await();
+        }
     }
 
     @Benchmark
@@ -52,7 +62,11 @@ public void mergeNSyncStreamsOfN(final InputThousand input) throws InterruptedEx
         });
         LatchedObserver<Integer> o = input.newLatchedObserver();
         Observable.merge(os).subscribe(o);
-        o.latch.await();
+        if (input.size == 1) {
+            while (o.latch.getCount() != 0);
+        } else {
+            o.latch.await();
+        }
     }
 
     @Benchmark
@@ -62,22 +76,34 @@ public void mergeNAsyncStreamsOfN(final InputThousand input) throws InterruptedE
         });
         LatchedObserver<Integer> o = input.newLatchedObserver();
         Observable.merge(os).subscribe(o);
-        o.latch.await();
+        if (input.size == 1) {
+            while (o.latch.getCount() != 0);
+        } else {
+            o.latch.await();
+        }
     }
 
     @Benchmark
     public void mergeTwoAsyncStreamsOfN(final InputThousand input) throws InterruptedException {
         LatchedObserver<Integer> o = input.newLatchedObserver();
         Observable<Integer> ob = Observable.range(0, input.size).subscribeOn(Schedulers.computation());
         Observable.merge(ob, ob).subscribe(o);
-        o.latch.await();
+        if (input.size == 1) {
+            while (o.latch.getCount() != 0);
+        } else {
+            o.latch.await();
+        }
     }
 
     @Benchmark
     public void mergeNSyncStreamsOf1(final InputForMergeN input) throws InterruptedException {
         LatchedObserver<Integer> o = input.newLatchedObserver();
         Observable.merge(input.observables).subscribe(o);
-        o.latch.await();
+        if (input.size == 1) {
+            while (o.latch.getCount() != 0);
+        } else {
+            o.latch.await();
+        }
     }
 
     @State(Scope.Thread)

src/perf/java/io/reactivex/RangePerf.java

@@ -61,7 +61,11 @@ public void rangeAsync(Blackhole bh) throws Exception {
 
         rangeAsync.subscribe(lo);
 
-        lo.latch.await();
+        if (times == 1) {
+            while (lo.latch.getCount() != 0);
+        } else {
+            lo.latch.await();
+        }
     }
 
     @Benchmark
@@ -70,7 +74,11 @@ public void rangePipeline(Blackhole bh) throws Exception {
 
         rangeAsyncPipeline.subscribe(lo);
 
-        lo.latch.await();
+        if (times == 1) {
+            while (lo.latch.getCount() != 0);
+        } else {
+            lo.latch.await();
+        }
     }
 
 }