Writing concurrent or asynchronous code can be quite difficult, reading it even more. But does it have to be that way? Let me introduce you to coroutines, a new and still experimental feature which simplifies your concurrent code by providing you with a nice syntax and high level abstractions that prevent misuse.
In this series of articles I want to provide you with some in depth knowledge about coroutines and how their fancy high level abstractions work. Some of the concepts might be a bit difficult to grasp at the start, which is completely normal, and you will eventually have a moment, where your mind is blown and it all becomes clear to you.
All the code snippets in this post are executable and can even be modified. So feel free to experiment around while reading through the article. The snippets utilize the embeddable Kotlin playground, which sadly does not support streaming output. Therefore running the snippets will not continuously display their respective output. Instead, you would wait for the full runtime of the snippet and then see the full output of the code at once.
Experimental status
Coroutines have existed in Kotlin since version 1.1 and are still experimental in 1.2.*. But don’t let that “experimental” scare you. Kotlin’s backwards compatibility guarantee also holds for them and by now they became pretty stable in terms of feature additions.
What is a coroutine?
A coroutine is a suspendable (a.k.a. resumable) function. For this to work the function needs to remember where it was suspended, so it can later be resumed at that point. Also suspending is something that coroutines actively do. That means a currently executed coroutine might figure that it needs to wait for something to be available. In that case it not only suspends itself, but also tells someone else when it wants to be resumed. This means we usually need some kind of runtime environment which takes care of resuming currently suspended coroutines when their time has come.
A very simple coroutine might look like this:
//sampleStart
suspend fun foo() {
println("Hello!")
delay(2000) // suspend for 2 seconds
println("Good bye!")
}
//sampleEnd
fun main(args: Array<String>) {
runBlocking {
foo()
}
}
The first thing we notice is the suspend keyword. This makes the function a coroutine. Coroutines cannot be called directly by normal functions. Why is that? When a function calls another, it expects for the function to return a result, when it is done. However, a coroutine might not get executed to the end. You could even discard a suspended coroutine without ever running it to the end. So we actually need some code that would take care of running the coroutine and also handling the different states it might get into.
Other than that we might also notice, that despite the function being suspended at the delay (which means some other code might get executed during that time on this thread or another one), it reads like synchronous code. We don’t need to deal with promises and later explicitely waiting for them, nor do we have to care for manually switching execution contexts. Instead, we have nice syntactical sugar and a comfortable supporting library (kotlinx.coroutines) with high level abstractions.
For the rest of this and also for future posts, we will actually have a look at the inner workings to get a better understanding for them, as well as on the high level abstractions provided for us.
Running a coroutine
In its simplest form the runtime environment could look like this:
(You may also hit the plus above the snippet to see the rest of it.)
// when coroutines are no longer experimental they will move to kotlin.coroutines.*
import kotlin.coroutines.experimental.Continuation
import kotlin.coroutines.experimental.CoroutineContext
import kotlin.coroutines.experimental.EmptyCoroutineContext
import kotlin.coroutines.experimental.startCoroutine
//sampleStart
// a very simple coroutine which does not suspend
suspend fun sayHello() {
println("Hello world!")
}
// we need a coroutine lambda to get started
fun <R> runBlocking(block: suspend () -> R): R {
var result: R? = null
// startCoroutine is an extension function on coroutines
// it creates the coroutine state for us and directly executes the coroutine
block.startCoroutine(completion = object : Continuation<R> {
// we don't do anything with the context yet, so we can just use the provided empty context
override val context: CoroutineContext = EmptyCoroutineContext
// this is called after the lambda finished execution
// that is, after it really reached the end
// if the lambda is suspended and never resumed, this function never gets called
override fun resume(value: R) {
result = value
}
// this function is called whenever an uncaught exception leaves the scope of the lambda
override fun resumeWithException(exception: Throwable) {
// we wrap the exception from the coroutine with our own, so it is later possible to determine where the coroutine was called.
throw RuntimeException("Coroutine resumed exceptionally!", exception)
}
})
// at this point we might not have a value as the coroutine might not have finished yet
return result!!
}
//sampleEnd
fun main(args: Array<String>) {
val answer = runBlocking {
// inside this block, which already is a coroutine in form of a lambda, we can call other coroutines directly
sayHello()
42 // this is the result we see in the resume function above
}
println("The answer is $answer.")
}
Now this extremely simple example only works because neither of the two coroutines (the lambda and the sayHello function) actually suspends. So our coroutines run perfectly synchronously, which means, when startCoroutine returns, we already have our result.
Resuming coroutines
For suspension to work we need several things:
- Some kind of queue where we can store currently suspended coroutines.
- Some runner which would execute said coroutines until all of them are finished.
Now how do we do this? A coroutine can suspend itself via the suspendCoroutine function. When a coroutine is suspended, we get a so called continuation, which is the current state of the coroutine. We can call resume on it, to continue the execution of the coroutine, or resumeExceptionally to inject an exception. To have our coroutine resumed at a later point in time we need to register that continuation in the queue mentioned above, so the runtime environment can resume it later on.
Let’s have a look at a very basic implementation for this:
import kotlin.coroutines.experimental.Continuation
import kotlin.coroutines.experimental.CoroutineContext
import kotlin.coroutines.experimental.createCoroutine
import kotlin.coroutines.experimental.suspendCoroutine
//sampleStart
// this is our runtime environment
// things we want to put in the coroutine's context need to inherit from CoroutineContext.Element
// also such an Element is already a CoroutineContext
// so try not to get confused with that...
class CoroutineRunner : CoroutineContext.Element {
// We have our companion object be the context key, so we can use the class name directly as the key when accessing the context
companion object : CoroutineContext.Key<CoroutineRunner>
// this is required by Element
// also keys in the coroutine context are compared by reference, not by value
// that's why every key needs to have a unique reference
// our companion object fulfills this trait as objects in Kotlin are singletons
override val key: CoroutineContext.Key<*> = CoroutineRunner
// here we store suspended coroutines
val coroutines = mutableSetOf<Continuation<Unit>>()
// this function runs suspended coroutines until no suspended coroutines are left
fun run() {
while (coroutines.isNotEmpty()) {
val current = coroutines.first()
println("==> Resume!")
current.resume(Unit)
coroutines.remove(current)
}
println("==> Finish!")
}
}
suspend fun yield() {
suspendCoroutine<Unit> { continuation ->
// calling yield will suspend the coroutine until the runtime environment reaches the continuation again
continuation.context[CoroutineRunner]!!.coroutines.add(continuation)
println("==> Suspend!")
}
}
suspend fun foo() {
println("Hello!")
yield()
println("Good bye!")
}
//sampleEnd
// runBlocking now does some more :)
fun <R> runBlocking(block: suspend () -> R): R {
var result: R? = null
val runner = CoroutineRunner()
// this time we use the extension function createCoroutine, which only create the coroutine state and returns a continuation
// so the created coroutine is initially suspended
// also we add this suspended coroutine to our runtime, so when we call the run function it will start that coroutine
runner.coroutines.add(block.createCoroutine(completion = object : Continuation<R> {
override val context: CoroutineContext = runner
override fun resume(value: R) {
result = value
}
override fun resumeWithException(exception: Throwable) {
throw RuntimeException("foo", exception)
}
}))
// do the thing
runner.run()
return result!!
}
fun main(args: Array<String>) {
val answer = runBlocking {
foo()
42
}
println("The answer is $answer.")
}
When you run the example, the printlns should make it clear, at which points the coroutines are suspended and when they are resumed.
Conditional execution
To close this article let’s get back to the starting example. To implement the delay function we would need to support some kind of conditional execution. This means, we want to be able to suspend our coroutine and tell the runtime, that we want to be resumed, when some condition becomes true. In the case of delay we want to be resumed, after the given time has elapsed. Other use cases would be waiting for data to be available from some network communication, or reacting to some external signals.
import java.lang.System.currentTimeMillis
import kotlin.coroutines.experimental.Continuation
import kotlin.coroutines.experimental.CoroutineContext
import kotlin.coroutines.experimental.createCoroutine
import kotlin.coroutines.experimental.suspendCoroutine
//sampleStart
// our example from the start
suspend fun foo() {
println("Hello!")
delay(5) // okay, I went down to 5 milliseconds as the println in delay otherwise really spams the output
println("Good bye!")
}
class CoroutineRunner : CoroutineContext.Element {
companion object : CoroutineContext.Key<CoroutineRunner>
override val key: CoroutineContext.Key<*> = CoroutineRunner
// this time we also have a function determining if the coroutine should be resumed
val coroutines = mutableMapOf<Continuation<Unit>, () -> Boolean>()
// the default is just a comfort thing so adding a coroutine without a condition directly resumes it on the next cycle
fun addCoroutine(continuation: Continuation<Unit>, condition: () -> Boolean = { true }) {
coroutines[continuation] = condition
}
fun run() {
while (coroutines.isNotEmpty()) {
// first we filter for coroutines whose condition is fulfilled
// as a new list is returned we can then safely remove the executed coroutines from the map
coroutines
.filter { (_, condition) -> condition() }
.forEach { (continuation, _) ->
println("==> Resume!")
coroutines.remove(continuation)
continuation.resume(Unit)
}
}
println("==> Finish!")
}
}
suspend fun delay(timeout: Int) {
suspendCoroutine<Unit> { continuation ->
// when calling delay we calculated when the coroutine should be resumed
// this value is then captured by the condition lambda
val resumeAfter = currentTimeMillis() + timeout
continuation.context[CoroutineRunner]!!.addCoroutine(continuation) {
val now = currentTimeMillis()
// if the time is right, we tell the runtime to resume the coroutine
if (now < resumeAfter) {
println("==> Skip! (${resumeAfter - now}ms to go...)")
false
} else true
}
println("==> Suspend!")
}
}
//sampleEnd
fun <R> runBlocking(block: suspend () -> R): R {
var result: R? = null
val runner = CoroutineRunner()
runner.addCoroutine(block.createCoroutine(completion = object : Continuation<R> {
override val context: CoroutineContext = runner
override fun resume(value: R) {
result = value
}
override fun resumeWithException(exception: Throwable) {
throw RuntimeException("foo", exception)
}
}))
runner.run()
return result!!
}
fun main(args: Array<String>) {
val answer = runBlocking {
foo()
42
}
println("The answer is $answer.")
}
Now this implementation is not really resource friendly as we basically do busy waiting, i.e. we are constantly running code checking whether we can continue any of our coroutines. This could be improved by registering for certain events and then having the program sleep until one of the events happens and the associated coroutine can be resumed. We will have a look into this in follow up posts, so stay tuned.
See also
