Reactor-core: Race condition in .bufferUntil causing "Could not emit buffer due to lack of requests"

[benwiles1]

Expected Behavior

.bufferUntil should not be able to cause "Could not emit buffer due to lack of requests" since it is deterministic, and not based on the rate of the subscriber like in .bufferTimeout.

Actual Behavior

.bufferUntil threw "Could not emit buffer due to lack of requests"

Steps to Reproduce

package test;

import java.time.Duration;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;

public class Test {
    public static void main(String[] args) {
        Hooks.onOperatorDebug();
        Flux.range(0, 1000)
                .bufferUntil(v -> true)
                .flatMap(v -> Mono.delay(Duration.ofMillis(10)))
                .blockLast();
    }
}

Possible Solution

Your Environment

• Reactor version(s) used: 3.3.0.RELEASE
• Other relevant libraries versions (eg. netty, ...): none
• JVM version (javar -version): JDK 11
• OS and version (eg uname -a): Windows 10

[bsideup]

Adding "onOperatorDebug" because, without the hook applied, it does not seem to fail

[bsideup]

Short update: managed to reproduce without Hooks.onOperatorDebug()

[UgiR]

For every buffer emission, REQUESTED is decremented.

When a request comes in for n buffers, multiple emissions can occur before REQUESTED is properly updated to reflect the request.

So while the loop in DrainUtils.postCompleteRequest is constantly failing to get a grasp on REQUESTED and update it with n, emit() is emitting buffers and decrementing REQUESTED until it hits -1.

Putting the call to postCompleteRequest() in request() within a synchronized block seems to fix this but I'm not sure if that is a good approach to fixing this.

[UgiR]

Hmm, maybe there's more to it. Here's a reproducer that fails reliably even when the update to REQUESTED is in a synchronized block. Note that changing flatMap's concurrency to 1 will allow the test to pass.

@Test
public void requestedCountContentionRaceCondition() {
    Mono<Void> source = Flux.range(0, 1000)
            .bufferUntil(i -> true)
            .limitRate(3)
            .flatMap(i -> Mono.delay(Duration.ofMillis(1)), 16)
            .then();

    StepVerifier.create(source)
            .verifyComplete();
}

[robotmrv]

I was able to reproduce similar issue with totally synchronous code

Flux.just(0, 1, 2)
        .bufferUntil(s -> true)
        .publishOn(Schedulers.immediate(), false, 1)
        .subscribe(new BaseSubscriber<Object>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(1);
            }
        });

It reproduces if there is operator with prefetch downstream the bufferUntil()

As far as i understand problem appears when reactor.core.publisher.FluxBufferPredicate.BufferPredicateSubscriber#request is invoked indirectly from or in parallel with reactor.core.publisher.FluxBufferPredicate.BufferPredicateSubscriber#emit and subscriber has limited demand.

I have naive fix for this :)
in reactor.core.publisher.FluxBufferPredicate.BufferPredicateSubscriber

boolean emit(C b) {
    if (fastpath) {
        actual.onNext(b);
        return false;
    }
    long r = REQUESTED.getAndDecrement(this);
    if(r > 0){
        boolean requestMore = r - 1 > 0;// just move return requested > 0 check before onNext()
        actual.onNext(b);
        return requestMore;//return requested > 0;
    }
    cancel();
    actual.onError(Exceptions.failWithOverflow("Could not emit buffer due to lack of requests"));
    return false;
}

[simonbasle]

Thanks for investigating @UgiR and @robotmrv 👍