SerializedObserver and SerializedSubscriber

Using the "queue and lock" implementation which won our performance and production testing.

Author: benjchristensen

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

@@ -7264,6 +7264,7 @@ public final <R> Observable<R> switchMap(Func1<? super T, ? extends Observable<?
      * @return an Observable that is a chronologically well-behaved version of the source Observable, and that
      *         synchronously notifies its {@link Observer}s
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-synchronize">RxJava Wiki: synchronize()</a>
+     * @deprecated Use {@link #serialize()} instead as it doesn't block threads while emitting notification.
      */
     public final Observable<T> synchronize() {
         return lift(new OperatorSynchronize<T>());
@@ -7288,6 +7289,7 @@ public final Observable<T> synchronize() {
      * @return an Observable that is a chronologically well-behaved version of the source Observable, and that
      *         synchronously notifies its {@link Observer}s
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-synchronize">RxJava Wiki: synchronize()</a>
+     * @deprecated Use {@link #serialize()} instead as it doesn't block threads while emitting notification.
      */
     public final Observable<T> synchronize(Object lock) {
         return lift(new OperatorSynchronize<T>(lock));

rxjava-core/src/main/java/rx/observers/SerializedObserver.java

@@ -0,0 +1,42 @@
+package rx.observers;
+
+import rx.Observer;
+
+/**
+ * Enforce single-threaded, serialized, ordered execution of onNext, onCompleted, onError.
+ * <p>
+ * When multiple threads are notifying they will be serialized by:
+ * <p>
+ * <li>Allowing only one thread at a time to emit</li>
+ * <li>Adding notifications to a queue if another thread is already emitting</li>
+ * <li>Not holding any locks or blocking any threads while emitting</li>
+ * <p>
+ * 
+ * @param <T>
+ */
+public class SerializedObserver<T> implements Observer<T> {
+    /*
+     * Facade to actual implementation until final decision is made
+     * on the implementation.
+     */
+    private final SerializedObserverViaQueueAndLock<T> actual;
+
+    public SerializedObserver(Observer<? super T> observer) {
+        this.actual = new SerializedObserverViaQueueAndLock<T>(observer);
+    }
+
+    @Override
+    public void onCompleted() {
+        actual.onCompleted();
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        actual.onError(e);
+    }
+
+    @Override
+    public void onNext(T t) {
+        actual.onNext(t);
+    }
+}

rxjava-core/src/main/java/rx/observers/SerializedObserverViaQueueAndCounter.java

@@ -5,7 +5,7 @@
 
 import rx.Observer;
 
-public class SerializedObserverViaQueueAndCounter<T> implements Observer<T> {
+/* package */class SerializedObserverViaQueueAndCounter<T> implements Observer<T> {
     private final Observer<? super T> actual;
     private final AtomicInteger count = new AtomicInteger();
     private final ConcurrentLinkedQueue<Object> queue = new ConcurrentLinkedQueue<Object>();

rxjava-core/src/main/java/rx/observers/SerializedObserverViaQueueAndLock.java

@@ -4,7 +4,7 @@
 
 import rx.Observer;
 
-public class SerializedObserverViaQueueAndLock<T> implements Observer<T> {
+/* package */ class SerializedObserverViaQueueAndLock<T> implements Observer<T> {
     private final Observer<? super T> actual;
 
     private boolean emitting = false;

rxjava-core/src/main/java/rx/observers/SerializedObserverViaStateMachine.java

@@ -4,7 +4,7 @@
 
 import rx.Observer;
 
-public class SerializedObserverViaStateMachine<T> implements Observer<T> {
+/* package */class SerializedObserverViaStateMachine<T> implements Observer<T> {
 
     private final AtomicReference<State> state = new AtomicReference<State>(State.createNew());
     private final Observer<? super T> s;

rxjava-core/src/main/java/rx/observers/SerializedSubscriber.java

@@ -3,12 +3,24 @@
 import rx.Observer;
 import rx.Subscriber;
 
+/**
+ * Enforce single-threaded, serialized, ordered execution of onNext, onCompleted, onError.
+ * <p>
+ * When multiple threads are notifying they will be serialized by:
+ * <p>
+ * <li>Allowing only one thread at a time to emit</li>
+ * <li>Adding notifications to a queue if another thread is already emitting</li>
+ * <li>Not holding any locks or blocking any threads while emitting</li>
+ * <p>
+ * 
+ * @param <T>
+ */
 public class SerializedSubscriber<T> extends Subscriber<T> {
 
     private final Observer<T> s;
 
     public SerializedSubscriber(Subscriber<? super T> s) {
-        this.s = new SerializedObserverViaQueueAndLock<T>(s);
+        this.s = new SerializedObserver<T>(s);
     }
 
     @Override

rxjava-core/src/main/java/rx/observers/SynchronizedObserver.java

@@ -23,7 +23,9 @@
  * This ONLY does synchronization. It does not involve itself in safety or subscriptions. See SafeSubscriber for that.
  * 
  * @param <T>
+ * @deprecated Use SerializedObserver instead as it doesn't block threads during event notification.
  */
+@Deprecated
 public final class SynchronizedObserver<T> implements Observer<T> {
 
     /**

rxjava-core/src/main/java/rx/observers/SynchronizedSubscriber.java

@@ -29,7 +29,9 @@
  * </ul>
  * 
  * @param <T>
+ * @deprecated Use SerializedSubscriber instead as it doesn't block threads during event notification.
  */
+@Deprecated
 public final class SynchronizedSubscriber<T> extends Subscriber<T> {
 
     private final Observer<? super T> observer;

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

@@ -14,10 +14,9 @@
 import rx.schedulers.Schedulers;
 
 public class OperatorSerializePerformance extends AbstractPerformanceTester {
-    //    static int reps = Integer.MAX_VALUE / 16384; // timeTwoStreams
-
-    static int reps = Integer.MAX_VALUE / 1024; // timeSingleStream
+    static int reps = Integer.MAX_VALUE / 16384; // timeTwoStreams
 
+    //    static int reps = Integer.MAX_VALUE / 1024; // timeSingleStream
     //    static int reps = 1000; // interval streams
 
     OperatorSerializePerformance() {
@@ -32,8 +31,9 @@ public static void main(String args[]) {
 
                 @Override
                 public void call() {
-                    //                    spt.timeTwoStreams();
-                    spt.timeSingleStream();
+                    spt.timeTwoStreams();
+                    //                    spt.timeSingleStream();
+                    //                    spt.timeTwoStreamsIntervals();
                 }
             });
         } catch (Exception e) {
@@ -43,24 +43,33 @@ public void call() {
     }
 
     /**
+     * 1 streams emitting in a tight loop. Testing for single-threaded overhead.
+     * 
+     * -> blocking synchronization (SynchronizedObserver)
      * 
-     * -> state machine technique
+     * Run: 10 - 58,186,310 ops/sec
+     * Run: 11 - 60,592,037 ops/sec
+     * Run: 12 - 58,099,263 ops/sec
+     * Run: 13 - 59,034,765 ops/sec
+     * Run: 14 - 58,231,548 ops/sec
+     * 
+     * -> state machine technique (SerializedObserverViaStateMachine)
      * 
      * Run: 10 - 34,668,810 ops/sec
      * Run: 11 - 32,874,312 ops/sec
      * Run: 12 - 33,389,339 ops/sec
      * Run: 13 - 35,269,946 ops/sec
      * Run: 14 - 34,165,013 ops/sec
      * 
-     * -> using queue and counter technique
+     * -> using queue and counter technique (SerializedObserverViaQueueAndCounter)
      * 
      * Run: 10 - 19,548,387 ops/sec
      * Run: 11 - 19,471,069 ops/sec
      * Run: 12 - 19,480,112 ops/sec
      * Run: 13 - 18,720,550 ops/sec
      * Run: 14 - 19,070,383 ops/sec
      * 
-     * -> using queue and lock technique
+     * -> using queue and lock technique (SerializedObserverViaQueueAndLock)
      * 
      * Run: 10 - 51,295,152 ops/sec
      * Run: 11 - 50,317,937 ops/sec
@@ -112,29 +121,41 @@ public void call(Integer t1) {
     }
 
     /**
-     * -> state machine technique
+     * 2 streams emitting in tight loops so very high contention.
+     * 
+     * -> blocking synchronization (SynchronizedObserver)
+     * 
+     * Run: 10 - 8,361,252 ops/sec
+     * Run: 11 - 7,184,728 ops/sec
+     * Run: 12 - 8,249,685 ops/sec
+     * Run: 13 - 6,831,595 ops/sec
+     * Run: 14 - 8,003,358 ops/sec
+     * 
+     * (faster because it allows each thread to be "single threaded" while blocking the other)
+     * 
+     * -> state machine technique (SerializedObserverViaStateMachine)
      * 
-     * Run: 10 - 3,432,256 ops/sec
-     * Run: 11 - 3,570,444 ops/sec
-     * Run: 12 - 3,791,137 ops/sec
-     * Run: 13 - 3,664,579 ops/sec
-     * Run: 14 - 5,211,156 ops/sec
+     * Run: 10 - 4,060,062 ops/sec
+     * Run: 11 - 3,561,131 ops/sec
+     * Run: 12 - 3,721,387 ops/sec
+     * Run: 13 - 3,693,909 ops/sec
+     * Run: 14 - 3,516,324 ops/sec
      * 
-     * -> using "observeOn" technique
+     * -> using queue and counter technique (SerializedObserverViaQueueAndCounter)
      * 
-     * Run: 10 - 3,995,336 ops/sec
-     * Run: 11 - 4,033,077 ops/sec
-     * Run: 12 - 4,510,978 ops/sec
-     * Run: 13 - 3,218,915 ops/sec
-     * Run: 14 - 3,938,549 ops/sec
+     * Run: 10 - 4,300,229 ops/sec
+     * Run: 11 - 4,395,995 ops/sec
+     * Run: 12 - 4,551,550 ops/sec
+     * Run: 13 - 4,443,235 ops/sec
+     * Run: 14 - 4,158,475 ops/sec
      * 
-     * -> using queue and lock technique
+     * -> using queue and lock technique (SerializedObserverViaQueueAndLock)
      * 
-     * Run: 10 - 5,348,090 ops/sec
-     * Run: 11 - 6,458,608 ops/sec
-     * Run: 12 - 5,430,743 ops/sec
-     * Run: 13 - 5,159,666 ops/sec
-     * Run: 14 - 6,129,682 ops/sec
+     * Run: 10 - 6,369,781 ops/sec
+     * Run: 11 - 6,933,872 ops/sec
+     * Run: 12 - 5,652,535 ops/sec
+     * Run: 13 - 5,503,716 ops/sec
+     * Run: 14 - 6,219,264 ops/sec
      */
     public long timeTwoStreams() {
 
@@ -198,11 +219,39 @@ public void call(Integer t1) {
     }
 
     /**
+     * 2 streams emitting once a millisecond. Slow emission so little to no contention.
+     * 
+     * -> blocking synchronization (SynchronizedObserver)
+     * 
+     * Run: 10 - 1,996 ops/sec
+     * Run: 11 - 1,996 ops/sec
+     * Run: 12 - 1,995 ops/sec
+     * Run: 13 - 1,997 ops/sec
+     * Run: 14 - 1,996 ops/sec
+     * 
+     * -> state machine technique (SerializedObserverViaStateMachine)
+     * 
      * Run: 10 - 1,996 ops/sec
      * Run: 11 - 1,996 ops/sec
      * Run: 12 - 1,996 ops/sec
      * Run: 13 - 1,996 ops/sec
      * Run: 14 - 1,996 ops/sec
+     * 
+     * -> using queue and counter technique (SerializedObserverViaQueueAndCounter)
+     * 
+     * Run: 10 - 1,996 ops/sec
+     * Run: 11 - 1,996 ops/sec
+     * Run: 12 - 1,996 ops/sec
+     * Run: 13 - 1,996 ops/sec
+     * Run: 14 - 1,995 ops/sec
+     * 
+     * -> using queue and lock technique (SerializedObserverViaQueueAndLock)
+     * 
+     * Run: 10 - 1,996 ops/sec
+     * Run: 11 - 1,996 ops/sec
+     * Run: 12 - 1,997 ops/sec
+     * Run: 13 - 1,996 ops/sec
+     * Run: 14 - 1,995 ops/sec
      */
     public long timeTwoStreamsIntervals() {