Run expensive / UI coroutines in parallel

Issue

Given the following code example:

@ExperimentalCoroutinesApi
fun main() {
    val inputChannel = Channel<Unit>()
    val outputChannel = Channel<String>()
    val engine = Engine(outputChannel)
    val calculator = Logger()

    runBlocking(Dispatchers.Default) {
        launch {
            engine.connect(inputChannel)
        }

        launch {
            calculator.connect(outputChannel)
        }

        inputChannel.send(Unit)
        inputChannel.send(Unit)
    }
}

class Engine(private val logger: SendChannel<String>) {

    private val pool = Executors.newCachedThreadPool().asCoroutineDispatcher()

    @ExperimentalCoroutinesApi
    suspend fun connect(input: ReceiveChannel<Unit>) {
        input.consumeEach {
            println("${Instant.now()} [${Thread.currentThread().name}] Engine - Received input")
            // What to do here so `calc()` runs in parallel, not blocking the channel?
            calc()
        }
    }

    suspend fun calc() {
        logger.send("Starting processing")
        for (i in 1..100) {
            delay(100)
            print(".")
        }
        println()
        logger.send("Finished processing")
    }
}

class Logger {

    @ExperimentalCoroutinesApi
    suspend fun connect(channel: ReceiveChannel<String>) {
        channel.consumeEach {
            println("${Instant.now()} [${Thread.currentThread().name}] Logger - $it")
        }
    }
}

The log output shows

2021-11-14T11:50:21.021357700Z [main] Engine - Received input
2021-11-14T11:50:21.025358900Z [main] Logger - Starting processing
....................................................................................................
2021-11-14T11:50:31.829252900Z [main] Engine - Received input
2021-11-14T11:50:31.829252900Z [main] Logger - Finished processing
2021-11-14T11:50:31.829252900Z [main] Logger - Starting processing
....................................................................................................
2021-11-14T11:50:42.686953100Z [main] Logger - Finished processing

that all things are happening on the same thread. However, I would like Engine to be able to calculate inputs in parallel (this is just a MWE in place of a much larger and more complex application that requires things to run in parallel).

What to I need to change in this code so that messages to inputChannel will spawn multiple ... "scopes" (?) in which calculations can happen and that can send messages to outputChannel during processing?

Solution

By using runBlocking() like this you start a single-threaded dispatcher and all child coroutines are dispatched on this single thread. If you need to perform mostly CPU-intensive calculations then use Dispatchers.Default instead:

runBlocking(Dispatchers.Default) { ... }

Then, if you need to perform blocking IO inside this code, enclose it in:

withContext(Dispatchers.IO) { ... }

Answered By - broot