scan infinite loop

[davidmoten]

The following test fails in 1.x (and 1.0.15):

    @Test(timeout=1000)
    public void testScanDoesNotHang() {
        Observable.range(0, Integer.MAX_VALUE)
                //
                .scan(1, new Func2<Integer, Integer, Integer>() {

                    @Override
                    public Integer call(Integer t1, Integer t2) {
                        return t1;
                    }
                })
                //
                .subscribe(new Subscriber<Integer>() {

                    int count = 0;

                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(Integer t) {
                        count++;
                        if (count == 2)
                            unsubscribe();
                    }
                });
    }

@akarnokd I've stepped through in a debugger and in OperatorScan L284 initiates a fast path emission in OnSubscriberRange which calls OperatorScan.emit() L289 but because the method containing L284 has set emitting to true the emission goes no further and an a loop occurs that fills the heap.

Though the test might look a bit contrived I just isolated the behaviour to scan for the test. I encountered the bug when this hung:

observable.scan(..).elementAt(n); //where n > 1

I'd rank this bug as nasty (not an improbable edge case like some concurrency bugs) and possibly deserving of a new release quickly.

[akarnokd]

You are right. Serializing the requesting/producer arrival with the normal event emission is unnecessary. I'll merge #3485 now but start working on the fix in my morning. I'll review other operators that might exhibit the same problem.

[davidmoten]

Thanks @akarnokd

[akarnokd]

See #3491 for the fix.