공유 변경 가능 상태와 동시성
공유 변경 가능 상태와 동시성 (Shared mutable state and concurrency)
코루틴은 Dispatchers.Default 같은 멀티스레드 디스패처를 사용해 병렬로 실행될 수 있어요. 이것은 일반적인 병렬성 문제들을 모두 제시해요. 가장 주요한 문제는 공유 변경 가능 상태에 대한 접근의 동기화예요. 코루틴 세계에서 이 문제에 대한 몇몇 해결책은 멀티스레드 세계의 해결책과 비슷하지만, 다른 것들은 독특해요.
출처: Kotlin 공식 문서
본문
문제
100개의 코루틴을 시작해 모두 같은 동작을 천 번씩 수행하게 해 볼게요. 추가 비교를 위해 완료 시간도 측정할 거예요.
suspend fun massiveRun(action: suspend () -> Unit) {
val n = 100 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
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 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
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
전체 코드는 여기에서 얻을 수 있어요. 결국 무엇을 출력할까요? "Counter = 100000"을 출력할 가능성은 극히 낮아요. 100개의 코루틴이 어떤 동기화도 없이 여러 스레드에서 counter를 동시에 증가시키기 때문이에요.
volatile은 도움이 되지 않아요
volatile로 변수를 만드는 것이 동시성 문제를 해결한다는 흔한 오해가 있어요. 한번 시도해 볼게요.
import kotlinx.coroutines.*
import kotlin.system.*
suspend fun massiveRun(action: suspend () -> Unit) {
val n = 100 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
repeat(n) {
launch {
repeat(k) { action() }
}
}
}
}
println("Completed ${n * k} actions in $time ms")
}
//sampleStart
@Volatile // Kotlin에서 `volatile`은 애노테이션
var counter = 0
fun main() = runBlocking {
withContext(Dispatchers.Default) {
massiveRun {
counter++
}
}
println("Counter = $counter")
}
//sampleEnd
전체 코드는 여기에서 얻을 수 있어요. 이 코드는 더 느리게 동작하지만, 여전히 결국 "Counter = 100000"을 항상 얻지는 못해요. volatile 변수는 해당 변수에 대한 선형화 가능한(linearizable, "원자적"을 뜻하는 기술 용어) 읽기와 쓰기를 보장하지만, 더 큰 동작(우리의 경우 증가)의 원자성을 제공하지 않기 때문이에요.
스레드 안전 데이터 구조
스레드와 코루틴 모두에 동작하는 일반적인 해결책은 스레드 안전(synchronized, linearizable, atomic이라고도 함) 데이터 구조를 사용하는 것이에요. 이 구조는 공유 상태에서 수행해야 하는 해당 연산에 필요한 모든 동기화를 제공해요. 단순한 카운터의 경우 원자적 incrementAndGet 연산을 가진 AtomicInteger 클래스를 사용할 수 있어요.
import kotlinx.coroutines.*
import java.util.concurrent.atomic.*
import kotlin.system.*
suspend fun massiveRun(action: suspend () -> Unit) {
val n = 100 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
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
전체 코드는 여기에서 얻을 수 있어요. 이것은 이 특정 문제에 대한 가장 빠른 해결책이에요. 단순한 카운터, 컬렉션, 큐, 그 외 표준 데이터 구조와 그것들에 대한 기본 연산에 잘 동작해요. 하지만 복잡한 상태나 바로 사용할 수 있는 스레드 안전 구현이 없는 복잡한 연산으로는 쉽게 확장되지 않아요.
세밀한 스레드 제한
스레드 제한(thread confinement)은 특정 공유 상태에 대한 모든 접근을 단일 스레드로 제한하는, 공유 변경 가능 상태 문제에 대한 접근 방식이에요. 이것은 일반적으로 모든 UI 상태가 단일 이벤트 디스패치/애플리케이션 스레드로 제한되는 UI 애플리케이션에서 사용돼요. 단일 스레드 컨텍스트를 사용하면 코루틴으로 쉽게 적용할 수 있어요.
import kotlinx.coroutines.*
import kotlin.system.*
suspend fun massiveRun(action: suspend () -> Unit) {
val n = 100 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
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 {
// 각 증가를 단일 스레드 컨텍스트로 제한
withContext(counterContext) {
counter++
}
}
}
println("Counter = $counter")
}
//sampleEnd
전체 코드는 여기에서 얻을 수 있어요. 이 코드는 세밀한 스레드 제한을 수행하기 때문에 아주 느리게 동작해요. 각 개별 증가가 withContext(counterContext) 블록을 사용해 멀티스레드 Dispatchers.Default 컨텍스트에서 단일 스레드 컨텍스트로 전환하기 때문이에요.
조잡한 스레드 제한
실무에서 스레드 제한은 큰 덩어리로 수행돼요. 예를 들어 상태를 갱신하는 큰 비즈니스 로직 조각들이 단일 스레드로 제한되죠. 다음 예시는 각 코루틴을 처음부터 단일 스레드 컨텍스트에서 실행해 그렇게 해요.
import kotlinx.coroutines.*
import kotlin.system.*
suspend fun massiveRun(action: suspend () -> Unit) {
val n = 100 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
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(counterContext) {
massiveRun {
counter++
}
}
println("Counter = $counter")
}
//sampleEnd
전체 코드는 여기에서 얻을 수 있어요. 이제 이것은 훨씬 빠르게 동작하고 올바른 결과를 만들어요.
상호 배제
이 문제에 대한 상호 배제(mutual exclusion) 해결책은 공유 상태의 모든 수정을 결코 동시에 실행되지 않는 임계 구역으로 보호하는 거예요. 차단 세계에서는 보통 synchronized나 ReentrantLock을 사용해요. 코루틴의 대안은 Mutex라고 불러요. 그것은 임계 구역을 구분하는 lock과 unlock 함수를 가져요. 핵심 차이는 Mutex.lock()이 일시 중단 함수라는 것이에요. 스레드를 막지 않죠.
또한 mutex.lock(); try { ... } finally { mutex.unlock() } 패턴을 편리하게 나타내는 withLock 확장 함수도 있어요.
import kotlinx.coroutines.*
import kotlinx.coroutines.sync.*
import kotlin.system.*
suspend fun massiveRun(action: suspend () -> Unit) {
val n = 100 // 시작할 코루틴 수
val k = 1000 // 각 코루틴이 동작을 반복하는 횟수
val time = measureTimeMillis {
coroutineScope { // 코루틴 스코프
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 {
// 각 증가를 lock으로 보호
mutex.withLock {
counter++
}
}
}
println("Counter = $counter")
}
//sampleEnd
전체 코드는 여기에서 얻을 수 있어요. 이 예시의 잠금은 세밀해서 그 대가를 치러요. 하지만 반드시 공유 상태를 주기적으로 수정해야 하는데, 그 상태가 제한될 자연스러운 스레드가 없는 상황에서는 좋은 선택이에요.