What is the relationship between non-blocking I/O and Kotlin coroutines?

Question

What is the relationship between Kotlin coroutines and non-blocking I/O? Does one imply the other? What happens if I use blocking I/O? How does this affect performance?

Answer 1

There are two aspects of it:

• If you use a non-blocking server to expose your API More details here: https://blog.allegro.tech/2020/02/webflux-and-coroutines.html or here: https://medium.com/@akarsh7791/non-blocking-i-o-with-netty-32ef20ab4b79

• If you use non blocking http clients or database drivers in the code that processes those requests (WebClient, apache async http client etc).

Doing just one of those things (eg coroutines+webflux+netty but blocking database driver under the hoods) beats the purpose

And one question from me: Have you perhaps tried KTOR with other non blocking engine, like WebClient? We've had issues with CIO when DNS resolves an adress to multiple IPs.

Answer 2

Coroutines are designed to contain non-blocking (i.e. CPU-bound) code. This is why the default coroutine dispatcher – Dispatchers.Default – has a total of max(2, num_of_cpus) threads to execute dispatched coroutines. For example, by default a highly concurrent program such as a web server running in a computer with 2 CPUs would have its compute capacity degraded by 50% while a thread blocks waiting on I/O to complete in a coroutine.

Non-blocking I/O is not a feature of coroutines though. Coroutines simply provide an easier programming model consisting of suspending functions instead of hard-to-read CompletableFuture<T> continuations in Java, and structured concurrency among other concepts.

---

To understand how coroutines and non-blocking I/O work together, here's a practical example:

server.js: A simple Node.js HTTP server that receives a request, and returns a response ~5s after.

const { createServer } = require("http");

let reqCount = 0;
const server = createServer(async (req, res) => {
    const { method, url } = req;
    const reqNumber = ++reqCount;
    console.log(`${new Date().toISOString()} [${reqNumber}] ${method} ${url}`);
    
    await new Promise((resolve) => setTimeout(resolve, 5000));
    res.end("Hello!\n");
    console.log(`${new Date().toISOString()} [${reqNumber}] done!`);
});

server.listen(8080);
console.log("Server started!");

main.kt: Sends 128 HTTP requests to the Node.js server using three implementations:

1. withJdkClientBlocking(): Invokes JDK11 java.net.http.HttpClient's blocking I/O methods inside of a coroutine dispatched by Dispatchers.IO.

import java.net.URI
import java.net.http.HttpClient as JDK11HttpClient
import java.net.http.HttpRequest as JDK11HttpRequest
import java.net.http.HttpResponse as JDK11HttpResponse
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext

fun withJdkClientBlocking() {
    println("Running with JDK11 client using blocking send()")

    val client = JDK11HttpClient.newHttpClient()
    runExample {
        // Sometimes you can't avoid coroutines with blocking I/O methods.
        // These must be always be dispatched by Dispatchers.IO.
        withContext(Dispatchers.IO) {
            // Kotlin compiler warns this is a blocking I/O method.
            val response = client.send(
                JDK11HttpRequest.newBuilder(URI("http://localhost:8080")).build(),
                JDK11HttpResponse.BodyHandlers.ofString()
            )
            // Return status code.
            response.statusCode()
        }
    }
}

2. withJdkClientNonBlocking(): Invokes JDK11 java.net.HttpClient non-blocking I/O methods. These methods return a CompletableFuture<T> whose results are consumed using CompletionStage<T>.await() interoperability extension function from kotlinx-coroutines-jdk8. Even though the I/O does not block any thread, the asynchronous request/response marshalling/unmarshalling runs on a Java Executor, so the example uses a single-threaded executor to illustrate how a single thread can handle many concurrent requests due to the non-blocking I/O.

import java.net.URI
import java.net.http.HttpClient as JDK11HttpClient
import java.net.http.HttpRequest as JDK11HttpRequest
import java.net.http.HttpResponse as JDK11HttpResponse
import java.util.concurrent.Executors
import kotlinx.coroutines.future.await

fun withJdkClientNonBlocking() {
    println("Running with JDK11 client using non-blocking sendAsync()")

    val httpExecutor = Executors.newSingleThreadExecutor()
    val client = JDK11HttpClient.newBuilder().executor(httpExecutor).build()
    try {
        runExample {
            // We use `.await()` for interoperability with `CompletableFuture`.
            val response = client.sendAsync(
                JDK11HttpRequest.newBuilder(URI("http://localhost:8080")).build(),
                JDK11HttpResponse.BodyHandlers.ofString()
            ).await()
            // Return status code.
            response.statusCode()
        }
    } finally {
        httpExecutor.shutdown()
    }
}

3. withKtorHttpClient() Uses Ktor, a non-blocking I/O HTTP client written with Kotlin and coroutines.

import io.ktor.client.engine.cio.CIO
import io.ktor.client.HttpClient as KtorClient
import io.ktor.client.request.get
import io.ktor.client.statement.HttpResponse as KtorHttpResponse

fun withKtorHttpClient() {
    println("Running with Ktor client")

    // Non-blocking I/O does not imply unlimited connections to a host.
    // You are still limited by the number of ephemeral ports (an other limits like file descriptors).
    // With no configurable thread limit, you can configure the max number of connections.
    // Note that HTTP/2 allows concurrent requests with a single connection.
    KtorClient(CIO) { engine { maxConnectionsCount = 128 } }.use { client ->
        runExample {
            // KtorClient.get() is a suspend fun, so suspension is implicit here
            val response = client.get<KtorHttpResponse>("http://localhost:8080")
            // Return status code.
            response.status.value
        }
    }
}

Putting it all together:

import kotlin.system.measureTimeMillis
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking

fun runExample(block: suspend () -> Int) {
    var successCount = 0
    var failCount = 0

    Executors.newSingleThreadExecutor().asCoroutineDispatcher().use { dispatcher ->
        measureTimeMillis {
            runBlocking(dispatcher) {
                val responses = mutableListOf<Deferred<Int>>()
                repeat(128) { responses += async { block() } }
                responses.awaitAll().forEach {
                    if (it in 200..399) {
                        ++successCount
                    } else {
                        ++failCount
                    }
                }
            }
        }.also {
            println("Successfully sent ${success + fail} requests in ${it}ms: $successCount were successful and $failCount failed.")
        }
    }
}

fun main() {
    withJdkClientBlocking()
    withJdkClientNonBlocking()
    withKtorHttpClient()
}

---

Example run:

main.kt (with # comments for clarification)

# There were ~6,454ms of overhead in this execution
Running with JDK11 client using blocking send()
Successfully sent 128 requests in 16454ms: 128 were successful and 0 failed.

# There were ~203ms of overhead in this execution
Running with JDK11 client using non-blocking sendAsync()
Successfully sent 128 requests in 5203ms: 128 were successful and 0 failed.

# There were ~862ms of overhead in this execution
Running with Ktor client
Successfully sent 128 requests in 5862ms: 128 were successful and 0 failed.

server.js (with # comments for clarification)

# These are the requests from JDK11's HttpClient blocking I/O.
# Notice how we only receive 64 requests at a time.
# This is because Dispatchers.IO has a limit of 64 threads by default, so main.kt can't send anymore requests until those are done and the Dispatchers.IO threads are released.
2022-07-24T17:59:29.107Z [1] GET /
(...)
2022-07-24T17:59:29.218Z [64] GET /
2022-07-24T17:59:34.124Z [1] done!
(...)
2022-07-24T17:59:34.219Z [64] done!
2022-07-24T17:59:35.618Z [65] GET /
(...)
2022-07-24T17:59:35.653Z [128] GET /
2022-07-24T17:59:40.624Z [65] done!
(...)
2022-07-24T17:59:40.655Z [128] done!

# These are the requests from JDK11's HttpClient non-blocking I/O.
# Notice how we receive all 128 requests at once.
2022-07-24T17:59:41.163Z [129] GET /
(...)
2022-07-24T17:59:41.257Z [256] GET /
2022-07-24T17:59:46.170Z [129] done!
(...)
2022-07-24T17:59:46.276Z [256] done!

# These are there requests from Ktor's HTTP client non-blocking I/O.
# Notice how we also receive all 128 requests at once.
2022-07-24T17:59:46.869Z [257] GET /
(...)
2022-07-24T17:59:46.918Z [384] GET /
2022-07-24T17:59:51.874Z [257] done!
(...)
2022-07-24T17:59:51.921Z [384] done!