Basic exception stacktrace recovery mechanism in JobSupport (~all builders) and channels

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

Fixes #493

Author: qwwdfsad

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

@@ -248,7 +248,8 @@ private class RunCompletion<in T>(
     override val context: CoroutineContext,
     delegate: Continuation<T>,
     resumeMode: Int
-) : AbstractContinuation<T>(delegate, resumeMode) {
-
+) : AbstractContinuation<T>(delegate, resumeMode) , CoroutineStackFrame {
+    override val callerFrame: CoroutineStackFrame? = delegate as CoroutineStackFrame?
+    override fun getStackTraceElement(): StackTraceElement? = null
     override val useCancellingState: Boolean get() = true
 }

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

@@ -261,10 +261,15 @@ private class DisposeOnCancel(private val handle: DisposableHandle) : CancelHand
 internal 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

@@ -13,9 +13,11 @@ private val UNDEFINED = Symbol("UNDEFINED")
 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 {
     private var _state: Any? = UNDEFINED
     public override var resumeMode: Int = 0
+    override val callerFrame: CoroutineStackFrame? = continuation as? CoroutineStackFrame
+    override fun getStackTraceElement(): StackTraceElement? = null
 
     override fun takeState(): Any? {
         val state = _state
@@ -44,8 +46,9 @@ internal class DispatchedContinuation<in T>(
             _state = value
             resumeMode = MODE_CANCELLABLE
             dispatcher.dispatch(context, this)
-        } else
+        } else {
             resumeUndispatched(value)
+        }
     }
 
     @Suppress("NOTHING_TO_INLINE") // we need it inline to save us an entry on the stack

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

@@ -956,7 +956,7 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
             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
 
             }
@@ -971,7 +971,7 @@ internal open class JobSupport constructor(active: Boolean) : Job, SelectClause0
             val state = this.state
             check(state !is Incomplete)
             if (state is CompletedExceptionally)
-                cont.resumeWithException(state.cause)
+                cont.resumeWithStackTrace(state.cause)
             else
                 cont.resume(state)
         })

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

@@ -174,8 +174,8 @@ public 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?.also { recoverStackTrace(it) } ?: return false
+            result is Closed<*> -> throw recoverStackTrace(result.sendException)
             else -> error("offerInternal returned $result")
         }
     }
@@ -192,7 +192,7 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
                 }
                 is Closed<*> -> {
                     helpClose(enqueueResult)
-                    cont.resumeWithException(enqueueResult.sendException)
+                    cont.resumeWithStackTrace(enqueueResult.sendException)
                     return@sc
                 }
             }
@@ -206,7 +206,7 @@ public 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")
@@ -405,7 +405,7 @@ public 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 {
@@ -417,7 +417,7 @@ public 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")
                 }
             }
@@ -571,7 +571,7 @@ public 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
     }
 
@@ -587,7 +587,7 @@ public 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) {
@@ -617,7 +617,7 @@ public 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
@@ -638,7 +638,7 @@ public 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) {
@@ -751,7 +751,7 @@ public 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
@@ -791,7 +791,7 @@ public abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E>
                                 block.startCoroutineUnintercepted(null, select.completion)
                             return
                         } else
-                            throw pollResult.closeCause
+                            throw recoverStackTrace(pollResult.closeCause)
                     }
                     else -> {
                         // selected successfully
@@ -850,7 +850,7 @@ public 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
@@ -871,7 +871,7 @@ public 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) {
@@ -884,7 +884,7 @@ public 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
@@ -904,7 +904,7 @@ public 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]"
     }
@@ -939,7 +939,7 @@ public abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E>
             val token = if (closed.closeCause == null)
                 cont.tryResume(false)
             else
-                cont.tryResumeWithException(closed.receiveException)
+                cont.tryResumeWithException(recoverStackTrace(closed.receiveException, cont))
             if (token != null) {
                 iterator.result = closed
                 cont.completeResume(token)
@@ -1052,7 +1052,7 @@ public 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,37 @@
+/*
+ * 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
+
+@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?
+}

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

@@ -20,3 +20,6 @@ public expect open class TestBase constructor() {
         block: suspend CoroutineScope.() -> Unit
     )
 }
+
+// Specific exception which stacktrace cannot be recovered by our machinery
+class TestException(message: String) : Exception(message)

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

@@ -0,0 +1,104 @@
+/*
+ * 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 kotlinx.coroutines.*
+import java.util.*
+import kotlin.coroutines.*
+
+internal actual fun <E : Throwable> recoverStackTrace(exception: E): E {
+    if (!DEBUG || exception is CancellationException) {
+        return exception
+    }
+
+    val copy = tryCopyException(exception) ?: return exception
+    return copy.sanitizeStackTrace()
+}
+
+private fun <E : Throwable> E.sanitizeStackTrace(): E {
+    val size = stackTrace.size
+
+    var lastIntrinsic = -1
+    for (i in 0 until size) {
+        val name = stackTrace[i].className
+        if ("kotlinx.coroutines.internal.ExceptionsKt" == name) {
+            lastIntrinsic = i
+        }
+    }
+
+    val startIndex = lastIntrinsic + 1
+    val trace = Array(size - lastIntrinsic) {
+        if (it == 0) {
+            artificialFrame("Current coroutine stacktrace")
+        } else {
+            stackTrace[startIndex + it - 1]
+        }
+    }
+
+    stackTrace = trace
+    return this
+}
+
+internal actual fun <E : Throwable> recoverStackTrace(exception: E, continuation: Continuation<*>): E {
+    if (!DEBUG || exception is CancellationException || continuation !is CoroutineStackFrame) {
+        return exception
+    }
+
+    val stacktrace = createStackTrace(continuation)
+    if (stacktrace.isEmpty()) return exception
+    val newException = tryCopyException(exception) ?: return exception
+    stacktrace.add(0, artificialFrame("Current coroutine stacktrace"))
+    newException.stackTrace = stacktrace.toTypedArray()
+    return newException
+}
+
+@Suppress("UNCHECKED_CAST")
+private fun <E : Throwable> tryCopyException(exception: E): E? {
+    /*
+     * Try to reflectively find constructor(), constructor(message, cause) or constructor(cause).
+     * First thing which comes in mind is why we should do it at all? Exceptions are share between coroutines,
+     * so we can't safely modify their stacktraces, thus making copy is our only hope
+     */
+    var newException: E? = null
+    try {
+        for (constructor in exception.javaClass.constructors.sortedByDescending { it.parameterTypes.size }) {
+            val parameters = constructor.parameterTypes
+            if (parameters.size == 2 && parameters[0] == String::class.java && parameters[1] == Throwable::class.java) {
+                newException = constructor.newInstance(exception.message, exception) as E
+            } else if (parameters.size == 1 && parameters[0] == Throwable::class.java) {
+                newException = constructor.newInstance(exception) as E
+            } else if (parameters.isEmpty()) {
+                newException = (constructor.newInstance() as E).also { it.initCause(exception) }
+            }
+
+            if (newException != null) {
+                break
+            }
+        }
+    } catch (e: Exception) {
+        // Do nothing
+    }
+    return newException
+}
+
+private fun createStackTrace(continuation: CoroutineStackFrame): ArrayList<StackTraceElement> {
+    val stack = ArrayList<StackTraceElement>()
+    continuation.getStackTraceElement()?.let { stack.add(it) }
+
+    var last = continuation
+    while (true) {
+        last = (last as? CoroutineStackFrame)?.callerFrame ?: break
+        last.getStackTraceElement()?.let { stack.add(it) }
+    }
+    return stack
+}
+
+
+public fun artificialFrame(message: String) = java.lang.StackTraceElement("\b\b\b($message", "\b", "\b", -1)
+
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+actual typealias CoroutineStackFrame = kotlin.coroutines.jvm.internal.CoroutineStackFrame
+
+actual typealias StackTraceElement = java.lang.StackTraceElement