Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/native/src/MultithreadedDispatchers.kt (Kotlin)

Code before:
        worker.requestTermination().result // Note: calling "result" blocks
    }

    private fun Long.toMicrosSafe(): Long {
        val result = this * 1000
        return if (result > this) result else Long.MAX_VALUE
    }
}

private const val IS_CLOSED_MASK = 1 shl 31

private class MultiWorkerDispatcher(
    private val name: String,
    private val workersCount: Int
) : CloseableCoroutineDispatcher() {
    private val tasksQueue = Channel<Runnable>(Channel.UNLIMITED)
    private val workers = atomicArrayOfNulls<Worker>(workersCount)

    // Number of active workers + isClosed flag in the highest bit
    private val controlState = atomic(0)
    private val activeWorkers: Int get() = controlState.value and (IS_CLOSED_MASK.inv())
    private var isClosed: Boolean
        get() = controlState.value.isClosed()
        set(value) {
            assert { value }
            controlState.value = controlState.value or IS_CLOSED_MASK
        }

    private inline fun Int.isClosed(): Boolean {
        return this and IS_CLOSED_MASK != 0
    }

    private fun workerRunLoop() = runBlocking {
        // NB: we leverage tail-call optimization in this loop, do not replace it with
        // .receive() without proper evaluation
        for (task in tasksQueue) {
            /**
             * Any unhandled exception here will pass through worker's boundary and will be properly reported.
             */
            task.run()
        }
    }

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        fun throwClosed(block: Runnable) {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule: $block")
        }

        if (activeWorkers != workersCount) {
            if (!tryAddWorker()) throwClosed(block) // Do not even try to send to avoid race
        }

Review comment (suggestion):
We may still leak some workers this way.
1. `controlState.value` is read in the `isClosed = true` statement.
2. Some new workers arrive. They increment `controlState.value` and do nothing else.
3. The old `controlState.value` is `or`-ed with `IS_CLOSED_MASK` and written to `controlState.value`.
4. The cleanup procedure is done.
5. The workers add themselves to the array and begin to do their thing.

Code after:
        worker.requestTermination().result // Note: calling "result" blocks
    }

    private fun Long.toMicrosSafe(): Long {
        val result = this * 1000
        return if (result > this) result else Long.MAX_VALUE
    }
}

private class MultiWorkerDispatcher(
    private val name: String,
    workersCount: Int
) : CloseableCoroutineDispatcher() {
    private val tasksQueue = Channel<Runnable>(Channel.UNLIMITED)
    private val workerPool = OnDemandAllocatingPool(workersCount) {
        Worker.start(name = "$name-$it").apply {
            executeAfter { workerRunLoop() }
        }
    }

    private fun workerRunLoop() = runBlocking {
        // NB: we leverage tail-call optimization in this loop, do not replace it with
        // .receive() without proper evaluation
        for (task in tasksQueue) {
            /**
             * Any unhandled exception here will pass through worker's boundary and will be properly reported.
             */
            task.run()
        }
    }

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        fun throwClosed(block: Runnable) {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule: $block")
        }

        if (!workerPool.allocate()) throwClosed(block) // Do not even try to send to avoid race

        tasksQueue.trySend(block).onClosed {
            throwClosed(block)
        }
    }

    override fun close() {
        val workers = workerPool.close()
        tasksQueue.close()
        /*
         * Here we cannot avoid waiting on `.result`, otherwise it will lead
         * to a native memory leak, including a pthread handle.
         */
        val requests = workers.map { it.requestTermination() }