Refine dispatching convention

Dispatching happens only via `suspendWithCurrentContinuation` calls
instead of each `resume` call

 #KT-15657 Fixed

compiler/backend/src/org/jetbrains/kotlin/codegen/coroutines/coroutineCodegenUtil.kt

@@ -60,6 +60,9 @@ const val COROUTINE_LABEL_FIELD_NAME = "label"
 const val SUSPEND_FUNCTION_CREATE_METHOD_NAME = "create"
 const val DO_RESUME_METHOD_NAME = "doResume"
 
+private val INTERNAL_COROUTINE_INTRINSICS_OWNER = "kotlin/jvm/internal/CoroutineIntrinsics"
+private val NORMALIZE_CONTINUATION_METHOD_NAME = "normalizeContinuation"
+
 data class ResolvedCallWithRealDescriptor(val resolvedCall: ResolvedCall<*>, val fakeContinuationExpression: KtExpression)
 
 @JvmField
@@ -207,6 +210,15 @@ fun createMethodNodeForSuspendCoroutineOrReturn(
 
     node.visitVarInsn(Opcodes.ALOAD, 0)
     node.visitVarInsn(Opcodes.ALOAD, 1)
+
+    node.visitMethodInsn(
+            Opcodes.INVOKESTATIC,
+            INTERNAL_COROUTINE_INTRINSICS_OWNER,
+            NORMALIZE_CONTINUATION_METHOD_NAME,
+            Type.getMethodDescriptor(AsmTypes.CONTINUATION, AsmTypes.CONTINUATION),
+            false
+    )
+
     node.visitMethodInsn(
             Opcodes.INVOKEINTERFACE,
             typeMapper.mapType(functionDescriptor.valueParameters[0]).internalName,

compiler/testData/codegen/box/coroutines/dispatchResume.kt

@@ -53,7 +53,7 @@ fun box(): String {
         val k = suspendWithValue("K")
         log += "$o$k;"
     }
-    if (result != "before 0;suspend(O);before 1;suspend(K);before 2;OK;after 2;after 1;after 0;") return "fail1: $result"
+    if (result != "suspend(O);before 0;suspend(K);before 1;OK;after 1;after 0;") return "fail1: $result"
 
     result = test {
         try {
@@ -64,7 +64,7 @@ fun box(): String {
             log += "${e.message};"
         }
     }
-    if (result != "before 0;error(OK);before 1;OK;after 1;after 0;") return "fail2: $result"
+    if (result != "error(OK);before 0;OK;after 0;") return "fail2: $result"
 
     return "OK"
 }

compiler/testData/codegen/box/coroutines/suspendFunctionAsCoroutine/dispatchResume.kt

@@ -71,12 +71,12 @@ fun box(): String {
     var result = test {
         test1()
     }
-    if (result != "before 0;suspend();before 1;suspend(O);before 2;before 3;suspend(K);before 4;OK;before 5;after 5;after 4;after 3;after 2;after 1;after 0;") return "fail1: $result"
+    if (result != "suspend();before 0;suspend(O);before 1;suspend(K);before 2;OK;after 2;after 1;after 0;") return "fail1: $result"
 
     result = test {
         test2()
     }
-    if (result != "before 0;suspend();before 1;suspend(O);before 2;before 3;error(OK);before 4;OK;before 5;after 5;after 4;after 3;after 2;after 1;after 0;") return "fail2: $result"
+    if (result != "suspend();before 0;suspend(O);before 1;error(OK);before 2;OK;after 2;after 1;after 0;") return "fail2: $result"
 
     return "OK"
 }

core/runtime.jvm/src/kotlin/jvm/internal/CoroutineImpl.kt

@@ -19,41 +19,25 @@ package kotlin.jvm.internal
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.SUSPENDED_MARKER
 
-private const val INTERCEPT_BIT_SET = 1 shl 31
-private const val INTERCEPT_BIT_CLEAR = INTERCEPT_BIT_SET.inv()
-
 abstract class CoroutineImpl : RestrictedCoroutineImpl, DispatchedContinuation<Any?> {
     private val _dispatcher: ContinuationDispatcher?
 
     override val dispatcher: ContinuationDispatcher?
         get() = _dispatcher
 
+    private var facade_: Continuation<Any?>? = null
+    val facade: Continuation<Any?> get() {
+        if (facade_ == null) {
+            facade_ = wrapContinuationIfNeeded(this, dispatcher)
+        }
+
+        return facade_!!
+    }
+
     // this constructor is used to create a continuation instance for coroutine
     constructor(arity: Int, completion: Continuation<Any?>?) : super(arity, completion) {
         _dispatcher = (completion as? DispatchedContinuation<*>)?.dispatcher
     }
-
-    override fun resume(value: Any?) {
-        if (_dispatcher != null) {
-            if (label and INTERCEPT_BIT_SET == 0) {
-                label = label or INTERCEPT_BIT_SET
-                if (_dispatcher.dispatchResume(value, this)) return
-            }
-            label = label and INTERCEPT_BIT_CLEAR
-        }
-        super.resume(value)
-    }
-
-    override fun resumeWithException(exception: Throwable) {
-        if (_dispatcher != null) {
-            if (label and INTERCEPT_BIT_SET == 0) {
-                label = label or INTERCEPT_BIT_SET
-                if (_dispatcher.dispatchResumeWithException(exception, this)) return
-            }
-            label = label and INTERCEPT_BIT_CLEAR
-        }
-        super.resumeWithException(exception)
-    }
 }
 
 abstract class RestrictedCoroutineImpl : Lambda, Continuation<Any?> {

core/runtime.jvm/src/kotlin/jvm/internal/CoroutineIntrinsics.kt

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2010-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+@file:kotlin.jvm.JvmName("CoroutineIntrinsics")
+package kotlin.jvm.internal
+
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.ContinuationDispatcher
+
+fun <T> normalizeContinuation(c: Continuation<T>): Continuation<T> {
+    if (c is CoroutineImpl) {
+        return c.facade
+    }
+
+    return c
+}
+
+internal fun <T> wrapContinuationIfNeeded(c: Continuation<T>, dispatcher: ContinuationDispatcher?): Continuation<T> {
+    if (dispatcher == null) return c
+    return DispatchedContinuationImpl(c, dispatcher)
+}
+
+private class DispatchedContinuationImpl<in T>(
+        private val c: Continuation<T>,
+        private val dispatcher: ContinuationDispatcher
+) : Continuation<T> {
+    override fun resume(value: T) {
+        if (!dispatcher.dispatchResume(value, c)) {
+            c.resume(value)
+        }
+    }
+
+    override fun resumeWithException(exception: Throwable) {
+        if (!dispatcher.dispatchResumeWithException(exception, c)) {
+            c.resumeWithException(exception)
+        }
+    }
+}

libraries/tools/binary-compatibility-validator/reference-public-api/kotlin-runtime.txt

@@ -471,10 +471,16 @@ public class kotlin/jvm/internal/CollectionToArray {
 public abstract class kotlin/jvm/internal/CoroutineImpl : kotlin/jvm/internal/RestrictedCoroutineImpl, kotlin/jvm/internal/DispatchedContinuation {
 	public fun <init> (ILkotlin/coroutines/Continuation;)V
 	public fun getDispatcher ()Lkotlin/coroutines/ContinuationDispatcher;
+	public final fun getFacade ()Lkotlin/coroutines/Continuation;
 	public fun resume (Ljava/lang/Object;)V
 	public fun resumeWithException (Ljava/lang/Throwable;)V
 }
 
+public final class kotlin/jvm/internal/CoroutineIntrinsics {
+	public static final fun normalizeContinuation (Lkotlin/coroutines/Continuation;)Lkotlin/coroutines/Continuation;
+	public static final fun wrapContinuationIfNeeded (Lkotlin/coroutines/Continuation;Lkotlin/coroutines/ContinuationDispatcher;)Lkotlin/coroutines/Continuation;
+}
+
 public abstract interface class kotlin/jvm/internal/DispatchedContinuation : kotlin/coroutines/Continuation {
 	public abstract fun getDispatcher ()Lkotlin/coroutines/ContinuationDispatcher;
 }