Basic exception stacktrace recovery mechanism

  * 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
  * Add kill-switch to disable stacktrace recovery, introduce method to recover stacktrace on the exceptional fast-path
  * Add `suspendCoroutineOrReturn` on exceptional fast-path in await in order to provide "real" stacktrace

Design rationale:
All recovery of *suspended* continuations takes place in Dispatched.kt file, the only place where all calls to "resume*" ends up, so we don't have to remember about stacktrace recovery in every primitive we are implementing. But to provide more accurate stacktraces we *have to* recover it on every fast-path for better debuggability.

Fixes #493

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -50,10 +50,12 @@ public final class kotlinx/coroutines/CancellableContinuation$DefaultImpls {
 	public static synthetic fun tryResume$default (Lkotlinx/coroutines/CancellableContinuation;Ljava/lang/Object;Ljava/lang/Object;ILjava/lang/Object;)Ljava/lang/Object;
 }
 
-public class kotlinx/coroutines/CancellableContinuationImpl : java/lang/Runnable, kotlinx/coroutines/CancellableContinuation {
+public class kotlinx/coroutines/CancellableContinuationImpl : java/lang/Runnable, kotlin/coroutines/jvm/internal/CoroutineStackFrame, kotlinx/coroutines/CancellableContinuation {
 	public fun <init> (Lkotlin/coroutines/Continuation;I)V
 	public fun completeResume (Ljava/lang/Object;)V
+	public fun getCallerFrame ()Lkotlin/coroutines/jvm/internal/CoroutineStackFrame;
 	public fun getContext ()Lkotlin/coroutines/CoroutineContext;
+	public fun getStackTraceElement ()Ljava/lang/StackTraceElement;
 	public fun getSuccessfulResult (Ljava/lang/Object;)Ljava/lang/Object;
 	public fun initCancellability ()V
 	protected fun nameString ()Ljava/lang/String;

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

@@ -218,10 +218,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,10 +81,12 @@ internal object UndispatchedEventLoop {
 internal class DispatchedContinuation<in T>(
     @JvmField val dispatcher: CoroutineDispatcher,
     @JvmField val continuation: Continuation<T>
-) : DispatchedTask<T>(MODE_ATOMIC_DEFAULT), Continuation<T> by continuation {
+) : DispatchedTask<T>(MODE_ATOMIC_DEFAULT), CoroutineStackFrame, Continuation<T> by continuation {
     @JvmField
     @Suppress("PropertyName")
     internal var _state: Any? = UNDEFINED
+    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)
 
@@ -167,7 +169,7 @@ internal class DispatchedContinuation<in T>(
     @Suppress("NOTHING_TO_INLINE") // we need it inline to save us an entry on the stack
     inline fun resumeUndispatchedWithException(exception: Throwable) {
         withCoroutineContext(context, countOrElement) {
-            continuation.resumeWithException(exception)
+            continuation.resumeWithStackTrace(exception)
         }
     }
 
@@ -190,7 +192,7 @@ internal fun <T> Continuation<T>.resumeCancellable(value: T) = when (this) {
 
 internal fun <T> Continuation<T>.resumeCancellableWithException(exception: Throwable) = when (this) {
     is DispatchedContinuation -> resumeCancellableWithException(exception)
-    else -> resumeWithException(exception)
+    else -> resumeWithStackTrace(exception)
 }
 
 internal fun <T> Continuation<T>.resumeDirect(value: T) = when (this) {
@@ -199,8 +201,8 @@ internal fun <T> Continuation<T>.resumeDirect(value: T) = when (this) {
 }
 
 internal fun <T> Continuation<T>.resumeDirectWithException(exception: Throwable) = when (this) {
-    is DispatchedContinuation -> continuation.resumeWithException(exception)
-    else -> resumeWithException(exception)
+    is DispatchedContinuation -> continuation.resumeWithStackTrace(exception)
+    else -> resumeWithStackTrace(exception)
 }
 
 internal abstract class DispatchedTask<in T>(
@@ -231,7 +233,7 @@ internal abstract class DispatchedTask<in T>(
                 else {
                     val exception = getExceptionalResult(state)
                     if (exception != null)
-                        continuation.resumeWithException(exception)
+                        continuation.resumeWithStackTrace(exception)
                     else
                         continuation.resume(getSuccessfulResult(state))
                 }
@@ -275,3 +277,7 @@ internal fun <T> DispatchedTask<T>.resume(delegate: Continuation<T>, useMode: In
         delegate.resumeMode(getSuccessfulResult(state), useMode)
     }
 }
+
+
+@Suppress("NOTHING_TO_INLINE")
+private inline fun Continuation<*>.resumeWithStackTrace(exception: Throwable) = resumeWith(Result.failure(recoverStackTrace(exception, this)))

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

@@ -242,8 +242,9 @@ 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)
+            val unwrapped = unwrap(exception)
+            if (unwrapped !== rootCause && unwrapped !is CancellationException && seenExceptions.add(exception)) {
+                rootCause.addSuppressedThrowable(unwrapped)
                 suppressed = true
             }
         }
@@ -1078,7 +1079,11 @@ 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) { // Slow path to recover stacktrace
+                    suspendCoroutineUninterceptedOrReturn<Unit> {
+                        throw recoverStackTrace(state.cause, it)
+                    }
+                }
                 return state
 
             }

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")
         }
     }
@@ -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
     }
 
@@ -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
@@ -759,7 +759,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
@@ -798,8 +798,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
@@ -858,7 +859,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
@@ -892,7 +893,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
@@ -944,10 +945,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)

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

@@ -0,0 +1,50 @@
+/*
+ * 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
+
+// Name conflict with recoverStackTrace
+@Suppress("NOTHING_TO_INLINE")
+internal expect suspend inline fun recoverAndThrow(exception: Throwable): Nothing
+
+/**
+ * The opposite of [recoverStackTrace].
+ * It is guaranteed that `unwrap(recoverStackTrace(e)) === e`
+ */
+internal expect fun <E: Throwable> unwrap(exception: E): E
+
+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

@@ -5,6 +5,7 @@
 package kotlinx.coroutines
 
 import kotlin.coroutines.*
+import kotlinx.coroutines.internal.*
 
 public expect open class TestBase constructor() {
     public val isStressTest: Boolean
@@ -23,13 +24,13 @@ 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)
 
-// Wrap context to avoid fast-paths on dispatcher comparison
 public fun wrapperDispatcher(context: CoroutineContext): CoroutineContext {
     val dispatcher = context[ContinuationInterceptor] as CoroutineDispatcher
     return object : CoroutineDispatcher() {

core/kotlinx-coroutines-core/src/Debug.kt

@@ -12,6 +12,20 @@ import kotlinx.coroutines.internal.*
  */
 public const val DEBUG_PROPERTY_NAME = "kotlinx.coroutines.debug"
 
+/**
+ * Name of the boolean property that controls stacktrace recovery (enabled by default) on JVM.
+ * Stacktrace recovery is enabled if both debug and stacktrace recovery modes are enabled.
+ *
+ * Stacktrace recovery mode wraps every exception into the exception of the same type with original exception
+ * as cause, but with stacktrace of the current coroutine.
+ * Exception is instantiated using reflection by using no-arg, cause or cause and message constructor.
+ * Stacktrace is not recovered if exception is an instance of [CancellationException] or [NonRecoverableThrowable].
+ *
+ * This mechanism is currently supported for channels, [async], [launch], [coroutineScope], [supervisorScope]
+ * and [withContext] builders.
+ */
+internal const val STACKTRACE_RECOVERY_PROPERTY_NAME = "kotlinx.coroutines.stacktrace.recovery"
+
 /**
  * Automatic debug configuration value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
  */
@@ -36,6 +50,8 @@ internal val DEBUG = systemProp(DEBUG_PROPERTY_NAME).let { value ->
     }
 }
 
+internal val RECOVER_STACKTRACE = systemProp(STACKTRACE_RECOVERY_PROPERTY_NAME, true)
+
 // internal debugging tools
 
 internal actual val Any.hexAddress: String