Change the workers to capture the stack trace for all subsequent scheduled actions to increase the readability of uncaught and fatal exceptions that bubble up to the schedulers.

abersnaze · abersnaze · commit e0929ee25327 · 2016-05-13T10:28:27.000-07:00
diff --git a/src/main/java/rx/internal/schedulers/ExecutorScheduler.java b/src/main/java/rx/internal/schedulers/ExecutorScheduler.java
@@ -46,6 +46,7 @@ public Worker createWorker() {
 
     /** Worker that schedules tasks on the executor indirectly through a trampoline mechanism. */
     static final class ExecutorSchedulerWorker extends Scheduler.Worker implements Runnable {
+        private final Throwable creationContext = new SchedulerContextException();
         final Executor executor;
         // TODO: use a better performing structure for task tracking
         final CompositeSubscription tasks;
@@ -68,7 +69,7 @@ public Subscription schedule(Action0 action) {
             if (isUnsubscribed()) {
                 return Subscriptions.unsubscribed();
             }
-            ScheduledAction ea = new ScheduledAction(action, tasks);
+            ScheduledAction ea = new ScheduledAction(action, tasks, creationContext);
             tasks.add(ea);
             queue.offer(ea);
             if (wip.getAndIncrement() == 0) {
@@ -150,7 +151,7 @@ public void call() {
                         ((ScheduledAction)s2).add(removeMas);
                     }
                 }
-            });
+            }, this.creationContext);
             // This will make sure if ea.call() gets executed before this line
             // we don't override the current task in mas.
             first.set(ea);
diff --git a/src/main/java/rx/internal/schedulers/NewThreadWorker.java b/src/main/java/rx/internal/schedulers/NewThreadWorker.java
@@ -33,6 +33,7 @@
  * @warn class description missing
  */
 public class NewThreadWorker extends Scheduler.Worker implements Subscription {
+    private final Throwable creationContext = new SchedulerContextException();
     private final ScheduledExecutorService executor;
     private final RxJavaSchedulersHook schedulersHook;
     volatile boolean isUnsubscribed;
@@ -233,7 +234,7 @@ public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit
      */
     public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
         Action0 decoratedAction = schedulersHook.onSchedule(action);
-        ScheduledAction run = new ScheduledAction(decoratedAction);
+        ScheduledAction run = new ScheduledAction(decoratedAction, creationContext);
         Future<?> f;
         if (delayTime <= 0) {
             f = executor.submit(run);
@@ -246,7 +247,7 @@ public ScheduledAction scheduleActual(final Action0 action, long delayTime, Time
     }
     public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit, CompositeSubscription parent) {
         Action0 decoratedAction = schedulersHook.onSchedule(action);
-        ScheduledAction run = new ScheduledAction(decoratedAction, parent);
+        ScheduledAction run = new ScheduledAction(decoratedAction, parent, creationContext);
         parent.add(run);
 
         Future<?> f;
@@ -262,7 +263,7 @@ public ScheduledAction scheduleActual(final Action0 action, long delayTime, Time
     
     public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit, SubscriptionList parent) {
         Action0 decoratedAction = schedulersHook.onSchedule(action);
-        ScheduledAction run = new ScheduledAction(decoratedAction, parent);
+        ScheduledAction run = new ScheduledAction(decoratedAction, parent, creationContext);
         parent.add(run);
         
         Future<?> f;
diff --git a/src/main/java/rx/internal/schedulers/ScheduledAction.java b/src/main/java/rx/internal/schedulers/ScheduledAction.java
@@ -19,6 +19,7 @@
 import java.util.concurrent.atomic.*;
 
 import rx.Subscription;
+import rx.exceptions.Exceptions;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action0;
 import rx.internal.util.SubscriptionList;
@@ -34,18 +35,22 @@ public final class ScheduledAction extends AtomicReference<Thread> implements Ru
     private static final long serialVersionUID = -3962399486978279857L;
     final SubscriptionList cancel;
     final Action0 action;
+    final Throwable creationContext;
 
-    public ScheduledAction(Action0 action) {
+    public ScheduledAction(Action0 action, Throwable creationContext) {
         this.action = action;
         this.cancel = new SubscriptionList();
+        this.creationContext = creationContext;
     }
-    public ScheduledAction(Action0 action, CompositeSubscription parent) {
+    public ScheduledAction(Action0 action, CompositeSubscription parent, Throwable creationContext) {
         this.action = action;
         this.cancel = new SubscriptionList(new Remover(this, parent));
+        this.creationContext = creationContext;
     }
-    public ScheduledAction(Action0 action, SubscriptionList parent) {
+    public ScheduledAction(Action0 action, SubscriptionList parent, Throwable creationContext) {
         this.action = action;
         this.cancel = new SubscriptionList(new Remover2(this, parent));
+        this.creationContext = creationContext;
     }
 
     @Override
@@ -61,6 +66,7 @@ public void run() {
             } else {
                 ie = new IllegalStateException("Fatal Exception thrown on Scheduler.Worker thread.", e);
             }
+            Exceptions.addCause(ie, creationContext);
             RxJavaPlugins.getInstance().getErrorHandler().handleError(ie);
             Thread thread = Thread.currentThread();
             thread.getUncaughtExceptionHandler().uncaughtException(thread, ie);
diff --git a/src/main/java/rx/internal/schedulers/SchedulerContextException.java b/src/main/java/rx/internal/schedulers/SchedulerContextException.java
@@ -0,0 +1,27 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.internal.schedulers;
+
+/**
+ * Used only for providing context around where work was scheduled should an error occur in a different thread.
+ */
+public class SchedulerContextException extends Exception {
+    public SchedulerContextException() {
+        super("Asynchronous work scheduled at");
+    }
+
+    private static final long serialVersionUID = 1L;
+}
diff --git a/src/test/java/rx/schedulers/ComputationSchedulerTests.java b/src/test/java/rx/schedulers/ComputationSchedulerTests.java
@@ -21,6 +21,7 @@
 
 import java.util.HashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 
@@ -30,6 +31,8 @@
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
+import rx.plugins.RxJavaErrorHandler;
+import rx.plugins.RxJavaPlugins;
 
 public class ComputationSchedulerTests extends AbstractSchedulerConcurrencyTests {
 
@@ -168,4 +171,48 @@ public void testCancelledTaskRetention() throws InterruptedException {
             w.unsubscribe();
         }
     }
+
+    @Test
+    public void testStackTraceAcrossThreads() throws Throwable {
+        final AtomicReference<Throwable> exceptionRef = new AtomicReference<Throwable>();
+        final CountDownLatch done = new CountDownLatch(1);
+
+        RxJavaPlugins.getInstance().reset();
+        RxJavaPlugins.getInstance().registerErrorHandler(new RxJavaErrorHandler() {
+            @Override
+            public void handleError(Throwable e) {
+                exceptionRef.set(e);
+                done.countDown();
+            }
+        });
+
+        try {
+            getScheduler().createWorker().schedule(new Action0() {
+                @Override
+                public void call() {
+                    throw new RuntimeException();
+                }
+            });
+        } catch (Exception e) {
+            exceptionRef.set(e);
+            done.countDown();
+        }
+
+        done.await();
+
+        Throwable exception = exceptionRef.get();
+        Throwable e = exception;
+        while (e != null) {
+            StackTraceElement[] st = e.getStackTrace();
+            for (StackTraceElement stackTraceElement : st) {
+                if (stackTraceElement.getClassName().contains(getClass().getName())) {
+                    // pass we found this class in the stack trace.
+                    return;
+                }
+            }
+            e = e.getCause();
+        }
+
+        throw exception;
+    }
 }
diff --git a/src/test/java/rx/schedulers/ImmediateSchedulerTest.java b/src/test/java/rx/schedulers/ImmediateSchedulerTest.java
@@ -17,12 +17,18 @@
 
 import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.junit.Test;
 
 import rx.Observable;
 import rx.Scheduler;
+import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
+import rx.plugins.RxJavaErrorHandler;
+import rx.plugins.RxJavaPlugins;
 
 public class ImmediateSchedulerTest extends AbstractSchedulerTests {
 
@@ -101,4 +107,48 @@ public void call(String t) {
             }
         });
     }
+
+    @Test
+    public void testStackTraceAcrossThreads() throws Throwable {
+        final AtomicReference<Throwable> exceptionRef = new AtomicReference<Throwable>();
+        final CountDownLatch done = new CountDownLatch(1);
+
+        RxJavaPlugins.getInstance().reset();
+        RxJavaPlugins.getInstance().registerErrorHandler(new RxJavaErrorHandler() {
+            @Override
+            public void handleError(Throwable e) {
+                exceptionRef.set(e);
+                done.countDown();
+            }
+        });
+
+        try {
+            getScheduler().createWorker().schedule(new Action0() {
+                @Override
+                public void call() {
+                    throw new RuntimeException();
+                }
+            });
+        } catch (Exception e) {
+            exceptionRef.set(e);
+            done.countDown();
+        }
+
+        done.await();
+
+        Throwable exception = exceptionRef.get();
+        Throwable e = exception;
+        while (e != null) {
+            StackTraceElement[] st = e.getStackTrace();
+            for (StackTraceElement stackTraceElement : st) {
+                if (stackTraceElement.getClassName().contains(getClass().getName())) {
+                    // pass we found this class in the stack trace.
+                    return;
+                }
+            }
+            e = e.getCause();
+        }
+
+        throw exception;
+    }
 }
diff --git a/src/test/java/rx/schedulers/IoSchedulerTest.java b/src/test/java/rx/schedulers/IoSchedulerTest.java
@@ -18,11 +18,16 @@
 
 import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.junit.Test;
 
 import rx.*;
 import rx.Scheduler.Worker;
 import rx.functions.*;
+import rx.plugins.RxJavaErrorHandler;
+import rx.plugins.RxJavaPlugins;
 
 public class IoSchedulerTest extends AbstractSchedulerConcurrencyTests {
 
@@ -83,4 +88,47 @@ public void testCancelledTaskRetention() throws InterruptedException {
         }
     }
 
+    @Test
+    public void testStackTraceAcrossThreads() throws Throwable {
+        final AtomicReference<Throwable> exceptionRef = new AtomicReference<Throwable>();
+        final CountDownLatch done = new CountDownLatch(1);
+
+        RxJavaPlugins.getInstance().reset();
+        RxJavaPlugins.getInstance().registerErrorHandler(new RxJavaErrorHandler() {
+            @Override
+            public void handleError(Throwable e) {
+                exceptionRef.set(e);
+                done.countDown();
+            }
+        });
+
+        try {
+            getScheduler().createWorker().schedule(new Action0() {
+                @Override
+                public void call() {
+                    throw new RuntimeException();
+                }
+            });
+        } catch (Exception e) {
+            exceptionRef.set(e);
+            done.countDown();
+        }
+
+        done.await();
+
+        Throwable exception = exceptionRef.get();
+        Throwable e = exception;
+        while (e != null) {
+            StackTraceElement[] st = e.getStackTrace();
+            for (StackTraceElement stackTraceElement : st) {
+                if (stackTraceElement.getClassName().contains(getClass().getName())) {
+                    // pass we found this class in the stack trace.
+                    return;
+                }
+            }
+            e = e.getCause();
+        }
+
+        throw exception;
+    }
 }
diff --git a/src/test/java/rx/schedulers/NewThreadSchedulerTest.java b/src/test/java/rx/schedulers/NewThreadSchedulerTest.java
@@ -27,6 +27,8 @@
 import rx.Scheduler;
 import rx.functions.Action0;
 import rx.internal.schedulers.ScheduledAction;
+import rx.plugins.RxJavaErrorHandler;
+import rx.plugins.RxJavaPlugins;
 import rx.subscriptions.Subscriptions;
 
 public class NewThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
@@ -83,4 +85,48 @@ public void call() {
             worker.unsubscribe();
         }
     }
+
+    @Test
+    public void testStackTraceAcrossThreads() throws Throwable {
+        final AtomicReference<Throwable> exceptionRef = new AtomicReference<Throwable>();
+        final CountDownLatch done = new CountDownLatch(1);
+
+        RxJavaPlugins.getInstance().reset();
+        RxJavaPlugins.getInstance().registerErrorHandler(new RxJavaErrorHandler() {
+            @Override
+            public void handleError(Throwable e) {
+                exceptionRef.set(e);
+                done.countDown();
+            }
+        });
+
+        try {
+            getScheduler().createWorker().schedule(new Action0() {
+                @Override
+                public void call() {
+                    throw new RuntimeException();
+                }
+            });
+        } catch (Exception e) {
+            exceptionRef.set(e);
+            done.countDown();
+        }
+
+        done.await();
+
+        Throwable exception = exceptionRef.get();
+        Throwable e = exception;
+        while (e != null) {
+            StackTraceElement[] st = e.getStackTrace();
+            for (StackTraceElement stackTraceElement : st) {
+                if (stackTraceElement.getClassName().contains(getClass().getName())) {
+                    // pass we found this class in the stack trace.
+                    return;
+                }
+            }
+            e = e.getCause();
+        }
+
+        throw exception;
+    }
 }
diff --git a/src/test/java/rx/schedulers/TrampolineSchedulerTest.java b/src/test/java/rx/schedulers/TrampolineSchedulerTest.java
