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

akarnokd · akarnokd · commit 046786af8663 · 2015-09-18T14:08:26.000+02:00
diff --git a/src/perf/java/io/reactivex/OperatorFlatMapPerf.java b/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
diff --git a/src/perf/java/io/reactivex/OperatorMergePerf.java b/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)
diff --git a/src/perf/java/io/reactivex/RangePerf.java b/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();
+        }
     }
 
 }

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,11 @@ public void rangeAsync(Blackhole bh) throws Exception {`
`61`	`61`
`62`	`62`	`rangeAsync.subscribe(lo);`
`63`	`63`
`64`		`- lo.latch.await();`
	`64`	`+ if (times == 1) {`
	`65`	`+ while (lo.latch.getCount() != 0);`
	`66`	`+ } else {`
	`67`	`+ lo.latch.await();`
	`68`	`+ }`
`65`	`69`	`}`
`66`	`70`
`67`	`71`	`@Benchmark`
`@@ -70,7 +74,11 @@ public void rangePipeline(Blackhole bh) throws Exception {`
`70`	`74`
`71`	`75`	`rangeAsyncPipeline.subscribe(lo);`
`72`	`76`
`73`		`- lo.latch.await();`
	`77`	`+ if (times == 1) {`
	`78`	`+ while (lo.latch.getCount() != 0);`
	`79`	`+ } else {`
	`80`	`+ lo.latch.await();`
	`81`	`+ }`
`74`	`82`	`}`
`75`	`83`
`76`	`84`	`}`