combineLatest back-pressure bug

[flatmap13]

The following program stops printing numbers around 511 where the expected behaviour is that it runs forever.

import java.util.concurrent.TimeUnit

import rx.lang.scala.Observable
import rx.lang.scala.schedulers.NewThreadScheduler

import scala.concurrent.duration.Duration

object CombineLatestTest {
  def main(args: Array[String]) {

    var left = Observable.timer(Duration(0, TimeUnit.MILLISECONDS), Duration(1, TimeUnit.MILLISECONDS))

    left.observeOn(NewThreadScheduler()).doOnEach(println(_)).combineLatest(Observable.never).subscribe()

    readLine()
  }
}

Without the observeOn it works as expected.

[benjchristensen]

Thanks for the report. I'll take a look. It may actually be related to combineLatest. ObserveOn may just be the thing that exposes it.

[headinthebox]

We ran into this in an Rx debugger demo where the UI suddenly froze, so I guess the Android folks may run into it as well at some point; the code above was the smallest retro we could find.

[benjchristensen]

Replicated with this unit test:

    @Test
    public void testWithCombineLatestIssue1717() {
        Observable<Long> timer = Observable.timer(0, 1, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.newThread())
                .doOnEach(new Action1<Notification<? super Long>>() {

                    @Override
                    public void call(Notification<? super Long> n) {
                        System.out.println(n);
                    }

                });

        TestSubscriber<Long> ts = new TestSubscriber<Long>();

        Observable.combineLatest(timer, Observable.<Void> never(), new Func2<Long, Void, Long>() {

            @Override
            public Long call(Long t1, Void t2) {
                System.out.println("t1: " + t1);
                return t1;
            }

        }).take(RxRingBuffer.SIZE*2).subscribe(ts);

        ts.awaitTerminalEvent();
        ts.assertNoErrors();
    }

It looks like combineLatest is not requesting more upwards.

[benjchristensen]

Actually ... this test should never pass.

Since the test is combining with never that will never emit and thus combineLatest will never emit anything.

It queues up 512 items from the timer, and then correctly waits for never to emit something, but it never does.

Modifying the test so it emits at least one value before hanging then allows it to push data through:

    @Test
    public void testWithCombineLatestIssue1717() {
        Observable<Long> timer = Observable.timer(0, 1, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.newThread())
                .doOnEach(new Action1<Notification<? super Long>>() {

                    @Override
                    public void call(Notification<? super Long> n) {
                        System.out.println(n);
                    }

                });

        TestSubscriber<Long> ts = new TestSubscriber<Long>();

        Observable.combineLatest(timer, Observable.just(1).concatWith(Observable.<Integer>never()), new Func2<Long, Integer, Long>() {

            @Override
            public Long call(Long t1, Integer t2) {
                return t1;
            }

        }).take(RxRingBuffer.SIZE*2).subscribe(ts);

        ts.awaitTerminalEvent();
        ts.assertNoErrors();
    }

Note the Observable.just(1).concatWith(Observable.<Integer>never()) code which will emit 1 and then never onComplete.

@GeorgiKhomeriki @headinthebox Do you expect combineLatest to emit even when one of the Observables never emits anything as never() does?

[benjchristensen]

If combineLatest is supposed to emit, even when a value is not yet emitted, is it supposed to be null?

[benjchristensen]

In 0.19.x before backpressure, this code just runs forever emitting values:

    @Test
    public void testWithCombineLatestIssue1717() {
        Observable<Long> timer = Observable.timer(0, 1, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.newThread())
                .doOnEach(new Action1<Notification<? super Long>>() {

                    @Override
                    public void call(Notification<? super Long> n) {
                        System.out.println(n);
                    }

                });

        TestSubscriber<Long> ts = new TestSubscriber<Long>();

        Observable.combineLatest(timer, Observable.<Void> never(), new Func2<Long, Void, Long>() {

            @Override
            public Long call(Long t1, Void t2) {
                System.out.println("t1: " + t1);
                return t1;
            }

        }).take(2000).subscribe(ts);

        ts.awaitTerminalEvent();
        ts.assertNoErrors();
    }

This code never prints anything in 0.19 as well (before backpressure):

    @Test
    public void testWithCombineLatestIssue1717() {
        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
        Observable.combineLatest(Observable.just(1), Observable.<Void> never(), new Func2<Integer, Void, Integer>() {

            @Override
            public Integer call(Integer t1, Void t2) {
                System.out.println("t1: " + t1 + " t2: " + t2);
                return t1;
            }

        }).subscribe(ts);
        ts.awaitTerminalEvent();
    }

Should it emit anything when Observable.never() won't emit?

[benjchristensen]

The introtorx.com site states:

Once both sequences have produced at least one value, the latest output from each sequence is passed to the resultSelector function every time either sequence produces a value.

http://www.introtorx.com/Content/v1.0.10621.0/12_CombiningSequences.html#CombiningMultipleSequences

[flatmap13]

@benjchristensen There is a subtle difference here, because we have a side-effect in the doOnEach operator, before the combineLatest. I think this side-effect should execute every time the left observable (timer) emits a value. After that, the combineLatest can block the event from propagating further down the line since it never receives a value from the other Observable. So to answer your question, combineLatest should indeed not emit any values in this case, but the doOnEach should keep on printing forever.

[headinthebox]

The example code is pretty nasty.

I am still puzzled what is exactly going on but as @GeorgiKhomeriki says, the trick is that there is a side-effect in the first input to a combineLatest that itself never produces a value (in the real use case, the other input is a stream that only receives very few values, and in our case few enough that the whole program froze).

Before back pressure, and without any observeOn, feeding stream xs into an combineLatest where the other input never produces was not an issue, combineLatest keeps the last value of xs and the side effects are happily executing as new values arrive.

[benjchristensen]

I found what's happening, it's odd. combineLatest behaves differently before it gets values from all Observables than it does after.

Before a value has been received from each Observable it does NOT obey backpressure and just acts like a BehaviorSubject dropping data and retaining only the last.

After a value has been received from each Observable however it then expects to never drop data and emit all combinations. This means backpressure is needed to achieve this otherwise unbounded buffers are needed, as in 0.19 and earlier (https://github.com/ReactiveX/RxJava/blob/0.19.x/rxjava-core/src/main/java/rx/internal/operators/OnSubscribeCombineLatest.java#L131).

The reason why observeOn affects it the way it does is that observeOn expects to receive a request for more data and combineLatest never does that. It is a bug in combineLatest that it doesn't request more after dropping data. That's bug 1.

Bug 2 is that the timer/observeOn relationship should result in a MissingBackpressureException being thrown because timer doesn't obey backpressure (by design). If observeOn exerts backpressure (as it does) and timer doesn't obey (as it won't) then an exception should be thrown.

[headinthebox]

Juicy!