Parallel Operator

This operator came out of discussions and work with @headinthebox to allow explicit and composable declaration of blocks of work that can be scheduled for parallel execution.

An Observable event stream will be sharded using groupBy using a value from Scheduler. degreeOfParallelism() (defaulting to number of CPU cores) and perform the defined work in parallel.

Instead of having various parallel operators like parallelMap, parallelFilter parallelScan etc this can work generically for any operators or sequence of operators.

Author: benjchristensen

language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/TestParallel.groovy

@@ -0,0 +1,21 @@
+package rx.lang.groovy
+
+import org.junit.Test
+
+import rx.Observable
+import rx.Scheduler
+import rx.concurrency.Schedulers
+import rx.util.functions.Func1
+
+class TestParallel {
+
+    @Test
+    public void testParallelOperator() {
+        Observable.range(0, 100)
+                .parallel({
+                    it.map({ return it; })
+                })
+                .toBlockingObservable()
+                .forEach({ println("T: " + it + " Thread: " + Thread.currentThread());  });
+    }
+}

rxjava-core/src/main/java/rx/Observable.java

@@ -34,10 +34,11 @@
 import rx.operators.OperationCache;
 import rx.operators.OperationCombineLatest;
 import rx.operators.OperationConcat;
+import rx.operators.OperationDebounce;
 import rx.operators.OperationDefer;
 import rx.operators.OperationDematerialize;
-import rx.operators.OperationDistinctUntilChanged;
 import rx.operators.OperationDistinct;
+import rx.operators.OperationDistinctUntilChanged;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
 import rx.operators.OperationFirstOrDefault;
@@ -53,6 +54,7 @@
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
+import rx.operators.OperationParallel;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
 import rx.operators.OperationScan;
@@ -67,7 +69,6 @@
 import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
 import rx.operators.OperationThrottleFirst;
-import rx.operators.OperationDebounce;
 import rx.operators.OperationTimestamp;
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationToObservableIterable;
@@ -1773,15 +1774,13 @@ public static <T> Observable<T> switchOnNext(Observable<? extends Observable<? e
      * its {@link Observer}s; it invokes {@code onCompleted} or {@code onError} only once; and it never invokes {@code onNext} after invoking either {@code onCompleted} or {@code onError}.
      * {@code synchronize} enforces this, and the Observable it returns invokes {@code onNext} and {@code onCompleted} or {@code onError} synchronously.
      * 
-     * @param observable
-     *            the source Observable
      * @param <T>
      *            the type of item emitted by the source Observable
      * @return an Observable that is a chronologically well-behaved version of the source
      *         Observable, and that synchronously notifies its {@link Observer}s
      */
-    public static <T> Observable<T> synchronize(Observable<? extends T> observable) {
-        return create(OperationSynchronize.synchronize(observable));
+    public Observable<T> synchronize() {
+        return create(OperationSynchronize.synchronize(this));
     }
 
 
@@ -3484,6 +3483,31 @@ public Observable<T> cache() {
         return create(OperationCache.cache(this));
     }
 
+    /**
+     * Perform work in parallel by sharding an {@code Observable<T>} on a {@link Schedulers#threadPoolForComputation()} {@link Scheduler} and return an {@code Observable<R>} with the output.
+     * 
+     * @param f
+     *            a {@link Func1} that applies Observable operators to {@code Observable<T>} in parallel and returns an {@code Observable<R>}
+     * @return an Observable with the output of the {@link Func1} executed on a {@link Scheduler}
+     */
+    public <R> Observable<R> parallel(Func1<Observable<T>, Observable<R>> f) {
+        return OperationParallel.parallel(this, f);
+    }
+
+    /**
+     * Perform work in parallel by sharding an {@code Observable<T>} on a {@link Scheduler} and return an {@code Observable<R>} with the output.
+     * 
+     * @param f
+     *            a {@link Func1} that applies Observable operators to {@code Observable<T>} in parallel and returns an {@code Observable<R>}
+     * @param s
+     *            a {@link Scheduler} to perform the work on.
+     * @return an Observable with the output of the {@link Func1} executed on a {@link Scheduler}
+     */
+    
+    public <R> Observable<R> parallel(final Func1<Observable<T>, Observable<R>> f, final Scheduler s) {
+        return OperationParallel.parallel(this, f, s);
+    }
+    
     /**
      * Returns a {@link ConnectableObservable}, which waits until its {@link ConnectableObservable#connect connect} method is called before it begins emitting
      * items to those {@link Observer}s that have subscribed to it.

rxjava-core/src/main/java/rx/Scheduler.java

@@ -212,12 +212,23 @@ public Subscription call(Scheduler scheduler, Void state) {
     }
 
     /**
-     * Returns the scheduler's notion of current absolute time in milliseconds.
+     * @return the scheduler's notion of current absolute time in milliseconds.
      */
     public long now() {
         return System.currentTimeMillis();
     }
 
+    /**
+     * Parallelism available to a Scheduler.
+     * <p>
+     * This defaults to {@code Runtime.getRuntime().availableProcessors()} but can be overridden for use cases such as scheduling work on a computer cluster.
+     * 
+     * @return the scheduler's available degree of parallelism.
+     */
+    public int degreeOfParallelism() {
+        return Runtime.getRuntime().availableProcessors();
+    }
+
     public static class UnitTest {
         @SuppressWarnings("unchecked") // mocking is unchecked, unfortunately
         @Test

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

@@ -0,0 +1,99 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import static org.junit.Assert.*;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Scheduler;
+import rx.concurrency.Schedulers;
+import rx.observables.GroupedObservable;
+import rx.util.functions.Action1;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+
+/**
+ * Identifies unit of work that can be executed in parallel on a given Scheduler.
+ */
+public final class OperationParallel<T> {
+
+    public static <T, R> Observable<R> parallel(Observable<T> source, Func1<Observable<T>, Observable<R>> f) {
+        return parallel(source, f, Schedulers.threadPoolForComputation());
+    }
+
+    public static <T, R> Observable<R> parallel(final Observable<T> source, final Func1<Observable<T>, Observable<R>> f, final Scheduler s) {
+        return Observable.defer(new Func0<Observable<R>>() {
+
+            @Override
+            public Observable<R> call() {
+                final AtomicInteger i = new AtomicInteger(0);
+                return source.groupBy(new Func1<T, Integer>() {
+
+                    @Override
+                    public Integer call(T t) {
+                        return i.incrementAndGet() % s.degreeOfParallelism();
+                    }
+
+                }).flatMap(new Func1<GroupedObservable<Integer, T>, Observable<R>>() {
+
+                    @Override
+                    public Observable<R> call(GroupedObservable<Integer, T> group) {
+                        return f.call(group.observeOn(s));
+                    }
+                }).synchronize();
+            }
+        });
+    }
+
+    public static class UnitTest {
+
+        @Test
+        public void testParallel() {
+            int NUM = 1000;
+            final AtomicInteger count = new AtomicInteger();
+            Observable.range(1, NUM).parallel(
+                    new Func1<Observable<Integer>, Observable<Integer[]>>() {
+
+                        @Override
+                        public Observable<Integer[]> call(Observable<Integer> o) {
+                            return o.map(new Func1<Integer, Integer[]>() {
+
+                                @Override
+                                public Integer[] call(Integer t) {
+                                    return new Integer[] { t, t * 99 };
+                                }
+
+                            });
+                        }
+                    }).toBlockingObservable().forEach(new Action1<Integer[]>() {
+
+                @Override
+                public void call(Integer[] v) {
+                    count.incrementAndGet();
+                    System.out.println("V: " + v[0] + " R: " + v[1] + " Thread: " + Thread.currentThread());
+                }
+
+            });
+
+            // just making sure we finish and get the number we expect
+            assertEquals(NUM, count.get());
+        }
+    }
+}