2.x: fix SubscriptionArbiter reentrancy/cancel bug (ReactiveX#4310)

akarnokd · web-flow · commit c0aff9c7b621 · 2016-08-08T15:53:22.000+02:00
* 2.x: fix SubscriptionArbiter reentrancy/cancel bug

* Use an infinite source with timeout as 1G element takes only 2 secs.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRedo.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRedo.java
@@ -22,7 +22,7 @@
 import io.reactivex.functions.*;
 import io.reactivex.internal.subscribers.flowable.ToNotificationSubscriber;
 import io.reactivex.internal.subscriptions.SubscriptionArbiter;
-import io.reactivex.processors.BehaviorProcessor;
+import io.reactivex.processors.*;
 
 // FIXME split and update to the Rsc version
 public final class FlowableRedo<T> extends Flowable<T> {
@@ -39,7 +39,7 @@ public FlowableRedo(Publisher<? extends T> source,
     public void subscribeActual(Subscriber<? super T> s) {
         
         // FIXE use BehaviorSubject? (once available)
-        BehaviorProcessor<Try<Optional<Object>>> subject = BehaviorProcessor.create();
+        FlowProcessor<Try<Optional<Object>>> subject = BehaviorProcessor.<Try<Optional<Object>>>create().toSerialized();
         
         final RedoSubscriber<T> parent = new RedoSubscriber<T>(s, subject, source);
 
@@ -70,13 +70,13 @@ static final class RedoSubscriber<T> extends AtomicBoolean implements Subscriber
         /** */
         private static final long serialVersionUID = -1151903143112844287L;
         final Subscriber<? super T> actual;
-        final BehaviorProcessor<Try<Optional<Object>>> subject;
+        final FlowProcessor<Try<Optional<Object>>> subject;
         final Publisher<? extends T> source;
         final SubscriptionArbiter arbiter;
         
         final AtomicInteger wip = new AtomicInteger();
         
-        public RedoSubscriber(Subscriber<? super T> actual, BehaviorProcessor<Try<Optional<Object>>> subject, Publisher<? extends T> source) {
+        public RedoSubscriber(Subscriber<? super T> actual, FlowProcessor<Try<Optional<Object>>> subject, Publisher<? extends T> source) {
             this.actual = actual;
             this.subject = subject;
             this.source = source;
diff --git a/src/main/java/io/reactivex/internal/subscriptions/SubscriptionArbiter.java b/src/main/java/io/reactivex/internal/subscriptions/SubscriptionArbiter.java
@@ -96,15 +96,14 @@ public final void setSubscription(Subscription s) {
             actual = s;
             
             long r = requested;
-            if (r != 0L) {
-                s.request(r);
-            }
             
-            if (decrementAndGet() == 0) {
-                return;
+            if (decrementAndGet() != 0) {
+                drainLoop();
             }
 
-            drainLoop();
+            if (r != 0L) {
+                s.request(r);
+            }
 
             return;
         }
@@ -133,15 +132,14 @@ public final void request(long n) {
                     }
                 }
                 Subscription a = actual;
-                if (a != null) {
-                    a.request(n);
-                }
 
-                if (decrementAndGet() == 0) {
-                    return;
+                if (decrementAndGet() != 0) {
+                    drainLoop();
                 }
 
-                drainLoop();
+                if (a != null) {
+                    a.request(n);
+                }
 
                 return;
             }
@@ -235,6 +233,9 @@ final void drain() {
     final void drainLoop() {
         int missed = 1;
 
+        long requestAmount = 0L;
+        Subscription requestTarget = null;
+        
         for (; ; ) {
 
             Subscription ms = missedSubscription.get();
@@ -287,15 +288,20 @@ final void drainLoop() {
                     }
                     actual = ms;
                     if (r != 0L) {
-                        ms.request(r);
+                        requestAmount = BackpressureHelper.addCap(requestAmount, r);
+                        requestTarget = ms;
                     }
-                } else if (mr != 0L && a != null) {
-                    a.request(mr);
+                } else if (a != null && mr != 0L) {
+                    requestAmount = BackpressureHelper.addCap(requestAmount, mr);
+                    requestTarget = a;
                 }
             }
 
             missed = addAndGet(-missed);
             if (missed == 0) {
+                if (requestAmount != 0L) {
+                    requestTarget.request(requestAmount);
+                }
                 return;
             }
         }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableConcatTest.java
@@ -1007,4 +1007,35 @@ public void startWith() throws Exception {
             ts.assertComplete();
         }
     }
+    
+    static final class InfiniteIterator implements Iterator<Integer>, Iterable<Integer> {
+
+        int count;
+        
+        @Override
+        public boolean hasNext() {
+            return true;
+        }
+
+        @Override
+        public Integer next() {
+            return count++;
+        }
+        
+        @Override
+        public void remove() {
+        }
+        
+        @Override
+        public Iterator<Integer> iterator() {
+            return this;
+        }
+    }
+    
+    @Test(timeout = 5000)
+    public void veryLongTake() {
+        Flowable.fromIterable(new InfiniteIterator()).concatWith(Flowable.<Integer>empty()).take(10)
+        .test()
+        .assertResult(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -96,15 +96,14 @@ public final void setSubscription(Subscription s) {`
`96`	`96`	`actual = s;`
`97`	`97`
`98`	`98`	`long r = requested;`
`99`		`- if (r != 0L) {`
`100`		`- s.request(r);`
`101`		`- }`
`102`	`99`
`103`		`- if (decrementAndGet() == 0) {`
`104`		`- return;`
	`100`	`+ if (decrementAndGet() != 0) {`
	`101`	`+ drainLoop();`
`105`	`102`	`}`
`106`	`103`
`107`		`- drainLoop();`
	`104`	`+ if (r != 0L) {`
	`105`	`+ s.request(r);`
	`106`	`+ }`
`108`	`107`
`109`	`108`	`return;`
`110`	`109`	`}`
`@@ -133,15 +132,14 @@ public final void request(long n) {`
`133`	`132`	`}`
`134`	`133`	`}`
`135`	`134`	`Subscription a = actual;`
`136`		`- if (a != null) {`
`137`		`- a.request(n);`
`138`		`- }`
`139`	`135`
`140`		`- if (decrementAndGet() == 0) {`
`141`		`- return;`
	`136`	`+ if (decrementAndGet() != 0) {`
	`137`	`+ drainLoop();`
`142`	`138`	`}`
`143`	`139`
`144`		`- drainLoop();`
	`140`	`+ if (a != null) {`
	`141`	`+ a.request(n);`
	`142`	`+ }`
`145`	`143`
`146`	`144`	`return;`
`147`	`145`	`}`
`@@ -235,6 +233,9 @@ final void drain() {`
`235`	`233`	`final void drainLoop() {`
`236`	`234`	`int missed = 1;`
`237`	`235`
	`236`	`+ long requestAmount = 0L;`
	`237`	`+ Subscription requestTarget = null;`
	`238`	`+`
`238`	`239`	`for (; ; ) {`
`239`	`240`
`240`	`241`	`Subscription ms = missedSubscription.get();`
`@@ -287,15 +288,20 @@ final void drainLoop() {`
`287`	`288`	`}`
`288`	`289`	`actual = ms;`
`289`	`290`	`if (r != 0L) {`
`290`		`- ms.request(r);`
	`291`	`+ requestAmount = BackpressureHelper.addCap(requestAmount, r);`
	`292`	`+ requestTarget = ms;`
`291`	`293`	`}`
`292`		`- } else if (mr != 0L && a != null) {`
`293`		`- a.request(mr);`
	`294`	`+ } else if (a != null && mr != 0L) {`
	`295`	`+ requestAmount = BackpressureHelper.addCap(requestAmount, mr);`
	`296`	`+ requestTarget = a;`
`294`	`297`	`}`
`295`	`298`	`}`
`296`	`299`
`297`	`300`	`missed = addAndGet(-missed);`
`298`	`301`	`if (missed == 0) {`
	`302`	`+ if (requestAmount != 0L) {`
	`303`	`+ requestTarget.request(requestAmount);`
	`304`	`+ }`
`299`	`305`	`return;`
`300`	`306`	`}`
`301`	`307`	`}`