Merge pull request #893 from benjchristensen/parallel

Change Parallel to use Long instead of Int

Author: benjchristensen

rxjava-core/src/main/java/rx/operators/OperatorParallel.java

@@ -29,33 +29,35 @@ public final class OperatorParallel<T, R> implements Operator<R, T> {
 
     private final Scheduler scheduler;
     private final Func1<Observable<T>, Observable<R>> f;
+    private final int degreeOfParallelism;
 
     public OperatorParallel(Func1<Observable<T>, Observable<R>> f, Scheduler scheduler) {
         this.scheduler = scheduler;
         this.f = f;
+        this.degreeOfParallelism = scheduler.degreeOfParallelism();
     }
 
     @Override
     public Subscriber<? super T> call(Subscriber<? super R> op) {
 
-        Func1<Subscriber<? super GroupedObservable<Integer, T>>, Subscriber<? super T>> groupBy =
-                new OperatorGroupBy<Integer, T>(new Func1<T, Integer>() {
+        Func1<Subscriber<? super GroupedObservable<Long, T>>, Subscriber<? super T>> groupBy =
+                new OperatorGroupBy<Long, T>(new Func1<T, Long>() {
 
-                    int i = 0;
+                    long i = 0;
 
                     @Override
-                    public Integer call(T t) {
-                        return i++ % scheduler.degreeOfParallelism();
+                    public Long call(T t) {
+                        return i++ % degreeOfParallelism;
                     }
 
                 });
 
-        Func1<Subscriber<? super Observable<R>>, Subscriber<? super GroupedObservable<Integer, T>>> map =
-                new OperatorMap<GroupedObservable<Integer, T>, Observable<R>>(
-                        new Func1<GroupedObservable<Integer, T>, Observable<R>>() {
+        Func1<Subscriber<? super Observable<R>>, Subscriber<? super GroupedObservable<Long, T>>> map =
+                new OperatorMap<GroupedObservable<Long, T>, Observable<R>>(
+                        new Func1<GroupedObservable<Long, T>, Observable<R>>() {
 
                             @Override
-                            public Observable<R> call(GroupedObservable<Integer, T> g) {
+                            public Observable<R> call(GroupedObservable<Long, T> g) {
                                 // Must use observeOn not subscribeOn because we have a single source behind groupBy.
                                 // The origin is already subscribed to, we are moving each group on to a new thread
                                 // but the origin itself can only be on a single thread.

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

@@ -0,0 +1,66 @@
+package rx.operators;
+
+import rx.Observable;
+import rx.functions.Action0;
+import rx.functions.Func1;
+import rx.perf.AbstractPerformanceTester;
+import rx.perf.IntegerSumObserver;
+
+public class OperatorParallelPerformance extends AbstractPerformanceTester {
+
+    private final static int REPS = 10000000;
+
+    OperatorParallelPerformance() {
+        super(REPS);
+    }
+
+    public static void main(String args[]) {
+
+        final OperatorParallelPerformance spt = new OperatorParallelPerformance();
+        try {
+            spt.runTest(new Action0() {
+
+                @Override
+                public void call() {
+                    spt.parallelSum();
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+    }
+
+    /**
+     * 
+     * Run: 10 - 11,220,888 ops/sec
+     * Run: 11 - 12,372,424 ops/sec
+     * Run: 12 - 11,028,921 ops/sec
+     * Run: 13 - 11,813,711 ops/sec
+     * Run: 14 - 12,098,364 ops/sec
+     * 
+     */
+    public long parallelSum() {
+
+        Observable<Integer> s = Observable.range(1, REPS).parallel(new Func1<Observable<Integer>, Observable<Integer>>() {
+
+            @Override
+            public Observable<Integer> call(Observable<Integer> l) {
+                return l.map(new Func1<Integer, Integer>() {
+
+                    @Override
+                    public Integer call(Integer i) {
+                        return i + 1;
+                    }
+
+                });
+            }
+
+        });
+        IntegerSumObserver o = new IntegerSumObserver();
+
+        s.subscribe(o);
+        return o.sum;
+    }
+
+}