Fix stacktrace recovery for Channel.send

Author: qwwdfsad

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

@@ -108,7 +108,6 @@ internal open class BufferedChannel<E>(
     // ## The send operations ##
     // #########################
 
-    // TODO onClosed misses stacktrace recovery mechanism even though it shouldn't
     override suspend fun send(element: E): Unit =
         sendImpl( // <-- this is an inline function
             element = element,
@@ -123,20 +122,26 @@ internal open class BufferedChannel<E>(
             // According to the `send(e)` contract, we need to call
             // `onUndeliveredElement(..)` handler and throw an exception
             // if the channel is already closed.
-            onClosed = { onClosedSend(element) },
+            onClosed = {
+                    // Use continuation in order to recover stacktrace
+                suspendCancellableCoroutine<Unit> {
+                    onClosedSend(element, it)
+                }
+            },
             // When `send(e)` decides to suspend, the corresponding
             // `onNoWaiterSuspend` function that creates a continuation
             // is called. The tail-call optimization is applied here.
             onNoWaiterSuspend = { segm, i, elem, s -> sendOnNoWaiterSuspend(segm, i, elem, s) }
         )
 
-    private fun onClosedSend(element: E) {
+    private fun onClosedSend(element: E, continuation: Continuation<Nothing>) {
         onUndeliveredElement?.callUndeliveredElementCatchingException(element)?.let {
             // If it crashes, add send exception as suppressed for better diagnostics
             it.addSuppressed(sendException)
-            throw recoverStackTrace(it)
+            continuation.resumeWithStackTrace(it)
+            return
         }
-        throw recoverStackTrace(sendException)
+        continuation.resumeWithStackTrace(sendException)
     }
 
     private suspend fun sendOnNoWaiterSuspend(
@@ -188,6 +193,7 @@ internal open class BufferedChannel<E>(
         override fun dispose() {
             segment.onCancellation(index)
         }
+
         override fun invoke(cause: Throwable?) = dispose()
     }
 
@@ -199,6 +205,7 @@ internal open class BufferedChannel<E>(
         override fun dispose() {
             segment.onSenderCancellationWithOnUndeliveredElement(index, context)
         }
+
         override fun invoke(cause: Throwable?) = dispose()
     }
 
@@ -741,7 +748,7 @@ internal open class BufferedChannel<E>(
         invokeOnCancellation(SenderOrReceiverCancellationHandler(segment, index).asHandler)
     }
 
-    private fun onClosedReceiveOnNoWaiterSuspend(cont : CancellableContinuation<E>) {
+    private fun onClosedReceiveOnNoWaiterSuspend(cont: CancellableContinuation<E>) {
         cont.resumeWithException(receiveException)
     }
 
@@ -2040,7 +2047,7 @@ internal open class BufferedChannel<E>(
                 val globalIndex = segment.id * SEGMENT_SIZE + index
                 if (globalIndex < receiversCounter) return -1
                 // Process the cell `segment[index]`.
-                cell_update@while (true) {
+                cell_update@ while (true) {
                     val state = segment.getState(index)
                     when {
                         // The cell is empty.
@@ -2644,7 +2651,7 @@ internal open class BufferedChannel<E>(
         append_elements@ while (true) {
             process_cell@ for (i in 0 until SEGMENT_SIZE) {
                 val globalCellIndex = segment.id * SEGMENT_SIZE + i
-                if (globalCellIndex >=s && globalCellIndex >= r) break@append_elements
+                if (globalCellIndex >= s && globalCellIndex >= r) break@append_elements
                 val cellState = segment.getState(i)
                 val element = segment.getElement(i)
                 val cellStateString = when (cellState) {

kotlinx-coroutines-core/jvm/test-resources/stacktraces/channels/testSendToChannel.txt

@@ -1,15 +1,20 @@
 java.util.concurrent.CancellationException: Channel was cancelled
+	at kotlinx.coroutines.channels.BufferedChannel.cancelImpl$(BufferedChannel.kt)
+	at kotlinx.coroutines.channels.BufferedChannel.cancel(BufferedChannel.kt)
+	at kotlinx.coroutines.channels.ReceiveChannel$DefaultImpls.cancel$default(Channel.kt)
+	at kotlinx.coroutines.exceptions.StackTraceRecoveryChannelsTest$testSendToChannel$1$job$1.invokeSuspend(StackTraceRecoveryChannelsTest.kt)
 	at _COROUTINE._BOUNDARY._(CoroutineDebugging.kt)
-	at kotlinx.coroutines.internal.StackTraceRecoveryKt.recoverStackTrace(StackTraceRecovery.kt)
-	at kotlinx.coroutines.channels.BufferedChannel.onClosedSend(BufferedChannel.kt)
-	at kotlinx.coroutines.channels.BufferedChannel.send$suspendImpl(BufferedChannel.kt)
-	at kotlinx.coroutines.channels.BufferedChannel.send(BufferedChannel.kt)
 	at kotlinx.coroutines.exceptions.StackTraceRecoveryChannelsTest.channelSend(StackTraceRecoveryChannelsTest.kt)
-	at kotlinx.coroutines.exceptions.StackTraceRecoveryChannelsTest.access$channelSend(StackTraceRecoveryChannelsTest.kt)
-	at kotlinx.coroutines.exceptions.StackTraceRecoveryChannelsTest$channelSend$1.invokeSuspend(StackTraceRecoveryChannelsTest.kt)
+	at kotlinx.coroutines.exceptions.StackTraceRecoveryChannelsTest$testSendToChannel$1.invokeSuspend(StackTraceRecoveryChannelsTest.kt)
 Caused by: java.util.concurrent.CancellationException: Channel was cancelled
 	at kotlinx.coroutines.channels.BufferedChannel.cancelImpl$(BufferedChannel.kt)
 	at kotlinx.coroutines.channels.BufferedChannel.cancel(BufferedChannel.kt)
 	at kotlinx.coroutines.channels.ReceiveChannel$DefaultImpls.cancel$default(Channel.kt)
 	at kotlinx.coroutines.exceptions.StackTraceRecoveryChannelsTest$testSendToChannel$1$job$1.invokeSuspend(StackTraceRecoveryChannelsTest.kt)
-	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
+	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt)
+	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt)
+	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt)
+	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt)
+	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt)
+	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
+	at kotlinx.coroutines.TestBase.runTest(TestBase.kt)