1.x: overhead reduction for merge and flatMap

Changes to the scalar fast-path was inspired by the Project Reactor's
flatMap which was in turn inspired by RxJava 2.x's implementation of
flatMap.

Author: akarnokd

src/main/java/rx/internal/operators/OnSubscribeRange.java

@@ -75,16 +75,16 @@ public void request(long n) {
          */
         void slowpath(long r) {
             long idx = index;
+            long e = 0L;
             while (true) {
                 /*
                  * This complicated logic is done to avoid touching the volatile `index` and `requested` values
                  * during the loop itself. If they are touched during the loop the performance is impacted significantly.
                  */
                 long fs = end - idx + 1;
-                long e = Math.min(fs, r);
                 final boolean complete = fs <= r;
 
-                fs = e + idx;
+                fs = Math.min(fs, r) + idx;
                 final Subscriber<? super Integer> o = this.o;
 
                 for (long i = idx; i != fs; i++) {
@@ -102,13 +102,19 @@ void slowpath(long r) {
                     return;
                 }
 
+                e -= fs - idx;
+                
                 idx = fs;
-                index = fs;
 
-                r = addAndGet(-e);
+                r = get() + e;
                 if (r == 0L) {
-                    // we're done emitting the number requested so return
-                    return;
+                    index = fs;
+                    r = addAndGet(e);
+                    if (r == 0L) {
+                        // we're done emitting the number requested so return
+                        return;
+                    }
+                    e = 0L;
                 }
             }
         }

src/main/java/rx/internal/operators/OperatorMerge.java

@@ -175,6 +175,10 @@ static final class MergeSubscriber<T> extends Subscriber<Observable<? extends T>
         /** An empty array to avoid creating new empty arrays in removeInner. */ 
         static final InnerSubscriber<?>[] EMPTY = new InnerSubscriber<?>[0];
 
+        int scalarEmission;
+        
+        final int scalarLimit;
+        
         public MergeSubscriber(Subscriber<? super T> child, boolean delayErrors, int maxConcurrent) {
             this.child = child;
             this.delayErrors = delayErrors;
@@ -183,6 +187,7 @@ public MergeSubscriber(Subscriber<? super T> child, boolean delayErrors, int max
             this.innerGuard = new Object();
             this.innerSubscribers = EMPTY;
             long r = Math.min(maxConcurrent, RxRingBuffer.SIZE);
+            scalarLimit = Math.max((int)r >> 1, 1);
             request(r);
         }
 
@@ -483,7 +488,11 @@ protected void emitScalar(T value, long r) {
                 if (r != Long.MAX_VALUE) {
                     producer.produced(1);
                 }
-                this.requestMore(1);
+                
+                if (++scalarEmission == scalarLimit) {
+                    scalarEmission = 0;
+                    this.requestMore(scalarLimit);
+                }
                 // check if some state changed while emitting
                 synchronized (this) {
                     skipFinal = true;