thecodex.expert · The Codex Family of Knowledge
Kotlin

Coroutines and Flow

Async code that reads like synchronous code — Kotlin coroutines suspend instead of block, and structured concurrency ensures nothing leaks.

kotlinx.coroutines Structured concurrency Last verified:
Canonical Definition

Kotlin coroutines are suspendable computations — they can pause (suspend) without blocking a thread, and resume later on a thread pool; structured concurrency (CoroutineScope) ensures every coroutine has a parent that cancels it when done; Flow is a cold asynchronous stream that emits multiple values sequentially.

The central idea

Waiting for a network call on a thread wastes that thread — it sits idle, consuming memory, doing nothing. Coroutines solve this: when a coroutine hits an await, it suspends (frees the thread for other work), and resumes when the result arrives. The code reads sequentially, but the thread is never blocked.

suspend functions and launch/async

Kotlincoroutines.kt
import kotlinx.coroutines.*

// suspend: this function can be paused without blocking a thread
suspend fun fetchUser(id: Int): String {
    delay(100)   // non-blocking delay — suspends coroutine, thread is free
    return "User$id"
}

suspend fun fetchPosts(userId: String): List<String> {
    delay(200)
    return listOf("Post1 by $userId", "Post2 by $userId")
}

fun main() = runBlocking {   // creates a coroutine scope for main (blocks the thread)
    // launch: fire-and-forget coroutine
    val job = launch {
        repeat(3) { i ->
            println("Counting $i")
            delay(50)
        }
    }

    // async: coroutine that returns a value (Deferred<T>)
    val userDeferred = async { fetchUser(42) }
    val user = userDeferred.await()   // suspends until result is ready
    println("Got: $user")

    // Sequential vs concurrent
    // Sequential (total ~300ms):
    val u1 = fetchUser(1)
    val p1 = fetchPosts(u1)

    // Concurrent (total ~200ms — both launch at same time):
    val u2Deferred = async { fetchUser(2) }
    val p2Deferred = async { fetchPosts("User2") }
    val u2 = u2Deferred.await()
    val p2 = p2Deferred.await()
    println("$u2: $p2")

    job.join()   // wait for the counting job to finish
}

Structured concurrency: CoroutineScope

Every coroutine belongs to a CoroutineScope. When the scope is cancelled (e.g. a user navigates away from a screen), all child coroutines are cancelled automatically. This prevents coroutine leaks — the Kotlin equivalent of memory leaks, where background work continues even after the requesting component is gone.

Kotlinstructured_concurrency.kt
import kotlinx.coroutines.*

class UserViewModel {
    // coroutineScope tied to this ViewModel's lifecycle
    private val scope = CoroutineScope(Dispatchers.Main + SupervisorJob())

    fun loadUser(id: Int) {
        scope.launch {
            try {
                val user = withContext(Dispatchers.IO) {
                    fetchUser(id)   // switch to IO thread pool for network
                }
                // Back on Main dispatcher here — update UI
                println("Loaded: $user")
            } catch (e: CancellationException) {
                throw e   // always re-throw CancellationException
            } catch (e: Exception) {
                println("Error: ${e.message}")
            }
        }
    }

    // Called when ViewModel is destroyed — cancels all child coroutines
    fun onCleared() {
        scope.cancel()
    }
}

// Dispatchers:
// Dispatchers.Main   — UI thread (Android/JavaFX)
// Dispatchers.IO     — thread pool for blocking I/O (network, disk)
// Dispatchers.Default — CPU-bound work (heavy computation)
// Dispatchers.Unconfined — no thread affinity (rarely used)

Flow: asynchronous streams

Flow<T> is a cold asynchronous stream — it emits multiple values over time. Unlike a List which returns all values at once, a Flow emits them lazily. StateFlow and SharedFlow are hot flows that retain state and share emissions between multiple collectors.

Kotlinflow.kt
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// flow builder: emit values lazily
fun numbers(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(100)
        emit(i)   // emit one value at a time
    }
}

fun main() = runBlocking {
    // collect: terminal operator — consumes the flow
    numbers()
        .filter { it % 2 == 0 }        // keep even numbers
        .map { it * it }               // square them
        .collect { println(it) }       // 4, 16

    // StateFlow: always has a current value — like LiveData
    val counter = MutableStateFlow(0)
    launch {
        repeat(3) {
            delay(100)
            counter.value++
        }
    }
    counter
        .take(4)   // take 4 emissions then cancel
        .collect { println("Counter: $it") }   // 0, 1, 2, 3
}

Coroutine context and dispatchers

A CoroutineContext is a set of elements associated with a coroutine — the dispatcher, the job, the coroutine name, and exception handler. withContext(Dispatchers.IO) switches the dispatcher for the duration of the block without creating a new coroutine. This is the standard pattern for switching from the main thread to an IO thread for network or disk operations and back. SupervisorJob() prevents one child coroutine's failure from cancelling all siblings — useful in ViewModels where independent operations should not cancel each other.

Kotlincontext.kt
import kotlinx.coroutines.*

suspend fun loadData(): String = withContext(Dispatchers.IO) {
    // This block runs on the IO thread pool
    Thread.sleep(100)   // simulated blocking I/O
    "data from disk"
}   // Automatically resumes on the original dispatcher

fun main() = runBlocking {
    // coroutineContext: access the current context
    println(coroutineContext[Job])

    // CoroutineExceptionHandler: handle uncaught exceptions
    val handler = CoroutineExceptionHandler { _, exception ->
        println("Caught: $exception")
    }

    val scope = CoroutineScope(Dispatchers.Default + handler)

    // SupervisorJob: sibling failures are independent
    val supervisor = SupervisorJob()
    val childScope = CoroutineScope(Dispatchers.Default + supervisor)

    val child1 = childScope.launch { throw RuntimeException("child1 failed") }
    val child2 = childScope.launch {
        delay(200)
        println("child2 completed")   // still runs despite child1 failing
    }

    delay(300)
    println(loadData())
}
Commonly confused
Coroutines are NOT threads. Creating 100,000 coroutines is fine. Creating 100,000 threads will crash the JVM. Coroutines are multiplexed onto a thread pool (the dispatcher). Dispatchers.IO uses a pool of up to 64 threads (by default); thousands of coroutines can share those threads by suspending when idle. Coroutines switch between threads at suspension points — not via context-switching like OS threads.
Always re-throw CancellationException. CancellationException is how coroutine cancellation propagates. Catching it and not rethrowing breaks structured concurrency — the cancelled coroutine keeps running while its parent thinks it stopped. Correct pattern: catch (e: Exception) { if (e is CancellationException) throw e; handleError(e) } or catch only specific exceptions.
Flow is cold; SharedFlow/StateFlow are hot. A regular flow {} starts fresh for each collector — the producer code runs again per collector. SharedFlow and StateFlow are hot — one producer, multiple collectors share the same emissions. Use StateFlow for UI state (always has a current value); use SharedFlow for one-time events like navigation commands.

CPS transformation: how suspend is compiled

Kotlin's suspend keyword is implemented by the compiler using Continuation Passing Style (CPS) transformation. Every suspend function gets an implicit Continuation<T> parameter appended — the continuation is the "rest of the computation" to run after the suspend point. The function returns either the result directly (if it completed without suspending) or a special COROUTINE_SUSPENDED marker (if it suspended). The coroutine's local variables and state are stored in a generated state machine class that implements Continuation. This is entirely compile-time transformation — there is no coroutine runtime object on the heap beyond the small state machine. The Kotlin coroutines library provides the scheduler (dispatcher), while the compiler generates the state machines.

Flow internals: cold streams and backpressure

A Flow<T> is just a functional interface with one method: suspend fun collect(collector: FlowCollector<T>). The flow builder creates an object that implements this interface; when collected, it executes the lambda body. This cold model means the producer doesn't run until there is a collector — no wasted computation. Backpressure is handled naturally: the producer suspends at each emit call until the collector has consumed the value. buffer() adds a channel between producer and consumer to decouple their speeds. conflate() drops intermediate values if the collector is slower than the producer. collectLatest cancels the previous collector block when a new value arrives — ideal for debouncing search queries.

Specification reference

Kotlin coroutines documentation. kotlinlang.org/docs/coroutines-overview.html. Elizarov, R. (2018). Structured Concurrency. elizarov.medium.com/structured-concurrency-722d765aa952. kotlinx.coroutines library source: github.com/Kotlin/kotlinx.coroutines. Kotlin Flow documentation: kotlinlang.org/docs/flow.html.

Sources

1
Kotlin documentation. Coroutines overview. kotlinlang.org/docs/coroutines-overview.html.
2
Kotlin documentation. Asynchronous Flow. kotlinlang.org/docs/flow.html.
3
Elizarov, R. (2018). Structured Concurrency. elizarov.medium.com. — Design rationale for structured concurrency.
4
kotlinx.coroutines. github.com/Kotlin/kotlinx.coroutines. — Official library source and documentation.
5
Jemerov, D. & Isakova, S. (2017). Kotlin in Action. Manning. — Chapter 15: coroutines.
Source confidence: High Last verified: Primary source: kotlinlang.org/docs/coroutines-overview.html