2.x: Fix buffer(open, close) not disposing indicators properly #5811

akarnokd · 2018-01-16T14:46:56Z

The PR fixes the resource management in the buffer operator that uses other reactive sources to indicate when a buffer starts and ends. Both Flowable and Observable implementations had to be fixed.

Fixes: #5809

simonbasle · 2018-01-16T15:11:57Z

@akarnokd let me know if I understand the spirit of this PR on FlowableBufferBoundary:

avoids keeping reference to / tracking buffers in subscribers other than the main
cancel source on open subscriber completion (if it completes and the last buffer's close publisher has completed, subscribers will have size 0 and we can safely cancel upstream as no new buffer will open at this point)
in main, distinguish a open subscriber error from a close subscriber error (due to 1., the later needs associated buffer cleanup)

Is that correct/exhaustive?

akarnokd · 2018-01-16T15:18:24Z

avoids keeping reference to / tracking buffers in subscribers other than the main

Yes.

cancel source on open subscriber completion (if it completes and the last buffer's close publisher has completed, subscribers will have size 0 and we can safely cancel upstream as no new buffer will open at this point

Yes.

in main, distinguish a open subscriber error from a close subscriber error (due to 1., the later needs associated buffer cleanup

No. If the openError or closeError is fired, both should cancel the main source, cancel the subscribers other than themselves, clean the buffer and then report the error. They could be converted into a common method of boundsError(Disposable d, Throwable ex).

codecov · 2018-01-16T15:19:45Z

Codecov Report

Merging #5811 into 2.x will increase coverage by 0.11%.
The diff coverage is 97.35%.

@@             Coverage Diff              @@
##                2.x    #5811      +/-   ##
============================================
+ Coverage     96.28%   96.39%   +0.11%     
+ Complexity     5817     5814       -3     
============================================
  Files           634      634              
  Lines         41647    41761     +114     
  Branches       5776     5796      +20     
============================================
+ Hits          40098    40255     +157     
+ Misses          611      583      -28     
+ Partials        938      923      -15

Impacted Files	Coverage Δ	Complexity Δ
...nal/operators/flowable/FlowableBufferBoundary.java	`96.05% <96.29%> (+18.85%)`	`2 <1> (ø)`	⬇️
...operators/observable/ObservableBufferBoundary.java	`98.31% <98.57%> (+20.45%)`	`2 <1> (ø)`	⬇️
.../operators/completable/CompletableConcatArray.java	`93.75% <0%> (-6.25%)`	`2% <0%> (ø)`
...nternal/operators/parallel/ParallelReduceFull.java	`91.17% <0%> (-3.93%)`	`2% <0%> (ø)`
...l/operators/observable/ObservableFlatMapMaybe.java	`84.96% <0%> (-2.62%)`	`2% <0%> (ø)`
...ain/java/io/reactivex/subjects/PublishSubject.java	`97.8% <0%> (-2.2%)`	`38% <0%> (-1%)`
.../internal/disposables/ListCompositeDisposable.java	`98% <0%> (-2%)`	`34% <0%> (-1%)`
...ex/internal/subscribers/InnerQueuedSubscriber.java	`96.07% <0%> (-1.97%)`	`18% <0%> (-1%)`
...ternal/operators/flowable/FlowableSampleTimed.java	`95.58% <0%> (-1.48%)`	`3% <0%> (ø)`
...a/io/reactivex/internal/util/QueueDrainHelper.java	`63.19% <0%> (-1.39%)`	`36% <0%> (-1%)`
... and 20 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 aa31330...04598ce. Read the comment docs.

simonbasle · 2018-01-16T16:08:18Z

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

+                for (C b : bufs.values()) {
+                    queue.offer(b);
+                }
+                bufs = null;


shouldn't this be buffers = null?

Yes, fixed. Also combined the error methods and changed to LinkedHashMap so that the final open buffers in onComplete remain in their order they were opened.

simonbasle

got it, looking better with the buffers null-out and the mutualized boundaryError 👍

davidmoten

gave quick scan only

davidmoten · 2018-01-16T22:30:58Z

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

+                }
+
+                emitted = e;
+                missed = addAndGet(-missed);


requested not getting reduced by emitted?

No longer necessary in these type of drain loops. requested can only grow and the emitted will follow it separately. This saves an atomic decrement here.

artem-zinnatullin · 2018-01-18T03:05:19Z

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

-            resources.add(bcs);
+        void boundaryError(Disposable subscriber, Throwable ex) {
+            SubscriptionHelper.cancel(upstream);
+            subscribers.delete(subscriber);


Can we just dispose() all subscribers here and remove dispose() call below?

Same in Observable version

It avoids disposing the caller which is known to have reached its terminal state.

artem-zinnatullin · 2018-01-18T03:08:34Z

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

+                            a.onError(ex);
+                            return;
+                        } else
+                        if (q.isEmpty()) {


nit: formatting, this line typically goes on the same line with else

artem-zinnatullin · 2018-01-18T03:18:23Z

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

+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                SubscriptionHelper.cancel(upstream);
+                if (errors.addThrowable(ex)) {


if/else block seems to be exact same as onError(), let's just call it here?

artem-zinnatullin · 2018-01-18T03:20:51Z

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

+        void boundaryError(Disposable subscriber, Throwable ex) {
+            SubscriptionHelper.cancel(upstream);
+            subscribers.delete(subscriber);
+            if (errors.addThrowable(ex)) {


if/else block seems to be exact same as onError(), let's just call it here?

Same in Observable version

artem-zinnatullin · 2018-01-18T03:23:50Z

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

+    @Test
+    @SuppressWarnings("unchecked")
+    public void boundaryOpenCloseDisposedOnComplete() {
+        PublishProcessor<Integer> pp0 = PublishProcessor.create();


nit: upstream

artem-zinnatullin · 2018-01-18T03:24:31Z

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

+    public void boundaryOpenCloseDisposedOnComplete() {
+        PublishProcessor<Integer> pp0 = PublishProcessor.create();
+
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();


nit: openingIndicator

artem-zinnatullin · 2018-01-18T03:24:44Z

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

+
+        PublishProcessor<Integer> pp1 = PublishProcessor.create();
+
+        PublishProcessor<Integer> pp2 = PublishProcessor.create();


nit: closingIndicator

and same in some tests below

artem-zinnatullin

lgtm

vanniktech

👍

2.x: Fix buffer(open, close) not disposing indicators properly

08fffce

akarnokd added Bug 2.x labels Jan 16, 2018

akarnokd added this to the 2.2 milestone Jan 16, 2018

akarnokd mentioned this pull request Jan 16, 2018

2.x: FlowableBufferBoundary doesn't stop opening new windows in overlapping case #5809

Closed

simonbasle reviewed Jan 16, 2018

View reviewed changes

Unify boundary error methods, fix cleanup, fix last buffer orders

4483c9e

simonbasle approved these changes Jan 16, 2018

View reviewed changes

davidmoten reviewed Jan 16, 2018

View reviewed changes

davidmoten approved these changes Jan 16, 2018

View reviewed changes

artem-zinnatullin reviewed Jan 18, 2018

View reviewed changes

Fix nits.

04598ce

artem-zinnatullin approved these changes Jan 18, 2018

View reviewed changes

akarnokd merged commit eb426fd into ReactiveX:2.x Jan 19, 2018

akarnokd deleted the BufferOpenCloseFix branch January 19, 2018 09:36

akarnokd mentioned this pull request Jan 19, 2018

2.1.9 release preparations #5815

Closed

vanniktech reviewed Jan 21, 2018

View reviewed changes

jared-bot mentioned this pull request Jan 25, 2018

Dependency update: rxjava xmartlabs/bigbang-template#51

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

2.x: Fix buffer(open, close) not disposing indicators properly #5811

2.x: Fix buffer(open, close) not disposing indicators properly #5811

akarnokd commented Jan 16, 2018

simonbasle commented Jan 16, 2018

akarnokd commented Jan 16, 2018

codecov bot commented Jan 16, 2018 •

edited

Loading

simonbasle Jan 16, 2018

akarnokd Jan 16, 2018

simonbasle left a comment

davidmoten left a comment

davidmoten Jan 16, 2018

akarnokd Jan 16, 2018

davidmoten Jan 16, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

akarnokd Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin Jan 18, 2018

artem-zinnatullin left a comment

vanniktech left a comment


		PublishProcessor<Integer> pp1 = PublishProcessor.create();

		PublishProcessor<Integer> pp2 = PublishProcessor.create();

2.x: Fix buffer(open, close) not disposing indicators properly #5811

2.x: Fix buffer(open, close) not disposing indicators properly #5811

Conversation

akarnokd commented Jan 16, 2018

simonbasle commented Jan 16, 2018

akarnokd commented Jan 16, 2018

codecov bot commented Jan 16, 2018 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

simonbasle left a comment

Choose a reason for hiding this comment

davidmoten 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

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

artem-zinnatullin left a comment

Choose a reason for hiding this comment

vanniktech left a comment

Choose a reason for hiding this comment

codecov bot commented Jan 16, 2018 •

edited

Loading