[코루틴의 정석] 11-1장. 코루틴 심화

11.1. 공유 상태를 사용하는 코루틴의 문제와 데이터 동기화

 

11.1.1. 가변 변수를 사용할 때의 문제점

 

멀티 스레드 환경에서 가변 변수에 동시에 접근해 값을 변경하면 데이터 동기화 문제가 발생할 수 있다.

var count = 0

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        count += 1
      }
    }
  }
  println("count = ${count}")
}

/*
// 결과1:
count = 9062 // 매번 다른 값이 나온다.
// 결과2:
count = 9019 // 매번 다른 값이 나온다.
// 결과3:
count = 8644 // 매번 다른 값이 나온다.
*/

원인은 두 가지로 요약될 수 있다.

 

1. 메모리 가시성(Memory VIsibility): 스레드가 변수를 변경 시킬때 메인 메모리가 아닌 CPU 캐시를 사용할 경우 CPU 캐시의 값이 메모리에 전파되는 데 약간의 시간이 걸려 CPU 캐시와 메인 메모리 간에 데이터 불일치 문제가 생긴다.
2. 경쟁 상태(Race Condition): 2개의 스레드가 동시에 값을 읽고 같은 업데이트 연산이 일어나므로 하나의 연산만 되고 나머지는 손실된다.

 

11.1.2. JVM의 메모리 공간이 하드웨어 메모리 구조와 연결되는 방식

 

 

JVM의 메모리 구조

 

스택 영역

• 원시 타입(Primitive Type) 데이터 저장
• 힙 영역(Heap Area)에 저장된 객체에 대한 참조 저장
  
  힙 영역
  
  
• JVM 스레드에서 공통으로 사용되는 메모리 공간
• 크고 복잡한 데이터 저장 (객체나 배열 등)

 

컴퓨터(하드웨어)의 메모리 구조

 

• CPU 레지스터
• CPU 캐시 메모리
• 메인 메모리

각 CPU는 CPU 캐시 메모리를 두며, 데이터 조회 시 공통 영역인 메인 메모리까지 가지 않고 CPU 캐시 메모리에서 데이터를 조회할 수 있도록 만들어 메모리 엑세스 속도를 향상시킨다.

 

 

 

 

 

 

 

JVM의 메모리 공간인 스택 영역과 힙 영역을 하드웨어 메모리의 구조와 연결한 모습은 다음과 같다.

JVM의 메모리 구조와 하드웨어 메모리 모델의 연결

 

11.1.3. 공유 상태에 대한 메모리 가시성 문제와 해결 방법

 

11.1.3.1. 공유 상태에 대한 메모리 가시성 문제

 

하나의 스레드가 다른 스레드가 변경된 상태를 확인하지 못하는 것을 메모리 가시성 문제라고 하며, 서로 다른 CPU에서 실행되는 스레드들에서 공유 상태를 조회하고 업데이트할 때 생기는 문제이다.

 

1. 메인 메모리에 count = 1000 상태 저장되어 있음

2. CPU 캐시 메모리에서 메인 메모리에서 count값을 읽어와 저장

3. 스레드는 이 값을 이용해 count 값을 증가시키는 연산을 실행

4. 연산이 완료되면 정보를 메인 스레드에 쓰지 않고 CPU 캐시 메모리에 씀

 

1. 메인 메모리에 count = 1000 상태 저장되어 있음

2. CPU 캐시 메모리에서 메인 메모리에서 count값을 읽어와 저장

3. 스레드는 이 값을 이용해 count 값을 증가시키는 연산을 실행

4. 연산이 완료되면 정보를 메인 스레드에 쓰지 않고 CPU 캐시 메모리에 씀

5. CPU 캐시 메모리의 데이터가 메인 메모리로 전파되지 않은 상태에서, 다른 CPU에서 메인 메모리의 count를 읽어옴

6. 각 CPU 캐시 메모리의 값이 메인 메모리로 플러시가 일어남

7. 연산은 두 번 일어나지만 메인 메모리의 count 값은 하나만 증가

 

 

11.1.3.2. @Volatile 사용해 공유 상태에 대한 메모리 가시성 문제 해결하기

 

@Volatile 어노테이션이 설정된 변수를 읽고 쓸 때는 CPU 캐시 메모리를 사용하지 않는다.

@Volatile
var count = 0

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        count += 1
      }
    }
  }
  println("count = ${count}")
}
/*
// 결과:
count = 9122
*/

 

따라서 스레드에서는 값을 메인 메모리에서 곧바로 조회해 오며, 값에 대한 연산도 메인 메모리에서 수행한다.

여전히 여러 스레드에서 메인 메모리의 count 변수에 동시 접근하여 count 변수의 값이 10000이 나오지 않는 것을 볼 수 있다.

 

 

 

11.1.4. 공유 상태에 대한 경쟁 상태 문제와 해결 방법

 

11.1.4.1. 공유 상태에 대한 경쟁 상태 문제

 

각 스레드에서 실행 중인 코루틴들이 count 변수에 동시에 접근할 수 있으므로 같은 연산이 중복으로 실행될 수 있다.
이런 경쟁 상태 문제를 해결하기 위해서는 하나의 변수에 스레드가 동시에 접근할 수 없도록 만들어야 한다.

 

 

 

11.1.4.2. Mutex, ReentrantLock 사용해 동시 접근 제한하기

 

공유 변수의 변경 가능 지점을 임계 영역(Critical Section)으로 만들어 동시 접근을 제한할 수 있다.

Mutex 객체나 ReentrantLock 객체를 통해 임계 영역을 만들 수 있다.

 

Mutex

 

• lock 함수가 호출되면 락이 획득된다.
• unlock이 호출돼 락이 해제될 때까지 다른 코루틴이 해당 임계 영역에 진입할 수 없다.

// Locks this mutex, suspending caller while the mutex is locked
public suspend fun lock(owner: Any? = null)

// Unlocks this mutex. Throws IllegalStateException if invoked on a mutex 
// that is not locked or was locked with a different owner token
public fun unlock(owner: Any? = null)

import kotlinx.coroutines.sync.Mutex

var count = 0
val mutex = Mutex()

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        mutex.lock() // 임계 영역 시작 지점
        count += 1
        mutex.unlock() // 임계 영역 종료 지점
      }
    }
  }
  println("count = ${count}")
}
/*
// 결과:
count = 10000
*/

lock-unlock 쌍을 직접 호출하기보다는 withLock 함수를 사용하는 것이 안전하다.

import kotlinx.coroutines.sync.Mutex

var count = 0
val mutex = Mutex()

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        mutex.withLock {
          count += 1
        }
      }
    }
  }
  println("count = ${count}")
}
/*
// 결과:
count = 10000
*/

 

ReentrantLock

 

Mutex는 주로 코틀린의 코루틴 환경에서 사용되는 반면, ReentrantLock은 자바의 스레드 기반 동시성 제어를 위해 설계된 클래스이다.

• Mutex 객체의 lock 함수는 이미 다른 코루틴에 의해 Mutex 객체에 락이 걸려 있으면 코루틴은 기존 락이 해제될 때까지 스레드를 양보하고 일시 중단한다.
• ReentrantLock 객체의 lock 함수는 락이 해제될 때까지 스레드를 블로킹하고 기다린다.

import java.util.concurrent.locks.ReentrantLock

var count = 0
val reentrantLock = ReentrantLock()

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        reentrantLock.lock() // 스레드를 블록하고 기존의 락이 해제될 때까지 기다림
        count += 1
        reentrantLock.unlock()
      }
    }
  }
  println("count = ${count}")
}
/*
// 결과:
count = 10000
*/

 

11.1.4.3. 원자성 있는 데이터 구조를 사용한 경쟁 상태 문제 해결

 

경쟁 상태를 해결하기 위해 원자성(Atomic) 있는 객체를 사용할 수 있으며, 내부적으로 하드웨어 수준의 CAS(Compare-And-Swap) 명령어를 사용하여 여러 스레드에서 동시에 접근하더라도 하나의 스레드만 접근할 수 있도록 제한하여 원자성을 보장한다.

 

Atomic 객체는 ReentrantLock 객체와 비슷하게 이미 다른 스레드의 코루틴에 Atomic 객체에 대한 연산을 실행 중인 경우 코루틴은 스레드를 블로킹하고 연산 중인 스레드가 연산을 모두 수행할 때까지 기다린다.

var count = AtomicInteger(0)

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        count.getAndUpdate { // 만약 다른 스레드가 연산을 실행 중이면 코루틴은 스레드를 블로킹 시키고 대기한다.
          it + 1 // count값 1 더하기
        }
      }
    }
  }
  println("count = ${count}")
}
/*
// 결과:
count = 10000
*/

AtomicReference를 사용해 복잡한 객체에 대한 원자성을 부여할 수 있다.

data class Counter(val name: String, val count: Int)
val atomicCounter = AtomicReference(Counter("MyCounter", 0)) // 원자성 있는 Counter 만들기

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        atomicCounter.getAndUpdate {
          it.copy(count = it.count + 1) // MyCounter의 count값 1 더하기
        }
      }
    }
  }
  println(atomicCounter.get())
}
/*
// 결과:
Counter(name=MyCounter, count=10000)
*/

 

Atomic 객체를 사용할 때 객체의 읽기와 쓰기를 따로 실행하면 안된다.

읽기와 쓰기를 따로 사용하면 값이 변경되기 전에 다른 스레드에서 값에 대한 접근이 가능해지기 때문이다.

getAndUpdate나 incrementAndGet과 같이 읽기와 쓰기를 함께 실행하는 함수를 사용해야 한다.

var count = AtomicInteger(0)

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch {
        val currentCount = count.get()
        // 위 코드와 아래 코드의 실행 사이에 다른 스레드가 count의 값을 읽거나 변경할 수 있다.
        count.set(currentCount + 1)
      }
    }
  }
  println("count = ${count}")
}
/*
// 결과:
count = 8399
*/

 

 

 

11.1.4.4. 공유 상태 변경을 위해 전용 스레드 사용하기

 

하나의 변수에 스레드가 동시에 접근할 수 없도록 공유 변수에 접근할 때 하나의 스레드만 사용하도록 강제할 수도 있다.

var count = 0
val countChangeDispatcher = newSingleThreadContext("CountChangeThread")

fun main() = runBlocking<Unit> {
  withContext(Dispatchers.Default) {
    repeat(10_000) {
      launch { // count 값을 변경 시킬 때만 사용
        increaseCount()
      }
    }
  }
  println("count = ${count}")
}

suspend fun increaseCount() = coroutineScope {
  withContext(countChangeDispatcher) {
    count += 1
  }
}
/*
// 결과:
count = 10000
*/