Basic exception stacktrace recovery mechanism in JobSupport, ScopeCoroutine and AbstractChannel

  * Implement CoroutineStackFrame in CancellableContinuationImpl, DispatchedContinuation and ScopeCoroutine
  * On coroutine resumption try to reflectively instantiate exception instance of the same type, but with augmented stacktrace
  * Recover stacktrace by walking over CoroutineStackFrame
  * Recover stacktrace on fast-path exceptions without CoroutineStackFrame walking to provide more context to an exception
  * Unwrap exceptions when doing aggregation in JobSupport

Fixes #493

Author: qwwdfsad

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

@@ -216,10 +216,15 @@ private class DisposeOnCancel(private val handle: DisposableHandle) : CancelHand
 internal open class CancellableContinuationImpl<in T>(
     delegate: Continuation<T>,
     resumeMode: Int
-) : AbstractContinuation<T>(delegate, resumeMode), CancellableContinuation<T>, Runnable {
+) : AbstractContinuation<T>(delegate, resumeMode), CancellableContinuation<T>, Runnable, CoroutineStackFrame {
 
     public override val context: CoroutineContext = delegate.context
 
+    override val callerFrame: CoroutineStackFrame?
+        get() = delegate as? CoroutineStackFrame
+
+    override fun getStackTraceElement(): StackTraceElement? = null
+
     override fun initCancellability() {
         initParentJobInternal(delegate.context[Job])
     }

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

@@ -81,11 +81,13 @@ internal object UndispatchedEventLoop {
 internal class DispatchedContinuation<in T>(
     @JvmField val dispatcher: CoroutineDispatcher,
     @JvmField val continuation: Continuation<T>
-) : Continuation<T> by continuation, DispatchedTask<T> {
+) : Continuation<T> by continuation, DispatchedTask<T>, CoroutineStackFrame {
     @JvmField
     @Suppress("PropertyName")
     internal var _state: Any? = UNDEFINED
     public override var resumeMode: Int = 0
+    override val callerFrame: CoroutineStackFrame? = continuation as? CoroutineStackFrame
+    override fun getStackTraceElement(): StackTraceElement? = null
     @JvmField // pre-cached value to avoid ctx.fold on every resumption
     internal val countOrElement = threadContextElements(context)
 

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

@@ -242,8 +242,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         val seenExceptions = identitySet<Throwable>(exceptions.size)
         var suppressed = false
         for (exception in exceptions) {
-            if (exception !== rootCause && exception !is CancellationException && seenExceptions.add(exception)) {
-                rootCause.addSuppressedThrowable(exception)
+            // Unwrap original exception from one with recovered stacktrace
+            val unwrapped = unwrap(exception)
+            if (unwrapped !== rootCause && unwrapped !is CancellationException && seenExceptions.add(exception)) {
+                rootCause.addSuppressedThrowable(unwrapped)
                 suppressed = true
             }
         }
@@ -1077,7 +1079,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             val state = this.state
             if (state !is Incomplete) {
                 // already complete -- just return result
-                if (state is CompletedExceptionally) throw state.cause
+                if (state is CompletedExceptionally) throw recoverStackTrace(state.cause)
                 return state
 
             }
@@ -1233,7 +1235,7 @@ private class ResumeAwaitOnCompletion<T>(
         check(state !is Incomplete)
         if (state is CompletedExceptionally) {
             // Resume with exception in atomic way to preserve exception
-            continuation.resumeWithExceptionMode(state.cause, MODE_ATOMIC_DEFAULT)
+            continuation.resumeWithExceptionMode(recoverStackTrace(state.cause, continuation), MODE_ATOMIC_DEFAULT)
         } else {
             // Resuming with value in a cancellable way (AwaitContinuation is configured for this mode).
             @Suppress("UNCHECKED_CAST")

common/kotlinx-coroutines-core-common/src/channels/AbstractChannel.kt

@@ -176,8 +176,8 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
             result === OFFER_SUCCESS -> true
             // We should check for closed token on offer as well, otherwise offer won't be linearizable
             // in the face of concurrent close()
-            result === OFFER_FAILED ->  throw closedForSend?.sendException ?: return false
-            result is Closed<*> -> throw result.sendException
+            result === OFFER_FAILED -> throw closedForSend?.sendException?.let { recoverStackTrace(it) } ?: return false
+            result is Closed<*> -> throw recoverStackTrace(result.sendException)
             else -> error("offerInternal returned $result")
         }
     }
@@ -194,7 +194,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
                 }
                 is Closed<*> -> {
                     helpClose(enqueueResult)
-                    cont.resumeWithException(enqueueResult.sendException)
+                    cont.resumeWithStackTrace(enqueueResult.sendException)
                     return@sc
                 }
             }
@@ -208,7 +208,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
                 offerResult === OFFER_FAILED -> continue@loop
                 offerResult is Closed<*> -> {
                     helpClose(offerResult)
-                    cont.resumeWithException(offerResult.sendException)
+                    cont.resumeWithStackTrace(offerResult.sendException)
                     return@sc
                 }
                 else -> error("offerInternal returned $offerResult")
@@ -408,7 +408,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
                 when {
                     enqueueResult === ALREADY_SELECTED -> return
                     enqueueResult === ENQUEUE_FAILED -> {} // retry
-                    enqueueResult is Closed<*> -> throw enqueueResult.sendException
+                    enqueueResult is Closed<*> -> throw recoverStackTrace(enqueueResult.sendException)
                     else -> error("performAtomicIfNotSelected(TryEnqueueSendDesc) returned $enqueueResult")
                 }
             } else {
@@ -420,7 +420,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
                         block.startCoroutineUnintercepted(receiver = this, completion = select.completion)
                         return
                     }
-                    offerResult is Closed<*> -> throw offerResult.sendException
+                    offerResult is Closed<*> -> throw recoverStackTrace(offerResult.sendException)
                     else -> error("offerSelectInternal returned $offerResult")
                 }
             }
@@ -574,7 +574,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
 
     @Suppress("UNCHECKED_CAST")
     private fun receiveResult(result: Any?): E {
-        if (result is Closed<*>) throw result.receiveException
+        if (result is Closed<*>) throw recoverStackTrace(result.receiveException)
         return result as E
     }
 
@@ -590,7 +590,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
             // hm... something is not right. try to poll
             val result = pollInternal()
             if (result is Closed<*>) {
-                cont.resumeWithException(result.receiveException)
+                cont.resumeWithStackTrace(result.receiveException)
                 return@sc
             }
             if (result !== POLL_FAILED) {
@@ -620,7 +620,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
     @Suppress("UNCHECKED_CAST")
     private fun receiveOrNullResult(result: Any?): E? {
         if (result is Closed<*>) {
-            if (result.closeCause != null) throw result.closeCause
+            if (result.closeCause != null) throw recoverStackTrace(result.closeCause)
             return null
         }
         return result as E
@@ -641,7 +641,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                 if (result.closeCause == null)
                     cont.resume(null)
                 else
-                    cont.resumeWithException(result.closeCause)
+                    cont.resumeWithStackTrace(result.closeCause)
                 return@sc
             }
             if (result !== POLL_FAILED) {
@@ -758,7 +758,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                 when {
                     pollResult === ALREADY_SELECTED -> return
                     pollResult === POLL_FAILED -> {} // retry
-                    pollResult is Closed<*> -> throw pollResult.receiveException
+                    pollResult is Closed<*> -> throw recoverStackTrace(pollResult.receiveException)
                     else -> {
                         block.startCoroutineUnintercepted(pollResult as E, select.completion)
                         return
@@ -797,8 +797,9 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                             if (select.trySelect(null))
                                 block.startCoroutineUnintercepted(null, select.completion)
                             return
-                        } else
-                            throw pollResult.closeCause
+                        } else {
+                            throw recoverStackTrace(pollResult.closeCause)
+                        }
                     }
                     else -> {
                         // selected successfully
@@ -857,7 +858,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
 
         private fun hasNextResult(result: Any?): Boolean {
             if (result is Closed<*>) {
-                if (result.closeCause != null) throw result.receiveException
+                if (result.closeCause != null) recoverStackTrace(throw result.receiveException)
                 return false
             }
             return true
@@ -878,7 +879,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                     if (result.closeCause == null)
                         cont.resume(false)
                     else
-                        cont.resumeWithException(result.receiveException)
+                        cont.resumeWithStackTrace(result.receiveException)
                     return@sc
                 }
                 if (result !== POLL_FAILED) {
@@ -891,7 +892,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         @Suppress("UNCHECKED_CAST")
         override suspend fun next(): E {
             val result = this.result
-            if (result is Closed<*>) throw result.receiveException
+            if (result is Closed<*>) throw recoverStackTrace(result.receiveException)
             if (result !== POLL_FAILED) {
                 this.result = POLL_FAILED
                 return result as E
@@ -911,7 +912,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
             if (closed.closeCause == null && nullOnClose)
                 cont.resume(null)
             else
-                cont.resumeWithException(closed.receiveException)
+                cont.resumeWithStackTrace(closed.receiveException)
         }
         override fun toString(): String = "ReceiveElement[$cont,nullOnClose=$nullOnClose]"
     }
@@ -943,10 +944,11 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         }
 
         override fun resumeReceiveClosed(closed: Closed<*>) {
-            val token = if (closed.closeCause == null)
+            val token = if (closed.closeCause == null) {
                 cont.tryResume(false)
-            else
-                cont.tryResumeWithException(closed.receiveException)
+            } else {
+                cont.tryResumeWithException(recoverStackTrace(closed.receiveException, cont))
+            }
             if (token != null) {
                 iterator.result = closed
                 cont.completeResume(token)
@@ -1056,7 +1058,7 @@ internal class SendElement(
 ) : LockFreeLinkedListNode(), Send {
     override fun tryResumeSend(idempotent: Any?): Any? = cont.tryResume(Unit, idempotent)
     override fun completeResumeSend(token: Any) = cont.completeResume(token)
-    override fun resumeSendClosed(closed: Closed<*>) = cont.resumeWithException(closed.sendException)
+    override fun resumeSendClosed(closed: Closed<*>) = cont.resumeWithStackTrace(closed.sendException)
     override fun toString(): String = "SendElement($pollResult)[$cont]"
 }
 

common/kotlinx-coroutines-core-common/src/internal/Exceptions.common.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.internal
+
+import kotlin.coroutines.*
+
+/**
+ * Tries to recover stacktrace for given [exception] and [continuation].
+ * Stacktrace recovery tries to restore [continuation] stack frames using its debug metadata with [CoroutineStackFrame] API
+ * and then reflectively instantiate exception of given type with original exception as a cause and
+ * sets new stacktrace for wrapping exception.
+ * Some frames may be missing due to tail-call elimination.
+ *
+ * Works only on JVM with enabled debug-mode.
+ */
+internal expect fun <E: Throwable> recoverStackTrace(exception: E, continuation: Continuation<*>): E
+
+/**
+ * Tries to recover stacktrace for given [exception]. Used in non-suspendable points of awaiting.
+ * Stacktrace recovery tries to instantiate exception of given type with original exception as a cause.
+ * Wrapping exception will have proper stacktrace as it's instantiated in the right context.
+ *
+ * Works only on JVM with enabled debug-mode.
+ */
+internal expect fun <E: Throwable> recoverStackTrace(exception: E): E
+
+/**
+ * The opposite of [recoverStackTrace].
+ * It is guaranteed that `unwrap(recoverStackTrace(e)) === e`
+ */
+internal expect fun <E: Throwable> unwrap(exception: E): E
+
+@Suppress("NOTHING_TO_INLINE")
+internal inline fun Continuation<*>.resumeWithStackTrace(exception: Throwable) = resumeWith(Result.failure(recoverStackTrace(exception, this)))
+
+expect class StackTraceElement
+
+internal expect interface CoroutineStackFrame {
+    public val callerFrame: CoroutineStackFrame?
+    public fun getStackTraceElement(): StackTraceElement?
+}
+
+/**
+ * Marker that indicates that stacktrace of the exception should not be recovered.
+ * Currently internal, but may become public in the future
+ */
+internal interface NonRecoverableThrowable

common/kotlinx-coroutines-core-common/src/internal/Scopes.kt

@@ -14,7 +14,9 @@ import kotlin.jvm.*
 internal open class ScopeCoroutine<in T>(
     context: CoroutineContext,
     @JvmField val uCont: Continuation<T> // unintercepted continuation
-) : AbstractCoroutine<T>(context, true) {
+) : AbstractCoroutine<T>(context, true), CoroutineStackFrame {
+    final override val callerFrame: CoroutineStackFrame? get() = uCont as CoroutineStackFrame?
+    final override fun getStackTraceElement(): StackTraceElement? = null
     override val defaultResumeMode: Int get() = MODE_DIRECT
 
     @Suppress("UNCHECKED_CAST")

common/kotlinx-coroutines-core-common/test/TestBase.common.kt

@@ -4,6 +4,8 @@
 
 package kotlinx.coroutines
 
+import kotlinx.coroutines.internal.*
+
 public expect open class TestBase constructor() {
     public val isStressTest: Boolean
     public val stressTestMultiplier: Int
@@ -21,8 +23,9 @@ public expect open class TestBase constructor() {
     )
 }
 
-public class TestException(message: String? = null) : Throwable(message)
-public class TestException1(message: String? = null) : Throwable(message)
-public class TestException2(message: String? = null) : Throwable(message)
-public class TestException3(message: String? = null) : Throwable(message)
-public class TestRuntimeException(message: String? = null) : RuntimeException(message)
+public class TestException(message: String? = null) : Throwable(message), NonRecoverableThrowable
+public class TestException1(message: String? = null) : Throwable(message), NonRecoverableThrowable
+public class TestException2(message: String? = null) : Throwable(message), NonRecoverableThrowable
+public class TestException3(message: String? = null) : Throwable(message), NonRecoverableThrowable
+public class TestRuntimeException(message: String? = null) : RuntimeException(message), NonRecoverableThrowable
+public class RecoverableTestException(message: String? = null) : Throwable(message)