Merge pull request #1463 from benjchristensen/merge-1420

Merge Bug: Missing Emissions

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/operators/BlockingOperatorToIterator.java

@@ -19,6 +19,7 @@
 import java.util.NoSuchElementException;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 
 import rx.Notification;
 import rx.Observable;
@@ -89,7 +90,13 @@ public T next() {
 
             private Notification<? extends T> take() {
                 try {
-                    return notifications.take();
+                    Notification<? extends T> n = notifications.poll(10000, TimeUnit.MILLISECONDS);
+                    if(n == null) {
+                        System.err.println("Timed out waiting for value. File a bug at github.com/Netflix/RxJava");
+                        throw new RuntimeException("Timed out waiting for value. File a bug at github.com/Netflix/RxJava");
+                    } else {
+                        return n;
+                    }
                 } catch (InterruptedException e) {
                     throw Exceptions.propagate(e);
                 }

rxjava-core/src/main/java/rx/internal/operators/OperatorMerge.java

@@ -44,6 +44,39 @@
  */
 public class OperatorMerge<T> implements Operator<T, Observable<? extends T>> {
 
+    /*
+     * benjchristensen => This class is complex and I'm not a fan of it despite writing it. I want to give some background
+     * as to why for anyone who wants to try and help improve it.
+     * 
+     * One of my first implementations that added backpressure support (Producer.request) was fairly elegant and used a simple
+     * queue draining approach. It was simple to understand as all onNext were added to their queues, then a single winner
+     * would drain the queues, similar to observeOn. It killed the Netflix API when I canaried it. There were two problems:
+     * (1) performance and (2) object allocation overhead causing massive GC pressure. Remember that merge is one of the most
+     * used operators (mostly due to flatmap) and is therefore critical to and a limiter of performance in any application.
+     * 
+     * All subsequent work on this class and the various fast-paths and branches within it have been to achieve the needed functionality
+     * while reducing or eliminating object allocation and keeping performance acceptable.
+     * 
+     * This has meant adopting strategies such as:
+     * 
+     * - ring buffers instead of growable queues
+     * - object pooling
+     * - skipping request logic when downstream does not need backpressure
+     * - ScalarValueQueue for optimizing synchronous single-value Observables
+     * - adopting data structures that use Unsafe (and gating them based on environment so non-Oracle JVMs still work)
+     * 
+     * It has definitely increased the complexity and maintenance cost of this class, but the performance gains have been significant.
+     * 
+     * The biggest cost of the increased complexity is concurrency bugs and reasoning through what's going on.
+     * 
+     * I'd love to have contributions that improve this class, but keep in mind the performance and GC pressure.
+     * The benchmarks I use are in the JMH OperatorMergePerf class. GC memory pressure is tested using Java Flight Recorder
+     * to track object allocation.
+     * 
+     * TODO There is still a known concurrency bug somewhere either in this class, in SubscriptionIndexedRingBuffer or their relationship.
+     * See https://github.com/Netflix/RxJava/issues/1420 for more information on this.
+     */
+
     public OperatorMerge() {
         this.delayErrors = false;
     }

rxjava-core/src/main/java/rx/internal/util/SubscriptionIndexedRingBuffer.java

@@ -57,7 +57,8 @@ public boolean isUnsubscribed() {
      * 
      * @return int index that can be used to remove a Subscription
      */
-    public int add(final T s) {
+    public synchronized int add(final T s) {
+        // TODO figure out how to remove synchronized here. See https://github.com/Netflix/RxJava/issues/1420
         if (unsubscribed == 1 || subscriptions == null) {
             s.unsubscribe();
             return -1;
@@ -116,13 +117,14 @@ public void unsubscribe() {
     public int forEach(Func1<T, Boolean> action) {
         return forEach(action, 0);
     }
-    
+
     /**
      * 
      * @param action
      * @return int of last index seen if forEach exited early
      */
-    public int forEach(Func1<T, Boolean> action, int startIndex) {
+    public synchronized int forEach(Func1<T, Boolean> action, int startIndex) {
+        // TODO figure out how to remove synchronized here. See https://github.com/Netflix/RxJava/issues/1420
         if (unsubscribed == 1 || subscriptions == null) {
             return 0;
         }