1.x: concat reduce overhead when streaming a source

Author: akarnokd

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

@@ -103,4 +103,27 @@ public static long addCap(long a, long b) {
         return u;
     }
 
+    /**
+     * Atomically subtracts a value from the requested amount unless it's at Long.MAX_VALUE.
+     * @param requested the requested amount holder
+     * @param n the value to subtract from the requested amount, has to be positive (not verified)
+     * @return the new requested amount
+     * @throws IllegalStateException if n is greater than the current requested amount, which
+     * indicates a bug in the request accounting logic
+     */
+    public static long produced(AtomicLong requested, long n) {
+        for (;;) {
+            long current = requested.get();
+            if (current == Long.MAX_VALUE) {
+                return Long.MAX_VALUE;
+            }
+            long next = current + n;
+            if (next < 0L) {
+                throw new IllegalStateException("More produced than requested: " + next);
+            }
+            if (requested.compareAndSet(current, next)) {
+                return next;
+            }
+        }
+    }
 }

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

@@ -112,9 +112,19 @@ public void onStart() {
         }
 
         private void requestFromChild(long n) {
-            if (n <=0) return;
+            if (n <= 0) return;
             // we track 'requested' so we know whether we should subscribe the next or not
-            long previous = BackpressureUtils.getAndAddRequest(requested, n);
+            
+            final AtomicLong requestedField = requested;
+            
+            long previous;
+            
+            if (requestedField.get() != Long.MAX_VALUE) {
+                previous = BackpressureUtils.getAndAddRequest(requestedField, n);
+            } else {
+                previous = Long.MAX_VALUE;
+            }
+            
             arbiter.request(n);
             if (previous == 0) {
                 if (currentSubscriber == null && wip.get() > 0) {
@@ -125,10 +135,6 @@ private void requestFromChild(long n) {
             } 
         }
 
-        private void decrementRequested() {
-            requested.decrementAndGet();
-        }
-
         @Override
         public void onNext(Observable<? extends T> t) {
             queue.add(nl.next(t));
@@ -167,8 +173,10 @@ void subscribeNext() {
                     child.onCompleted();
                 } else if (o != null) {
                     Observable<? extends T> obs = nl.getValue(o);
+                    
                     currentSubscriber = new ConcatInnerSubscriber<T>(this, child, arbiter);
                     current.set(currentSubscriber);
+
                     obs.unsafeSubscribe(currentSubscriber);
                 }
             } else {
@@ -179,6 +187,11 @@ void subscribeNext() {
                 }
             }
         }
+        
+        void produced(long c) {
+            arbiter.produced(c);
+            BackpressureUtils.produced(requested, c);
+        }
     }
 
     static class ConcatInnerSubscriber<T> extends Subscriber<T> {
@@ -187,6 +200,8 @@ static class ConcatInnerSubscriber<T> extends Subscriber<T> {
         private final ConcatSubscriber<T> parent;
         private final AtomicInteger once = new AtomicInteger();
         private final ProducerArbiter arbiter;
+        
+        long produced;
 
         public ConcatInnerSubscriber(ConcatSubscriber<T> parent, Subscriber<T> child, ProducerArbiter arbiter) {
             this.parent = parent;
@@ -196,9 +211,9 @@ public ConcatInnerSubscriber(ConcatSubscriber<T> parent, Subscriber<T> child, Pr
 
         @Override
         public void onNext(T t) {
+            produced++;
+            
             child.onNext(t);
-            parent.decrementRequested();
-            arbiter.produced(1);
         }
 
         @Override
@@ -212,6 +227,11 @@ public void onError(Throwable e) {
         @Override
         public void onCompleted() {
             if (once.compareAndSet(0, 1)) {
+                // signal the production count at once instead of one by one
+                long c = produced;
+                if (c != 0L) {
+                    parent.produced(c);
+                }
                 // terminal completion to parent so it continues to the next
                 parent.completeInner();
             }
@@ -221,6 +241,5 @@ public void onCompleted() {
         public void setProducer(Producer producer) {
             arbiter.setProducer(producer);
         }
-
     }
 }

src/perf/java/rx/operators/ConcatPerf.java

@@ -0,0 +1,75 @@
+/**
+ * Copyright 2014 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 java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.infra.Blackhole;
+
+import rx.Observable;
+import rx.jmh.LatchedObserver;
+
+/**
+ * Benchmark typical atomic operations on volatile fields and AtomicXYZ classes.
+ * <p>
+ * gradlew benchmarks "-Pjmh=-f 1 -tu s -bm thrpt -wi 5 -i 5 -r 1 .*ConcatPerf.*"
+ * <p>
+ * gradlew benchmarks "-Pjmh=-f 1 -tu ns -bm avgt -wi 5 -i 5 -r 1 .*ConcatPerf.*"
+ */
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Thread)
+public class ConcatPerf {
+
+    Observable<Integer> source;
+    
+    Observable<Integer> baseline;
+    
+    @Param({"1", "1000", "1000000"})
+    int count;
+    
+    @Setup
+    public void setup() {
+        Integer[] array = new Integer[count];
+        
+        for (int i = 0; i < count; i++) {
+            array[i] = 777;
+        }
+        
+        baseline = Observable.from(array);
+        
+        source = Observable.concat(baseline, Observable.<Integer>empty());
+    }
+    
+    @Benchmark
+    public void normal(Blackhole bh) {
+        source.subscribe(new LatchedObserver<Integer>(bh));
+    }
+
+    @Benchmark
+    public void baseline(Blackhole bh) {
+        baseline.subscribe(new LatchedObserver<Integer>(bh));
+    }
+}