Handle fatal exceptions in continuation machinery using handleViaException instead of rethrowing it to provide faster feedback loop

Fixes #808
Fixes #773

Author: qwwdfsad

kotlinx-coroutines-core/common/src/Dispatched.kt

@@ -51,7 +51,7 @@ private fun DispatchedTask<*>.resumeUnconfined() {
     }
 }
 
-private inline fun runUnconfinedEventLoop(
+private inline fun DispatchedTask<*>.runUnconfinedEventLoop(
     eventLoop: EventLoop,
     block: () -> Unit
 ) {
@@ -64,10 +64,10 @@ private inline fun runUnconfinedEventLoop(
         }
     } catch (e: Throwable) {
         /*
-         * This exception doesn't happen normally, only if user either submitted throwing runnable
-         * or if we have a bug in implementation. Throw an exception that better explains the problem.
+         * This exception doesn't happen normally, only if we have a bug in implementation.
+         * Report it as a fatal exception.
          */
-        throw DispatchException("Unexpected exception in unconfined event loop", e)
+        handleFatalException(e, null)
     } finally {
         eventLoop.decrementUseCount(unconfined = true)
     }
@@ -216,6 +216,7 @@ internal abstract class DispatchedTask<in T>(
 
     public final override fun run() {
         val taskContext = this.taskContext
+        var exception: Throwable? = null
         try {
             val delegate = delegate as DispatchedContinuation<T>
             val continuation = delegate.continuation
@@ -234,11 +235,43 @@ internal abstract class DispatchedTask<in T>(
                 }
             }
         } catch (e: Throwable) {
-            throw DispatchException("Unexpected exception running $this", e)
+            // This instead of runCatching to have nicer stacktrace and debug experience
+            exception = e
         } finally {
-            taskContext.afterTask()
+            val result = runCatching { taskContext.afterTask() }
+            handleFatalException(exception, result.exceptionOrNull())
         }
     }
+
+    /**
+     * Machinery that handles fatal exceptions in kotlinx.coroutines.
+     * There are two kinds of fatal exceptions:
+     *
+     * 1) Exceptions from kotlinx.coroutines code. Such exceptions indicate that either
+     *    the library or the compiler has a bug that breaks internal invariants.
+     *    They usually have specific workarounds, but require careful study of the cause and should
+     *    be reported to the maintainers and fixed on the library's side anyway.
+     *
+     * 2) Exceptions from [ThreadContextElement.updateThreadContext] and [ThreadContextElement.restoreThreadContext].
+     *    While a user code can trigger such exception by providing an improper implementation of [ThreadContextElement],
+     *    we can't ignore it because it may leave coroutine in the inconsistent state.
+     *    If you encounter such exception, you can either disable this context element or wrap it into
+     *    another context element that catches all exceptions and handles it in the application specific manner.
+     *
+     * Fatal exception handling can be intercepted with [CoroutineExceptionHandler] element in the context of
+     * a failed coroutine, but such exceptions should be reported anyway.
+     */
+    internal fun handleFatalException(exception: Throwable?, finallyException: Throwable?) {
+        if (exception === null && finallyException === null) return
+        if (exception !== null && finallyException !== null) {
+            exception.addSuppressedThrowable(finallyException)
+        }
+
+        val cause = exception ?: finallyException
+        val reason = DispatchException("Fatal exception in coroutines machinery for $this. " +
+                "Please read KDoc to 'handleFatalException' method and report this incident to maintainers", cause!!)
+        handleExceptionViaHandler(this.delegate.context, reason)
+    }
 }
 
 internal fun DispatchedContinuation<Unit>.yieldUndispatched(): Boolean =

kotlinx-coroutines-core/jvm/src/ThreadContextElement.kt

@@ -56,7 +56,7 @@ public interface ThreadContextElement<S> : CoroutineContext.Element {
      * when the context of the coroutine this element.
      * The result of this function is the old value of the thread-local state that will be passed to [restoreThreadContext].
      * This method should handle its own exceptions and do not rethrow it. Thrown exceptions will leave coroutine which
-     * context is updated in an undefined state.
+     * context is updated in an undefined state and may crash an application.
      *
      * @param context the coroutine context.
      */
@@ -69,7 +69,7 @@ public interface ThreadContextElement<S> : CoroutineContext.Element {
      * The value of [oldState] is the result of the previous invocation of [updateThreadContext] and it should
      * be restored in the thread-local state by this function.
      * This method should handle its own exceptions and do not rethrow it. Thrown exceptions will leave coroutine which
-     * context is updated in an undefined state.
+     * context is updated in an undefined state and may crash an application.
      *
      * @param context the coroutine context.
      * @param oldState the value returned by the previous invocation of [updateThreadContext].

kotlinx-coroutines-core/jvm/test/FailingCoroutinesMachineryTest.kt

@@ -0,0 +1,135 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import org.junit.*
+import org.junit.Test
+import org.junit.runner.*
+import org.junit.runners.*
+import java.util.concurrent.*
+import kotlin.coroutines.*
+import kotlin.test.*
+
+@RunWith(Parameterized::class)
+class FailingCoroutinesMachineryTest(
+    private val element: CoroutineContext.Element,
+    private val dispatcher: CoroutineDispatcher
+) : TestBase() {
+
+    private var caught: Throwable? = null
+    private val latch = CountDownLatch(1)
+    private var exceptionHandler = CoroutineExceptionHandler { _, t -> caught = t;latch.countDown() }
+    private val lazyOuterDispatcher = lazy { newFixedThreadPoolContext(1, "") }
+
+    private object FailingUpdate : ThreadContextElement<Unit> {
+        private object Key : CoroutineContext.Key<MyElement>
+
+        override val key: CoroutineContext.Key<*> get() = Key
+
+        override fun restoreThreadContext(context: CoroutineContext, oldState: Unit) {
+        }
+
+        override fun updateThreadContext(context: CoroutineContext) {
+            throw TestException("Prevent a coroutine from starting right here for some reason")
+        }
+
+        override fun toString() = "FailingUpdate"
+    }
+
+    private object FailingRestore : ThreadContextElement<Unit> {
+        private object Key : CoroutineContext.Key<MyElement>
+
+        override val key: CoroutineContext.Key<*> get() = Key
+
+        override fun restoreThreadContext(context: CoroutineContext, oldState: Unit) {
+            throw TestException("Prevent a coroutine from starting right here for some reason")
+        }
+
+        override fun updateThreadContext(context: CoroutineContext) {
+        }
+
+        override fun toString() = "FailingRestore"
+    }
+
+    private object ThrowingDispatcher : CoroutineDispatcher() {
+        override fun dispatch(context: CoroutineContext, block: Runnable) {
+            throw TestException()
+        }
+
+        override fun toString() = "ThrowingDispatcher"
+    }
+
+    private object ThrowingDispatcher2 : CoroutineDispatcher() {
+        override fun dispatch(context: CoroutineContext, block: Runnable) {
+            block.run()
+        }
+
+        override fun isDispatchNeeded(context: CoroutineContext): Boolean {
+            throw TestException()
+        }
+
+        override fun toString() = "ThrowingDispatcher2"
+    }
+
+    @After
+    fun tearDown() {
+        runCatching { (dispatcher as? ExecutorCoroutineDispatcher)?.close() }
+        if (lazyOuterDispatcher.isInitialized()) lazyOuterDispatcher.value.close()
+    }
+
+    companion object {
+        @JvmStatic
+        @Parameterized.Parameters(name = "Element: {0}, dispatcher: {1}")
+        fun dispatchers(): List<Array<Any>> {
+            val elements = listOf<Any>(FailingRestore, FailingUpdate)
+            val dispatchers = listOf<Any>(
+                Dispatchers.Unconfined,
+                Dispatchers.Default,
+                Executors.newFixedThreadPool(1).asCoroutineDispatcher(),
+                Executors.newScheduledThreadPool(1).asCoroutineDispatcher(),
+                ThrowingDispatcher, ThrowingDispatcher2
+            )
+
+            return elements.flatMap { element ->
+                dispatchers.map { dispatcher ->
+                    arrayOf(element, dispatcher)
+                }
+            }
+        }
+    }
+
+    @Test
+    fun testElement() = runTest {
+        launch(NonCancellable + dispatcher + exceptionHandler + element) {}
+        checkException()
+    }
+
+    @Test
+    fun testNestedElement() = runTest {
+        launch(NonCancellable + dispatcher + exceptionHandler) {
+            launch(element) {  }
+        }
+        checkException()
+    }
+
+    @Test
+    fun testNestedDispatcherAndElement() = runTest {
+        launch(lazyOuterDispatcher.value + NonCancellable + exceptionHandler) {
+            launch(element + dispatcher) {  }
+        }
+        checkException()
+    }
+
+    private fun checkException() {
+        latch.await(2, TimeUnit.SECONDS)
+        val e = caught
+        assertNotNull(e)
+        // First condition -- failure in context element
+        val firstCondition = e is DispatchException && e.cause is TestException
+        // Second condition -- failure from isDispatchNeeded (#880)
+        val secondCondition = e is TestException
+        assertTrue(firstCondition xor secondCondition)
+    }
+}