Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/src/scheduling/Dispatcher.kt (Kotlin)

Code before:
internal open class SchedulerCoroutineDispatcher(
    private val corePoolSize: Int = CORE_POOL_SIZE,
    private val maxPoolSize: Int = MAX_POOL_SIZE,
    private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
    private val schedulerName: String = "CoroutineScheduler",
) : ExecutorCoroutineDispatcher() {

    override val executor: Executor
        get() = coroutineScheduler

    // This is variable for test purposes, so that we can reinitialize from clean state
    private var coroutineScheduler = createScheduler()

    private fun createScheduler() =
        CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)

    override fun dispatch(context: CoroutineContext, block: Runnable): Unit = coroutineScheduler.dispatch(block)

    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() {
        coroutineScheduler.close()
    }

    // fot tests only
    @Synchronized
    internal fun usePrivateScheduler() {
        coroutineScheduler.shutdown(1_000L)
        coroutineScheduler = createScheduler()
    }

    // for tests only
    @Synchronized
    internal fun shutdown(timeout: Long) {
        coroutineScheduler.shutdown(timeout)
    }

    // for tests only
    internal fun restore() = usePrivateScheduler() // recreate scheduler
}


Review comment (security):
What you're proposing is better than what we have now, but it's still not ideal. We are just one step farther from seeing the same effect. `LoggingDispatcher(Dispatchers.IO)` will exhibit this behavior, as the author of `LoggingDispatcher` can't override `dispatchYield`. That's a problem we're also facing with `Delay` (people can't wrap `Delay` implementations and expect the same behavior).

I see the following options of dealing with this problem for good:
* Make the `YieldContext` marker used by `Dispatchers.Unconfined` available to all dispatchers (and not just those not needing a dispatch) and check its presence when `dispatch`'ing to the coroutine scheduler; remove `dispatchYield` completely. Probably too slow to check that on every dispatch?
* Apply the proposed fix, but also design and publish some support for wrapping coroutine dispatchers that takes care to propagate `dispatchYield`.
* Make `dispatchYield` a valid optimization in a wide range of scenarios (including things like "skip suspensions on `Dispatchers.Main` if the looper is empty anyway") by recognizing the coroutine not being yet dispatched to the correct execution context as a separate state. EDIT: ... and make `dispatchYield` public, as it will no longer be deeply tied to our implementation details.

Thoughts?

Code after:
internal open class SchedulerCoroutineDispatcher(
    private val corePoolSize: Int = CORE_POOL_SIZE,
    private val maxPoolSize: Int = MAX_POOL_SIZE,
    private val idleWorkerKeepAliveNs: Long = IDLE_WORKER_KEEP_ALIVE_NS,
    private val schedulerName: String = "CoroutineScheduler",
) : ExecutorCoroutineDispatcher() {

    override val executor: Executor
        get() = coroutineScheduler

    // This is variable for test purposes, so that we can reinitialize from clean state
    private var coroutineScheduler = createScheduler()

    private fun createScheduler() =
        CoroutineScheduler(corePoolSize, maxPoolSize, idleWorkerKeepAliveNs, schedulerName)

    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 close() {
        coroutineScheduler.close()
    }

    // fot tests only
    @Synchronized
    internal fun usePrivateScheduler() {
        coroutineScheduler.shutdown(1_000L)
        coroutineScheduler = createScheduler()
    }

    // for tests only
    @Synchronized
    internal fun shutdown(timeout: Long) {
        coroutineScheduler.shutdown(timeout)
    }