Revert "Document and fix dispatchYield codepath (Kotlin#4255)"

This reverted introduces regression IJPL-182130. We did not get this deadlock before, so it seems safe to revert the commit

This reverts commit d30af7c.

Author: knisht

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

@@ -222,9 +222,6 @@ public abstract class CoroutineDispatcher :
      * Though the `yield` marker may be passed as a part of [context], this
      * is a separate method for performance reasons.
      *
-     * Implementation note: this entry-point is used for `Dispatchers.IO` and [Dispatchers.Default]
-     * unerlying implementations, see overrides for this method.
-     *
      * @suppress **This an internal API and should not be used from general code.**
      */
     @InternalCoroutinesApi

kotlinx-coroutines-core/jvm/src/scheduling/CoroutineScheduler.kt

@@ -410,14 +410,14 @@ internal class CoroutineScheduler(
      * this [block] may execute blocking operations (IO, system calls, locking primitives etc.)
      *
      * [taskContext] -- concurrency context of given [block].
-     * [fair] -- whether this [dispatch] call is fair.
-     * If `true` then the task will be dispatched in a FIFO manner.
+     * [tailDispatch] -- whether this [dispatch] call is the last action the (presumably) worker thread does in its current task.
+     * If `true`, then  the task will be dispatched in a FIFO manner and no additional workers will be requested,
+     * but only if the current thread is a corresponding worker thread.
      * Note that caller cannot be ensured that it is being executed on worker thread for the following reasons:
      *   - [CoroutineStart.UNDISPATCHED]
-     *   - Concurrent [close] that effectively shutdowns the worker thread.
-     * Used for [yield].
+     *   - Concurrent [close] that effectively shutdowns the worker thread
      */
-    fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, fair: Boolean = false) {
+    fun dispatch(block: Runnable, taskContext: TaskContext = NonBlockingContext, tailDispatch: Boolean = false) {
         trackTask() // this is needed for virtual time support
         val task = createTask(block, taskContext)
         val isBlockingTask = task.isBlocking
@@ -426,18 +426,20 @@ internal class CoroutineScheduler(
         val stateSnapshot = if (isBlockingTask) incrementBlockingTasks() else 0
         // try to submit the task to the local queue and act depending on the result
         val currentWorker = currentWorker()
-        val notAdded = currentWorker.submitToLocalQueue(task, fair)
+        val notAdded = currentWorker.submitToLocalQueue(task, tailDispatch)
         if (notAdded != null) {
             if (!addToGlobalQueue(notAdded)) {
                 // Global queue is closed in the last step of close/shutdown -- no more tasks should be accepted
                 throw RejectedExecutionException("$schedulerName was terminated")
             }
         }
+        val skipUnpark = tailDispatch && currentWorker != null
         // Checking 'task' instead of 'notAdded' is completely okay
         if (isBlockingTask) {
             // Use state snapshot to better estimate the number of running threads
-            signalBlockingWork(stateSnapshot)
+            signalBlockingWork(stateSnapshot, skipUnpark = skipUnpark)
         } else {
+            if (skipUnpark) return
             signalCpuWork()
         }
     }
@@ -453,7 +455,8 @@ internal class CoroutineScheduler(
     }
 
     // NB: should only be called from 'dispatch' method due to blocking tasks increment
-    private fun signalBlockingWork(stateSnapshot: Long) {
+    private fun signalBlockingWork(stateSnapshot: Long, skipUnpark: Boolean) {
+        if (skipUnpark) return
         if (tryUnpark()) return
         // Use state snapshot to avoid accidental thread overprovision
         if (tryCreateWorker(stateSnapshot)) return
@@ -526,7 +529,7 @@ internal class CoroutineScheduler(
      * Returns `null` if task was successfully added or an instance of the
      * task that was not added or replaced (thus should be added to global queue).
      */
-    private fun Worker?.submitToLocalQueue(task: Task, fair: Boolean): Task? {
+    private fun Worker?.submitToLocalQueue(task: Task, tailDispatch: Boolean): Task? {
         if (this == null) return task
         /*
          * This worker could have been already terminated from this thread by close/shutdown and it should not
@@ -538,7 +541,7 @@ internal class CoroutineScheduler(
             return task
         }
         mayHaveLocalTasks = true
-        return localQueue.add(task, fair = fair)
+        return localQueue.add(task, fair = tailDispatch)
     }
 
     private fun currentWorker(): Worker? = (Thread.currentThread() as? Worker)?.takeIf { it.scheduler == this }

kotlinx-coroutines-core/jvm/src/scheduling/Dispatcher.kt

@@ -124,21 +124,11 @@ internal open class SchedulerCoroutineDispatcher(
 
     override fun dispatch(context: CoroutineContext, block: Runnable): Unit = coroutineScheduler.dispatch(block)
 
-    override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit {
-        /*
-         * 'dispatchYield' implementation is needed to address the scheduler's scheduling policy.
-         * By default, the scheduler dispatches tasks in a semi-LIFO order, meaning that for the
-         * task sequence [#1, #2, #3], the scheduling of task #4 will produce
-         * [#4, #1, #2, #3], allocates new worker and makes #4 stealable after some time.
-         * On a fast enough system, it means that `while (true) { yield() }` might obstruct the progress
-         * of the system and potentially starve it.
-         * To mitigate that, `dispatchYield` is a dedicated entry point that produces [#1, #2, #3, #4]
-         */
-        coroutineScheduler.dispatch(block, fair = true)
-    }
-
-    internal fun dispatchWithContext(block: Runnable, context: TaskContext, fair: Boolean) {
-        coroutineScheduler.dispatch(block, context, fair)
+    override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit =
+        coroutineScheduler.dispatch(block, tailDispatch = true)
+
+    internal fun dispatchWithContext(block: Runnable, context: TaskContext, tailDispatch: Boolean) {
+        coroutineScheduler.dispatch(block, context, tailDispatch)
     }
 
     override fun close() {

kotlinx-coroutines-core/jvm/test/scheduling/CoroutineSchedulerTest.kt

@@ -1,6 +1,7 @@
 package kotlinx.coroutines.scheduling
 
 import kotlinx.coroutines.testing.*
+import kotlinx.coroutines.*
 import org.junit.Test
 import java.lang.Runnable
 import java.util.concurrent.*
@@ -79,7 +80,7 @@ class CoroutineSchedulerTest : TestBase() {
                 it.dispatch(Runnable {
                     expect(2)
                     finishLatch.countDown()
-                }, fair = true)
+                }, tailDispatch = true)
             })
 
             startLatch.countDown()