From f295e0bda611e32623264a22281171a08b034254 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Thu, 17 Oct 2019 13:52:53 +0200 Subject: [PATCH] 3.x: Fix concurrent clear() calls when fused chains are canceled --- .../operators/flowable/FlowableGroupBy.java | 3 +- .../FlowableOnBackpressureBuffer.java | 2 +- .../rxjava3/processors/UnicastProcessor.java | 8 +- .../rxjava3/subjects/UnicastSubject.java | 5 +- .../flowable/FlowableGroupByTest.java | 84 ++++++++++++++++++- .../FlowableOnBackpressureBufferTest.java | 40 ++++++++- .../processors/UnicastProcessorTest.java | 39 ++++++++- .../rxjava3/subjects/UnicastSubjectTest.java | 38 ++++++++- 8 files changed, 205 insertions(+), 14 deletions(-) diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java index 5bda5c7530..60b9437f05 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupBy.java @@ -268,7 +268,7 @@ public void cancel(K key) { if (groupCount.decrementAndGet() == 0) { upstream.cancel(); - if (getAndIncrement() == 0) { + if (!outputFused && getAndIncrement() == 0) { queue.clear(); } } @@ -601,7 +601,6 @@ void drainFused() { for (;;) { if (a != null) { if (cancelled.get()) { - q.clear(); return; } diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java index 4485174e29..df04ac54db 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java @@ -150,7 +150,7 @@ public void cancel() { cancelled = true; upstream.cancel(); - if (getAndIncrement() == 0) { + if (!outputFused && getAndIncrement() == 0) { queue.clear(); } } diff --git a/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java b/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java index d5c1c1b072..a3715103d5 100644 --- a/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java +++ b/src/main/java/io/reactivex/rxjava3/processors/UnicastProcessor.java @@ -345,7 +345,6 @@ void drainFused(Subscriber a) { for (;;) { if (cancelled) { - q.clear(); downstream.lazySet(null); return; } @@ -548,10 +547,11 @@ public void cancel() { doTerminate(); - if (!enableOperatorFusion) { - if (wip.getAndIncrement() == 0) { + downstream.lazySet(null); + if (wip.getAndIncrement() == 0) { + downstream.lazySet(null); + if (!enableOperatorFusion) { queue.clear(); - downstream.lazySet(null); } } } diff --git a/src/main/java/io/reactivex/rxjava3/subjects/UnicastSubject.java b/src/main/java/io/reactivex/rxjava3/subjects/UnicastSubject.java index b709a96d6c..be824a7708 100644 --- a/src/main/java/io/reactivex/rxjava3/subjects/UnicastSubject.java +++ b/src/main/java/io/reactivex/rxjava3/subjects/UnicastSubject.java @@ -418,7 +418,6 @@ void drainFused(Observer a) { if (disposed) { downstream.lazySet(null); - q.clear(); return; } boolean d = done; @@ -556,7 +555,9 @@ public void dispose() { downstream.lazySet(null); if (wip.getAndIncrement() == 0) { downstream.lazySet(null); - queue.clear(); + if (!enableOperatorFusion) { + queue.clear(); + } } } } diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupByTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupByTest.java index be36af652e..891e4f86f5 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupByTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableGroupByTest.java @@ -30,12 +30,13 @@ import com.google.common.cache.*; import io.reactivex.rxjava3.core.*; -import io.reactivex.rxjava3.exceptions.TestException; +import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.flowables.GroupedFlowable; import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.Functions; -import io.reactivex.rxjava3.internal.fuseable.QueueFuseable; +import io.reactivex.rxjava3.internal.fuseable.*; import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; import io.reactivex.rxjava3.processors.PublishProcessor; import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.subjects.PublishSubject; @@ -2332,4 +2333,83 @@ public void accept(GroupedFlowable g) throws Throwable { ts2.assertFailure(TestException.class, 1); } + + @Test + public void fusedNoConcurrentCleanDueToCancel() { + for (int j = 0; j < TestHelper.RACE_LONG_LOOPS; j++) { + List errors = TestHelper.trackPluginErrors(); + try { + final PublishProcessor pp = PublishProcessor.create(); + + final AtomicReference>> qs = new AtomicReference>>(); + + final TestSubscriber ts2 = new TestSubscriber(); + + pp.groupBy(Functions.identity(), Functions.identity(), false, 4) + .subscribe(new FlowableSubscriber>() { + + boolean once; + + @Override + public void onNext(GroupedFlowable g) { + if (!once) { + try { + GroupedFlowable t = qs.get().poll(); + if (t != null) { + once = true; + t.subscribe(ts2); + } + } catch (Throwable ignored) { + // not relevant here + } + } + } + + @Override + public void onError(Throwable t) { + } + + @Override + public void onComplete() { + } + + @Override + public void onSubscribe(Subscription s) { + @SuppressWarnings("unchecked") + QueueSubscription> q = (QueueSubscription>)s; + qs.set(q); + q.requestFusion(QueueFuseable.ANY); + q.request(1); + } + }) + ; + + Runnable r1 = new Runnable() { + @Override + public void run() { + qs.get().cancel(); + qs.get().clear(); + } + }; + Runnable r2 = new Runnable() { + @Override + public void run() { + ts2.cancel(); + } + }; + + for (int i = 0; i < 100; i++) { + pp.onNext(i); + } + + TestHelper.race(r1, r2); + + if (!errors.isEmpty()) { + throw new CompositeException(errors); + } + } finally { + RxJavaPlugins.reset(); + } + } + } } diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java index 79e0d7120b..3d32d2c808 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java @@ -15,6 +15,7 @@ import static org.junit.Assert.*; +import java.util.List; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicBoolean; @@ -24,11 +25,15 @@ import io.reactivex.rxjava3.core.*; import io.reactivex.rxjava3.exceptions.*; import io.reactivex.rxjava3.functions.*; +import io.reactivex.rxjava3.internal.functions.Functions; import io.reactivex.rxjava3.internal.fuseable.QueueFuseable; import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription; +import io.reactivex.rxjava3.observers.TestObserver; +import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.processors.PublishProcessor; import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.subscribers.*; -import io.reactivex.rxjava3.testsupport.TestSubscriberEx; +import io.reactivex.rxjava3.testsupport.*; public class FlowableOnBackpressureBufferTest extends RxJavaTest { @@ -308,4 +313,37 @@ public void fusionRejected() { ts.assertFusionMode(QueueFuseable.NONE) .assertEmpty(); } + + @Test + public void fusedNoConcurrentCleanDueToCancel() { + for (int j = 0; j < TestHelper.RACE_LONG_LOOPS; j++) { + List errors = TestHelper.trackPluginErrors(); + try { + final PublishProcessor pp = PublishProcessor.create(); + + TestObserver to = pp.onBackpressureBuffer(4, false, true) + .observeOn(Schedulers.io()) + .map(Functions.identity()) + .observeOn(Schedulers.single()) + .firstOrError() + .test(); + + for (int i = 0; pp.hasSubscribers(); i++) { + pp.onNext(i); + } + + to + .awaitDone(5, TimeUnit.SECONDS) + ; + + if (!errors.isEmpty()) { + throw new CompositeException(errors); + } + + to.assertResult(0); + } finally { + RxJavaPlugins.reset(); + } + } + } } diff --git a/src/test/java/io/reactivex/rxjava3/processors/UnicastProcessorTest.java b/src/test/java/io/reactivex/rxjava3/processors/UnicastProcessorTest.java index 96229747dc..a552783994 100644 --- a/src/test/java/io/reactivex/rxjava3/processors/UnicastProcessorTest.java +++ b/src/test/java/io/reactivex/rxjava3/processors/UnicastProcessorTest.java @@ -16,16 +16,20 @@ import static org.junit.Assert.*; import java.util.List; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import io.reactivex.rxjava3.core.Observable; import io.reactivex.rxjava3.disposables.Disposable; -import io.reactivex.rxjava3.exceptions.TestException; +import io.reactivex.rxjava3.exceptions.*; +import io.reactivex.rxjava3.internal.functions.Functions; import io.reactivex.rxjava3.internal.fuseable.QueueFuseable; import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription; +import io.reactivex.rxjava3.observers.TestObserver; import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.subscribers.TestSubscriber; import io.reactivex.rxjava3.testsupport.*; @@ -439,4 +443,37 @@ public void unicastSubscriptionBadRequest() { RxJavaPlugins.reset(); } } + + @Test + public void fusedNoConcurrentCleanDueToCancel() { + for (int j = 0; j < TestHelper.RACE_LONG_LOOPS; j++) { + List errors = TestHelper.trackPluginErrors(); + try { + final UnicastProcessor us = UnicastProcessor.create(); + + TestObserver to = us + .observeOn(Schedulers.io()) + .map(Functions.identity()) + .observeOn(Schedulers.single()) + .firstOrError() + .test(); + + for (int i = 0; us.hasSubscribers(); i++) { + us.onNext(i); + } + + to + .awaitDone(5, TimeUnit.SECONDS) + ; + + if (!errors.isEmpty()) { + throw new CompositeException(errors); + } + + to.assertResult(0); + } finally { + RxJavaPlugins.reset(); + } + } + } } diff --git a/src/test/java/io/reactivex/rxjava3/subjects/UnicastSubjectTest.java b/src/test/java/io/reactivex/rxjava3/subjects/UnicastSubjectTest.java index 28c1d48de2..ba2deb2c91 100644 --- a/src/test/java/io/reactivex/rxjava3/subjects/UnicastSubjectTest.java +++ b/src/test/java/io/reactivex/rxjava3/subjects/UnicastSubjectTest.java @@ -17,16 +17,19 @@ import static org.mockito.Mockito.mock; import java.util.List; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import io.reactivex.rxjava3.core.Observable; import io.reactivex.rxjava3.disposables.*; -import io.reactivex.rxjava3.exceptions.TestException; +import io.reactivex.rxjava3.exceptions.*; +import io.reactivex.rxjava3.internal.functions.Functions; import io.reactivex.rxjava3.internal.fuseable.QueueFuseable; import io.reactivex.rxjava3.observers.TestObserver; import io.reactivex.rxjava3.plugins.RxJavaPlugins; +import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.testsupport.*; public class UnicastSubjectTest extends SubjectTest { @@ -457,4 +460,37 @@ public void drainFusedFailFastEmpty() { to.assertEmpty(); } + + @Test + public void fusedNoConcurrentCleanDueToCancel() { + for (int j = 0; j < TestHelper.RACE_LONG_LOOPS; j++) { + List errors = TestHelper.trackPluginErrors(); + try { + final UnicastSubject us = UnicastSubject.create(); + + TestObserver to = us + .observeOn(Schedulers.io()) + .map(Functions.identity()) + .observeOn(Schedulers.single()) + .firstOrError() + .test(); + + for (int i = 0; us.hasObservers(); i++) { + us.onNext(i); + } + + to + .awaitDone(5, TimeUnit.SECONDS) + ; + + if (!errors.isEmpty()) { + throw new CompositeException(errors); + } + + to.assertResult(0); + } finally { + RxJavaPlugins.reset(); + } + } + } }