BufferTimeout with fair backpressure rework

[chemicL]

The fair backpressure variant of the bufferTimeout operator has been reworked to use a state machine with a minimum number of volatile variables eliminating potential data races, such as skipping the delivery when onNext and timeout happen concurrently or cancellation happens while onNext is delivered, etc.

Resolves #3531

[OlegDokuka]

The issue sits deeper and as I suggested first requires complete rework.

To reproduce the issue try to modify discard setup as follows:

@Tag("slow")
public class OnDiscardShouldNotLeakTest {

	private static final int NB_ITERATIONS = 100_000;
	// add DiscardScenarios here to test more operators
	private static final DiscardScenario[] SCENARIOS = new DiscardScenario[] {
			DiscardScenario.fluxSource("bufferTimeout", f -> f.bufferTimeout(2, Duration.ofNanos(1), true).flatMapIterable(Function.identity())),
	};
       ...

the problem appears when multiple threads comes into play. If you have race between scheduled timeout task trying to flash the window and cancellation ,then some elements might be undischarged

[OlegDokuka]

Although the above could be false negative, since the async task can be unawaited properly. Can you please doublecheck @chemicL

[OlegDokuka]

I just remembered that it was one of the reasons why all the discard tests are in stress tests, therefore it could make sense to port part of them for bufferTimeout

[chemicL]

@OlegDokuka the commit 0334959 adds significant improvements to the test suite that helps catch the racy situations. In the next commit I will add temporary fixes for the identified issues, but afterwards will follow up with a state machine implementation to eliminate the last one with timeout racing with draining.

[chemicL]

For the latest changes, I added a JMH benchmark. The idea is to simply test a non-contended, single threaded case of pulling 100 items and packaging them in 1, 10, and 100-item buffers.

Devising a contended usage scenario with a concurrent, asymmetric JMH benchmark doesn't seem worthwhile as the dominating factor should never be two actors competing but rather one actor winning a potential race and performing the bulk of the work. We can't approach this performance evaluation the same way as a regular queue benchmark with a randezvous scenario of handing over between two threads. In the Queue benchmark case, this operation is the essence of the work being done. In contrast, with reactive-streams it is not the case as there is the idea of work stealing where one actor winning a race (which can contend in a lock-free manner for a short moment) is performing a drain operation and the Queue is a means to an end and we are not evaluating its behaviour under load, but the operator's algorithm itself.

Having said the above, I compared the new backpressured variant with the current (broken) implementation and also with the simpler, yet not-backpressured variant. Here are my results on an M1 MacBook Pro (10-core, 32GB):

This PR, fairBackpressure = true

Benchmark                             (bufferSize)  Mode  Cnt     Score     Error  Units
FluxBufferTimeoutBenchmark.oneByOne              1  avgt    5  6259.514 ± 214.272  ns/op
FluxBufferTimeoutBenchmark.oneByOne             10  avgt    5  4418.668 ± 595.467  ns/op
FluxBufferTimeoutBenchmark.oneByOne            100  avgt    5  3938.607 ± 518.551  ns/op
FluxBufferTimeoutBenchmark.unlimited             1  avgt    5  5764.457 ± 399.358  ns/op
FluxBufferTimeoutBenchmark.unlimited            10  avgt    5  4072.673 ± 169.380  ns/op
FluxBufferTimeoutBenchmark.unlimited           100  avgt    5  3797.930 ±  94.629  ns/op

Current 3.5.x, fairBackpressure = true

Benchmark                             (bufferSize)  Mode  Cnt      Score      Error  Units
FluxBufferTimeoutBenchmark.oneByOne              1  avgt    5  23836.157 ± 2138.635  ns/op
FluxBufferTimeoutBenchmark.oneByOne             10  avgt    5   8935.440 ±   43.610  ns/op
FluxBufferTimeoutBenchmark.oneByOne            100  avgt    5   7182.099 ±  129.899  ns/op
FluxBufferTimeoutBenchmark.unlimited             1  avgt    5  18836.570 ± 1281.322  ns/op
FluxBufferTimeoutBenchmark.unlimited            10  avgt    5   8231.299 ±   55.848  ns/op
FluxBufferTimeoutBenchmark.unlimited           100  avgt    5   7197.949 ±  160.759  ns/op

Current 3.5.x, fairBackpressure = false

Benchmark                             (bufferSize)  Mode  Cnt      Score      Error  Units
FluxBufferTimeoutBenchmark.oneByOne              1  avgt    5  18376.986 ±  795.211  ns/op
FluxBufferTimeoutBenchmark.oneByOne             10  avgt    5   3458.388 ±  200.223  ns/op
FluxBufferTimeoutBenchmark.oneByOne            100  avgt    5   2533.851 ±  440.186  ns/op
FluxBufferTimeoutBenchmark.unlimited             1  avgt    5  15716.505 ± 1168.297  ns/op
FluxBufferTimeoutBenchmark.unlimited            10  avgt    5   3040.167 ±  144.305  ns/op
FluxBufferTimeoutBenchmark.unlimited           100  avgt    5   2470.696 ±  375.516  ns/op

To summarize the above, the new implementation passes the JCStress tests devised to catch the discovered issues while maintaining a better performance characteristic. The non-prefetching and incapable of respecting backpressure variant is more performant and that's understood as it doesn't need to keep track of much accounting and is a simpler implementation.

[chemicL]

With this change it's much easier to compare the state changes as they are stacked on top of each other so the differences are immediately visible as opposed to a horizontal presentation.

[chemicL]

This is a behaviour change, yet it is not dramatic or disruptive. Before, a request would force requesting whatever is remaining from the upstream to fulfill the prefetch capacity. Now it will do that only once the replenishMark is reached.