Buffer timeout with fair backpressure

[chemicL]

This change adds a variant of bufferTimeout that honors backpressure of a slow downstream Subscriber better than the current implementation (which just errors in the face of backpressure and timeouts).

The problem with the existing implementation is that it requests n * maxSize items from the upstream with the assumption that the items will fill a buffer which was requested. However, if timeouts occur, the amount of buffers created and delivered can exceed the accumulated demand. In such case, a backpressure error happens.

The new variant uses prefetching with 4 * maxSize items from the upstream. Internally it uses a queue, which is used to satisfy the downstream demand with no more buffers than have been requested. It uses a watermark to replenish the queue to be ready to deliver buffers in an efficient manner when the demand increases.

[almogtavor]

@chemicL Can you update us regarding what's left in this PR? This feature would really help me and I'd like to help if possible

[chemicL]

@almogtavor I'm waiting for @OlegDokuka to find some time to review the change. Also, I needed to move on to a different subject and couldn't spend more time thinking about JCStress tests. If you have ideas how to improve confidence in the concurrent execution of this operator, the ideas what scenarios to test would be useful. So far I ran the tough stress test suite against it on an x86 PC and an M1 ARM Mac against JDK 8, 11, and 17 and wasn't able to get it to an inconsistent state. Let me work with the team and try to get this feature in.

[almogtavor]

@OlegDokuka @chemicL can you tell me approximately when this will get reviewed? This feature will help a lot 😊

[patpatpat123]

Upvoting this, I am also blocked by this, this fix would help. Cheering for you guys

[patpatpat123]

Hello team, I heard there is a workaround using a (window operation + concatMap) instead of buffer timeout, which can still yield the exception as of 3.5.x.

Would it be possible to share the workaround please?

[OlegDokuka]

@patpatpat123

Hello team, I heard there is a workaround using a (window operation + concatMap) instead of buffer timeout, which can still yield the exception as of 3.5.x.

do you mean you see issues in windowTimeout, if so can you open an issue which reproduces the failure.

Thanks

[patpatpat123]

So basically, in order to perform some internal batch operation within a flux, for a number or a duration, (on purpose not using the words bufferTimeout or windowTimeout here) one has the following possibilities:

use bufferTimeout -> very high likelihood of encountering "Could not emit buffer due to lack of requests"
use windowTimeout + concatMap collectList -> very noticeable performance degradation

May I ask if my understanding is correct please?

[chemicL]

use bufferTimeout -> very high likelihood of encountering "Could not emit buffer due to lack of requests"

As long as the consumer is slower than the producer, yes.

use windowTimeout + concatMap collectList -> very noticeable performance degradation

The performance impact depends on the workload, but in general yes, windowTimeout has a lot more to do as for every window it goes through the entire reactive lifecycle to drain the results, while the buffer is a simple data structure that can be navigated in imperative manner.

For this operator, we aim to deliver it in 3.5.7 release.

[patpatpat123]

Very clear explanation @chemicL .
Many thanks

[chemicL]

Build failure attributed to a flaky test.