Start of Performance Testing Operators

These are examples of how we should proceed. Very likely though we want to use something like Caliper: https://code.google.com/p/caliper/

Author: benjchristensen

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

@@ -0,0 +1,49 @@
+package rx.operators;
+
+import java.util.Arrays;
+
+import rx.Observable;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.LongSumObserver;
+import rx.util.functions.Action0;
+
+public class OperatorFromIterablePerformance extends AbstractPerformanceTester {
+
+    public static void main(String args[]) {
+
+        final OperatorFromIterablePerformance spt = new OperatorFromIterablePerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    spt.timeTenLongs();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * Observable.from(Iterable)
+     * 
+     * Run: 10 - 10,629,658 ops/sec
+     * Run: 11 - 9,573,775 ops/sec
+     * Run: 12 - 10,589,787 ops/sec
+     * Run: 13 - 10,514,141 ops/sec
+     * Run: 14 - 9,765,586 ops/sec
+     */
+    public long timeTenLongs() {
+
+        Observable<Long> s = Observable.from(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L));
+        LongSumObserver o = new LongSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+
+}

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

@@ -0,0 +1,55 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.LongSumObserver;
+import rx.util.functions.Action0;
+import rx.util.functions.Func1;
+
+public class OperatorMapPerformance extends AbstractPerformanceTester {
+
+    public static void main(String args[]) {
+
+        final OperatorMapPerformance spt = new OperatorMapPerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    spt.timeMapPlusOne();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * Observable.from(1L).map((l) -> { l+1})
+     * 
+     * Run: 10 - 7,377,982 ops/sec
+     * Run: 11 - 7,714,715 ops/sec
+     * Run: 12 - 7,783,579 ops/sec
+     * Run: 13 - 7,693,372 ops/sec
+     * Run: 14 - 7,567,777 ops/sec
+     */
+    public long timeMapPlusOne() {
+
+        Observable<Long> s = Observable.from(1L).map(new Func1<Long, Long>() {
+
+            @Override
+            public Long call(Long l) {
+                return l + 1;
+            }
+
+        });
+        LongSumObserver o = new LongSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+
+}

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

@@ -0,0 +1,74 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.IntegerSumObserver;
+import rx.util.functions.Action0;
+
+public class OperatorMergePerformance extends AbstractPerformanceTester {
+
+    public static void main(String args[]) {
+
+        final OperatorMergePerformance spt = new OperatorMergePerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    //                    spt.timeMergeAandBwithSingleItems();
+                    spt.timeMergeAandBwith100Items();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * Observable.merge(from(1), from(1))
+     * 
+     * Run: 10 - 2,308,617 ops/sec
+     * Run: 11 - 2,309,602 ops/sec
+     * Run: 12 - 2,318,590 ops/sec
+     * Run: 13 - 2,270,100 ops/sec
+     * Run: 14 - 2,312,006 ops/sec
+     * 
+     */
+    public long timeMergeAandBwithSingleItems() {
+
+        Observable<Integer> sA = Observable.from(1);
+        Observable<Integer> sB = Observable.from(2);
+        Observable<Integer> s = Observable.merge(sA, sB);
+
+        IntegerSumObserver o = new IntegerSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+
+    /**
+     * Observable.merge(range(0, 100), range(100, 200))
+     * 
+     * Run: 10 - 340,049 ops/sec
+     * Run: 11 - 339,059 ops/sec
+     * Run: 12 - 348,899 ops/sec
+     * Run: 13 - 350,953 ops/sec
+     * Run: 14 - 352,228 ops/sec
+     */
+    public long timeMergeAandBwith100Items() {
+
+        Observable<Integer> sA = Observable.range(0, 100);
+        Observable<Integer> sB = Observable.range(100, 200);
+        Observable<Integer> s = Observable.merge(sA, sB);
+
+        IntegerSumObserver o = new IntegerSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+}

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

@@ -0,0 +1,48 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.IntegerSumObserver;
+import rx.util.functions.Action0;
+
+public class OperatorTakePerformance extends AbstractPerformanceTester {
+
+    public static void main(String args[]) {
+
+        final OperatorTakePerformance spt = new OperatorTakePerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    spt.timeTake5();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * Observable.range(0, 10).take(5);
+     * 
+     * Run: 10 - 3,951,557 ops/sec
+     * Run: 11 - 3,981,329 ops/sec
+     * Run: 12 - 3,988,949 ops/sec
+     * Run: 13 - 3,925,971 ops/sec
+     * Run: 14 - 4,033,468 ops/sec
+     */
+    public long timeTake5() {
+
+        Observable<Integer> s = Observable.range(0, 10).take(5);
+
+        IntegerSumObserver o = new IntegerSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+
+}

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

@@ -0,0 +1,88 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.IntegerSumObserver;
+import rx.util.functions.Action0;
+import rx.util.functions.Func2;
+
+public class OperatorZipPerformance extends AbstractPerformanceTester {
+
+    public static void main(String args[]) {
+
+        final OperatorZipPerformance spt = new OperatorZipPerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    //                    spt.timeZipAandBwithSingleItems();
+                    spt.timeZipAandBwith100Items();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * Observable.zip(from(1), from(1), {a, b -> a+b})
+     * 
+     * Run: 10 - 1,419,071 ops/sec
+     * Run: 11 - 1,418,867 ops/sec
+     * Run: 12 - 1,420,459 ops/sec
+     * Run: 13 - 1,409,877 ops/sec
+     * Run: 14 - 1,426,019 ops/sec
+     */
+    public long timeZipAandBwithSingleItems() {
+
+        Observable<Integer> sA = Observable.from(1);
+        Observable<Integer> sB = Observable.from(2);
+        Observable<Integer> s = Observable.zip(sA, sB, new Func2<Integer, Integer, Integer>() {
+
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return t1 + t2;
+            }
+
+        });
+
+        IntegerSumObserver o = new IntegerSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+
+    /**
+     * Observable.zip(range(0, 100), range(100, 200), {a, b -> a+b})
+     * 
+     * ... really slow ...
+     * 
+     * Run: 0 - 30,698 ops/sec
+     * Run: 1 - 31,061 ops/sec
+     * 
+     */
+    public long timeZipAandBwith100Items() {
+
+        Observable<Integer> sA = Observable.range(0, 100);
+        Observable<Integer> sB = Observable.range(100, 200);
+        Observable<Integer> s = Observable.zip(sA, sB, new Func2<Integer, Integer, Integer>() {
+
+            @Override
+            public Integer call(Integer t1, Integer t2) {
+                return t1 + t2;
+            }
+
+        });
+
+        IntegerSumObserver o = new IntegerSumObserver();
+
+        for (long l = 0; l < REPETITIONS; l++) {
+            s.subscribe(o);
+        }
+        return o.sum;
+    }
+}

rxjava-core/src/perf/java/rx/perf/AbstractPerformanceTester.java

@@ -0,0 +1,47 @@
+package rx.perf;
+
+import rx.util.functions.Action0;
+
+public abstract class AbstractPerformanceTester {
+
+    public static final int REPETITIONS = 5 * 1000 * 1000;
+    public static final int NUM_PRODUCERS = 1;
+
+    public final void runTest(Action0 action) throws InterruptedException {
+        for (int runNum = 0; runNum < 15; runNum++) {
+            System.gc();
+            Thread.sleep(1000L);
+
+            final long start = System.nanoTime();
+
+            action.call();
+
+            long duration = System.nanoTime() - start;
+            long opsPerSec = (REPETITIONS * NUM_PRODUCERS * 1000L * 1000L * 1000L) / duration;
+            System.out.printf("Run: %d - %,d ops/sec \n",
+                    Integer.valueOf(runNum),
+                    Long.valueOf(opsPerSec));
+        }
+    }
+
+    /**
+     * Baseline ops/second without a subject.
+     * 
+     * Perf along this order of magnitude:
+     * 
+     * Run: 10 - 316,235,532 ops/sec
+     * Run: 11 - 301,886,792 ops/sec
+     * Run: 12 - 310,472,228 ops/sec
+     * Run: 13 - 313,469,797 ops/sec
+     * Run: 14 - 305,380,809 ops/sec
+     */
+    public long baseline() {
+        LongSumObserver o = new LongSumObserver();
+        for (long l = 0; l < REPETITIONS; l++) {
+            o.onNext(l);
+        }
+        o.onCompleted();
+        return o.sum;
+    }
+    
+}

rxjava-core/src/perf/java/rx/perf/IntegerSumObserver.java

@@ -0,0 +1,23 @@
+package rx.perf;
+
+import rx.Observer;
+
+public class IntegerSumObserver implements Observer<Integer> {
+
+    public int sum = 0;
+
+    @Override
+    public void onCompleted() {
+
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        throw new RuntimeException(e);
+    }
+
+    @Override
+    public void onNext(Integer l) {
+        sum += l;
+    }
+}