Take Doesn't Reduce Large RequestN

[benjchristensen]

When a child subscriber submits a large requestN value, such as Long.MAX_VALUE, the 'take' operator does not reduce it as expected.

For example, in the following where a default subscribe happens and requests Long.MAX_VALUE up, it is expected that the take(10) would adjust the requested size to the max value that take will permit through.

someFlowable.take(10).subscribe(s)

Here is a unit test:

    @Test
    public void testDoesntRequestMoreThanNeededFromUpstream2() throws InterruptedException {
        final AtomicLong requests = new AtomicLong();
        TestSubscriber<Long> ts = new TestSubscriber<Long>();
        Flowable.interval(100, TimeUnit.MILLISECONDS)
            .doOnRequest(new LongConsumer() {
                @Override
                public void accept(long n) {
                    System.out.println(n);
                    requests.addAndGet(n);
            }})
            .take(2)
            .subscribe(ts);
        
        ts.awaitTerminalEvent();
        ts.assertComplete();
        ts.assertNoErrors();
        assertEquals(2, requests.get());
    }

This errors with:

java.lang.AssertionError: expected:<2> but was:<9223372036854775807>
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.failNotEquals(Assert.java:834)
	at org.junit.Assert.assertEquals(Assert.java:645)
	at org.junit.Assert.assertEquals(Assert.java:631)
	at io.reactivex.internal.operators.flowable.FlowableTakeTest.testDoesntRequestMoreThanNeededFromUpstream2(FlowableTakeTest.java:419)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:497)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:86)
	at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:459)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:678)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:382)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:192)

Is there a reason that take in RxJava 2 does not behave this way and reduce the requestN value to the limit?

[akarnokd]

If the downstream is requesting more than N, the operator requests unbounded so the upstream can go on a fast-path when it emits. Take will then stop the upstream at N and there is no overflow. If the downstream requested M < N, take will also just request M.

Try this:

Flowable.interval(100, TimeUnit.MILLISECONDS)
            .doOnRequest(new LongConsumer() {
                @Override
                public void accept(long n) {
                    System.out.println(n);
                    requests.addAndGet(n);
            }})
            .take(2)
            .rebatchRequests(1)         // <-----------------------------------------
            .subscribe(ts);

[benjchristensen]

That doesn't make sense to me and differs from v1 behavior. If the upstream is across a network boundary that means someone has to know that take doesn't actually reduce the requested amount and therefore always needs to use both take and rebatch.

I don't buy the argument that if downstream requests more that this should be done, as it is normal for the downstream subscriber to use the default value and take to be used to constrain the output.

If unconstrained output was wanted I'd use an Observable, but I'm using Flowable so expect it to be constrained.

[akarnokd]

differs from v1 behavior

New major versions have the opportunity to change behavior on some level.

If the upstream is across a network boundary that means someone has to know that take doesn't actually reduce the requested amount and therefore always needs to use both take and rebatch.

Is this for the ReactiveSocket library? As an IO boundary provider, you should never trust the downstream's request amount the same way as observeOn doesn't trust it and requests on its own pace.

[benjchristensen]

No it's not for ReactiveSocket. But it's for Reactive Streams APIs that involve IO.

I understand a major version can change behavior, but I don't understand why this would change. Please explain why the very small perf improvement in the source is worth changing this behavior? The point of Flowable is to constrain emission. If I wanted unconstrained I'd use Observable.

Why should the source not be restricted to the maximum that take will accept for real world use cases and not just fabricated micro benchmarks?

[akarnokd]

§3.17: A Subscription MUST support an unbounded number of calls to request and MUST support a demand (sum requested - sum delivered) up to 2^63-1 (java.lang.Long.MAX_VALUE). A demand equal or greater than 2^63-1 (java.lang.Long.MAX_VALUE) MAY be considered by the Publisher as “effectively unbounded”[3].

It is completely legal for an operator to go unbounded if it ensures the rest of the rules towards its downstream consumer. Running unbounded means less overhead from operators which means more time to process the actual items, especially those just came over from an async boundary (observeOn).

If I wanted unconstrained I'd use Observable

You are still thinking about requests as an end-to-end way of communicating with some server and sending a message to it requesting the next item based on consumption at the very end. Reactive-Streams backpressure is different. It is a negotiation between subsequent operators. Some may not interfere it, some might change the timing or amount, some might go unbounded. It is about honoring the downstream's amount and working out an amount for the upstream.

For example:

source.observeOn(Schedulers.computation()).take(5).subscribe(...)

Will the source consumed in an unbounded manner just because take(5) goes unbounded?

No.

take(5) and observeOn negotiate that observeOn is free to emit as fast as it can and as many items as available because take can handle it.

source and observeOn negotiate that due to the async boundary, batching is desired thus observeOn requests the default amount (128) and keeps requesting once take consumed what's already became available over the boundary. This requires source to honor that initial 128 and any subsequent requests (96).

[benjchristensen]

It is completely legal for an operator to go unbounded if it ensures the rest of the rules towards its downstream consumer.

Yes I know that. I was involved in arguing for inclusion of that rule in the Reactive Streams spec and know the benefits.

You are still thinking about requests as an end-to-end way of communicating with some server and sending a message to it requesting the next item based on consumption at the very end. Reactive-Streams backpressure is different.

No I'm not. Remember I was part of the founding group for Reactive Streams, so I understand the semantics. Early in the discussions it was assumed every operator hop was async, but we argued that hops could be synchronous and therefore would benefit from optimizations in the spec, such as rule 3.17 and 2.2.

Some may not interfere it, some might change the timing or amount, some might go unbounded. It is about honoring the downstream's amount and working out an amount for the upstream.

I understand that, and am not questioning that. I'm questioning a simple implementation choice, not a contractual or semantic problem. Everything that the take operator is doing is perfectly legal. I am purely questioning the decision to not alter the upstream request(n) as it seems a legitimate choice to reduce to the limit.

But I won't bother attempting discussion any further. It's not important enough to me.

[davidmoten]

thanks for raising this @benjchristensen. I wasn't aware of it.

So I'm assuming then that if I've got some synchronous network requesting source (that can't effectively emit partially e.g. a web service request that returns a list of items) and I don't want that source to transfer an unnecessarily large list in reaction to a take(2) then I suppose I would have to do this:

source.rebatchRequests(2).take(2);

This is certainly an uncommon use case for me but I did like that in RxJava 1.x upstream requests were constrained generally to what was required by downstream with the exception of a few well-known operators like flatMap and observeOn. Perhaps now I can't make any assumptions and when I want to constrain requests I need to use rebatchRequests in lots of places. We'll see how that goes I suppose.

[davidmoten]

@akarnokd or @benjchristensen can you reopen this issue? I think it needs more discussion.

[benjchristensen]

The use case is fairly common for me. Consider this:

Library:

public Flowable<Data> getDataFromNetwork(Input args) {
   return SomeFlowable(...)
               // ensure over network we do no more than request(128)
               .rebatchRequests(128);
}

Then a consumer that I don't control:

getDataFromNetwork(...).take(10).subscribe(...)

In that case I would prefer to only request 10 items over the network, but in the getDataFromNetwork method I don't know what my consumers will request. And the consumer calling it doesn't and shouldn't know the implementation details. So, the way take is currently implemented, I will have to just request(128), start emitting over the network, over produce and asyhnchronously race a cancel back over the network. Most likely I will end up producing all 128, sending many or most of them over the network, and then dropping 118 of them on the floor.

[davidmoten]

Yep, I'm not a fan of that behaviour and I'm sure I'll encounter it tying together service calls as our organisation moves more logic behind services. There's a broader consideration here I suppose which is how much should users of the RxJava api know or expect of the request behaviour of operators. We don't really have a guiding principle in place that helps us here. For me I'd like to see operators only request what they need and where for performance considerations it makes sense to request in batches (across async boundaries for instance) then it should be configurable. flatMap and observeOn are configurable in this regard.

I've had a quick look through the list of operators and I've identified the operators that I'd prefer only requested what they need. To have constrained requesting behaviour on these operators as well as the ability to configure flatMap, observeOn and call rebatchRequests is enough control over request patterns for my needs.

Note that all these below request Long.MAX_VALUE:

elementAt
firstXXX
take

The operators above are the only operators I saw that have a definite upper bound that is less than Long.MAX_VALUE on the number of items required to complete.

[akarnokd]

Anything that has the @BackpressureSupport(UNBOUNDED_IN), including subscribe(Consumer). It is the duty of the getDataFromNetwork(...) to turn any downstream request pattern into a reasonable amount for its level, for example, by rebatchRequests internally.