Range concurrency & off-by-one bugfix

Author: akarnokd

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

@@ -45,15 +45,13 @@ private static final class RangeProducer implements Producer {
         // accessed by REQUESTED_UPDATER
         private volatile int requested;
         private static final AtomicIntegerFieldUpdater<RangeProducer> REQUESTED_UPDATER = AtomicIntegerFieldUpdater.newUpdater(RangeProducer.class, "requested");
-        private volatile int index;
+        private int index;
         private final int end;
-        private final int start;
 
         private RangeProducer(Subscriber<? super Integer> o, int start, int end) {
             this.o = o;
             this.index = start;
             this.end = end;
-            this.start = start;
         }
 
         @Override
@@ -76,17 +74,20 @@ public void request(int n) {
                          * 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.
                          */
-                        int numLeft = start + (end - index);
-                        int e = Math.min(numLeft, requested);
-                        boolean completeOnFinish = numLeft < requested;
-                        int stopAt = e + index;
-                        for (int i = index; i < stopAt; i++) {
+                        int r = requested;
+                        int idx = index;
+                        int numLeft = end - idx + 1;
+                        int e = Math.min(numLeft, r);
+                        boolean completeOnFinish = numLeft <= r;
+                        int stopAt = e + idx;
+                        for (int i = idx; i < stopAt; i++) {
                             if (o.isUnsubscribed()) {
                                 return;
                             }
                             o.onNext(i);
                         }
-                        index += e;
+                        index = stopAt;
+                        
                         if (completeOnFinish) {
                             o.onCompleted();
                             return;

rxjava-core/src/test/java/rx/internal/operators/OnSubscribeRangeTest.java

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
@@ -130,4 +131,63 @@ public void testNoBackpressure() {
         ts.assertReceivedOnNext(list);
         ts.assertTerminalEvent();
     }
+    void testWithBackpressureOneByOne(int start) {
+        Observable<Integer> source = Observable.range(start, 100);
+        
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ts.request(1);
+        source.subscribe(ts);
+        
+        List<Integer> list = new ArrayList<Integer>(100);
+        for (int i = 0; i < 100; i++) {
+            list.add(i + start);
+            ts.request(1);
+        }
+        ts.assertReceivedOnNext(list);
+        ts.assertTerminalEvent();
+    }
+    void testWithBackpressureAllAtOnce(int start) {
+        Observable<Integer> source = Observable.range(start, 100);
+        
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ts.request(100);
+        source.subscribe(ts);
+        
+        List<Integer> list = new ArrayList<Integer>(100);
+        for (int i = 0; i < 100; i++) {
+            list.add(i + start);
+        }
+        ts.assertReceivedOnNext(list);
+        ts.assertTerminalEvent();
+    }
+    @Test
+    public void testWithBackpressure1() {
+        for (int i = 0; i < 100; i++) {
+            testWithBackpressureOneByOne(i);
+        }
+    }
+    @Test
+    public void testWithBackpressureAllAtOnce() {
+        for (int i = 0; i < 100; i++) {
+            testWithBackpressureAllAtOnce(i);
+        }
+    }
+    @Test
+    public void testWithBackpressureRequestWayMore() {
+        Observable<Integer> source = Observable.range(50, 100);
+        
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
+        ts.request(150);
+        source.subscribe(ts);
+        
+        List<Integer> list = new ArrayList<Integer>(100);
+        for (int i = 0; i < 100; i++) {
+            list.add(i + 50);
+        }
+        
+        ts.request(50); // and then some
+        
+        ts.assertReceivedOnNext(list);
+        ts.assertTerminalEvent();
+    }
 }