1.x: range and flatMap overhead reduction

Author: akarnokd

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

@@ -25,90 +25,96 @@
  */
 public final class OnSubscribeRange implements OnSubscribe<Integer> {
 
-    private final int start;
-    private final int end;
+    private final int startIndex;
+    private final int endIndex;
 
-    public OnSubscribeRange(int start, int end) {
-        this.start = start;
-        this.end = end;
+    public OnSubscribeRange(int startIndex, int endIndex) {
+        this.startIndex = startIndex;
+        this.endIndex = endIndex;
     }
 
     @Override
-    public void call(final Subscriber<? super Integer> o) {
-        o.setProducer(new RangeProducer(o, start, end));
+    public void call(final Subscriber<? super Integer> childSubscriber) {
+        childSubscriber.setProducer(new RangeProducer(childSubscriber, startIndex, endIndex));
     }
 
     private static final class RangeProducer extends AtomicLong implements Producer {
         /** */
         private static final long serialVersionUID = 4114392207069098388L;
 
-        private final Subscriber<? super Integer> o;
-        private final int end;
-        private long index;
+        private final Subscriber<? super Integer> childSubscriber;
+        private final int endOfRange;
+        private long currentIndex;
 
-        private RangeProducer(Subscriber<? super Integer> o, int start, int end) {
-            this.o = o;
-            this.index = start;
-            this.end = end;
+        private RangeProducer(Subscriber<? super Integer> childSubscriber, int startIndex, int endIndex) {
+            this.childSubscriber = childSubscriber;
+            this.currentIndex = startIndex;
+            this.endOfRange = endIndex;
         }
 
         @Override
-        public void request(long n) {
+        public void request(long requestedAmount) {
             if (get() == Long.MAX_VALUE) {
                 // already started with fast-path
                 return;
             }
-            if (n == Long.MAX_VALUE && compareAndSet(0L, Long.MAX_VALUE)) {
+            if (requestedAmount == Long.MAX_VALUE && compareAndSet(0L, Long.MAX_VALUE)) {
                 // fast-path without backpressure
                 fastpath();
-            } else if (n > 0L) {
-                long c = BackpressureUtils.getAndAddRequest(this, n);
+            } else if (requestedAmount > 0L) {
+                long c = BackpressureUtils.getAndAddRequest(this, requestedAmount);
                 if (c == 0L) {
                     // backpressure is requested
-                    slowpath(n);
+                    slowpath(requestedAmount);
                 }
             }
         }
 
         /**
          * 
          */
-        void slowpath(long r) {
-            long idx = index;
+        void slowpath(long requestedAmount) {
+            long positionInRange = currentIndex;
+            long emitted = 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;
+                long remainingAmountOrEndIndex = endOfRange - positionInRange + 1;
+                final boolean complete = remainingAmountOrEndIndex <= requestedAmount;
 
-                fs = e + idx;
-                final Subscriber<? super Integer> o = this.o;
+                remainingAmountOrEndIndex = Math.min(remainingAmountOrEndIndex, requestedAmount) + positionInRange;
+                final Subscriber<? super Integer> childSubscriber = this.childSubscriber;
 
-                for (long i = idx; i != fs; i++) {
-                    if (o.isUnsubscribed()) {
+                for (long runningIndex = positionInRange; runningIndex != remainingAmountOrEndIndex; runningIndex++) {
+                    if (childSubscriber.isUnsubscribed()) {
                         return;
                     }
-                    o.onNext((int) i);
+                    childSubscriber.onNext((int) runningIndex);
                 }
 
                 if (complete) {
-                    if (o.isUnsubscribed()) {
+                    if (childSubscriber.isUnsubscribed()) {
                         return;
                     }
-                    o.onCompleted();
+                    childSubscriber.onCompleted();
                     return;
                 }
 
-                idx = fs;
-                index = fs;
+                emitted -= remainingAmountOrEndIndex - positionInRange;
 
-                r = addAndGet(-e);
-                if (r == 0L) {
-                    // we're done emitting the number requested so return
-                    return;
+                positionInRange = remainingAmountOrEndIndex;
+                
+                requestedAmount = get() + emitted;
+                if (requestedAmount == 0L) {
+                    currentIndex = remainingAmountOrEndIndex;
+                    requestedAmount = addAndGet(emitted);
+                    if (requestedAmount == 0L) {
+                        // we're done emitting the number requested so return
+                        return;
+                    }
+                    emitted = 0L;
                 }
             }
         }
@@ -117,18 +123,18 @@ void slowpath(long r) {
          * 
          */
         void fastpath() {
-            final long end = this.end + 1L;
-            final Subscriber<? super Integer> o = this.o;
-            for (long i = index; i != end; i++) {
-                if (o.isUnsubscribed()) {
+            final long end = this.endOfRange + 1L;
+            final Subscriber<? super Integer> childSubscriber = this.childSubscriber;
+            for (long runningIndex = currentIndex; runningIndex != end; runningIndex++) {
+                if (childSubscriber.isUnsubscribed()) {
                     return;
                 }
-                o.onNext((int) i);
+                childSubscriber.onNext((int) runningIndex);
             }
-            if (!o.isUnsubscribed()) {
-                o.onCompleted();
+            if (!childSubscriber.isUnsubscribed()) {
+                childSubscriber.onCompleted();
             }
         }
     }
 
-}
+}

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

@@ -177,14 +177,28 @@ 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;
             this.maxConcurrent = maxConcurrent;
             this.nl = NotificationLite.instance();
             this.innerGuard = new Object();
             this.innerSubscribers = EMPTY;
-            request(maxConcurrent == Integer.MAX_VALUE ? Long.MAX_VALUE : maxConcurrent);
+            long r;
+            int lim;
+            if (maxConcurrent == Integer.MAX_VALUE) {
+                r = Long.MAX_VALUE;
+                lim = Integer.MAX_VALUE;
+            } else {
+                r = maxConcurrent;
+                lim = Math.max(maxConcurrent >> 1, 1);
+            }
+            this.scalarLimit = lim;
+            request(r);
         }
 
         Queue<Throwable> getOrCreateErrorQueue() {
@@ -488,7 +502,10 @@ 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;