Reactor-core: Improving Flux.limitRate performance

[OlegDokuka]

Motivation

The current implementation of the limitRate operator uses publishOn with Scheduler.immediate which means that every element in the stream will be enqueued/drained from the given Queue. Obviously, it is an overhead which significantly impacts the performance. As we all know the main idea of the limitRate operator is to limit request size up to highTild (upper possible level) and then keep the request rate stable during the lifetime of the stream. The most common case when the limitRate operator can be applied is when the downstream requests Long.MAX_VALUE. It all leads to the fact that having the queue in the middle is unreasonably expansive.

Desired solution

The desired outcome is having the "native" limitRate impl which does not involve queue and eager prefetching in order to achieve kind of prefetch-replenish behavior.

Considered alternatives

During the exercising proper internal flow control (transport-publisher interaction) in RSocket I ended up with light-weight impl of rateLimitaer which is up to 5 times faster than publishOn based one.

You can find the ported implementation of the operator in the following gist:
https://gist.github.com/OlegDokuka/ad069d21ff972d8f50add46370f2f7b4

[OlegDokuka]

One may argue that limitRate keeping the same size of the request and send it with the same rate and similar behaviors could be achieved with limitRequest operator. I would disagree since limitRequest just split the request into equal batches and send them in the loop. It means that in case of async upstream, e.g : Flux.range(...)....suscribeOn(...).limitRequest() all of those requests will be scheduled as separate tasks which leads to the fact that producing rate will have gaps while the next task is delivering. In contrast, the mentioned implementation is aimed to keep rate stable and preserve exactly the same behaviors as publishOn(Schedulers.immediate()) except prefetch and cases when downstream requests lower number of elements than highTild (in such cases the request will be directly propagated)

[OlegDokuka]

Since we are moving to 3.3.x, which can include braking changes in terms of operators, I would appreciate migration to the mentioned solution. Moreover, one can always achieve previous behavior by using publishOn(Scheduler.immediate) explicitly as well as projects like RSocket can benefit the usage of overhead less limitRate operators without reimplementing it internaly

[simonbasle]

Since we are moving to 3.3.x, which can include braking changes in terms of operators, I would appreciate migration to the mentioned solution.

Can you expand on what the breaking change / behavior change is? Does the linked solution only work with requests > highTide?

[OlegDokuka]

Since we are moving to 3.3.x, which can include braking changes in terms of operators, I would appreciate migration to the mentioned solution.

Can you expand on what the breaking change / behavior change is? Does the linked solution only work with requests > highTide?

1) The first braking change is related to requests > highTide behavior only, lower requests are bypassing directly
2) Absence of the eager prefetch request, so those who relied on that previously might be suffered

[simonbasle]

@OlegDokuka note limitRequest puts a hard cap on the request amount. With eg. limitRequest(5), the following request pattern:

request(1)
request(3)
request(3)
request(3)
request(3)

is translated to:

request(1) //4 remaining
request(3) //1 remaining
request(1) //0 remaining
//rest is ignored, actually upon receiving last onNext the operator cancels upstream and completes

[OlegDokuka]

Yeah, I see

[simonbasle]

@OlegDokuka the case where a downstream makes a lot of small requests and the limitRate operator is introduced with a higher highTide and lowTide in order to rebatch into larger requests (ie. prefetch) is also legit. This case _needs_ a queue, since the downstream might be under-requesting compared to what upstream pushes.

I'm not sure which case is the most common: putting a floor on requests (ie. prefetching to have a regular "shape") or putting a ceiling on requests (the one you're optimizing for).

Maybe this optimization can be introduced as an opt-in overload like limitRequest(int highTide, int lowTide, boolean passthroughLowerRequests)?

[OlegDokuka]

Sounds good, guessing it can workout. I will going to send a PR if you do not mind

[simonbasle]

bump @OlegDokuka: do you think you will have time to submit a PR?

[maxim-bandurko-lsvt]

@OlegDokuka This improvement is very good, as it is natural backpressure control. But, it comes to moment, that there are no ways to track when requested batch is completed. So, it is just basically re-process eager request to multiple sub requests that are not controlled.

This improvement can help to implement some kind reactive "pageable" for spring data reactive, but to make it happen, would be nice to have at least some operation that can modify limited flux. Like, use case:

Flux.just(1,2,3,4,5,6,7,8,9).collectList().flatMapMany(
(batch) -> {
return Flux.just(batch);
}
).limitRate(2, 2, true);

Or may be some other logic. Result should be ability to track when requested batch is finished and do some action/modification.

As an idea, could be passing one more argument, like BiFunction, that is doing something with finished batch:

.limitRate(2, 2, true, 
(T batch, Long processed) -> {
......
return batch;
} 
);

Or even better, as that is what I am using at current moment as my custom solution for implementation of kind "reactive pageable" in spring data r2dbc:

Flux.fromPageable(
(T batch, LimitRateSubscriber subs) - {
return repository.getByKey("someKey", subs.produced, subs.limitRate)
}
2, //Long limitRate
4, //Long offset
)

This allows to build data queries with offset and fetch values.

[maxim-bandurko-lsvt]

@OlegDokuka Never mind on my previous comment. Just realized that a lot of databases already have Pageable implemented from Flux. It is just a matter of time when it be implemented to r2dbc.

Your improvement is exactly that is needed!