3.x: Fix Observable.flatMap with maxConcurrency hangs (#6946)

* 3.x: Fix Observable.flatMap with maxConcurrency hangs * Verify Flowable
ReactiveX · Apr 4, 2020 · c478097 · c478097
1 parent 14a9525
commit c478097
Show file tree

Hide file tree

Showing 3 changed files with 75 additions and 12 deletions.
diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableFlatMap.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableFlatMap.java
@@ -318,6 +318,7 @@ void drainLoop() {
                 if (checkTerminate()) {
                     return;
                 }
+                int innerCompleted = 0;
                 SimplePlainQueue<U> svq = queue;
 
                 if (svq != null) {
@@ -333,9 +334,18 @@ void drainLoop() {
                         }
 
                         child.onNext(o);
+                        innerCompleted++;
                     }
                 }
 
+                if (innerCompleted != 0) {
+                    if (maxConcurrency != Integer.MAX_VALUE) {
+                        subscribeMore(innerCompleted);
+                        innerCompleted = 0;
+                    }
+                    continue;
+                }
+
                 boolean d = done;
                 svq = queue;
                 InnerObserver<?, ?>[] inner = observers.get();
@@ -353,7 +363,6 @@ void drainLoop() {
                     return;
                 }
 
-                int innerCompleted = 0;
                 if (n != 0) {
                     int j = Math.min(n - 1, lastIndex);
 
@@ -415,27 +424,33 @@ void drainLoop() {
 
                 if (innerCompleted != 0) {
                     if (maxConcurrency != Integer.MAX_VALUE) {
-                        while (innerCompleted-- != 0) {
-                            ObservableSource<? extends U> p;
-                            synchronized (this) {
-                                p = sources.poll();
-                                if (p == null) {
-                                    wip--;
-                                    continue;
-                                }
-                            }
-                            subscribeInner(p);
-                        }
+                        subscribeMore(innerCompleted);
+                        innerCompleted = 0;
                     }
                     continue;
                 }
+
                 missed = addAndGet(-missed);
                 if (missed == 0) {
                     break;
                 }
             }
         }
 
+        void subscribeMore(int innerCompleted) {
+            while (innerCompleted-- != 0) {
+                ObservableSource<? extends U> p;
+                synchronized (this) {
+                    p = sources.poll();
+                    if (p == null) {
+                        wip--;
+                        continue;
+                    }
+                }
+                subscribeInner(p);
+            }
+        }
+
         boolean checkTerminate() {
             if (disposed) {
                 return true;

diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMapTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableFlatMapTest.java
@@ -1478,4 +1478,28 @@ public void innerCompletesAfterOnNextInDrainThenCancels() {
         .requestMore(1)
         .assertValuesOnly(1);
     }
+
+    @Test(timeout = 5000)
+    public void mixedScalarAsync() {
+        for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
+            Flowable
+            .range(0, 20)
+            .flatMap(
+                    integer -> {
+                        if (integer % 5 != 0) {
+                            return Flowable
+                                    .just(integer);
+                        }
+
+                        return Flowable
+                                .just(-integer)
+                                .observeOn(Schedulers.computation());
+                    },
+                    false,
+                    1
+            )
+            .ignoreElements()
+            .blockingAwait();
+        }
+    }
 }
diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableFlatMapTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableFlatMapTest.java
@@ -1240,4 +1240,28 @@ public void fusedInnerCrash2() {
 
         to.assertFailure(TestException.class, 1, 2);
     }
+
+    @Test(timeout = 5000)
+    public void mixedScalarAsync() {
+        for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) {
+            Observable
+            .range(0, 20)
+            .flatMap(
+                    integer -> {
+                        if (integer % 5 != 0) {
+                            return Observable
+                                    .just(integer);
+                        }
+
+                        return Observable
+                                .just(-integer)
+                                .observeOn(Schedulers.computation());
+                    },
+                    false,
+                    1
+            )
+            .ignoreElements()
+            .blockingAwait();
+        }
+    }
 }