Fix Flowable.blockingSubscribe is unbounded and can lead to OOME #6026

RomanWuattier · 2018-05-29T11:57:26Z

Create and bound new blockingSubscribe overloads to bufferSize.

Create new overloads with bufferSize
Create a boundedConsumer
Create a BoundedSubsciber

codecov · 2018-05-29T12:11:57Z

Codecov Report

Merging #6026 into 2.x will decrease coverage by 0.03%.
The diff coverage is 94.44%.

@@             Coverage Diff             @@
##               2.x    #6026      +/-   ##
===========================================
- Coverage     98.3%   98.27%   -0.04%     
- Complexity    6175     6193      +18     
===========================================
  Files          665      666       +1     
  Lines        44729    44801      +72     
  Branches      6205     6206       +1     
===========================================
+ Hits         43973    44030      +57     
- Misses         222      233      +11     
- Partials       534      538       +4

Impacted Files	Coverage Δ	Complexity Δ
...ava/io/reactivex/internal/functions/Functions.java	`100% <100%> (ø)`	`36 <1> (+1)`	⬆️
src/main/java/io/reactivex/Flowable.java	`100% <100%> (ø)`	`564 <3> (+3)`	⬆️
.../operators/flowable/FlowableBlockingSubscribe.java	`93.02% <100%> (-4.28%)`	`10 <1> (ø)`
...ctivex/internal/subscribers/BoundedSubscriber.java	`92.59% <92.59%> (ø)`	`15 <15> (?)`
.../operators/observable/ObservableFlatMapSingle.java	`88.8% <0%> (-5.98%)`	`2% <0%> (ø)`
...ternal/operators/observable/ObservablePublish.java	`96.46% <0%> (-3.54%)`	`11% <0%> (ø)`
...a/io/reactivex/internal/util/QueueDrainHelper.java	`97.22% <0%> (-2.78%)`	`56% <0%> (-2%)`
...tivex/internal/schedulers/TrampolineScheduler.java	`96.1% <0%> (-2.6%)`	`6% <0%> (ø)`
...activex/internal/observers/QueueDrainObserver.java	`97.43% <0%> (-2.57%)`	`21% <0%> (-1%)`
... and 30 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f3c8862...8bd5b13. Read the comment docs.

akarnokd

There are some small missing pieces and one logical error.

akarnokd · 2018-05-29T12:35:36Z

src/main/java/io/reactivex/Flowable.java

+     * </dl>
+     * @param onNext the callback action for each source value
+     * @param bufferSize the size of the buffer
+     * @since 2.0


Please designate as 2.1.15 - experimental here and to the other methods.

akarnokd · 2018-05-29T12:35:55Z

src/main/java/io/reactivex/Flowable.java

+     */
+    @BackpressureSupport(BackpressureKind.FULL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final void blockingSubscribe(Consumer<? super T> onNext, int bufferSize) {


Please add @Experimental annotation to these new methods.

akarnokd · 2018-05-29T12:38:36Z

src/main/java/io/reactivex/internal/subscribers/BoundedSubscriber.java

+    final Action onComplete;
+    final Consumer<? super Subscription> onSubscribe;
+
+    private int bufferSize;


Please no private instance fields.

akarnokd · 2018-05-29T12:38:58Z

src/main/java/io/reactivex/internal/subscribers/BoundedSubscriber.java

+    final Consumer<? super Subscription> onSubscribe;
+
+    private int bufferSize;
+    private final AtomicInteger internalBuffer;


Why is this atomic?

akarnokd · 2018-05-29T12:39:36Z

src/main/java/io/reactivex/internal/subscribers/BoundedSubscriber.java

+                if (internalBuffer.getAndIncrement() < bufferSize) {
+                    onNext.accept(t);
+                }
+                if (internalBuffer.get() == bufferSize) {


Why would you end the flow when the bufferSize number of elements have been received?

akarnokd · 2018-05-29T12:40:15Z

src/test/java/io/reactivex/internal/operators/flowable/FlowableBlockingTest.java

+
+        assertEquals(Arrays.asList(1, 2, 3, 100), list);
+    }
+


Please add an unit test that transfers 1 million items.

davidmoten · 2018-05-29T23:13:40Z

src/main/java/io/reactivex/internal/subscribers/BoundedSubscriber.java

+
+    @Override
+    public void onComplete() {
+        if (get() != SubscriptionHelper.CANCELLED) {


if we are going to check the value of get() may as well do a compareAndSet loop for concurrent cancellation?

Don't. That level of precision it is not really worth it here.

akarnokd · 2018-05-30T10:47:29Z

src/main/java/io/reactivex/internal/subscribers/BoundedSubscriber.java

+                onNext.accept(t);
+
+                int c = consumed + 1;
+                if (c == bufferSize) {


It is better to use a low-watermark than requesting only when the entire buffer has been consumed. Define a final int limit field, in the constructor set it as this.limit = bufferSize - (bufferSize >> 2);, then compare and request as limit.

akarnokd

One algorithmic adjustment and good to go.

outdated

akarnokd · 2018-05-30T10:56:22Z

Also it would be great if you copied the unit tests that crash the callbacks so that the coverage on the new class gets very high.

akarnokd · 2018-05-30T10:59:24Z

https://codecov.io/gh/ReactiveX/RxJava/pull/6026/diff?src=pr&el=tree#diff-c3JjL21haW4vamF2YS9pby9yZWFjdGl2ZXgvaW50ZXJuYWwvc3Vic2NyaWJlcnMvQm91bmRlZFN1YnNjcmliZXIuamF2YQ==

RomanWuattier · 2018-06-01T09:31:52Z

I created a BoundedSubscriberTest.java to trigger Consumer and Action exceptions, but the code coverage remains unchanged. Did I miss a test?

akarnokd · 2018-06-01T09:55:05Z

src/main/java/io/reactivex/internal/operators/flowable/FlowableBlockingSubscribe.java

+     * @param bufferSize the number of elements to prefetch from the source Publisher
+     * @param <T> the value type
+     */
+    @SuppressWarnings("ResultOfMethodCallIgnored")


Why is this needed. It never appears anywhere else in RxJava. Please remove.

akarnokd · 2018-06-01T09:55:50Z

src/main/java/io/reactivex/internal/subscribers/BoundedSubscriber.java

+        this.onError = onError;
+        this.onComplete = onComplete;
+        this.onSubscribe = onSubscribe;
+        this.bufferSize = ((Functions.BoundedConsumer) onSubscribe).getBufferSize();


Why is this cast needed? Can't you could simply pass the bufferSize into the constructor?

akarnokd · 2018-06-01T09:57:37Z

src/test/java/io/reactivex/internal/subscribers/BoundedSubscriberTest.java

+            }, new Consumer<Throwable>() {
+                @Override
+                public void accept(Throwable throwable) throws Exception {
+                    throw new TestException("Inner");


Place a breakpoint here and debug the test. It should reveal why the code coverage doesn't seem to execute that code in BoundedSubscriber.

akarnokd · 2018-06-02T13:42:25Z

src/test/java/io/reactivex/internal/subscribers/BoundedSubscriberTest.java

@@ -1,11 +1,11 @@
 /**
 * Copyright (c) 2016-present, RxJava Contributors.
- *
+ * <p>


Please don't change the license headers.

akarnokd · 2018-06-02T13:45:43Z

src/test/java/io/reactivex/internal/subscribers/BoundedSubscriberTest.java

@@ -0,0 +1,388 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ * <p>


Please don't change the license headers.

akarnokd · 2018-06-02T14:03:59Z

src/test/java/io/reactivex/internal/subscribers/BoundedSubscriberTest.java

+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.testng.annotations.Test;


Please use org.junit.Test as the TestNG tests are not part of the code coverage.

akarnokd

Fix headers and use org.junit.Test please.

akarnokd previously requested changes May 29, 2018

View reviewed changes

davidmoten reviewed May 29, 2018

View reviewed changes

akarnokd reviewed May 30, 2018

View reviewed changes

akarnokd added the 2.x label May 30, 2018

akarnokd added this to the 2.2 milestone May 30, 2018

akarnokd added the Enhancement label May 30, 2018

RomanWuattier force-pushed the fix-unbounded-flowable-blocking-subscribe branch 9 times, most recently from 62a9816 to 74e7cde Compare May 30, 2018 21:06

akarnokd requested changes Jun 1, 2018

View reviewed changes

akarnokd reviewed Jun 2, 2018

View reviewed changes

akarnokd requested changes Jun 2, 2018

View reviewed changes

Fix Flowable.blockingSubscribe is unbounded and can lead to OOME

8bd5b13

RomanWuattier force-pushed the fix-unbounded-flowable-blocking-subscribe branch from db252ee to 8bd5b13 Compare June 2, 2018 14:49

akarnokd approved these changes Jun 9, 2018

View reviewed changes

akarnokd merged commit fc0ca6e into ReactiveX:2.x Jun 14, 2018

akarnokd mentioned this pull request Jun 17, 2018

2.1.15 release preparations #6048

Closed

jared-bot mentioned this pull request Jul 2, 2018

Dependency update: RxJava xmartlabs/bigbang-template#105

Closed

jared-bot mentioned this pull request Jul 23, 2018

Dependency update: RxJava xmartlabs/bigbang-template#113

Closed

jared-bot mentioned this pull request Jul 31, 2018

Dependency update: RxJava xmartlabs/bigbang-template#117

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix Flowable.blockingSubscribe is unbounded and can lead to OOME #6026

Fix Flowable.blockingSubscribe is unbounded and can lead to OOME #6026

RomanWuattier commented May 29, 2018

codecov bot commented May 29, 2018 •

edited

Loading

akarnokd left a comment

akarnokd May 29, 2018

akarnokd May 29, 2018

akarnokd May 29, 2018

akarnokd May 29, 2018

akarnokd May 29, 2018

akarnokd May 29, 2018

davidmoten May 29, 2018

akarnokd May 30, 2018

akarnokd May 30, 2018

akarnokd left a comment

akarnokd commented May 30, 2018

akarnokd commented May 30, 2018

RomanWuattier commented Jun 1, 2018

akarnokd Jun 1, 2018

akarnokd Jun 1, 2018

akarnokd Jun 1, 2018

akarnokd Jun 2, 2018

akarnokd Jun 2, 2018

akarnokd Jun 2, 2018

akarnokd left a comment

Fix Flowable.blockingSubscribe is unbounded and can lead to OOME #6026

Fix Flowable.blockingSubscribe is unbounded and can lead to OOME #6026

Conversation

RomanWuattier commented May 29, 2018

codecov bot commented May 29, 2018 • edited Loading

Codecov Report

akarnokd left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akarnokd left a comment

Choose a reason for hiding this comment

akarnokd commented May 30, 2018

akarnokd commented May 30, 2018

RomanWuattier commented Jun 1, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akarnokd left a comment

Choose a reason for hiding this comment

codecov bot commented May 29, 2018 •

edited

Loading