StackOverflowError is swallowed

[zsxwing]

The following code should have thrown StackOverflowError:

        final PublishSubject<Integer> a = PublishSubject.create();
        final PublishSubject<Integer> b = PublishSubject.create();
        a.subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(Integer args) {
                System.out.println(args);
            }
        });
        b.subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(Integer args) {
                System.out.println(args);
            }
        });
        a.subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(Integer args) {
                b.onNext(args + 1);
            }
        });
        b.subscribe(new Observer<Integer>() {

            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }

            @Override
            public void onNext(Integer args) {
                a.onNext(args + 1);
            }
        });
        a.onNext(1);

However, StackOverflowError is swallowed.
The problem is the following line in SafeObserver: https://github.com/Netflix/RxJava/blob/0b1b6e7a91d89ad63263b80747f0bd4683fe4ba2/rxjava-core/src/main/java/rx/operators/SafeObserver.java#L117

            RxJavaPlugins.getInstance().getErrorHandler().handleError(e);

When StackOverflowError is thrown, there is few stack space. However, this line will generate too big stack frame and cause the thread crash. If I comment out this line, I can observe StackOverflowError.

Reported by Samuel at https://groups.google.com/forum/#!topic/rxjava/eraZ-32w1gQ

[duarten]

The proper fix seems to be to immediately re-throw any VirtualMachineError by introducing a more specialized catch.

[benjchristensen]

Not surprised we ended up with something like this happening after migrating from Exception to Throwable: #296

I agree @duarten that SafeObserver should catch and re-throw at least the VirtualMachineError class of errors: http://docs.oracle.com/javase/7/docs/api/java/lang/VirtualMachineError.html

Any others we should include? ThreadDeath looks fun :-) The LinkageError exceptions also seem like we should never catch them: http://docs.oracle.com/javase/7/docs/api/java/lang/LinkageError.html

[duarten]

Maybe it's better not to catch any of the Error subclasses?

[zsxwing]

Re-throwing the Error still has a problem when Schedulers.io() is involved. The Error will be swallowed by the ExecutorService.

[benjchristensen]

In general not catching Error and Throwable is worse than catching them, as in an async system that means they are being thrown on some random thread in the background, won't call onError and thus can lock up a system since an Observable won't be terminated.

So it may only be the StackOverflow case that needs to be special-cased since it is a unique case of an infinite loop that means we won't ever successfully exit.

[duarten]

But swallowing them is not good either, and then that will put the onus on the users to re-throw those exceptions, because there's really no way to handle them. If there are uncaught Errors won't the JVM shutdown even if there are non-daemon threads?

Interestingly, Scala defines a NonFatal exception as:

object NonFatal {
   def apply(t: Throwable): Boolean = t match {
     case _: StackOverflowError => true // StackOverflowError ok even though it is a VirtualMachineError
     case _: VirtualMachineError | _: ThreadDeath | _: InterruptedException | _: LinkageError | _: ControlThrowable | _: NotImplementedError => false
     case _ => true
   }
}

which they use in their Futures library to check if it should be handed to the user for handling.

[benjchristensen]

It won't be swallowing them, they are or should all be passed via onError to the user provided Observer. StackOverflow is 'swallowed' because of a recursive loop.

[duarten]

Sorry, my point is that users may not be expecting to be handed such exception, and may swallow them themselves, by for example just logging them. The correct behaviour, IMHO, would be to let them bring down the JVM asap.

[benjchristensen]

Background on Throwable being caught and emitted via onError can be seen here: #296

The Scala definition seems good. I think we should follow that same logic for defining something Fatal and immediately re-throwing it as opposed to passing via onError.

[duarten]

That sounds good!

[zsxwing]

Just found a potential problem in this method:

    protected void _onError(Throwable e) {
        try {
            RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
            actual.onError(e);
        } catch (Throwable e2) {
            e2.printStackTrace();
            if (e2 instanceof OnErrorNotImplementedException) {
                /*
                 * onError isn't implemented so throw
                 * 
                 * https://github.com/Netflix/RxJava/issues/198
                 * 
                 * Rx Design Guidelines 5.2
                 * 
                 * "when calling the Subscribe method that only has an onNext argument, the OnError behavior will be
                 * to rethrow the exception on the thread that the message comes out from the observable sequence.
                 * The OnCompleted behavior in this case is to do nothing."
                 */
                try {
                    subscription.unsubscribe();
                } catch (Throwable unsubscribeException) {
                    RxJavaPlugins.getInstance().getErrorHandler().handleError(unsubscribeException);
                    throw new RuntimeException("Observer.onError not implemented and error while unsubscribing.", new CompositeException(Arrays.asList(e, unsubscribeException)));
                }
                throw (OnErrorNotImplementedException) e2;
            } else {
                /*
                 * throw since the Rx contract is broken if onError failed
                 * 
                 * https://github.com/Netflix/RxJava/issues/198
                 */
                RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);
                try {
                    subscription.unsubscribe();
                } catch (Throwable unsubscribeException) {
                    RxJavaPlugins.getInstance().getErrorHandler().handleError(unsubscribeException);
                    throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError and during unsubscription.", new CompositeException(Arrays.asList(e, e2, unsubscribeException)));
                }

                throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
            }
        }
        // if we did not throw about we will unsubscribe here, if onError failed then unsubscribe happens in the catch
        try {
            subscription.unsubscribe();
        } catch (RuntimeException unsubscribeException) {
            RxJavaPlugins.getInstance().getErrorHandler().handleError(unsubscribeException);
            throw unsubscribeException;
        }
    }

If RxJavaPlugins.getInstance().getErrorHandler().handleError(e); throws a Throwable, RxJavaPlugins.getInstance().getErrorHandler().handleError(e2); also may throw one. So there may be a recursive loop. Is it necessary to make a distinction between Throwables from RxJavaPlugins.getInstance().getErrorHandler().handleError(e); and actual.onError(e);.

Sorry, I made a mistake. It's not a recursive loop.

[zsxwing]

A thought to solve the swallowed StackOverflowError: can we call actual.onError(e); before RxJavaPlugins.getInstance().getErrorHandler().handleError(e);?

[zsxwing]

Just dig deeply and found StackOverflowError is swallowed by SafeObserver.
In this example, there are 4 observers and wrapped by 4 SafeObservers. Here I call them so1, so2, so3, so4 in order. When the StackOverflowError is thrown, the thread call stack is something like:

so1.onNext
so4.onNext
so3.onNext
so4.onNext
so3.onNext
so4.onNext
...

Next,

so1.onError(throw a new StackOverflowError from "RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);")
so1.onNext
so4.onNext
so3.onNext
so4.onNext
so3.onNext
so4.onNext
...

Next,

so4.onNext
so3.onNext
so4.onNext
so3.onNext
so4.onNext
...

Next,

so4.onError(throw a new StackOverflowError from "RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);")
so4.onNext
so3.onNext
so4.onNext
so3.onNext
so4.onNext
...

Next,

so3.onNext
so4(isFinished = true).onNext
so3.onNext
so4(isFinished = true).onNext
...

Next,

so3.onError(throw a new StackOverflowError from "RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);")
so3.onNext
so4(isFinished = true).onNext
so3(isFinished = true).onNext
so4(isFinished = true).onNext
...

Next,

so4(isFinished = true).onNext
so3(isFinished = true).onNext
so4(isFinished = true).onNext
...

Next,

so4(isFinished = true).onError (StackOverflowError is swallowed here)
so4(isFinished = true).onNext
so3(isFinished = true).onNext
so4(isFinished = true).onNext
...

Here StackOverflowError is sent to so4.onError again, so4 will swallow it since isFinished is true.

[zsxwing]

So if RxJavaPlugins.getInstance().getErrorHandler().handleError(e); always throw a StackOverflowError, actual.onError(e); will be skipped. No matter what we do in the catch clause(e.g, rethrow the StackOverflowError), it will be swallowed by SafeObserver in this example.

[benjchristensen]

Dealing with in pull request #839

[benjchristensen]

Should be fixed in #839