Estado mutable compartido: Mutex, atómicos y confinamiento
Una carrera no necesita dos hilos. Necesita un punto de suspensión entre el chequeo y la acción — y un dispatcher de un solo hilo no te va a salvar.
Explorar la explicación completa 8 ideas
El malentendido más peligroso sobre corrutinas es que un dispatcher de un solo hilo vuelve seguro al estado compartido. No lo hace. Las corrutinas se entrelazan en los **puntos de suspensión**, y cualquier código con la forma «leé un valor, suspendé, escribilo de vuelta» puede tener a todas las demás corrutinas corriendo en el hueco. `val current = stock; delay(1); stock = current - 1` va a sobrevender en `Dispatchers.Main`, en `runTest`, en un solo hilo, de forma determinista. Lo que rompe las operaciones compuestas es la concurrencia, no el paralelismo, y las corrutinas te dan muchísima concurrencia.
El modo de fallo tiene nombre — la **actualización perdida** — y una firma distintiva. Cinco corrutinas leen un stock de tres, todas calculan `3 - 1`, todas escriben `2`. Cinco ventas, tres unidades, y un conteo final de dos: un estado aritméticamente imposible, que es la versión educada. La mayoría de las actualizaciones perdidas no se ven imposibles; simplemente subcuentan una métrica en silencio o aplican un descuento dos veces, y nada alerta porque nunca se lanza una excepción.
La primera herramienta es `Mutex`. `mutex.withLock { … }` es el lock nativo de corrutinas: **suspende** en vez de bloquear un hilo, así que es seguro en un event loop donde un `ReentrantLock` aparcaría un hilo de Netty y sería una caída de verdad. Es `lock()` más `try/finally { unlock() }`, así que un cuerpo que lanza no puede filtrar el lock. Dos propiedades vale memorizar: **no es reentrante** — llamar a una función con `withLock` desde adentro de otro `withLock` del mismo mutex deadlockea para siempre, sin excepción ni timeout — y protege solo lo que está *adentro* del bloque, así que un lock sobre la escritura pero no sobre la lectura es exactamente el bug del que partiste.
La segunda herramienta es un **atómico**: `AtomicInteger`, `AtomicReference` o el `atomicfu` de Kotlin. Son perfectos para un *único* compare-and-set — `counter.incrementAndGet()` es correcto, sin locks y barato. Solos no sirven para «leer, decidir y después escribir», porque la decisión ocurre fuera de la operación atómica. La forma honesta es `updateAndGet { … }` (o un bucle de reintento con CAS), que re-ejecuta la transformación hasta ganar; cualquier cosa más larga que eso quiere un lock.
La tercera herramienta es el **confinamiento**, y suele ser la mejor. Dale al estado un único dueño — una corrutina que lee comandos de un `Channel` y es el único código que toca los datos — y mandale mensajes en vez de compartir una referencia. No hay lock que olvidar, ni sección crítica que hacer mal, y los invariantes viven en un lugar que podés leer de arriba abajo. Cuesta un salto de mensaje, y deja de ser cómodo cuando un llamador necesita una respuesta síncrona (que recuperás mandando un `CompletableDeferred` junto con el pedido). Este es el modelo de actores, y `kotlinx.coroutines` supo traer un builder `actor { }`; el patrón sobrevivió al builder y hoy se escribe con un `Channel` corriente.
La cuarta herramienta es no compartir nada. La mayoría del «estado compartido» en un servidor con scope por request no necesita compartirse: construí el resultado de forma inmutable, devolvelo, y dejá que el llamador combine. `coroutineScope { val a = async { … }; val b = async { … }; combinar(a.await(), b.await()) }` tiene dos operaciones concurrentes y cero estado mutable compartido, y no necesita sincronización de ningún tipo. Cuando te encontrás buscando un lock dentro de un handler de request, vale preguntarse si la variable necesita sobrevivir a la llamada.
Una palabra sobre `@Volatile`, porque es la respuesta equivocada que más se ofrece. `@Volatile` arregla la **visibilidad** — garantiza que una escritura de un hilo la vea otro — y no hace nada por la **atomicidad**. `@Volatile var count = 0; count++` sigue siendo una actualización perdida, porque `++` es una lectura, una suma y una escritura. Si la operación es compuesta, la volatilidad es irrelevante; si es una única escritura de un valor calculado aparte, la volatilidad es exactamente lo que necesitás.
La decisión, comprimida. Un contador, una operación → un atómico. Una operación compuesta corta sobre un estado → un `Mutex`. Un conjunto de estado relacionado con invariantes reales → confinalo detrás de un canal. Estado que no necesita sobrevivir al request → no lo compartas. Y elijas lo que elijas, poder decir *por qué* en una frase, porque «usamos un Mutex» no es una respuesta y «la lectura y la escritura tienen que ser un solo paso indivisible, así que ambas están dentro del lock» sí lo es.
Aprendamos juntos.
Pregúntame sobre este curso o por dónde empezar.
import kotlinx.coroutines.*import kotlinx.coroutines.sync.*import kotlinx.coroutines.test.*class Inventory(var stock: Int) {val sold = mutableListOf<Int>()private val mutex = Mutex()// check ... suspend ... act: every buyer reads the same stale value.suspend fun buyUnsafe(id: Int) {val current = stockdelay(1)if (current > 0) { stock = current - 1; sold.add(id) }}// The lock spans the READ as well as the write, so each buyer sees what// the previous one wrote.suspend fun buySafe(id: Int) = mutex.withLock {val current = stockdelay(1)if (current > 0) { stock = current - 1; sold.add(id) }}}@OptIn(ExperimentalCoroutinesApi::class)fun main() = runTest {val unsafe = Inventory(3)coroutineScope { (1..5).forEach { id -> launch { unsafe.buyUnsafe(id) } } }println("unsafe: sold " + unsafe.sold.size + " from 3, stock left " + unsafe.stock)val safe = Inventory(3)coroutineScope { (1..5).forEach { id -> launch { safe.buySafe(id) } } }println("safe: sold " + safe.sold.size + " from 3, stock left " + safe.stock)}
The lost update, reproduced deterministically on a single-threaded test dispatcher — and fixed by putting the whole read-modify-write inside one critical section.
Arena IDEimport kotlinx.coroutines.*import java.util.concurrent.atomic.AtomicIntegerfun main() = runBlocking {// Correct: the whole operation is one CAS.val good = AtomicInteger(0)coroutineScope { repeat(1_000) { launch(Dispatchers.Default) { good.incrementAndGet() } } }println("incrementAndGet = " + good.get())// Also correct for a compound TRANSFORMATION: updateAndGet retries the CAS.val capped = AtomicInteger(0)coroutineScope {repeat(1_000) { launch(Dispatchers.Default) { capped.updateAndGet { if (it < 500) it + 1 else it } } }}println("updateAndGet cap = " + capped.get())// WRONG: the decision happens OUTSIDE the atomic operation.val bad = AtomicInteger(0)coroutineScope {repeat(1_000) { launch(Dispatchers.Default) { val v = bad.get(); if (v < 500) bad.set(v + 1) } }}println("read-then-set = " + bad.get() + " (target was 500)")}
Atomics are for ONE indivisible operation. incrementAndGet is correct; a manual read-decide-write around an atomic is the same lost update with extra ceremony.
Arena IDEimport kotlinx.coroutines.*import kotlinx.coroutines.channels.*sealed interface Cmddata class Reserve(val id: Int, val reply: CompletableDeferred<Boolean>) : Cmddata class Report(val reply: CompletableDeferred<Int>) : Cmdfun CoroutineScope.inventory(initial: Int): SendChannel<Cmd> {val cmds = Channel<Cmd>(Channel.UNLIMITED)launch {var stock = initial // owned by THIS coroutine alonefor (cmd in cmds) when (cmd) {is Reserve -> {val ok = stock > 0if (ok) stock--cmd.reply.complete(ok)}is Report -> cmd.reply.complete(stock)}}return cmds}fun main() = runBlocking {val inv = inventory(3)val granted = (1..5).map { id ->val reply = CompletableDeferred<Boolean>()inv.send(Reserve(id, reply))reply.await()}.count { it }val left = CompletableDeferred<Int>().also { inv.send(Report(it)) }.await()println("granted = " + granted + ", stock left = " + left)inv.close()}
Confinement: one coroutine owns the state, everyone else sends messages. No lock, and a CompletableDeferred carries the answer back when a caller needs one.
Arena IDEimport kotlinx.coroutines.*import kotlinx.coroutines.sync.*class Ledger {private val mutex = Mutex()private var total = 0suspend fun add(n: Int) = mutex.withLock { total += n }// BUG: add() takes the same mutex we are already holding.suspend fun addTwice(n: Int) = mutex.withLock {add(n)add(n)}suspend fun addTwiceFixed(n: Int) = mutex.withLock { total += 2 * n }suspend fun total(): Int = mutex.withLock { total }}fun main() = runBlocking {val ledger = Ledger()val deadlocked = withTimeoutOrNull(200) { ledger.addTwice(5) } == nullprintln("reentrant call deadlocked: " + deadlocked)val ok = Ledger()ok.addTwiceFixed(5)println("non-reentrant version total: " + ok.total())}
The trap that has no exception and no timeout: Mutex is NOT reentrant, so a locking function called from inside the lock deadlocks forever. withTimeoutOrNull is how you prove it without hanging the demo.
Arena IDE🧠 Comprueba tu comprensión
0/1 · 0/1 answered1. Your service runs on `Dispatchers.Main.immediate` — a single thread. A method reads a counter, calls a suspending repository, then writes the counter back. Is it safe?