ScheduledAction Swallows Errors

[benjchristensen]

ScheduledAction (used by observeOn and other things doing scheduling) swallows errors because FutureTask.run() swallows errors inside a Future.

This means something like OnErrorNotImplemented on the Subscriber side of an observeOn will throw and be swallowed and everything will fail silently.

The following code fails silently:

Observable.error(new RuntimeException()).observeOn(Schedulers.computation()).subscribe();

[benjchristensen]

I've been researching this and there isn't a whole lot of choice in what to do when an Exception is thrown on a random Scheduler.Worker thread beyond catching and printing the error using System.err.

I can't throw the exception anywhere else as the whole issue here is an unhandled exception being thrown.

Thus, I'm going to catch the exceptions in ScheduledAction and e.printStackTrace() them with a warning about exceptions being thrown.

This is an edge case that an app should not allow ... and the reason why Observables propagate errors as events ... to avoid this very case, so seeing these types of messages in log output signals a "bad thing" that should be resolved.

This would NOT be good for system performance to allow these types of errors to be logged, but I feel this is far better to yell and scream about bad code rather than silently swallowing errors as it does now.

[benjchristensen]

An error such as this will be printed:

java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
    at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:46)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: rx.exceptions.OnErrorNotImplementedException
    at rx.Observable$36.onError(Observable.java:7387)
    at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:128)
    at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:97)
    at rx.internal.operators.NotificationLite.accept(NotificationLite.java:144)
    at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.pollQueue(OperatorObserveOn.java:177)
    at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.access$0(OperatorObserveOn.java:161)
    at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber$2.call(OperatorObserveOn.java:153)
    at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:43)
    ... 7 more
Caused by: java.lang.RuntimeException
    at rx.exceptions.ErrorsOverAsyncBoundaries.testSubscriberOnErrorFails(ErrorsOverAsyncBoundaries.java:43)

[benjchristensen]

I have merged the change. If anyone can suggest a better approach that would be great, please let me know here.

[headinthebox]

Best of the worst.

[dlew]

The added logging is nice, but why not continue throwing the error if it's fatal (like OnErrorNotImplemented)? Minus observeOn your program would crash because you aren't handling the exception - it wouldn't just log the exception.

[benjchristensen]

@dlew how would you solve this differently? The error is occurring on a separate thread, so the most violent death that can occur is killing that thread inside the ThreadPoolExecutor. The error can not propagate to the main thread and thus does not kill the JVM.

The Throwable gets caught inside java.util.concurrent.FutureTask:

    public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex); //  <------------- THIS IS WHERE THE EXCEPTION GOES
                }
                if (ran)
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

One can use ThreadPoolExecutor.afterExecute to extract the caught error, but rethrowing only causes the ThreadPoolExecutor to kill the thread, not the JVM since it can't propagate to the main thread (the nature of async): http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html#afterExecute(java.lang.Runnable,%20java.lang.Throwable)

The only alternative I can think of is to reverse schedule events up the Observable chain, like having an onError on the Observable.OnSubscribe and having a Scheduler on the producing side, not just the consuming side. That however is a crazy amount of overhead and infrastructure for a degenerate case.

If you can submit a pull request that offers a better solution I'd appreciate that.

[dlew]

Thanks, I understand the problem a lot better now. I'll have to think on it; not sure there is a good solution.

[abersnaze]

I know this isn't possible for even 1.0 but could Subscription schedule(Action0 action) be changed to
Observable<Void> schedule(Action0 action). That way the caller can get both the ability to subscribe to get the subscription and notification of completion or failure.

[benjchristensen]

A calling thread could not get a reference, as void onNext is the signature the caller invokes. It is on the wrong side of the thread where the exception is caught, scheduled from inside the onNext. The calling (parent) thread has already left by the time the child thread throws the exception.

There is no issue catching the exception on the child thread, it's what to do with it that is the question.

[mttkay]

There is a related discussion here: #969

I haven't checked in a while, but IIRC part of the issue was that some operators like ObserveOn weren't forwarding errors?

[benjchristensen]

ObserveOn propagates errors downstream. We just can't send them upstream across thread boundaries.

[loganj]

It would be great if we could add another plugin that gets called in place of the printStackTrace. Similar to RxJavaErrorHandler, but only for otherwise unhandled errors.

[benjchristensen]

Would you prefer that over this:

Thread.getDefaultUncaughtExceptionHandler().uncaughtException(Thread.currentThread(), e);

I've never that so have no idea how it behaves especially with thread pools.

[loganj]

No, you're right, we can just use the built-in handler mechanism directly.

We go directly Thread.getDefaultUncaughtExceptionHandler() today, but that's probably wrong. That handler is the global last-resort handler for all threads. It looks like what we really want is Thread.currentThread().getUncaughtExceptionHandler().

I'll get a PR together.

[benjchristensen]

That sounds good, since then we are free to add listeners at any of the 3 levels, but just registering the default global listener (most likely) will still work and it will propagate to it.

http://docs.oracle.com/javase/7/docs/api/java/lang/Thread.UncaughtExceptionHandler.html

When a thread is about to terminate due to an uncaught exception the Java Virtual Machine will query the thread for its UncaughtExceptionHandler using Thread.getUncaughtExceptionHandler() and will invoke the handler's uncaughtException method, passing the thread and the exception as arguments. If a thread has not had its UncaughtExceptionHandler explicitly set, then its ThreadGroup object acts as its UncaughtExceptionHandler. If the ThreadGroup object has no special requirements for dealing with the exception, it can forward the invocation to the default uncaught exception handler.

Thanks for bringing this to my attention @loganj I honestly wasn't even aware of this capability in the JVM and my use cases don't generally result in uncaught exceptions so I hadn't ever gone looking for it.

[loganj]

No problem, we're really good at making exceptions happen.

I'm basically done, but not sure if I should I apply the same handling to Schedulers that don't use ScheduledAction. ExecutorScheduler seems to need it. What about ImmediateScheduler and TrampolineScheduler? Do people rely on their current throwing behavior? Is that behavior contractual?

[benjchristensen]

I have not tried this issue on TrampolineScheduler but that may be okay as it's all on the same thread. ExecutorScheduler probably does need this, but let's get it solved for ScheduledAction first and then we can go fix that one in a separate PR.

[benjchristensen]

I have confirmed this fix is working nicely after the changes by @loganj in #1766