Kotlin § concurrency

Coroutines

Kotlin’s coroutines are lightweight, suspendable computations — substantially cheaper than threads, supporting structured concurrency, cancellation, and substantial composition. The principal mechanisms: suspend functions — admit suspension at suspension points (typically await-style operations); coroutine builders (launch, async, runBlocking) for starting coroutines; CoroutineScope for structured concurrency; Dispatchers (Main, IO, Default) for thread-pool selection; Flow for asynchronous streams; Channel for hot streams. The substantial integration with the JVM’s existing concurrency primitives admits substantial interop. The combination — suspend functions, coroutine builders, structured concurrency via scopes, dispatchers for thread management, Flow for async streams, channels for producer-consumer patterns — is the substance of Kotlin’s concurrency model.

`suspend` functions

A function marked suspend admits suspension:

suspend fun fetchData(url: String): Data {
    delay(1000)                                    // suspending — non-blocking
    return Data.fromUrl(url)
}

The suspend admits the function being paused and resumed; admit calling other suspend functions and substantial async APIs.

suspend functions can only be called from:

Other suspend functions.
A coroutine builder (launch, async, runBlocking).

Coroutine builders

`launch`

For fire-and-forget coroutines:

import kotlinx.coroutines.*

runBlocking {
    val job = launch {
        delay(1000)
        println("World!")
    }

    println("Hello,")
    job.join()                                     // wait for completion
}
// Output:
// Hello,
// World!

The launch returns a Job — admit cancellation, joining, and substantial introspection.

`async`

For coroutines returning a value:

runBlocking {
    val a = async { fetchA() }
    val b = async { fetchB() }
    val c = async { fetchC() }

    val result = a.await() + b.await() + c.await()
    println(result)
}

The async returns a Deferred<T> — await() retrieves the result.

For concurrent execution, async admits substantial parallelism — the three fetches run simultaneously.

`runBlocking`

For blocking the calling thread until coroutines complete:

fun main() = runBlocking {
    val data = fetchData()
    process(data)
}

The runBlocking is conventional only at:

main() of executables.
Tests.
Top-level invocations into coroutine code.

Structured concurrency

CoroutineScope admits structured concurrency — child coroutines are tied to a parent scope:

suspend fun loadData(): Result = coroutineScope {
    val a = async { fetchA() }
    val b = async { fetchB() }

    Result(a.await(), b.await())
    // If either fetch fails, the other is cancelled
    // The scope completes only when all children complete
}

The coroutineScope admits “all children must complete; on failure, cancel siblings”.

For supervisor scope (one child failing doesn’t cancel siblings):

suspend fun loadAll() = supervisorScope {
    val a = async { fetchA() }
    val b = async { fetchB() }
    val c = async { fetchC() }

    val results = listOf(
        runCatching { a.await() }.getOrNull(),
        runCatching { b.await() }.getOrNull(),
        runCatching { c.await() }.getOrNull()
    )
    results
}

Dispatchers

The Dispatchers admit selecting the thread pool for coroutine execution:

withContext(Dispatchers.IO) {
    // I/O thread pool — blocking I/O (file, network)
    file.readText()
}

withContext(Dispatchers.Default) {
    // CPU-intensive thread pool
    expensiveComputation()
}

withContext(Dispatchers.Main) {
    // Main thread (Android UI)
    updateUI()
}

withContext(Dispatchers.Unconfined) {
    // Inherits the calling thread; rare
}

The withContext admits switching dispatchers; conventional in I/O-heavy code:

suspend fun loadFile(path: String): String = withContext(Dispatchers.IO) {
    File(path).readText()
}

Cancellation

Coroutines admit cooperative cancellation:

val job = launch {
    while (isActive) {                             // check cancellation
        doWork()
    }
}

delay(2000)
job.cancel()                                       // signal cancellation
job.join()                                         // wait for cleanup

The isActive admits cooperative cancellation; substantial work between suspension points may also be checked via ensureActive().

For cancellation with cleanup:

val job = launch {
    try {
        while (isActive) {
            processStep()
        }
    } finally {
        cleanup()                                  // runs on cancel too
    }
}

The conventional discipline:

Cooperate with cancellation — admit isActive checks at substantial work boundaries.
Use withContext(NonCancellable) for cleanup that must complete.

Job hierarchy

Coroutines form a parent-child hierarchy:

val parent = launch {
    val child1 = launch { doWork1() }
    val child2 = launch { doWork2() }
    val child3 = launch { doWork3() }

    child1.join()
    child2.join()
    child3.join()
}

parent.cancel()                                    // cancels parent and all children

The structured concurrency admits substantial cancellation propagation — admit substantial leak prevention.

Exception handling

Coroutines propagate exceptions through the parent-child hierarchy:

val handler = CoroutineExceptionHandler { _, exception ->
    println("Caught $exception")
}

val job = GlobalScope.launch(handler) {
    throw RuntimeException("oops")
}

// With supervisorScope:
supervisorScope {
    val job = launch(handler) { throw RuntimeException() }
    // siblings unaffected
}

For value-returning coroutines, exceptions propagate through await():

val deferred = async {
    throw RuntimeException("oops")
}

try {
    deferred.await()
} catch (e: RuntimeException) {
    println("caught: ${e.message}")
}

`Flow` — async streams

Flow<T> represents an asynchronous stream of values:

fun fetchEvents(): Flow<Event> = flow {
    var page = 1
    while (true) {
        val events = api.fetchPage(page)
        if (events.isEmpty()) break
        emit(events)                               // emit each event
        page++
    }
}

// Collect:
fetchEvents().collect { event ->
    process(event)
}

// With operators:
fetchEvents()
    .filter { it.isImportant }
    .map { transform(it) }
    .take(100)
    .collect { handle(it) }

Flow operators are substantially lazy — work happens only on collect.

Flow operators

flow.map { transform(it) }
flow.filter { predicate(it) }
flow.take(10)
flow.drop(5)
flow.distinctUntilChanged()
flow.debounce(300.milliseconds)
flow.throttleFirst(1.seconds)
flow.combine(otherFlow) { a, b -> /* ... */ }
flow.zip(otherFlow) { a, b -> /* ... */ }
flow.flatMapConcat { other }
flow.flatMapMerge { other }                        // concurrent
flow.flatMapLatest { other }                       // cancel on new
flow.onEach { sideEffect(it) }
flow.onCompletion { /* cleanup */ }
flow.catch { /* error handling */ }
flow.retry(3)
flow.fold(initial) { acc, value -> /* ... */ }
flow.toList()

`StateFlow` and `SharedFlow`

For hot flows (always active, multiple subscribers):

class CounterViewModel : ViewModel() {
    private val _count = MutableStateFlow(0)
    val count: StateFlow<Int> = _count.asStateFlow()

    fun increment() {
        _count.value++
    }
}

class EventBus {
    private val _events = MutableSharedFlow<Event>()
    val events: SharedFlow<Event> = _events.asSharedFlow()

    suspend fun emit(event: Event) = _events.emit(event)
}

The StateFlow admits a current value; SharedFlow admits event-style broadcasting.

Channels

Channel<T> admits producer-consumer patterns:

val channel = Channel<Int>()

launch {
    for (n in 1..5) {
        channel.send(n)
    }
    channel.close()
}

for (n in channel) {
    println(n)
}

The conventional channel forms:

Channel.RENDEZVOUS (default) — no buffer; senders suspend until receivers.
Channel.BUFFERED — bounded buffer.
Channel.UNLIMITED — unbounded buffer.
Channel.CONFLATED — keep only the latest value.

val ch = Channel<Int>(capacity = 10)

val producer = launch {
    for (n in 1..100) {
        ch.send(n)
    }
    ch.close()
}

val consumer = launch {
    for (n in ch) {
        process(n)
    }
}

Common patterns

Concurrent fetches

suspend fun loadDashboard(): DashboardData = coroutineScope {
    val users = async { fetchUsers() }
    val posts = async { fetchPosts() }
    val comments = async { fetchComments() }

    DashboardData(
        users = users.await(),
        posts = posts.await(),
        comments = comments.await()
    )
}

Bounded parallelism

suspend fun processAll(items: List<Item>): List<Result> = coroutineScope {
    val semaphore = Semaphore(5)                   // max 5 concurrent

    items.map { item ->
        async {
            semaphore.withPermit { process(item) }
        }
    }.awaitAll()
}

Timeout

suspend fun fetchWithTimeout(url: String): Data = withTimeout(5.seconds) {
    fetchData(url)
}

// Or with substitute:
suspend fun fetchOrDefault(url: String): Data = withTimeoutOrNull(5.seconds) {
    fetchData(url)
} ?: Data.empty()

Retry

suspend fun <T> retry(attempts: Int = 3, delayMs: Long = 1000, block: suspend () -> T): T {
    var lastError: Throwable? = null
    repeat(attempts) {
        try {
            return block()
        } catch (e: Throwable) {
            if (e is CancellationException) throw e
            lastError = e
            if (it < attempts - 1) delay(delayMs * (1 shl it))
        }
    }
    throw lastError!!
}

Concurrent map

suspend fun <T, R> List<T>.mapConcurrently(
    concurrency: Int = 5,
    transform: suspend (T) -> R
): List<R> = coroutineScope {
    val semaphore = Semaphore(concurrency)
    map { item ->
        async { semaphore.withPermit { transform(item) } }
    }.awaitAll()
}

val results = items.mapConcurrently { fetch(it) }

Producer-consumer

fun produceNumbers() = flow {
    for (n in 1..100) {
        emit(n)
        delay(100)
    }
}

suspend fun consume(numbers: Flow<Int>) {
    numbers.collect { n ->
        process(n)
    }
}

State propagation

class ViewModel {
    private val _state = MutableStateFlow<UiState>(UiState.Loading)
    val state: StateFlow<UiState> = _state.asStateFlow()

    fun loadData() {
        viewModelScope.launch {
            _state.value = UiState.Loading
            try {
                val data = fetch()
                _state.value = UiState.Success(data)
            } catch (e: Exception) {
                _state.value = UiState.Error(e.message ?: "unknown")
            }
        }
    }
}

Event bus

object EventBus {
    private val _events = MutableSharedFlow<Event>(replay = 0)

    suspend fun emit(event: Event) {
        _events.emit(event)
    }

    fun subscribe(): SharedFlow<Event> = _events.asSharedFlow()
}

// Subscribe:
launch {
    EventBus.subscribe().collect { event ->
        handle(event)
    }
}

// Emit:
EventBus.emit(Event.UserLoggedIn(user))

Cancellation propagation

val job = launch {
    val data = withContext(Dispatchers.IO) {
        // I/O work
        loadFile(path)
    }
    process(data)
}

delay(1.seconds)
job.cancel()
job.join()

Transform with Flow operators

val searchResults: Flow<List<Result>> = queryFlow
    .debounce(300.milliseconds)
    .filter { it.length >= 3 }
    .distinctUntilChanged()
    .flatMapLatest { query ->
        flow { emit(api.search(query)) }
    }
    .catch { emit(emptyList()) }

Async let

suspend fun loadUserAndPosts(userId: Int): Pair<User, List<Post>> {
    return coroutineScope {
        async { api.getUser(userId) } to async { api.getPosts(userId) }
    }.let { (user, posts) -> user.await() to posts.await() }
}

// Or simpler:
suspend fun loadUserAndPosts(userId: Int): Pair<User, List<Post>> = coroutineScope {
    val user = async { api.getUser(userId) }
    val posts = async { api.getPosts(userId) }
    user.await() to posts.await()
}

Channel-based pipeline

fun produceItems(scope: CoroutineScope) = scope.produce {
    for (i in 1..100) {
        send(i)
    }
}

fun processItems(scope: CoroutineScope, input: ReceiveChannel<Int>) = scope.produce {
    for (item in input) {
        send(item * 2)
    }
}

runBlocking {
    val source = produceItems(this)
    val processed = processItems(this, source)

    for (result in processed) {
        println(result)
    }
}

Mutex for shared state

class Counter {
    private var count = 0
    private val mutex = Mutex()

    suspend fun increment() {
        mutex.withLock {
            count++
        }
    }

    suspend fun get(): Int = mutex.withLock { count }
}

Coroutine scope in classes

class DataLoader {
    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)

    fun load() {
        scope.launch {
            val data = fetchData()
            cache(data)
        }
    }

    fun close() {
        scope.cancel()
    }
}

Flow with state

val state = flowOf(initialState)
    .scan(initialState) { acc, action ->
        reduce(acc, action)
    }
    .stateIn(scope, SharingStarted.Eagerly, initialState)

Interop with callbacks

suspend fun fetchAsCallback(url: String): String = suspendCancellableCoroutine { cont ->
    val request = api.fetchAsync(url, callback = { result, error ->
        if (error != null) cont.resumeWithException(error)
        else cont.resume(result)
    })

    cont.invokeOnCancellation { request.cancel() }
}

The suspendCancellableCoroutine admits substantial integration with callback-based APIs.

Periodic work

val timerJob = launch {
    while (isActive) {
        doWork()
        delay(60.seconds)
    }
}

A note on the conventional discipline

The contemporary Kotlin coroutines advice:

Use suspend functions for substantial async work.
Use coroutineScope for structured concurrency.
Use supervisorScope when sibling failures should not cancel.
Use withContext(Dispatchers.IO) for blocking I/O.
Use withContext(Dispatchers.Default) for CPU-bound work.
Use async/await for parallel value-returning coroutines.
Use launch for fire-and-forget.
Use Flow for async streams.
Use StateFlow for state observation.
Use SharedFlow for event broadcasting.
Use Channel for producer-consumer.
Cooperate with cancellation — check isActive.
Don’t catch CancellationException in routine handlers.
Use Mutex for shared mutable state.
Use withTimeout for time-bounded operations.

The combination — suspend functions, structured concurrency via scopes, dispatchers for thread management, the substantial Flow API for async streams, channels for producer-consumer, the Job hierarchy for cancellation propagation — is the substance of Kotlin’s concurrency model. The discipline produces correct, performant, type-safe concurrent code with substantial protection against the conventional concurrency pitfalls (data races, leaks, unhandled cancellation).

suspend functions