Perf Tests with JMH

Author: benjchristensen

rxjava-core/src/perf/java/rx/jmh/Baseline.java

@@ -0,0 +1,35 @@
+package rx.jmh;
+
+import org.openjdk.jmh.annotations.GenerateMicroBenchmark;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.logic.BlackHole;
+
+import rx.functions.Func1;
+
+public class Baseline {
+
+    @GenerateMicroBenchmark
+    public void forLoopInvokingFunction(BlackHole bh, Input input) {
+        for (int value = 0; value < input.size; value++) {
+            bh.consume(IDENTITY_FUNCTION.call(value));
+        }
+    }
+
+    private static final Func1<Integer, Integer> IDENTITY_FUNCTION = new Func1<Integer, Integer>() {
+        @Override
+        public Integer call(Integer value) {
+            return value;
+        }
+    };
+
+    @State(Scope.Thread)
+    public static class Input {
+
+        @Param({ "1024", "1048576" })
+        public int size;
+
+    }
+
+}

rxjava-core/src/perf/java/rx/operators/OperatorMapPerf.java

@@ -1,7 +1,5 @@
 package rx.operators;
 
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.concurrent.CountDownLatch;
 
 import org.openjdk.jmh.annotations.GenerateMicroBenchmark;
@@ -21,14 +19,7 @@
 public class OperatorMapPerf {
 
     @GenerateMicroBenchmark
-    public void measureBaseline(BlackHole bh, Input input) {
-        for (Integer value : input.values) {
-            bh.consume(IDENTITY_FUNCTION.call(value));
-        }
-    }
-
-    @GenerateMicroBenchmark
-    public void measureMap(Input input) throws InterruptedException {
+    public void mapIdentityFunction(Input input) throws InterruptedException {
         input.observable.lift(MAP_OPERATOR).subscribe(input.observer);
 
         input.awaitCompletion();
@@ -46,26 +37,20 @@ public Integer call(Integer value) {
     @State(Scope.Thread)
     public static class Input {
 
-        @Param({"1", "1024", "1048576"})
+        @Param({ "1", "1024", "1048576" })
         public int size;
 
-        public Collection<Integer> values;
         public Observable<Integer> observable;
         public Observer<Integer> observer;
 
         private CountDownLatch latch;
 
         @Setup
         public void setup() {
-            values = new ArrayList<Integer>();
-            for(int i = 0; i < size; i ++) {
-                values.add(i);
-            }
-
             observable = Observable.create(new OnSubscribe<Integer>() {
                 @Override
                 public void call(Subscriber<? super Integer> o) {
-                    for (Integer value : values) {
+                    for (int value = 0; value < size; value++) {
                         if (o.isUnsubscribed())
                             return;
                         o.onNext(value);

rxjava-core/src/perf/java/rx/operators/OperatorSerializePerf.java

@@ -0,0 +1,86 @@
+package rx.operators;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.openjdk.jmh.annotations.GenerateMicroBenchmark;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.logic.BlackHole;
+
+import rx.Observable;
+import rx.Observable.OnSubscribe;
+import rx.Observer;
+import rx.Subscriber;
+import rx.observers.TestSubscriber;
+
+public class OperatorSerializePerf {
+
+    @GenerateMicroBenchmark
+    public void noSerializationSingleThreaded(Input input) {
+        input.observable.subscribe(input.subscriber);
+    }
+
+    @GenerateMicroBenchmark
+    public void serializedSingleStream(Input input) {
+        input.observable.serialize().subscribe(input.subscriber);
+    }
+
+    @GenerateMicroBenchmark
+    public void synchronizedSingleStream(Input input) {
+        input.observable.synchronize().subscribe(input.subscriber);
+    }
+
+    @State(Scope.Thread)
+    public static class Input {
+
+        @Param({ "1024", "1048576" })
+        public int size;
+
+        public Observable<Integer> observable;
+        public TestSubscriber<Integer> subscriber;
+
+        private CountDownLatch latch;
+
+        @Setup
+        public void setup() {
+            observable = Observable.create(new OnSubscribe<Integer>() {
+                @Override
+                public void call(Subscriber<? super Integer> o) {
+                    for (int value = 0; value < size; value++) {
+                        if (o.isUnsubscribed())
+                            return;
+                        o.onNext(value);
+                    }
+                    o.onCompleted();
+                }
+            });
+
+            final BlackHole bh = new BlackHole();
+            latch = new CountDownLatch(1);
+
+            subscriber = new TestSubscriber<Integer>(new Observer<Integer>() {
+                @Override
+                public void onCompleted() {
+                    latch.countDown();
+                }
+
+                @Override
+                public void onError(Throwable e) {
+                    throw new RuntimeException(e);
+                }
+
+                @Override
+                public void onNext(Integer value) {
+                    bh.consume(value);
+                }
+            });
+
+        }
+
+        public void awaitCompletion() throws InterruptedException {
+            latch.await();
+        }
+    }
+}