"Suspend Your Disbelief" — Coroutines Fundamentals

// Fragment of story/VirtualThreadFanOut.kt
fun main() {
    val completed = AtomicInteger()
    Executors.newVirtualThreadPerTaskExecutor().use { executor -> // close() waits for submitted tasks
        val calls = (1..40).map { supplier ->
            CompletableFuture.supplyAsync({ blockingQuote(supplier, completed) }, executor)
        }
        val import = CompletableFuture.allOf(*calls.toTypedArray())
        Thread.sleep(50)
        import.cancel(true) // the buying team cancels the import
        println("import cancelled: ${import.isCancelled}") // -> import cancelled: true
    }
    // Cancelling the combined future cancelled nothing behind it.
    println("supplier calls that ran anyway: ${completed.get()} of 40") // -> supplier calls that ran anyway: 40 of 40
}

import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger

// A suspend function can pause without blocking its thread. delay() suspends; Thread.sleep() would block.
suspend fun quote(supplier: Int): Long {
    delay(200)
    return 16_000L + supplier * 10
}

fun main() = runBlocking { // bridges blocking code (main) and suspending code; never use it inside coroutines
    // launch: fire and forget, returns a Job.
    val job = launch { println("indexing started") }
    job.join() // -> indexing started

    // async: returns a Deferred<T>; await() suspends until the value is ready.
    val first = async { quote(1) }
    val second = async { quote(2) }
    println("best ${minOf(first.await(), second.await())}") // -> best 16010

    // 10,000 concurrent waits, and how many threads did the waiting.
    val threads = ConcurrentHashMap.newKeySet<String>()
    val done = AtomicInteger()
    (1..10_000).map { supplier ->
        async {
            threads += Thread.currentThread().name.substringBefore(" @") // strip the debug-mode coroutine name
            quote(supplier)
            done.incrementAndGet()
        }
    }.awaitAll()
    println("${done.get()} quotes on ${threads.size} thread") // -> 10000 quotes on 1 thread
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking

suspend fun fetchPrice(sku: String): Long {
    delay(10)
    return if (sku == "SHW-1001") 16_500 else 72_000
}

suspend fun fetchStock(sku: String): Int {
    delay(10)
    return if (sku == "SHW-1001") 120 else 35
}

// Two suspension points. The compiler turns this function into a state machine with three states.
suspend fun shelfLine(sku: String): String {
    val price = fetchPrice(sku) // state 0 -> suspends, resumes in state 1
    val units = fetchStock(sku) // state 1 -> suspends, resumes in state 2
    return "$sku ₹${price / 100} x$units" // state 2
}

fun main() = runBlocking {
    println(shelfLine("SHW-1001")) // -> SHW-1001 ₹165 x120
}

// javap -p StateMachineKt
public static final Object fetchPrice(String, Continuation<? super Long>);
public static final Object shelfLine(String, Continuation<? super String>);

// javap -p 'StateMachineKt$shelfLine$1'
final class StateMachineKt$shelfLine$1 extends ContinuationImpl {
  Object L$0;      // a saved local: sku
  long J$0;        // a saved local: price
  Object result;   // the value the last suspension resumed with
  int label;       // which state to resume in
  public final Object invokeSuspend(Object);  // re-enters shelfLine
}

// javap -c StateMachineKt.shelfLine (simplified)
  63: getfield      label
  66: tableswitch   { 0: 92;  1: 124;  2: 180 }
  // state 0: save sku, set label = 1, call fetchPrice; if it returned COROUTINE_SUSPENDED, return it
 103: putfield      L$0
 109: putfield      label
 112: invokestatic  fetchPrice:(Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 123: areturn
  // state 1: restore sku, save price, set label = 2, call fetchStock
 159: putfield      J$0
 165: putfield      label
 168: invokestatic  fetchStock:(Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 179: areturn
  // state 2: restore price and sku, build the string, return it

import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.async
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.supervisorScope
import java.io.IOException

suspend fun quote(supplier: String, ms: Long): Long {
    delay(ms)
    if (supplier == "flaky") throw IOException("$supplier: HTTP 503")
    return ms * 100
}

// coroutineScope: all children or nothing. One failure cancels the siblings and rethrows.
suspend fun allQuotes(): List<Long> = coroutineScope {
    val a = async { quote("fast", 50) }
    val b = async { quote("flaky", 100) }
    val c = async {
        try {
            quote("slow", 300)
        } finally {
            println("slow quote cancelled") // runs when the failing sibling cancels it
        }
    }
    listOf(a.await(), b.await(), c.await())
}

fun main(): Unit = runBlocking { // ": Unit": the last expression (a Job) is not main's result
    try {
        allQuotes()
    } catch (e: IOException) {
        println("failed: ${e.message}")
    }
    // -> slow quote cancelled
    // -> failed: flaky: HTTP 503

    // supervisorScope: a child's failure stays with that child. Nobody catches the flaky one here;
    // its exception goes to the handler, and its siblings carry on.
    val report = CoroutineExceptionHandler { _, e -> println("reported: ${e.message}") }
    supervisorScope {
        launch { printQuote("fast", 50) }
        launch(report) { printQuote("flaky", 100) }
        launch { printQuote("slow", 150) }
    }
    // -> fast: 5000
    // -> reported: flaky: HTTP 503
    // -> slow: 15000
}

suspend fun printQuote(supplier: String, ms: Long) = println("$supplier: ${quote(supplier, ms)}")

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import java.util.concurrent.atomic.AtomicInteger

fun threadName() = Thread.currentThread().name.substringBefore(" @") // drop the debug-mode coroutine suffix

fun main() = runBlocking {
    val caller = threadName() // 'main' when run directly; the test runner's thread under test

    // Default: CPU-bound work, one thread per core. IO: blocking calls, up to 64 threads by default.
    // They share one pool of worker threads; withContext switches and switches back.
    val onDefault = withContext(Dispatchers.Default) { threadName() }
    val onIo = withContext(Dispatchers.IO) { threadName() }
    println(onDefault.startsWith("DefaultDispatcher-worker")) // -> true
    println(onIo.startsWith("DefaultDispatcher-worker")) // -> true
    println(threadName() == caller) // -> true

    // limitedParallelism: a view of a dispatcher that runs at most N coroutines at a time.
    // Use it as a bulkhead, e.g. for a supplier API that allows two concurrent calls.
    val supplierApi = Dispatchers.IO.limitedParallelism(2)
    val inFlight = AtomicInteger()
    val maxInFlight = AtomicInteger()
    (1..10).map {
        async(supplierApi) {
            val now = inFlight.incrementAndGet()
            maxInFlight.accumulateAndGet(now, ::maxOf)
            Thread.sleep(20) // a blocking client call, fine on IO
            inFlight.decrementAndGet()
        }
    }.awaitAll()
    println("max concurrent calls: ${maxInFlight.get()}") // -> max concurrent calls: 2
}

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.delay
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.withTimeoutOrNull

fun busyWork(millis: Long) {
    val until = System.nanoTime() + millis * 1_000_000
    while (System.nanoTime() < until) { /* CPU work with no suspension point */ }
}

fun main() = runBlocking {
    // Cancellation is cooperative: a loop that never suspends or checks keeps running.
    var stubbornRounds = 0
    val stubborn = launch(Dispatchers.Default) {
        repeat(5) {
            busyWork(20)
            stubbornRounds++
        }
    }
    delay(30)
    stubborn.cancel()
    stubborn.join()
    println("stubborn finished ${stubbornRounds} of 5") // -> stubborn finished 5 of 5

    // ensureActive() (or isActive, or any suspending call) is a cancellation check.
    var politeRounds = 0
    val polite = launch(Dispatchers.Default) {
        repeat(5) {
            ensureActive()
            busyWork(20)
            politeRounds++
        }
    }
    delay(30)
    polite.cancel()
    polite.join()
    println("polite stopped early: ${politeRounds < 5}") // -> polite stopped early: true

    // Cleanup in finally; suspending cleanup needs NonCancellable.
    val indexer = launch {
        try {
            delay(1_000)
        } finally {
            withContext(NonCancellable) {
                delay(10) // would throw immediately in a cancelled coroutine without NonCancellable
                println("index lock released") // -> index lock released
            }
        }
    }
    delay(10)
    indexer.cancel()
    indexer.join()

    // Timeouts: withTimeout throws TimeoutCancellationException; withTimeoutOrNull returns null.
    println(withTimeoutOrNull(50) { delay(500); "quote" }) // -> null
    try {
        withTimeout(50) { delay(500) }
    } catch (e: TimeoutCancellationException) { // catch the timeout, never CancellationException itself
        println("timed out: ${e::class.simpleName}") // -> timed out: TimeoutCancellationException
    }
    println("still active: $isActive") // -> still active: true
}

import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.joinAll
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.supervisorScope
import java.io.IOException

fun main() = runBlocking {
    // Catching at await() is not enough: the failed child has already failed its parent scope.
    try {
        coroutineScope {
            val failing = async { throw IOException("supplier down") }
            try {
                failing.await()
            } catch (e: IOException) {
                println("await threw: ${e.message}") // -> await threw: supplier down
            }
        }
    } catch (e: IOException) {
        println("scope failed anyway: ${e.message}") // -> scope failed anyway: supplier down
    }

    // In a supervisorScope, a child's failure is only reported through its own await().
    supervisorScope {
        val failing = async { throw IOException("supplier down") }
        try {
            failing.await()
        } catch (e: IOException) {
            println("handled: ${e.message}") // -> handled: supplier down
        }
    }

    // launch has nobody to hand its exception to, so it goes up the job tree. At the root,
    // a CoroutineExceptionHandler is the last resort, the coroutine version of an uncaught-exception handler.
    val handler = CoroutineExceptionHandler { _, e -> println("handler saw: ${e.message}") }
    val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default + handler)
    val jobs = listOf(
        scope.launch { throw IOException("index rebuild failed") },
        scope.launch { delay(50); println("other job still ran") },
    )
    jobs.joinAll()
    // -> handler saw: index rebuild failed
    // -> other job still ran
}

// Fragment of quotes/SupplierQuotes.kt
// Step 1: a supplier API client with a suspending call. In Part 14 this becomes a real HTTP client.
data class Quote(val supplier: String, val sku: String, val paise: Long)

class SupplierClient(
    val name: String,
    private val latency: Duration,
    private val failing: Boolean = false,
    private val completedCalls: AtomicInteger,
) {
    suspend fun quote(sku: String): Quote {
        delay(latency) // the round trip: suspends, holds no thread
        if (failing) throw IOException("$name: HTTP 503")
        completedCalls.incrementAndGet()
        return Quote(name, sku, 16_000 + latency.inWholeMilliseconds)
    }
}

// Fragment of quotes/SupplierQuotes.kt
// Step 2: one supplier, with its own timeout. A slow or failing supplier yields null, not an exception.
suspend fun SupplierClient.quoteOrNull(sku: String, timeout: Duration): Quote? =
    try {
        withTimeoutOrNull(timeout) { quote(sku) }
    } catch (e: IOException) {
        null
    }

// Fragment of quotes/SupplierQuotes.kt
// Step 3: fan out inside a scope. Every call is a child of this function: if the caller is
// cancelled, all of them are; the function can't return while any of them is still running.
suspend fun bestQuote(sku: String, suppliers: List<SupplierClient>, maxInFlight: Int): Quote? = coroutineScope {
    val permits = Semaphore(maxInFlight) // a bulkhead: at most maxInFlight calls at once
    suppliers
        .map { supplier -> async { permits.withPermit { supplier.quoteOrNull(sku, 300.milliseconds) } } }
        .awaitAll()
        .filterNotNull()
        .minByOrNull { it.paise }
}

// Fragment of quotes/SupplierQuotes.kt
// Step 4: forty suppliers, two failing and one too slow, on the caller's single thread.
fun main() = runBlocking {
    val completed = AtomicInteger()
    val suppliers = (0 until 40).map { i ->
        when (i) {
            7, 23 -> SupplierClient("supplier-$i", 80.milliseconds, failing = true, completedCalls = completed)
            31 -> SupplierClient("supplier-$i", 2_000.milliseconds, completedCalls = completed)
            else -> SupplierClient("supplier-$i", (50 + i * 5).milliseconds, completedCalls = completed)
        }
    }

    val best = bestQuote("SHW-1001", suppliers, maxInFlight = 10)
    println("best: ${best?.supplier} at ${best?.paise}") // -> best: supplier-0 at 16050
    println("answered: ${completed.get()} of 40") // -> answered: 37 of 40

    // Step 5: cancellation reaches every call in flight.
    completed.set(0)
    val import = launch { bestQuote("SHW-2040", suppliers, maxInFlight = 40) }
    delay(30)
    import.cancel()
    import.join()
    println("after cancel: ${completed.get()} calls completed, import cancelled: ${import.isCancelled}") // -> after cancel: 0 calls completed, import cancelled: true
}

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.asContextElement
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext

// Stands in for SLF4J's MDC or Spring Security's SecurityContextHolder: state kept per thread.
val currentStore = ThreadLocal<String>()

suspend fun logLine(message: String): String = withContext(Dispatchers.Default) {
    delay(1) // a suspension point; the code may resume on any worker thread
    "[store=${currentStore.get()}] $message"
}

private val lock = Any()

fun main() = runBlocking {
    // 1. A ThreadLocal set on one thread is not there after the coroutine moves threads.
    currentStore.set("PUN-014")
    println(logLine("price updated")) // -> [store=null] price updated
    // asContextElement makes the coroutine carry the value and install it on whichever thread runs it.
    withContext(currentStore.asContextElement("PUN-014")) {
        println(logLine("price updated")) // -> [store=PUN-014] price updated
    }
    currentStore.remove()

    // 2. catch (e: Exception) catches CancellationException too, so this loop ignores cancellation.
    var rounds = 0
    val stubborn = launch {
        repeat(5) {
            try {
                delay(20)
            } catch (e: Exception) {
                // a Java habit: log and carry on. Here it swallows the cancellation.
            }
            rounds++
        }
    }
    delay(30)
    stubborn.cancel()
    stubborn.join()
    println("rounds completed after cancel: $rounds") // -> rounds completed after cancel: 5

    // 3. A lock held across a suspension point would be released on another thread, so it doesn't compile.
    synchronized(lock) {
        // delay(10)  // error: the 'delay' suspension point is inside a critical section.
        println("synchronized blocks can't suspend") // -> synchronized blocks can't suspend
    }
}

import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.async
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.supervisorScope
import java.io.IOException

suspend fun supplierQuote(sku: String): Long {
    delay(10)
    throw IOException("no quote for $sku")
}

// Run with -Dkotlinx.coroutines.debug (this module's tests set it): coroutine names appear in thread names,
// and exceptions rethrown across coroutines get the awaiting coroutine's frames added back.
fun main(): Unit = runBlocking(CoroutineName("nightly-import")) {
    println(Thread.currentThread().name) // -> … @nightly-import#…

    supervisorScope {
        val quote = async(CoroutineName("quote-SHW-1001")) { supplierQuote("SHW-1001") }
        try {
            quote.await()
        } catch (e: IOException) {
            e.stackTraceToString().lines().take(5).forEach(::println)
            // -> java.io.IOException: no quote for SHW-1001
            // ->   at com.shelfwise.part10.debugging.DebuggingKt.supplierQuote(Debugging.kt:12)
            // ->   at com.shelfwise.part10.debugging.DebuggingKt$supplierQuote$1.invokeSuspend(Debugging.kt)
            // ->   at _COROUTINE._BOUNDARY._(CoroutineDebugging.kt:42)
            // ->   at com.shelfwise.part10.debugging.DebuggingKt$main$1$1.invokeSuspend(Debugging.kt:23)
        }
    }
}