2.x: add limit() to limit both item count and request amount #5655

akarnokd · 2017-10-09T21:45:33Z

This PR adds the operator limit to limit the number of items as well as the total request amount.

Some asynchronous boundary-like sources, such as network libraries that translate from/to Flowable/Publisher, may not modulate or limit downstream requests and thus the other side may produce items unnecessarily.

Alternatives, such as take(N) is designed to go unbounded if its downstream requests more than N to improve performance of local flows and rebatchRequests(M) will keep requesting once 75% of M has been received.

Note that requests are negotiated on an operator boundary and limit's amount may not be preserved further upstream. For example, source.observeOn(Schedulers.computation()).limit(5) will still request the default (128) elements from the given source.

Related discussion: #5077.
Blog: Java 9 Flow API: taking and skipping - Limiting the request amount

codecov · 2017-10-09T22:01:14Z

Codecov Report

Merging #5655 into 2.x will decrease coverage by 0.03%.
The diff coverage is 100%.

@@             Coverage Diff              @@
##                2.x    #5655      +/-   ##
============================================
- Coverage     96.24%   96.21%   -0.04%     
+ Complexity     5853     5850       -3     
============================================
  Files           635      636       +1     
  Lines         41651    41703      +52     
  Branches       5768     5778      +10     
============================================
+ Hits          40089    40123      +34     
- Misses          616      623       +7     
- Partials        946      957      +11

Impacted Files	Coverage Δ	Complexity Δ
src/main/java/io/reactivex/Flowable.java	`100% <100%> (ø)`	`525 <2> (+2)`	⬆️
...vex/internal/operators/flowable/FlowableLimit.java	`100% <100%> (ø)`	`2 <2> (?)`
...ernal/operators/flowable/FlowableFlatMapMaybe.java	`88.88% <0%> (-7.73%)`	`2% <0%> (ø)`
.../operators/flowable/FlowableBlockingSubscribe.java	`91.89% <0%> (-5.41%)`	`9% <0%> (-1%)`
...rnal/subscribers/SinglePostCompleteSubscriber.java	`94.87% <0%> (-5.13%)`	`14% <0%> (-1%)`
.../io/reactivex/internal/schedulers/IoScheduler.java	`89.24% <0%> (-3.23%)`	`9% <0%> (ø)`
...a/io/reactivex/internal/util/QueueDrainHelper.java	`60.28% <0%> (-2.84%)`	`32% <0%> (-2%)`
...activex/internal/observers/QueueDrainObserver.java	`61.53% <0%> (-2.57%)`	`12% <0%> (-1%)`
...ternal/operators/completable/CompletableMerge.java	`96.42% <0%> (-2.39%)`	`2% <0%> (ø)`
.../operators/observable/ObservableFlatMapSingle.java	`88.8% <0%> (-2.24%)`	`2% <0%> (ø)`
... and 26 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 e02cb58...a7e3037. Read the comment docs.

artem-zinnatullin · 2017-10-10T11:10:35Z

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

+    @BackpressureSupport(BackpressureKind.SPECIAL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @CheckReturnValue
+    public final Flowable<T> limit(long count) {


I like this strict version of take(), but thinking about API design I'm afraid that this additional operator could add unnecessary confusion for the users.

Maybe an overload of take() with boolean flag that controls "strictness" of the upstream requests would work? (or something better if you have it on your mind)

artem-zinnatullin · 2017-10-10T11:14:43Z

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

+
+        @Override
+        public void onNext(T t) {
+            long r = remaining;


what's the point of local r? It looks like you can just do everything directly on remaining here

I avoid re-reading fields.

hm, it's not volatile, so it should be normally in CPU register/cache, but yeah local var has slightly more chances to be there

Although local var creates indirection between read/writes, so overall I think it's neither win nor loss in performance :)

There is interesting question on SO, but answers seem to only compare speed of access to the field when it's already initialized, not taking overall overhead of additional variable https://stackoverflow.com/questions/21613098/java-local-vs-instance-variable-access-speed

The JIT may load it into a register but not everything runs on HotSpot.

artem-zinnatullin · 2017-10-10T11:23:28Z

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

+                remaining = --r;
+                actual.onNext(t);
+                if (r == 0L) {
+                    upstream.cancel();


What about this:

if (r > 0L) { remaining = --r; final boolean completed = r == 0L; if (completed) { upstream.cancel(); } actual.onNext(t); if (completed) { actual.onComplete(); } }

This way we cancel upstream before delivering onNext notification, it could be important in case if onNext triggers long running chain of computations thus delaying cancellation of the upstream (in current version).

Though, current implementations of take() also does this so it might be unwanted behavior change, up2u, we can move this change to 3.x

Or it could trigger an interrupted thread and fail code in the downstream unexpectedly.

Good point!

But doesn't it mean that actual.onComplete() should happen before upstream.cancel() then?

It's way less likely to happen. There were issues around onNext before.

ok, although I think probability is the same here, upstream.cancel() has to be called in both ways of call ordering and possibly cause problems. But in current order we guarantee delivery of onNext which is probably good :)

artem-zinnatullin · 2017-10-10T11:34:55Z

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

+
+public class FlowableLimitTest implements LongConsumer, Action {
+
+    final List<Long> requests = Collections.synchronizedList(new ArrayList<Long>());


nit: I don't think you need synchronization here, I don't see any concurrent access to this list

Just in case.

artem-zinnatullin · 2017-10-10T11:40:07Z

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

+
+    @Override
+    public void run() throws Exception {
+        requests.add(-100L);


super nit: List<Object>, private static final Object CANCELLED = new Object(), requests.add(CANCELLED)

assertEquals(5, requests.get(0)); assertEquals(CANCELLED, requests.get(1));

-100 is fine.

vanniktech · 2017-10-10T17:35:54Z

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

+
+        assertEquals(2, requests.size());
+        assertEquals(5, requests.get(0).intValue());
+        assertEquals(-100, requests.get(1).intValue());


let's use the constant here?

akarnokd · 2017-10-10T20:07:28Z

Updated. I'm not too fond of a take(long, boolean) overload. I suggest having limit as the name. I think the feature it provides is mainly useful for network/IO library writers and power users.

vanniktech · 2017-10-10T20:53:35Z

I also prefer to have a dedicated method instead of magic Boolean operators.

2.x: add limit() to limit both item count and request amount

e256e08

akarnokd added 2.x Enhancement labels Oct 9, 2017

artem-zinnatullin reviewed Oct 10, 2017

View reviewed changes

Address some feedback.

76e09ac

vanniktech approved these changes Oct 10, 2017

View reviewed changes

Use the cancelled constant value.

a7e3037

artem-zinnatullin approved these changes Oct 11, 2017

View reviewed changes

akarnokd merged commit 0f4a5e7 into ReactiveX:2.x Oct 18, 2017

akarnokd deleted the Limit2x branch October 18, 2017 13:39

This was referenced Oct 18, 2017

2.1.6 release preparation #5681

Closed

Take Doesn't Reduce Large RequestN #5077

Closed

jared-bot mentioned this pull request Dec 12, 2017

Dependency update: rxjava xmartlabs/bigbang-template#33

Closed

jared-bot mentioned this pull request Dec 27, 2017

Dependency update: rxjava xmartlabs/bigbang-template#45

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

2.x: add limit() to limit both item count and request amount #5655

2.x: add limit() to limit both item count and request amount #5655

akarnokd commented Oct 9, 2017

codecov bot commented Oct 9, 2017 •

edited

Loading

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

artem-zinnatullin Oct 10, 2017

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

artem-zinnatullin Oct 10, 2017

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

artem-zinnatullin Oct 10, 2017

akarnokd Oct 10, 2017

vanniktech Oct 10, 2017

akarnokd commented Oct 10, 2017

vanniktech commented Oct 10, 2017


		public class FlowableLimitTest implements LongConsumer, Action {

		final List<Long> requests = Collections.synchronizedList(new ArrayList<Long>());

2.x: add limit() to limit both item count and request amount #5655

2.x: add limit() to limit both item count and request amount #5655

Conversation

akarnokd commented Oct 9, 2017

codecov bot commented Oct 9, 2017 • edited Loading

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akarnokd commented Oct 10, 2017

vanniktech commented Oct 10, 2017

codecov bot commented Oct 9, 2017 •

edited

Loading