Coroutines

In the previous lesson a sequence carried the recipe for a computation, and the computation ran when the terminal operation arrived. Some computations, though, wait for another job to finish: reading a file, a network response, a timer. During that wait the caller should not be held up. Kotlin offers the coroutine for this situation: a job can pause midway, release control and then continue where it left off.

Code that uses coroutines imports the coroutines library, kotlinx.coroutines. This library is not in the language itself but in a separate package; in this lesson and the next the line import kotlinx.coroutines.* is needed, which the earlier lessons did not need. Three constructs are enough for this lesson. A suspending function marked with suspend is a function that can pause in its body, and it is called only from a coroutine or from another suspending function. launch starts a new coroutine and returns a Job; the result is not awaited, only its completion is awaited with join(). async starts a new coroutine and returns a Deferred that carries the result; the result is taken with await(). All three must run inside a coroutine scope, which in this lesson runBlocking builds: it holds the thread that called the main function until all the coroutines inside it have finished.

A coroutine is not a thread. A thread is a unit of execution that the operating system schedules; a coroutine is a job that the library runs on a dispatcher (Dispatchers), and it can run in turn on the same thread. In this lesson no measurement depends on time or on a thread name: timing is not read from the output, only the order and the results of the jobs. The measuring question stays as in the earlier lessons: what makes these two values equal?

// Criterion.kt — classifies the criterion that makes two values equal
class Criterion(val label: String, first: Any?, twin: Any?, other: Any?) {
    val sameObject = first === twin
    val equalTwin = first == twin
    val equalOther = first == other

    val name: String = when {
        sameObject -> "singleton"
        !equalTwin -> "identity"
        !equalOther -> "content"
        else -> "partial-content"
    }

    fun report() {
        println("$label: same-object=$sameObject, twin=$equalTwin, other=$equalOther -> $name")
    }
}

fun summarize(checks: List<Criterion>) {
    for (name in listOf("singleton", "identity", "content", "partial-content")) {
        println("$name: ${checks.count { it.name == name }}")
    }
}

Order: launch, join, async, await

The block below defines a suspending function named doubled; inside it the yield() call sets up a point where the coroutine can hand the turn to another. main measures three things. First, whether the launch block runs before or after the line that follows the launch call: both add a record to the same list. Second, collecting the results of two async jobs with await(). Third, taking in order the results of three jobs started inside coroutineScope.

// Main.kt — launch, async and await
import kotlinx.coroutines.*

suspend fun doubled(n: Int): Int {
    yield()
    return n * 2
}

fun main() = runBlocking {
    val log = mutableListOf<String>()
    val job = launch { log.add("launch") }
    log.add("main")
    job.join()
    println(log)
    val first = async { doubled(21) }
    val second = async { doubled(4) }
    println(first.await() + second.await())
    val total = coroutineScope {
        val parts = listOf(1, 2, 3).map { async { doubled(it) } }
        parts.map { it.await() }
    }
    println(total)
}

[main, launch]
50
[2, 4, 6]

The first line shows an important result: the launch block did not run as soon as the launch call returned. The main record was added first, and the launch record came after it. Because runBlocking works on a single thread, the new coroutine waits in the queue until the coroutine that started it reaches a suspension point; here that point is the job.join() call. That the order comes out this way depends on the dispatcher; so the lesson counts as guaranteed not this order but that both records are in the list after the join() call.

In the second line the two async jobs produce the results 21 * 2 and 4 * 2; await() takes them and the total comes out 50. In the third line coroutineScope does not return until all the jobs started inside it are finished; the results of the three asyncs are taken with await() in the list in the order they were started: [2, 4, 6]. The order of the results follows the starting order, because the parts list makes the await() calls in that order. Which of the jobs finished first cannot be read from this output, and no attempt is made to read it.

CC20. These measurements are for a runBlocking scope and a single Int result type. The order of the launch and main records on another dispatcher, the order in a function with no suspension point, and the real finishing order among the jobs were not measured.

The Criterion: Identity

async returns not a result but the promise of a result. This promise is a Deferred object, and each async call builds a new one. The block below compares two separate async { 1000 } calls with a third whose value differs. The second measurement asks the same question with a constant small value. In both, the blocks do not call await(), to see that the question concerns a Deferred, not a result.

// Main.kt — two Deferreds
import kotlinx.coroutines.*

fun main() = runBlocking {
    val checks = listOf(
        Criterion("async-1000", async { 1000 }, async { 1000 }, async { 2000 }),
        Criterion("async-1", async { 1 }, async { 1 }, async { 2 }),
    )
    for (check in checks) check.report()
    summarize(checks)
}

async-1000: same-object=false, twin=false, other=false -> identity
async-1: same-object=false, twin=false, other=false -> identity
singleton: 0
identity: 2
content: 0
partial-content: 0

CC21. These measurements are for Deferred objects built with async. The equality of Job objects built with launch and of promises built directly, such as CompletableDeferred, was not measured.

The Bounding Measurement

Taking the result changes the equality. The block below measures three things side by side: the Deferred objects themselves, their results taken with await(), and two CoroutineName("a") elements that can be put in the context of the same coroutine. 1000 was chosen for the results; small integers can be shared during boxing and could come out singleton, and for 1000 the lesson does not depend on this detail, the result being whatever the sandbox prints.

// Main.kt — result and context element
import kotlinx.coroutines.*

fun main() = runBlocking {
    val a = async { 1000 }
    val b = async { 1000 }
    val c = async { 2000 }
    val checks = listOf(
        Criterion("deferred", a, b, c),
        Criterion("result", a.await(), b.await(), c.await()),
        Criterion("name", CoroutineName("a"), CoroutineName("a"), CoroutineName("b")),
    )
    for (check in checks) check.report()
    summarize(checks)
}

deferred: same-object=false, twin=false, other=false -> identity
result: same-object=false, twin=true, other=false -> content
name: same-object=false, twin=true, other=false -> content
singleton: 0
identity: 1
content: 2
partial-content: 0

The bounding measurement is this: equality is decided not by the job but by the type that carries the job’s result. A Deferred is compared by identity; the Int taken with await() is compared by content. The same jobs a and b were measured once as objects and once as results, and the criterion turned from identity to content. CoroutineName is also compared by content: two elements that carry the same name are equal, and one with a different name is not. This shows that an element in a coroutine context carries a value, not an identity.

The practical consequence: if two jobs have to be compared, the results should be compared, not the Deferreds; for that await() must be called first. That call too can be made only inside a coroutine; await() is a suspending function.

CC22. These measurements are for Int results and CoroutineName elements. The equality of other result types, of Dispatchers elements, and of the combination of several context elements was not measured.

Lazy Start and Cancellation

async and launch by default join the queue at once. If start = CoroutineStart.LAZY is given, the coroutine is built but not started: its body does not run at all until await(), join() or start() is called. This resembles the idea of the sequence in the previous lesson, but here what is lazy is not the result of a computation but the start of a job. Cancellation can end a job that has not started without it running at all. The block below first builds a lazy async and looks at it and at the list before and after await(). Then it builds a lazy launch, cancels it and only after that waits. The orders are read only from the points that the await() and join() calls guarantee.

// Main.kt — lazy start and cancellation
import kotlinx.coroutines.*

fun main() = runBlocking {
    val log = mutableListOf<String>()
    val lazy = async(start = CoroutineStart.LAZY) { log.add("body"); 7 }
    println(log.size)
    println(lazy.isActive)
    println(lazy.await())
    println(log)
    println(lazy.isCompleted)
    val cancelled = launch(start = CoroutineStart.LAZY) { log.add("never-runs") }
    cancelled.cancel()
    cancelled.join()
    println(cancelled.isCancelled)
    println(log)
}

0
false
7
[body]
true
true
[body]

When the lazy async was built the list was empty and the job was not active: the body did not run at all until the await() call. await() started the body, gave the result 7, and the body record dropped into the list. After that the job counted as completed. The lazy launch, because it was cancelled, never entered its body: the never-runs record is not in the last line, but the job was marked isCancelled. The result is this: that a coroutine has been started and that it has been finished are different things, and Job lets you query this state. Cancelling a job whose result will never be used can also prevent the side effects of its body.

CC23. These measurements are for jobs built with CoroutineStart.LAZY in a runBlocking scope. The effect of cancellation in the middle of a running job and the order of starting with start() were not measured.

Waiting for All the Children

When several launches are started, it is not necessary to wait for each one’s completion separately. A coroutineScope { } block does not return until all the jobs started inside it are finished; joinAll() waits for all of those in a given list of Jobs. The guarantee of both ways is the same: after the return, the side effects of the jobs are visible. But the order among the jobs is not part of this guarantee. So the block below measures not the order but the result: after the return it sorts the list and counts how many elements it holds.

// Main.kt — coroutineScope and joinAll
import kotlinx.coroutines.*

fun main() = runBlocking {
    val log = mutableListOf<Int>()
    coroutineScope {
        for (i in 1..3) {
            launch {
                yield()
                log.add(i)
            }
        }
    }
    println(log.sorted())
    val jobs = List(4) { i ->
        launch {
            yield()
            log.add(i * 10)
        }
    }
    jobs.joinAll()
    println(log.size)
    println(jobs.all { it.isCompleted })
}

[1, 2, 3]
7
true

When the coroutineScope block returned, all three records were in the list. The measurement printed the sorted list, because the order in which the jobs finish is not guaranteed. In the second group four jobs were awaited with joinAll() and the list held seven elements in all; all of them appeared completed. The distinction between guarantee and order matters in practice: a scope or join is used to wait for a result, and logic that depends on the finishing order is built wrongly.

CC24. These measurements are for launch jobs and an Int list. The behavior of the scope and the sibling jobs when a child job throws an exception was not measured in this lesson.

Context and Name

Every coroutine carries a context, and context elements are read by key. coroutineContext[CoroutineName] gives the name if there is one, and null if not. Inside a withContext(CoroutineName("worker")) { } block the context carries the new name, and when the block ends the old context returns. If a context element is given to an async or launch call, the child job carries it. The block below defines a small suspending function that reads the name and calls it from four places. The name is not a thread name; it is only a label put in the context, which is why it can be measured.

// Main.kt — reading the name from the context
import kotlinx.coroutines.*
import kotlin.coroutines.coroutineContext

suspend fun currentName(): String? = coroutineContext[CoroutineName]?.name

fun main() = runBlocking {
    println(currentName())
    println(withContext(CoroutineName("worker")) { currentName() })
    println(currentName())
    val named = async(CoroutineName("child")) { currentName() }
    println(named.await())
}

null
worker
null
child

In the first call there was no name in the runBlocking context and the result is null. In the withContext block the name became worker. After the block ended the name is null again: the context change was specific to the block. In the last line the child async read its own name, child. This is why the context is a clean way to carry information between coroutines: without passing a parameter, but with a scope that is defined in a specific way.

CC25. These measurements are for a CoroutineName element and a runBlocking scope. Reading a Job element or a Dispatchers element from the context, a child inheriting elements from the parent context, and combining two elements with + were not measured.

Job States and Repeated await

Every launch and async carries a Job, and the Job says at what stage the work is with three properties: isActive shows that it is in progress, isCompleted that it has finished, isCancelled that it has been cancelled. To say that a job has finished is guaranteed after join() or await() returns; the intermediate states between starting and the first suspension point depend on the dispatcher and are not measured in this lesson. await() can also be called more than once: the result is computed once and later calls give the same value. The block below measures first the state after join(), then two await() calls on the same Deferred and how many times the body ran.

// Main.kt — job states and repeated await
import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch { yield() }
    job.join()
    println(job.isActive)
    println(job.isCompleted)
    println(job.isCancelled)
    var runs = 0
    val deferred = async { runs++; "result" }
    println(deferred.await())
    println(deferred.await())
    println(runs)
}

false
true
false
result
result
1

After join() returned, the job was no longer active, it was completed and it had not been cancelled. The Deferred was awaited twice and gave the same value both times; the counter rose once, so the body ran only once. await() does not restart the job, it reads the result. This also means that once the result of a Deferred is computed it is stored, and every await() reads that stored value. This is the opposite of the flow case: a cold flow runs again at each collecting, while a Deferred shares the result of a single computation.

CC26. These measurements are for the states after join() and await(). The isActive value between starting and the first suspension, and the state of a job that ends with an error, were not measured.

Summary

• A function marked suspend is called only from a coroutine or from another suspending function; launch starts a job without a result, async starts a job with a result.
• A launch call does not run at once; inside runBlocking it waits until the coroutine that started it reaches a suspension point. join() and await() guarantee that the job has finished.
• The Deferred objects of two separate async calls are not equal. The criterion of this lesson: identity.
• The bounding measurement: the await() results of the same jobs give the content criterion; CoroutineName elements are also compared by content.

Next Step

This lesson dealt with jobs that produce a single value. A job can instead let values flow over time. The next lesson builds this with flows and asks the equality question there too: are two flows that emit the same values equal, and how are equal values that arrive one after another handled in a flow?