Support yield in immediate dispatchers

As special internal fun CoroutineDispatcher.isDispatchSupported is
introduced to special-case Unconfined dispatcher that cannot be
dispatched to. All other dispatcher return `true`. This gives a little
more freedom in distinguishing dispatchers

           | isDispNeeded | isDispSupported
Regular    | true         | true
Immediate  | depends      | true
Unconfined | false        | false

So yield now checks for "isDispatchSupported" and works properly for
immediate dispatchers.

Fixes #1474

Author: elizarov

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

@@ -79,6 +79,15 @@ public abstract class CoroutineDispatcher :
      */
     public abstract fun dispatch(context: CoroutineContext, block: Runnable)
 
+    /**
+     * Reserved for [Dispatchers.Unconfined] that returns `false` here because its
+     * [dispatch] method throws [UnsupportedOperationException]. It is used in [yield] implementation.
+     *
+     * @suppress **This an internal API and should not be used from general code.**
+     */
+    @InternalCoroutinesApi
+    public open fun isDispatchSupported(context: CoroutineContext): Boolean = true
+
     /**
      * Dispatches execution of a runnable `block` onto another thread in the given `context`
      * with a hint for the dispatcher that the current dispatch is triggered by a [yield] call, so that the execution of this

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

@@ -11,6 +11,7 @@ import kotlin.coroutines.*
  */
 internal object Unconfined : CoroutineDispatcher() {
     override fun isDispatchNeeded(context: CoroutineContext): Boolean = false
+    override fun isDispatchSupported(context: CoroutineContext): Boolean = false
     override fun dispatch(context: CoroutineContext, block: Runnable) { throw UnsupportedOperationException() }
     override fun toString(): String = "Unconfined"
 }

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

@@ -19,7 +19,9 @@ public suspend fun yield(): Unit = suspendCoroutineUninterceptedOrReturn sc@ { u
     val context = uCont.context
     context.checkCompletion()
     val cont = uCont.intercepted() as? DispatchedContinuation<Unit> ?: return@sc Unit
-    if (!cont.dispatcher.isDispatchNeeded(context)) {
+    // Special case for Unconfined dispatcher that does not support the concept of "dispatch" and where
+    // "dispatchYield" function cannot be used.
+    if (!cont.dispatcher.isDispatchSupported(context)) {
         return@sc if (cont.yieldUndispatched()) COROUTINE_SUSPENDED else Unit
     }
     cont.dispatchYield(Unit)

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlin.coroutines.*
+import kotlin.test.*
+
+class ImmediateYieldTest : TestBase() {
+
+    // See https://github.com/Kotlin/kotlinx.coroutines/issues/1474
+    @Test
+    fun testImmediateYield() = runTest {
+        expect(1)
+        launch(ImmediateDispatcher(coroutineContext[ContinuationInterceptor])) {
+            expect(2)
+            yield()
+            expect(4)
+        }
+        expect(3) // after yield
+        yield() // yield back
+        finish(5)
+    }
+
+    // imitate immediate dispatcher
+    private class ImmediateDispatcher(job: ContinuationInterceptor?) : CoroutineDispatcher() {
+        val delegate: CoroutineDispatcher = job as CoroutineDispatcher
+
+        override fun isDispatchNeeded(context: CoroutineContext): Boolean = false
+
+        override fun dispatch(context: CoroutineContext, block: Runnable) =
+            delegate.dispatch(context, block)
+    }
+}