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?
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!