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

Code before:
}

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)
    private val activeWorkers = atomic(0)

    private fun workerRunLoop() = runBlocking {
        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) {
        if (activeWorkers.value != workersCount) {
            tryAddWorker()
        }

        tasksQueue.trySend(block).onClosed {
            throw IllegalStateException("Dispatcher $name was closed, attempted to schedule : $block")
        }
    }

    private fun tryAddWorker() {
        activeWorkers.loop { curWorkers ->
            if (curWorkers >= workersCount) return
            if (activeWorkers.compareAndSet(curWorkers, curWorkers + 1)) {
                val worker = Worker.start(name = "$name-$curWorkers")
                worker.executeAfter { workerRunLoop() }
                workers[curWorkers].value = worker
                return
            }
        }
    }

    override fun close() {
        tasksQueue.close()
        /*
         * Here we cannot avoid not to wait on `.result`, otherwise it will lead
         * to a native memory leak, including pthread handle.
         */
        val requests = Array(workersCount) {
            workers[it].value?.requestTermination()
        }
        requests.map { it?.result }

Review comment (suggestion):
I would really like to provide a proper keep-alive here, but it doesn't seem adequately possible: we **must** terminate workers and we **must** await termination in a blocking way, otherwise native resources leak.

The preliminary decision is to get rid of workers and design something better (i.e. `Thread`) in K/N

Code after:
    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() }
        requests.map { it.result }
    }
}