2.x: Improve the Observable/Flowable cache() operators #6275

akarnokd · 2018-10-31T10:51:18Z

This PR rewrites the Observable.cache and Flowable.cache operators to allocate less and be more up-to-date algorithmically.

I've also added comments to help understand its inner workings in case someone is interested.

Resolves: #6270

codecov · 2018-10-31T11:06:53Z

Codecov Report

Merging #6275 into 2.x will decrease coverage by <.01%.
The diff coverage is 98.25%.

@@             Coverage Diff              @@
##                2.x    #6275      +/-   ##
============================================
- Coverage     98.22%   98.22%   -0.01%     
- Complexity     6213     6257      +44     
============================================
  Files           667      667              
  Lines         44905    44887      -18     
  Branches       6228     6213      -15     
============================================
- Hits          44110    44089      -21     
- Misses          251      255       +4     
+ Partials        544      543       -1

Impacted Files	Coverage Δ	Complexity Δ
src/main/java/io/reactivex/Observable.java	`100% <100%> (ø)`	`541 <2> (ø)`	⬇️
...internal/operators/observable/ObservableCache.java	`95.9% <98.18%> (+2.2%)`	`34 <33> (+25)`	⬆️
...vex/internal/operators/flowable/FlowableCache.java	`98.48% <98.27%> (-0.15%)`	`38 <37> (+27)`
.../operators/flowable/FlowableBlockingSubscribe.java	`93.02% <0%> (-4.66%)`	`10% <0%> (-1%)`
...a/io/reactivex/internal/util/QueueDrainHelper.java	`95.83% <0%> (-4.17%)`	`55% <0%> (-3%)`
...va/io/reactivex/internal/util/LinkedArrayList.java	`97.22% <0%> (-2.78%)`	`8% <0%> (-1%)`
...nternal/operators/observable/ObservableCreate.java	`95.72% <0%> (-2.57%)`	`2% <0%> (ø)`
...activex/internal/observers/QueueDrainObserver.java	`97.43% <0%> (-2.57%)`	`21% <0%> (-1%)`
...ernal/operators/flowable/FlowableFromIterable.java	`95.18% <0%> (-2.14%)`	`5% <0%> (ø)`
.../io/reactivex/disposables/CompositeDisposable.java	`98.14% <0%> (-1.86%)`	`39% <0%> (-1%)`
... and 19 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 fba8b61...0dc6be2. Read the comment docs.

vanniktech · 2018-10-31T12:54:11Z

src/test/java/io/reactivex/internal/operators/observable/ObservableCacheTest.java

@@ -35,7 +35,7 @@
 public class ObservableCacheTest {
    @Test
    public void testColdReplayNoBackpressure() {
-        ObservableCache<Integer> source = (ObservableCache<Integer>)ObservableCache.from(Observable.range(0, 1000));
+        ObservableCache<Integer> source = (ObservableCache<Integer>)new ObservableCache<Integer>(Observable.range(0, 1000), 16);


Is the cast really needed now?

So the test can access the the package-private methods verifying the internal state.

Either I'm missing something or

ObservableCache<Integer> source = (ObservableCache<Integer>)new ObservableCache<Integer>(Observable.range(0, 1000), 16);

can just be

ObservableCache<Integer> source = new ObservableCache<Integer>(Observable.range(0, 1000), 16);

You are right. Updated.

vanniktech · 2018-10-31T12:54:36Z

src/test/java/io/reactivex/internal/operators/observable/ObservableCacheTest.java

@@ -351,4 +351,23 @@ public void run() {
            .assertSubscribed().assertValueCount(500).assertComplete().assertNoErrors();
        }
    }
+
+    @Test
+    public void cancelledUpFront() {


Flowable has this test?

Yes it has.

artem-zinnatullin · 2018-10-31T14:34:51Z

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

+    void remove(CacheSubscription<T> consumer) {
+        for (;;) {
+            CacheSubscription<T>[] current = subscribers.get();
+            int n = current.length;


artem-zinnatullin · 2018-10-31T19:54:46Z

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

+        long index = consumer.index;
+        int offset = consumer.offset;
+        Node<T> node = consumer.node;
+        AtomicLong requested = consumer.requested;


Maybe remove this reference? It looks like the value is captured here and it's a bit misleading

I doubt you win much by capturing it here

It is accessed every turn as it doubles as the cancelled indicator.

artem-zinnatullin · 2018-11-01T00:38:02Z

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

+    @SuppressWarnings("unchecked")
+    void remove(CacheSubscription<T> consumer) {
+        for (;;) {
+            CacheSubscription<T>[] current = subscribers.get();


artem-zinnatullin · 2018-11-01T00:39:06Z

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

+        int offset = consumer.offset;
+        Node<T> node = consumer.node;
+        AtomicLong requested = consumer.requested;
+        Subscriber<? super T> downstream = consumer.downstream;


Would be great to mark what's possible as final here to improve readability

It's not our style.

Yea, I think I was overexcited to push code readability a bit further after recent rewrites you've done with multi-character variable names and comments :D

artem-zinnatullin · 2018-11-01T00:40:23Z

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

@@ -951,11 +951,11 @@ public void accept(String v) {
    @Test
    public void testUnsubscribeSource() throws Exception {
        Action unsubscribe = mock(Action.class);
-        Flowable<Integer> f = Flowable.just(1).doOnCancel(unsubscribe).cache();
+        Flowable<Integer> f = Flowable.just(1).doOnCancel(unsubscribe).replay().autoConnect();


2.x: Improve the Observable/Flowable cache() operators

dc6a161

akarnokd added Enhancement Performance 2.x labels Oct 31, 2018

akarnokd added this to the 2.2 backlog milestone Oct 31, 2018

vanniktech approved these changes Oct 31, 2018

View reviewed changes

akarnokd added 2 commits October 31, 2018 14:25

Remove unnecessary casting.

c70acb4

Remove another unnecessary cast.

0dc6be2

vanniktech approved these changes Oct 31, 2018

View reviewed changes

artem-zinnatullin approved these changes Nov 1, 2018

View reviewed changes

akarnokd merged commit c3cfb5a into ReactiveX:2.x Nov 1, 2018

akarnokd deleted the CacheImprovements branch November 1, 2018 06:31

akarnokd mentioned this pull request Nov 17, 2018

2.2.4 release preparations #6312

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

2.x: Improve the Observable/Flowable cache() operators #6275

2.x: Improve the Observable/Flowable cache() operators #6275

akarnokd commented Oct 31, 2018 •

edited

codecov bot commented Oct 31, 2018 •

edited

vanniktech Oct 31, 2018

akarnokd Oct 31, 2018

vanniktech Oct 31, 2018

akarnokd Oct 31, 2018

vanniktech Oct 31, 2018

akarnokd Oct 31, 2018

artem-zinnatullin Oct 31, 2018

artem-zinnatullin Oct 31, 2018

akarnokd Nov 1, 2018

artem-zinnatullin Nov 1, 2018

artem-zinnatullin Nov 1, 2018

akarnokd Nov 1, 2018

artem-zinnatullin Nov 1, 2018

artem-zinnatullin Nov 1, 2018

2.x: Improve the Observable/Flowable cache() operators #6275

2.x: Improve the Observable/Flowable cache() operators #6275

Conversation

akarnokd commented Oct 31, 2018 • edited

codecov bot commented Oct 31, 2018 • edited

Codecov Report

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

akarnokd commented Oct 31, 2018 •

edited

codecov bot commented Oct 31, 2018 •

edited