Chapter 1: Safety - Item 2: Eliminate critical sections

여러 스레드가 공유된 상태(shared state) 를 동시에 수정하게 되면,
예상하지 못한 결과가 발생할 수 있습니다.

이 문제는 이전 항목에서 이미 논의되었지만,
여기에서는 이를 더 자세히 설명하고,
Kotlin/JVM 환경에서 이 문제를 어떻게 다루는지를 설명하고자 합니다

The problem with threads and shared state

제가 이 글을 쓰고 있는 동안에도 제 컴퓨터에서는 많은 일들이 동시에 일어나고 있습니다.
음악이 재생되고 있고, IntelliJ는 이 장의 내용을 화면에 표시하고 있으며, Slack은 메시지를 보여주고 있고, 브라우저는 데이터를 다운로드하고 있습니다.

 

이 모든 것이 가능한 이유는 운영체제가 스레드(thread) 라는 개념을 도입했기 때문입니다.
운영체제는 여러 스레드의 실행을 스케줄링하며, 각 스레드는 서로 독립적인 실행 흐름을 가집니다.
설령 단일 코어 CPU를 사용하고 있더라도, 운영체제는 하나의 스레드를 아주 짧은 시간 동안 실행한 뒤 다른 스레드로 전환하는 방식을 반복함으로써 여러 스레드를 동시에 실행하는 것처럼 보이게 할 수 있습니다.

이를 타임 슬라이싱(time slicing) 이라고 합니다.

 

더 나아가, 현대의 컴퓨터는 여러 개의 코어를 가지고 있기 때문에 운영체제는 실제로 여러 작업을 서로 다른 스레드에서 동시에 실행할 수 있습니다.

 

이 과정에서 가장 큰 문제는, 운영체제가 언제 한 스레드의 실행을 멈추고 다른 스레드로 전환할지

우리가 확실히 알 수 없다는 점입니다.

다음 예제를 생각해 보겠습니다. 1,000개의 스레드를 생성하고, (thread{})

각 스레드가 하나의 가변 변수(mutable variable)를 증가시키도록 합니다.

문제는 값 하나를 증가시키는 작업이 단일 단계가 아니라는 점입니다.

이 작업은 현재 값을 읽고, 증가된 새 값을 만들고, 그 값을 다시 변수에 할당하는 여러 단계로 이루어져 있습니다.

 

만약 운영체제가 이 단계들 사이에서 스레드를 전환해 버리면, 일부 증가 연산이 누락될 수 있습니다.
이 때문에 아래의 코드는 1,000을 출력할 가능성이 매우 낮습니다. 실제로 테스트해 보았을 때, 981이 출력되었습니다.

 

kotlin

var num = 0
for (i in 1..1000) {
    thread {
        Thread.sleep(10)
        num += 1
    }
}
Thread.sleep(5000)
print(num) // 1000이 될 가능성이 매우 낮음
// 매번 다른 숫자

이 문제를 더 잘 이해하기 위해, 두 개의 스레드를 실행했을 때 발생할 수 있는 다음과 같은 상황을 생각해 보겠습니다.
한 스레드가 변수의 값을 0으로 읽은 뒤, CPU가 다른 스레드로 실행을 전환합니다. 그러면 두 번째 스레드도 동일하게 값 0을 읽고, 이를 증가시켜 변수에 1을 설정합니다. 이후 운영체제가 다시 첫 번째 스레드로 전환하면, 첫 번째 스레드는 역시 변수에 1을 설정하게 됩니다. 이 경우 증가 연산 하나가 사라진 셈이 됩니다.

 

이처럼 일부 연산이 누락되는 것은 실제 애플리케이션에서 심각한 문제가 될 수 있으며,

이 문제는 그보다 훨씬 더 심각한 결과로 이어질 수도 있습니다.
연산이 어떤 순서로 실행될지 알 수 없을 때, 객체가 잘못된 상태를 가지게 될 위험이 생깁니다. 이는 재현하거나 수정하기가 매우 어려운 버그로 이어지는 경우가 많습니다.

 

이러한 문제는 한 스레드가 리스트의 요소를 순회하고 있는 동안, 다른 스레드가 그 리스트에 요소를 추가하는 상황을 떠올리면 잘 이해할 수 있습니다. 기본 컬렉션들은 순회 중에 요소가 수정되는 것을 허용하지 않기 때문에, 이런 경우 ConcurrentModificationException 예외가 발생하게 됩니다.

 

kotlin

var numbers = mutableListOf<Int>()
for (i in 1..1000) {
    thread {
        Thread.sleep(1)
        numbers.add(i)
    }
    thread {
        Thread.sleep(1)
        print(numbers.sum()) // sum이 리스트를 반복함
        // 종종 ConcurrentModificationException
    }
}

 

여러 개의 스레드를 사용하는 디스패처에서 여러 코루틴을 실행할 때에도 동일한 문제가 발생합니다.
코루틴을 사용할 경우에도, 이 문제를 해결하기 위해서는 스레드에서 사용하던 것과 같은 기법들을 적용할 수 있습니다. 다만, 코루틴에는 이에 더해 전용으로 제공되는 도구들도 있으며, 이에 대해서는 《Kotlin Coroutines: Deep Dive》에서 자세히 설명한 바 있습니다.

 

앞선 장에서 설명했듯이, 가변성(mutability)을 사용하지 않는다면 이러한 문제들 대부분은 발생하지 않습니다. 그러나 실제 애플리케이션에서는 가변성을 완전히 피하기가 어려운 경우가 많기 때문에, 공유 상태(shared state)를 어떻게 다뤄야 하는지를 반드시 익혀야 합니다.

 

여러 스레드에 의해 수정될 수 있는 공유 상태가 존재한다면, 해당 상태에 대한 모든 연산이 올바르게 실행되도록 보장해야 합니다. 이를 위해 각 플랫폼은 서로 다른 도구들을 제공하므로, 이제 Kotlin/JVM에서 가장 중요한 도구들에 대해 살펴보겠습니다.

Synchronization in Kotlin/JVM

Kotlin/JVM 플랫폼에서 공유 상태(shared state)를 다루기 위한 가장 중요한 도구는 동기화(synchronization) 입니다.
동기화는 한 번에 하나의 스레드만 특정 코드 블록을 실행할 수 있도록 보장해 주는 메커니즘입니다.

 

이는 synchronized 함수를 기반으로 하며,
락 객체(lock object) 와 동기화할 코드가 담긴 람다 표현식을 필요로 합니다.
이 메커니즘은 동일한 락을 사용하는 동기화 블록에는 동시에 하나의 스레드만 진입할 수 있도록 보장합니다.

 

어떤 스레드가 동기화 블록에 도달했을 때, 이미 다른 스레드가 같은 락으로 동기화 블록을 실행 중이라면, 해당 스레드는 기존 스레드의 실행이 끝날 때까지 대기하게 됩니다.

다음 예제는 이러한 동기화를 사용하여 num 변수가 올바르게 증가되도록 보장하는 방법을 보여줍니다.

 

kotlin

val lock = Any()
var num = 0
for (i in 1..1000) {
    thread {
        Thread.sleep(10)
        synchronized(lock) {
            num += 1
        }
    }
}
Thread.sleep(1000)
print(num) // 1000

 

실제 애플리케이션에서는 동기화가 필요한 모든 함수들을 하나의 클래스 안에서 동기화 블록으로 감싸는 경우가 많습니다.
아래 예제는 Counter 클래스에 포함된 모든 연산을 동기화하는 방법을 보여줍니다.

 

 

 

kotlin

class Counter {
    private val lock = Any()
    private var num = 0
    
    fun inc() = synchronized(lock) {
        num += 1
    }
    
    fun dec() = synchronized(lock) {
        num -= 1
    }
    
    // Synchronization is not necessary; however,
    // without it, getter might serve stale value
    fun get(): Int = num
}

 

일부 클래스에서는 상태의 서로 다른 부분에 대해 여러 개의 락(lock)을 사용하는 경우도 있지만,

이는 올바르게 구현하기가 훨씬 어렵기 때문에 실제로는 많이 사용되지 않습니다.

 

Kotlin 코루틴을 사용할 때에는 synchronized를 사용하는 대신,
《Kotlin Coroutines: Deep Dive》에서 설명했듯이 단일 스레드로 제한된 디스패처나 Mutex 를 사용하는 편이 일반적입니다.

 

또한 스레드 전환(thread switching)은 비용이 없는 작업이 아니라는 점을 기억해야 합니다.
일부 클래스의 경우에는 여러 스레드를 사용하고 그 실행을 동기화하는 것보다,
하나의 스레드에서 모든 작업을 처리하는 것이 더 효율적일 수 있습니다.

Atomic objects

우리는 변수 증가 문제에 대한 논의로 시작했습니다. 일반적인 정수 증가 연산은 여러 단계로 이루어져 있기 때문에,

그 중간에 운영체제가 스레드를 전환하면 잘못된 결과가 발생할 수 있습니다.

 

일부 연산, 예를 들어 단순한 값 할당과 같은 경우는 프로세서의 단일 단계로 수행되므로 항상 올바르게 실행됩니다.

하지만 이러한 연산은 극히 단순한 경우에만 해당하며, 본질적으로 원자적(atomic) 인 연산은 매우 제한적입니다.

 

이와 달리 Java는 일반적으로 사용되는 클래스들을 원자적 연산을 보장하는 형태로 제공하는 원자 클래스(atomic classes) 집합을 제공합니다.
대표적으로 AtomicInteger, AtomicLong, AtomicBoolean, AtomicReference 등이 있으며,

이들 클래스는 원자적으로 실행됨이 보장된 메서드들을 제공합니다.

 

예를 들어 AtomicInteger는 값을 증가시키고 그 결과를 반환하는 incrementAndGet 메서드를 제공합니다.
아래 예제는 AtomicInteger를 사용하여 변수를 올바르게 증가시키는 방법을 보여줍니다.

 

 

kotlin

val num = AtomicInteger(0)
for (i in 1..1000) {
    thread {
        Thread.sleep(10)
        num.incrementAndGet()
    }
}
Thread.sleep(5000)
print(num.get()) // 1000

원자 객체(atomic objects)는 성능이 뛰어나며, 상태가 단일 값이거나 서로 독립적인 몇 개의 값으로만 구성된 단순한 경우에는 큰 도움이 됩니다.
그러나 더 복잡한 경우에는 이것만으로는 충분하지 않습니다.

 

예를 들어, 여러 객체에 대한 여러 연산을 하나의 일관된 흐름으로 동기화해야 하는 상황에서는 원자 객체만으로는 이를 처리할 수 없습니다.
이러한 경우에는 동기화 블록(synchronization block) 을 사용해야 합니다.

 

Concurrent collections

Java는 동시성(concurrency)을 지원하는 컬렉션들도 함께 제공합니다.
그중에서도 가장 중요한 것은 ConcurrentHashMap으로, 이는 HashMap의 스레드 안전(thread-safe) 버전입니다.

 

ConcurrentHashMap의 모든 연산은 충돌을 걱정하지 않고 안전하게 사용할 수 있습니다.
또한 이를 순회(iteration)할 때는, 순회가 시작된 시점의 상태에 대한 스냅샷(snapshot) 을 기준으로 동작하므로 ConcurrentModificationException 예외가 발생하지 않습니다.

 

충돌로 구조가 깨지진 않으나 항상 가장 최신 상태를 보장한다는 의미는 아니며,
순회 중에 발생한 변경 사항이 결과에 반영되지 않을 수도 있습니다.

 

map.toList()는 내부적으로 iterator로 순회하며,

동시에 다른 스레드에서 계속 값이 추가되고 있기 때문에

조회시 최신 상태를 반영할수도, 안할수도 있습니다.

 

 

kotlin

val map = ConcurrentHashMap<Int, String>()
for (i in 1..1000) {
    thread {
        Thread.sleep(1)
        map.put(i, "E$i")
    }
    thread {
        Thread.sleep(1)
        print(map.toList().sumOf { it.first })
    }
}

동시성 집합(concurrent set)이 필요할 때는,
ConcurrentHashMap의 newKeySet을 사용하는 것이 널리 쓰이는 방법입니다.
이는 값으로 Unit을 사용하는 ConcurrentHashMap을 감싼 래퍼(wrapper) 입니다.

 

newKeySet은 MutableSet 인터페이스를 구현하고 있기 때문에,
일반적인 Set과 동일한 방식으로 사용할 수 있습니다.

 

kotlin

val set = ConcurrentHashMap.newKeySet<Int>()
for (i in 1..1000) {
    thread {
        Thread.sleep(1)
        set += i
    }
}
Thread.sleep(5000)
println(set.size)

 

리스트 대신에는, 중복을 허용하는 동시성 컬렉션이 필요할 때 보통 ConcurrentLinkedQueue를 사용합니다.
이 컬렉션은 동시 접근 환경에서도 안전하게 사용할 수 있습니다.

 

이와 같은 도구들이 JVM에서 가변 상태(mutable state) 문제를 다루기 위해 사용할 수 있는 핵심적인 수단들입니다.

 

물론 코드 동기화를 지원하는 다른 도구들을 제공하는 라이브러리들도 존재합니다.
그중에는 AtomicFU와 같이 멀티플랫폼에서 사용할 수 있는 원자 객체를 제공하는 Kotlin 멀티플랫폼 라이브러리도 있습니다.

 

kotlin

// Using AtomicFU
val num = atomic(0)
for (i in 1..1000) {
    thread {
        Thread.sleep(10)
        num.incrementAndGet()
    }
}
Thread.sleep(5000)
print(num.value) // 1000

이제 시점을 다시 가변 상태(mutable state)에 대한 보다 일반적인 문제로 돌려서,
일반적인 상황에서 이를 어떻게 다뤄야 하는지를 설명해 보겠습니다.

 

 

Do not leak mutation points

다음 예제들처럼 공개된 상태(public state)를 표현하기 위해 가변 객체를 그대로 노출하는 것은
특히 위험한 상황입니다.
다음 예제를 한 번 살펴보시기 바랍니다.

 

 

kotlin

data class User(val name: String)

class UserRepository {
    private val users: MutableList<User> = mutableListOf()
    
    fun loadAll(): MutableList<User> = users
    
    //...
}

 

loadAll을 사용하면 UserRepository의 private 상태를 외부에서 수정할 수 있게 됩니다.

 

kotlin

val userRepository = UserRepository()

val users = userRepository.loadAll()
users.add(User("Kirill"))
//...

print(userRepository.loadAll()) // [User(name=Kirill)]

이러한 상황은 의도하지 않은 수정이 발생할 수 있다는 점에서 특히 위험합니다.
따라서 가장 먼저 해야 할 일은, 가변 객체를 읽기 전용 타입으로 업캐스팅(upcasting)하는 것입니다.
이 경우에는 MutableList를 List로 업캐스팅하는 것을 의미합니다.

 

 

 

kotlin

data class User(val name: String)

class UserRepository {
    private val users: MutableList<User> = mutableListOf()
    
    fun loadAll(): List<User> = users
    
    //...
}

 

하지만 주의해야 합니다. 위와 같은 구현만으로는 이 클래스를 안전하게 만든다고 할 수 없습니다.
겉으로 보기에는 읽기 전용 리스트를 반환하는 것처럼 보이지만, 실제로는 가변 리스트에 대한 참조이기 때문에 그 값이 변경될 수 있습니다.

이로 인해 개발자들이 심각한 실수를 저지를 가능성이 생기게 됩니다.

 

kotlin

data class User(val name: String)

class UserRepository {
    private val users: MutableList<User> = mutableListOf()
    
    fun loadAll(): List<User> = users
    
    fun add(user: User) {
        users += user
    }
}

class UserRepositoryTest {
    fun `should add elements`() {
        val repo = UserRepository()
        val oldElements = repo.loadAll()
        repo.add(User("B"))
        val newElements = repo.loadAll()
        assert(oldElements != newElements)
        // 이 assertion은 실패합니다. 두 참조가
        // 동일한 객체를 가리키며, 서로 같기 때문입니다
    }
}

둘째로, 한 스레드가 loadAll을 통해 반환된 리스트를 읽고 있는 동안,
다른 스레드가 동시에 해당 리스트를 수정하는 상황을 생각해 보아야 합니다.

 

다른 스레드가 순회(iteration) 중인 가변 컬렉션을 수정하는 것은 허용되지 않는 동작이며,
이러한 경우 예상하지 못한 예외가 발생하게 됩니다.

 

 

kotlin

val repo = UserRepository()
thread {
    for (i in 1..10000) repo.add(User("User$i"))
}
thread {
    for (i in 1..10000) {
        val list = repo.loadAll()
        for (e in list) {
            /* no-op */
        }
    }
}
// ConcurrentModificationException

 

이 문제를 해결하는 방법에는 두 가지가 있습니다.
첫 번째 방법은 실제 객체에 대한 참조를 반환하는 대신, 객체의 복사본을 반환하는 것입니다.
이를 방어적 복사(defensive copying) 라고 합니다.

 

다만 복사를 수행하는 도중에, 다른 스레드가 동시에 리스트에 새로운 요소를 추가하고 있다면 충돌이 발생할 수 있다는 점에 유의해야 합니다.
따라서 객체에 대한 멀티스레드 접근을 지원하려면, 이 복사 작업 역시 동기화되어야 합니다.

 

컬렉션은 toList와 같은 변환 함수를 사용해 복사할 수 있으며,
데이터 클래스의 경우에는 copy 메서드를 사용하여 복사할 수 있습니다.

 

 

 

kotlin

class UserRepository {
    private val users: MutableList<User> = mutableListOf()
    private val LOCK = Any()
    
    fun loadAll(): List<User> = synchronized(LOCK) {
        users.toList()
    }
    
    fun add(user: User) = synchronized(LOCK) {
        users += user
    }
  }

 

더 간단한 방법은 읽기 전용 리스트(read-only list)를 사용하는 것입니다.
이 방식은 보안을 확보하기가 더 쉽고, 객체의 변경 사항을 추적할 수 있는 방법도 더 많이 제공해 줍니다.

 

 

kotlin

class UserRepository {
    private var users: List<User> = listOf()
    
    fun loadAll(): List<User> = users
    
    fun add(user: User) {
        users = users + user
    }
}

이 방식을 사용할 경우, 멀티스레드 접근을 제대로 지원하기 위해서는 리스트를 수정하는 연산만 동기화하면 됩니다.
이로 인해 요소를 추가하는 작업은 다소 느려질 수 있지만, 리스트를 조회하는 작업은 더 빠르게 수행할 수 있습니다.

이는 쓰기보다 읽기 작업이 더 많은 경우에 매우 좋은 트레이드오프가 됩니다.

 

kotlin

class UserRepository {
    private var users: List<User> = listOf()
    private val LOCK = Any()
    
    fun loadAll(): List<User> = users
    
    fun add(user: User) = synchronized(LOCK) {
        users = users + user
    }
}