onBackpressureBlock() causes hangs

[wendigo]

Given this piece of code:

        Observable.create(subscriber -> {
            long counter = 0;

            while (counter++ < Long.MAX_VALUE) {
                System.out.println("onNext(" + counter + ")");

                if (!subscriber.isUnsubscribed()) {
                    subscriber.onNext(counter);
                } else {
                    break;
                }

                System.out.println("Subscribe.onNext(" + value + ")");
            }
        })
        .onBackpressureBlock(2)
        .subscribeOn(Schedulers.newThread())
        .observeOn(Schedulers.computation())
        .subscribe(System.out::println);

On version 1.0.7 above code will hang on producing element on position RxRingBuffer.SIZE + 2 (maxQueueLength):

...
afterOnNext(127)
onNext(128)
afterOnNext(128)
Subscribe.onNext(128)
onNext(129)
afterOnNext(129)
onNext(130)
afterOnNext(130)
onNext(131)

If we switch order of subscribe/observeOn (observeOn before subscribeOn) - it will work as expected :)

Also it will throw MissingBackpressureException without onBackpressure* (as expected).

[benjchristensen]

One of us will take a look. Did this just start happening in 1.0.7?

[wendigo]

I have came across this behaviour when .onBackpressureBlock() was introduced but I was not able to reproduce it nor fully understand under which circumstances it can be observed. Now it seems that this minimal use case demonstrates this bug.

[wendigo]

Note: lock happens on ArrayBlockingQueue in OperatorOnBackpressureBlock:58:

queue.put(nl.next(t));

queue contains maxQueueSize last produced elements just before it hangs.

[wendigo]

Next trail: BackpressureDrainManager was requested for n = RxRingBuffer.size elements. When manager.drain() is called n+1 time requestedCount is equal to 0 in this and subsequent calls. No more request() calls were made.

[zsxwing]

The problem is .subscribeOn(Schedulers.newThread()). Because the new thread is blocked, the new request (https://github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/internal/operators/OperatorSubscribeOn.java#L90) won't be scheduled to run.

[wendigo]

If I use any other scheduler I will get exactly same behaviour. Also I don't understand why order of .observeOn/.subscribeOn solves this problem.

[zsxwing]

See #2653

You need to move the subscribeOn before the onBackpressureBlock.

[benjchristensen]

onBackpressureBlock still feels so wrong. Perhaps it is too easy to break an app with it and is a failed experiment?

@zsxwing @akarnokd Is there something we can do to make this less likely to break an app, or is the addition of onBackpressureBlock fundamentally a bad idea?

[akarnokd]

Blocking and backpressure-continuations don't play well together: the same issue happened in #2466 with a different kind of blocking implementation. The solution is to move the subscribeOn before the blocking part (which can't be done if the very first source is the blocking one). We can put an emphasis on this fact in the javadoc of onBackpressureBuffer but if one isn't in control of the downstream, such downstream can introduce subscribeOn and the trouble is back. I'd say lets drop the operator for now.

[zsxwing]

+1 removing it, since it's too easy to hang an app.

[akarnokd]

Added PR #2970 which deprecates this operator, plus a javadoc update warning about the issue.