obsidian/Kotlin/Корутины/Разделяемое состояние.md

title, description, tags

title	description	tags
Разделяемое состояние	Mutex, Semaphore, атомарные типы, конфайнмент потока и подход «не разделять состояние вообще»	kotlin coroutines

Корутины не отменяют гонок. Если несколько корутин меняют одну переменную, нужна синхронизация - причём такая, которая приостанавливает, а не блокирует поток.

import kotlinx.coroutines.*

fun main() = runBlocking {
    var counter = 0
    withContext(Dispatchers.Default) {
        List(100) { launch { repeat(1000) { counter++ } } }.forEach { it.join() }
    }
    println("потеряно инкрементов: ${counter < 100_000}")      // потеряно инкрементов: true
}

counter++ - это чтение, увеличение и запись: три операции, между которыми может вклиниться другая корутина. На одном потоке проблемы не было бы, но Dispatchers.Default использует все ядра.

Способы, по возрастанию цены

Подход	Стоимость	Когда применять
не разделять состояние	нулевая	всегда, когда возможно
неизменяемые данные	нулевая	состояние целиком заменяется
атомарные типы	очень низкая	счётчики, флаги, одна переменная
MutableStateFlow.update	низкая	наблюдаемое состояние
конфайнмент в один поток	низкая	группа связанных переменных
Mutex	средняя	составная операция, вызов suspend внутри
Semaphore	средняя	ограничение числа одновременных операций

Порядок не случаен: первые два варианта убирают проблему, остальные её решают.

Атомарные типы

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicReference

fun main() = runBlocking {
    val counter = AtomicInteger(0)
    withContext(Dispatchers.Default) {
        List(100) { launch { repeat(1000) { counter.incrementAndGet() } } }.forEach { it.join() }
    }
    println("точный результат: ${counter.get()}")        // точный результат: 100000

    val ref = AtomicReference(listOf<String>())
    withContext(Dispatchers.Default) {
        List(10) { i -> launch { ref.updateAndGet { it + "элемент $i" } } }.forEach { it.join() }
    }
    println("добавлено без потерь: ${ref.get().size}")      // добавлено без потерь: 10
}

Атомарные типы работают через CAS (compare-and-set) - никаких блокировок, только повтор при конфликте. Это самый дешёвый способ, но он покрывает только операции с одной переменной.

Mutex

Mutex - взаимное исключение для корутин. В отличие от synchronized, он приостанавливает, а не блокирует поток.

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {
    val mutex = Mutex()
    var counter = 0

    withContext(Dispatchers.Default) {
        List(100) { launch { repeat(1000) { mutex.withLock { counter++ } } } }
            .forEach { it.join() }
    }
    println("результат точный: ${counter == 100_000}")      // результат точный: true
}

Главное преимущество - внутри withLock можно вызывать suspend-функции:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

class Cache {
    private val mutex = Mutex()
    private val data = mutableMapOf<Int, String>()

    suspend fun get(id: Int): String = mutex.withLock {
        data[id] ?: load(id).also { data[id] = it }
    }

    private suspend fun load(id: Int): String {
        delay(10)                            // suspend внутри критической секции
        return "значение $id"
    }
}

fun main() = runBlocking {
    val cache = Cache()
    val results = List(5) { async { cache.get(1) } }.awaitAll()
    println("все получили одно значение: ${results.toSet().size == 1}")      // все получили одно значение: true
}

С synchronized так нельзя: suspend-вызов внутри synchronized-блока не компилируется, потому что приостановка освободила бы поток, а монитор остался бы захваченным.

Mutex не реентерабелен

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {
    val mutex = Mutex()

    val deadlocked = withTimeoutOrNull(100) {
        mutex.withLock {
            mutex.withLock {              // вторичный захват тем же владельцем
                "недостижимо"
            }
        }
    }
    println("повторный захват повис: ${deadlocked == null}")      // повторный захват повис: true
}

Это осознанное решение: реентерабельность в корутинах определить корректно нельзя - «тот же владелец» для корутины понятие нечёткое, ведь она может перемещаться между потоками. Практический вывод: функция под withLock не должна вызывать другую функцию, которая берёт тот же мьютекс. Приватные методы, вызываемые из критической секции, не должны блокироваться сами.

Semaphore

Ограничивает число одновременных операций:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import java.util.concurrent.atomic.AtomicInteger

fun main() = runBlocking {
    val semaphore = Semaphore(permits = 3)
    val inFlight = AtomicInteger(0)
    val maxInFlight = AtomicInteger(0)

    withContext(Dispatchers.Default) {
        List(20) {
            launch {
                semaphore.withPermit {
                    val now = inFlight.incrementAndGet()
                    maxInFlight.updateAndGet { m -> maxOf(m, now) }
                    delay(20)                   // ожидание ответа API
                    inFlight.decrementAndGet()
                }
            }
        }.forEach { it.join() }
    }
    println("одновременно не больше 3: ${maxInFlight.get() <= 3}")      // одновременно не больше 3: true
}

Важное отличие от limitedParallelism: семафор считает операции в полёте, включая приостановленные. Именно он нужен, чтобы не превысить лимит внешнего API. Подробнее: Диспетчеры и контекст

Конфайнмент в один поток

Если состояние меняется только в одном потоке, синхронизация не нужна вовсе:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val single = Dispatchers.Default.limitedParallelism(1)
    var counter = 0

    withContext(Dispatchers.Default) {
        List(100) {
            launch {
                repeat(1000) {
                    withContext(single) { counter++ }        // все изменения в одном потоке
                }
            }
        }.forEach { it.join() }
    }
    println("точный результат: ${counter == 100_000}")      // точный результат: true
}

Работает, но дорого: 100 000 переключений контекста. Правильная версия конфайнмента - грубая, когда весь блок работы уходит в один поток:

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    val single = Dispatchers.Default.limitedParallelism(1)
    var counter = 0

    val coarse = measureTimeMillis {
        withContext(single) {                     // переключение один раз
            repeat(100_000) { counter++ }
        }
    }
    println("результат: $counter")
    println("грубый конфайнмент быстр: ${coarse < 500}")
    // результат: 100000
    // грубый конфайнмент быстр: true
}

Актор: состояние внутри корутины

Самый чистый подход - состояние не разделяется вообще, им владеет одна корутина:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel

sealed interface AccountCommand {
    data class Deposit(val amount: Int) : AccountCommand
    data class Withdraw(val amount: Int, val ok: CompletableDeferred<Boolean>) : AccountCommand
    data class Balance(val reply: CompletableDeferred<Int>) : AccountCommand
}

fun main() = runBlocking {
    val commands = Channel<AccountCommand>()

    launch {
        var balance = 0                       // никто больше не видит эту переменную
        for (cmd in commands) {
            when (cmd) {
                is AccountCommand.Deposit -> balance += cmd.amount
                is AccountCommand.Withdraw -> {
                    val enough = balance >= cmd.amount
                    if (enough) balance -= cmd.amount
                    cmd.ok.complete(enough)
                }
                is AccountCommand.Balance -> cmd.reply.complete(balance)
            }
        }
    }

    repeat(100) { commands.send(AccountCommand.Deposit(10)) }

    val ok = CompletableDeferred<Boolean>()
    commands.send(AccountCommand.Withdraw(500, ok))
    println("списание прошло: ${ok.await()}")

    val tooMuch = CompletableDeferred<Boolean>()
    commands.send(AccountCommand.Withdraw(10_000, tooMuch))
    println("превышение отклонено: ${!tooMuch.await()}")

    val balance = CompletableDeferred<Int>()
    commands.send(AccountCommand.Balance(balance))
    println("баланс: ${balance.await()}")
    commands.close()
    // списание прошло: true
    // превышение отклонено: true
    // баланс: 500
}

Проверка «хватает ли денег» и списание происходят атомарно без всякой блокировки - просто потому, что их выполняет одна корутина. Подробнее: Channels

StateFlow для наблюдаемого состояния

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.update

fun main() = runBlocking {
    val state = MutableStateFlow(0)
    withContext(Dispatchers.Default) {
        List(100) { launch { repeat(1000) { state.update { v -> v + 1 } } } }
            .forEach { it.join() }
    }
    println("результат точный: ${state.value == 100_000}")      // результат точный: true
}

update - это CAS-цикл, поэтому он и атомарен, и даёт оповещение подписчикам. Если состояние нужно наблюдать, это лучший вариант. Подробнее: StateFlow и SharedFlow

Как выбрать

graph TD
    A["Состояние меняют<br/>несколько корутин"] --> B{"Можно вообще<br/>не разделять?"}
    B -->|"да"| C["локальная переменная<br/>или актор"]
    B -->|"нет"| D{"За состоянием<br/>наблюдают?"}
    D -->|"да"| E["MutableStateFlow<br/>+ update"]
    D -->|"нет"| F{"Одна переменная?"}
    F -->|"да"| G["AtomicInteger / AtomicReference"]
    F -->|"нет, составная операция"| H{"Нужен suspend<br/>внутри секции?"}
    H -->|"да"| I["Mutex.withLock"]
    H -->|"нет"| J["Mutex или<br/>конфайнмент в поток"]

Неизменяемость убирает проблему

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.update

data class Stats(val requests: Int = 0, val errors: Int = 0) {
    val successRate: Double
        get() = if (requests == 0) 0.0 else (requests - errors).toDouble() / requests
}

fun main() = runBlocking {
    val stats = MutableStateFlow(Stats())

    withContext(Dispatchers.Default) {
        List(100) { i ->
            launch {
                stats.update { it.copy(requests = it.requests + 1, errors = it.errors + i % 10 / 9) }
            }
        }.forEach { it.join() }
    }
    val final = stats.value
    println("запросов: ${final.requests}, ошибок: ${final.errors}")
    println("согласованный снимок: ${final.requests == 100}")
    // запросов: 100, ошибок: 10
    // согласованный снимок: true
}

Два связанных поля обновляются одним атомарным copy, поэтому несогласованного состояния («запросов 5, ошибок 7») не возникает никогда. Это главный довод за неизменяемые data class в качестве состояния. Подробнее: Data class

Коллекции

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import java.util.concurrent.ConcurrentHashMap

fun main() = runBlocking {
    // обычный MutableList не потокобезопасен
    val unsafe = mutableListOf<Int>()
    withContext(Dispatchers.Default) {
        List(100) { i -> launch { repeat(100) { unsafe.add(i) } } }.forEach { it.join() }
    }
    println("размер может быть неверным: ${unsafe.size <= 10_000}")

    // вариант 1: Mutex
    val mutex = Mutex()
    val guarded = mutableListOf<Int>()
    withContext(Dispatchers.Default) {
        List(100) { i -> launch { repeat(100) { mutex.withLock { guarded.add(i) } } } }
            .forEach { it.join() }
    }
    println("с Mutex точно: ${guarded.size == 10_000}")

    // вариант 2: потокобезопасная коллекция из JDK
    val concurrent = ConcurrentHashMap<Int, Int>()
    withContext(Dispatchers.Default) {
        List(100) { i -> launch { concurrent[i] = i } }.forEach { it.join() }
    }
    println("ConcurrentHashMap точно: ${concurrent.size == 100}")
    // размер может быть неверным: true
    // с Mutex точно: true
    // ConcurrentHashMap точно: true
}

В Kotlin нет своих потокобезопасных коллекций для JVM - используются java.util.concurrent. Но учтите: ConcurrentHashMap безопасен для отдельных операций, а составные («прочитать, изменить, записать») всё равно требуют compute/merge или мьютекса. Подробнее: Изменяемость

Что не нужно делать

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

fun main() = runBlocking {
    var counter = 0
    val lock = Any()

    val time = measureTimeMillis {
        withContext(Dispatchers.Default) {
            List(100) { launch { repeat(1000) { synchronized(lock) { counter++ } } } }
                .forEach { it.join() }
        }
    }
    println("результат верный: ${counter == 100_000}")
    println("но поток блокировался: ${time >= 0}")
    // результат верный: true
    // но поток блокировался: true
}

synchronized работает и даёт правильный результат, но блокирует поток диспетчера: пока корутина ждёт монитор, поток не может выполнять другие корутины. На ограниченном диспетчере это прямой путь к deadlock, а suspend-вызов внутри synchronized вообще не скомпилируется.

[!info] Отличия от Java

Java	Kotlin
synchronized (блокирует поток)	Mutex.withLock (приостанавливает)
ReentrantLock	Mutex (не реентерабелен)
Semaphore из j.u.c. (блокирует)	kotlinx.coroutines.sync.Semaphore
AtomicInteger	он же, или MutableStateFlow
volatile	@Volatile или StateFlow
ThreadLocal для конфайнмента	элемент контекста корутины
ConcurrentHashMap	он же
нет аналога	актор на Channel

synchronized занимает поток, Mutex - нет. Это принципиальное различие. В пуле из 8 потоков восемь корутин, ждущих монитор, парализуют весь диспетчер - включая корутины, которым этот монитор не нужен. Mutex приостанавливает корутину и освобождает поток:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {
    val mutex = Mutex()
    val single = Dispatchers.Default.limitedParallelism(1)
    var progress = ""
    withContext(single) {
        launch { mutex.withLock { delay(30); progress += "A" } }
        launch { mutex.withLock { progress += "B" } }
        launch { progress += "C" }        // не ждёт мьютекс
    }
    println("порядок выполнения: $progress")
    // порядок выполнения: CAB
}

На одном потоке корутина C выполнилась первой, пока A держала мьютекс и ждала на delay. С synchronized и Thread.sleep поток был бы занят и C стояла бы в очереди.

suspend внутри synchronized не компилируется. Это не ограничение, а защита: приостановка освободила бы поток, а монитор остался бы захваченным - потенциальный deadlock, который JVM не заметит. Mutex снимает запрет, потому что владение не привязано к потоку.

ReentrantLock реентерабелен, Mutex - нет. В Java повторный захват тем же потоком разрешён, что делает рекурсивные вызовы под локом безопасными. В корутинах понятие «тот же владелец» не определено: корутина мигрирует между потоками, а withLock в дочерней корутине - это уже другой владелец. Поэтому Mutex принципиально не реентерабелен, и рекурсию под локом приходится исключать архитектурно.

volatile против StateFlow. @Volatile в Kotlin есть и работает так же: гарантирует видимость записи между потоками, но не атомарность составных операций. MutableStateFlow даёт и видимость, и атомарный update, и оповещение - три вещи вместо одной.

Актор не имеет аналога в Java. Паттерн «состояние внутри потока, снаружи только сообщения» в Java требует BlockingQueue + выделенный поток + «отравленную пилюлю» для остановки + ручную реализацию ответов (обычно через CompletableFuture в сообщении). В Kotlin это Channel + launch + CompletableDeferred - примерно двадцать строк и никаких потоков.

Проблемы те же. Корутины не отменяют ни гонок, ни deadlock'ов, ни инверсии приоритетов. Два мьютекса, захваченные в разном порядке, дают взаимную блокировку точно так же, как два synchronized:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {
    val a = Mutex()
    val b = Mutex()
    val result = withTimeoutOrNull(100) {
        coroutineScope {
            launch { a.withLock { delay(20); b.withLock { } } }
            launch { b.withLock { delay(20); a.withLock { } } }
        }
    }
    println("классический deadlock: ${result == null}")      // классический deadlock: true
    coroutineContext.cancelChildren()
}

Разница только в том, что корутинный deadlock можно разорвать таймаутом, а поток, стоящий на мониторе, прервать нельзя.

Сравнить: Взаимодействие между потоками в Java, HashMap в Java, JVM

[!warning] Подводные камни Mutex не реентерабелен - повторный захват это deadlock:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

class Service {
    private val mutex = Mutex()
    suspend fun outer() = mutex.withLock { inner() }
    suspend fun inner() = mutex.withLock { "никогда" }      // тот же мьютекс
}

fun main() = runBlocking {
    val result = withTimeoutOrNull(100) { Service().outer() }
    println("повисло: ${result == null}")      // повисло: true
}

Публичный метод, берущий лок, не должен вызывать другой публичный метод того же класса. Выносите логику в приватные методы без блокировки.

synchronized в корутине блокирует поток диспетчера. Результат будет верным, но пропускная способность упадёт, а на limitedParallelism(1) возможен deadlock. Используйте Mutex.

suspend внутри synchronized не компилируется - и это хорошо. Попытка обойти запрет через runBlocking внутри synchronized создаёт именно тот deadlock, от которого защищает компилятор.

Отмена внутри withLock освобождает лок, но работа может остаться незавершённой:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun main() = runBlocking {
    val mutex = Mutex()
    var step = 0
    val job = launch {
        mutex.withLock {
            step = 1
            delay(1000)          // здесь отмена
            step = 2              // не выполнится
        }
    }
    delay(20)
    job.cancelAndJoin()
    println("лок свободен: ${!mutex.isLocked}, состояние на середине: ${step == 1}")
    // лок свободен: true, состояние на середине: true
}

Критическая секция прервалась на середине, оставив несогласованное состояние. Либо делайте секцию атомарной (одна операция copy), либо восстанавливайте инвариант в finally.

AtomicInteger защищает одну переменную, а не инвариант:

import kotlinx.coroutines.*
import java.util.concurrent.atomic.AtomicInteger

fun main() = runBlocking {
    val total = AtomicInteger(0)
    val errors = AtomicInteger(0)
    withContext(Dispatchers.Default) {
        List(100) { launch { total.incrementAndGet(); errors.incrementAndGet() } }
            .forEach { it.join() }
    }
    println("каждый атомарен: ${total.get()} ${errors.get()}")
    // каждый атомарен: 100 100
    }

Итог верный, но в любой момент между двумя инкрементами снимок несогласован: читатель может увидеть total = 50, errors = 49. Для связанных значений нужен один объект состояния.

ConcurrentHashMap не делает составные операции атомарными. if (!map.containsKey(k)) map[k] = v - гонка. Нужны computeIfAbsent, putIfAbsent или merge.

Обычный mutableListOf в корутинах небезопасен, даже если кажется, что обращения редки. ArrayList.add при конкурентном вызове может потерять элемент или бросить ArrayIndexOutOfBoundsException из-за состояния расширения массива.

Мелкозернистый конфайнмент убивает производительность. withContext(single) { counter++ } внутри цикла - это переключение контекста на каждую операцию. Выносите весь блок работы в один withContext.

Два мьютекса в разном порядке - deadlock, как и в Java. Правило то же: фиксированный порядок захвата или единый лок.

Semaphore с утечкой разрешений исчерпывается. Ручная пара acquire()/release() без try/finally при исключении потеряет разрешение навсегда. Используйте withPermit.

limitedParallelism(1) - не замена Mutex. Он гарантирует один поток для своих задач, но состояние может меняться и из других контекстов. Конфайнмент работает только если все обращения идут через этот диспетчер.

@Volatile не делает ++ атомарным. Он даёт только видимость записи. @Volatile var counter = 0 с counter++ из нескольких корутин теряет инкременты точно так же, как обычное поле.

Актор сериализует все операции. Если команд много, одна корутина станет узким местом - её очередь будет расти. Для высокой нагрузки нужны шардирование по ключу или атомарные типы.

Далее: Корутины против потоков