Управляемое состояние и конкурентность

Проблема

Предположим, мы запускаем сто корутин, которые выполняют одно и то же действие тысячу раз. Мы также измерим время их завершения для дальнейшего сравнения:

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

Мы начнём с очень простого действия, которое увеличивает общее изменяемое значение с использованием многопотокового Dispatchers.Default.

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

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

//sampleStart
var counter = 0

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            counter++
        }
    }
    println("Counter = $counter")
}
//sampleEnd    

Что он выведет в итоге? В высшей степени маловероятно, что он когда-либо выведет «Счётчик = 100000», потому что сто корутин увеличивают counter одновременно из нескольких потоков без какой-либо синхронизации.

Изменчивые переменные не помогают

Существует распространённое заблуждение, что создание переменной volatile решает проблему конкурентности. Давайте попробуем:

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

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

//sampleStart
@Volatile // in Kotlin `volatile` is an annotation 
var counter = 0

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            counter++
        }
    }
    println("Counter = $counter")
}
//sampleEnd    

Этот код работает медленнее, но мы всё равно не всегда получаем «Счётчик = 100000» в конце, потому что изменчивые переменные гарантируют линейно-выполнимые (это технический термин для «атомарные») чтение и запись соответствующей переменной, но не обеспечивают атомарность более крупных действий (инкремента в нашем случае).

Потокобезопасные структуры данных

Общим решением, которое работает как для потоков, так и для корутин, является использование потокобезопасной (также известной как синхронизированной, линейно-выполнимой или атомарной) структуры данных, которая обеспечивает необходимую синхронизацию для соответствующих операций, которые нужно выполнить над общим состоянием. В случае простого счётчика мы можем использовать AtomicInteger класс, который имеет атомарные incrementAndGet операции:

import kotlinx.coroutines.*
import java.util.concurrent.atomic.*
import kotlin.system.*

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

//sampleStart
val counter = AtomicInteger()

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            counter.incrementAndGet()
        }
    }
    println("Counter = $counter")
}
//sampleEnd    

Это самое быстрое решение для этой конкретной задачи. Оно работает для простых счётчиков, коллекций, очередей и других стандартных структур данных и основных операций над ними. Однако масштабирование до сложного состояния или сложных операций, для которых нет готовых потокобезопасных реализаций, затруднительно.

Точная привязка к потоку

Привязка к потоку — подход к проблеме общего изменяемого состояния, где весь доступ к конкретному общему состоянию ограничен одним потоком. Обычно используется в приложениях с графическим интерфейсом, где всё состояние графического интерфейса привязано к единственному потоку обработки событий/приложения. Легко реализуется с помощью корутин с использованием контекста одиночного потока.

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

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

//sampleStart
val counterContext = newSingleThreadContext("CounterContext")
var counter = 0

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            // confine each increment to a single-threaded context
            withContext(counterContext) {
                counter++
            }
        }
    }
    println("Counter = $counter")
}
//sampleEnd      

Этот код работает очень медленно, потому что он реализует точную привязку к потоку. Каждый отдельный инкремент переключается с многопотокового контекста Dispatchers.Default на контекст одиночного потока с использованием блока withContext(counterContext).

Групповая привязка к потоку

На практике привязка к потоку выполняется большими блоками, например, большие фрагменты логики обновления состояния привязаны к одному потоку. Следующий пример делает это таким образом, выполняя каждую корутину в первую очередь в контексте одиночного потока.

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

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

//sampleStart
val counterContext = newSingleThreadContext("CounterContext")
var counter = 0

fun main() = runBlocking {
    // confine everything to a single-threaded context
    withContext(counterContext) {
        massiveRun {
            counter++
        }
    }
    println("Counter = $counter")
}
//sampleEnd     

Теперь это работает намного быстрее и даёт правильный результат.

Взаимное исключение

Решение проблемы с помощью взаимного исключения заключается в защите всех модификаций общего состояния с помощью критической секции, которая никогда не выполняется одновременно. В блокирующем мире для этого обычно используются synchronized или ReentrantLock. Альтернатива корутин называется Mutex. Она имеет функции lock и unlock для определения критической секции. Ключевое отличие в том, что Mutex.lock() — это приостанавливаемая функция. Она не блокирует поток.

Также существует расширяющая функция withLock, которая удобно представляет mutex.lock(); try { ... } finally { mutex.unlock() } шаблон:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
import kotlin.system.*

suspend fun massiveRun(action: suspend () -> Unit) {
    val n = 100  // number of coroutines to launch
    val k = 1000 // times an action is repeated by each coroutine
    val time = measureTimeMillis {
        coroutineScope { // scope for coroutines 
            repeat(n) {
                launch {
                    repeat(k) { action() }
                }
            }
        }
    }
    println("Completed ${n * k} actions in $time ms")    
}

//sampleStart
val mutex = Mutex()
var counter = 0

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            // protect each increment with lock
            mutex.withLock {
                counter++
            }
        }
    }
    println("Counter = $counter")
}
//sampleEnd    

Блокировка в этом примере мелкозернистая, поэтому она имеет цену. Тем не менее, это хороший выбор в некоторых ситуациях, когда вам абсолютно необходимо периодически изменять какое-то общее состояние, но нет естественного потока, к которому это состояние привязано.