공유 상태를 사용하는 코루틴의 문제와 해결책, 코루틴의 동작 방식

공유 상태를 사용하는 코루틴의 문제와 해결책

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

• 스레드 간에 데이터를 전달하거나 자원을 공유하는 경우에 가변 변수를 통해 공유하고 업데이트 하는 경우 여러 스레드에서 가변 변수에 동시에 접근해 값을 변경하며 데이터 손실이나 불일치로 버그가 발생할 수 있다. → 공유 상태를 사용할 때의 데이터 동기화 문제

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

• JVM은 스레드마다 스택 영역이라 불리는 메모리 공간을 갖고 있고, 이 스택 영역에는 원시 타입의 데이터나 힙 영역에 저장된 객체에 대한 참조가 저장된다.
• 힙 영역은 JVM에 올라간 스레드들에서 공통으로 사용되는 메모리 공간으로 복잡한 데이터(객체, 배열 등)가 저장된다.

• 컴퓨터는 CPU 레지스터, CPU 캐시 메모리, 메인 메모리 영역을 구성된다.
• 각 CPU는 CPU 캐시 메모리를 중간에 두고, 데이터 조회 시 메인 메모리까지 가지 않고, CPU 캐시 메모리를 조회할 수 있게 해 메모리 접근 속도를 향상 시킨다.

• JVM의 스택 영역과 힙 영역에 저장되는 데이터는 컴퓨터의 각 메모리 공간 중 어디에나 저장될 수 있다.
• 이로 인해 공유 상태의 메모리 가시성 문제 와 공유 상태에 대한 경쟁 상태 문제 가 발생할 수 있다.

---

공유 상태의 메모리 가시성 문제

• 하나의 스레드가 다른 스레드가 변경한 상태를 확인하지 못하는 문제
• 서로 다른 CPU에서 실행되는 스레드들에서 공유 상태를 조화하고 업데이트 할 때 생긴다.

[해결 방법]

• 변수 선언 시 @Volatile 어노테이션을 사용해 변수의 값을 읽고 쓸 때 CPU 캐시 메모리를 사용하지 않게 한다.

→ 공유 상태에 대한 경쟁 상태 문제 때문에 문제가 확실히 해결되지 않음

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

• 여러 스레드가 동시에 하나의 값에 접근하면서 발생하는 문제

[해결 방법]

1. Mutex 사용

• 공유 변수의 변경 가능 지점을 Mutex 객체를 사용하여 임계 영역으로 만들어 동시 접근을 제한할 수 있다.
• 코루틴이 Mutex 객체의 lock 일시 중단 함수를 호출하면 락이 획득되고, 해당 Mutex 객체에 대해 unlock이 호출될 때까지 다른 코루틴이 해당 임계 영역에 진입할 수 없다.

var count = 0
val mutex = Mutex()

fun main() = runBlocking<Unit> {
    withContext(Dispatchers.Default) {
        repeat(10000) {
            launch {
                mutex.lock()
                count += 1
                mutex.unlock()
            }
        }
    }
    println("count = $count")
}

• lock 함수와 unlock 함수를 직접 호출하는 것은 코드가 복잡해질수록 문제가 발생할 수 있다.

👉 lock-unlock 쌍을 직접 호출하는 대신 withLock 함수를 호출하는 것이 안전하다.

launch {
    mutex.withLock {
        count += 1
    }
}

Mutex 사용이 권장되는 이유

• ReentrantLock의 lock 함수는 만약 특정 스레드에서 락을 획득했다면, 다른 스레드에서 lock 함수를 호출할 경우 해당 스레드를 락이 해제될 때까지 블로킹 시킨다.
• Mutex의 lock 함수는 일시 중단 함수로, 만약 특정 코루틴이 락을 획득했다면, 다른 코루틴에서 lock 함수를 호출할 경우 해당 코루틴은 락이 해제될 때까지 일시 중단됨(스레드 블로킹 X)

2. 전용 스레드 사용

• 경쟁 상태 문제가 생기는 이유는 복수의 스레드가 공유 상태에 동시에 접근하기 때문
• 공유 상태 접근 시 하나의 스레드(전용 스레드)만 사용하며 경쟁 상태 문제를 해결할 수 있다.

val countDispatcher = newSingleThreadContext("전용 스레드")
// Dispatchers.IO.limitedParallelism(1)
// Dispatchers.Default.limitedParallelism(1)

---

원자성 있는 데이터 구조 사용해 경쟁 상태 문제 해결하기

원자성 있는 객체란?

• 여러 스레드가 동시에 접근하더라도 안전하게 값을 변경하거나 읽을 수 있도록 하는 객체
• ex) AtomicInteger
• 만약 다른 스레드에서 이미 연산을 실행하고 있다면 스레드가 블로킹 됨
• 원자성 있는 객체를 사용할 때, 읽기와 쓰기를 따로 실행하면, 연산이 손실될 수 있다.

코루틴에 실행 옵션 설정하기

• 코루틴에 실행 옵션을 주기 위해서는 launch나 async 코루틴 빌더의 start 인자로 CoroutineStart 옵션을 전달하며 된다.

1. CoroutineStart.DEFAULT
   • 기본 실행 옵션
   • 코루틴 빌더 함수를 호출한 즉시 코루틴이 생성되고 코루틴의 실행이 CoroutineDispatcher에 요청된다.
   • 코루틴 빌더 함수를 호출한 코루틴은 계속해서 실행된다.
2. CoroutineStart.ATOMIC
   • 생성 상태의 코루틴에 취소가 요청되면 해당 코루틴은 취소된다.(일반적)
   • 생성 상태일 때 취소되지 않는다.(실행 옵션이 CoroutineStart.ATOMIC인 경우)
3. CoroutineStart.UNDISPATCHED
   • CoroutineDispatcher 객체의 작업 대기열을 거치지 않고 호출자의 스레드에서 즉시 실행된다.
   • 일시 중단 후 재개될 때는 CoroutineDispatcher에 실행 요청된다.

4. CoroutineStart.LAZY

• 즉시 실행 요청되지 않는 코루틴

---

무제한 디스패처

• 코루틴을 자신을 실행시킨 스레드에서 즉시 실행하도록 만드는 디스패처
• Dispatchers.Unconfined

fun main() = runBlocking<Unit>(Dispatchers.IO) {
    println("runBlocking 코루틴 실행 스레드: ${Thread.currentThread().name}")
    launch(Dispatchers.Unconfined) {
        println("launch 코루틴 실행 스레드: ${Thread.currentThread().name}")
    }
}

• 무제한 디스패처를 사용해 실행된 코루틴은 중단 시점 이후의 재개를 코루틴을 재개 시킨 스레드에서 한다.
  • 어떤 스레드에서 재개될지 파악하기 어렵기 때문에 비동기 작업이 불안정해진다.
  • 일반적인 상황에서 사용하는 것을 권장하지 않음
• CoroutineStart.UNDISPATCHED 옵션이 적용된 코루틴은 재개 시 CoroutineDispatcher에 실행 요청된다.

---

코루틴의 동작 방식과 Continuation

• 일반적인 코드가 동작할 때는 작업이 스레드를 점유해 코드 라인이 순서대로 동작한다.
• 코루틴은 Continuation Passing Style 이라 불리는 프로그래밍 방식을 통해 실행 정보를 저장하고 전달한다.
  • 이어서 하는 작업을 전달하는 방식

Continuation Passing Style

• 코루틴의 일시 중단 시점에 남은 작업 정보가 Continuation 객체에 저장된다.
• Continuation의 resumeWith 함수가 호출되면 저장된 작업 정보가 복원돼 남은 작업들이 마저 실행된다. → resumeWith 함수는 코루틴의 재개를 일으킨다.

• suspendCancellableCoroutine 함수를 사용하면 코루틴이 일시 중단되고, 함수 람다식의 수신객체인 CancellableContinuation에 resume 함수가 호출되면 재개된다.

fun main() = runBlocking<Unit> {
    val result = suspendCancellableCoroutine<String> { continuation: CancellableContinuation<String> ->
        thread {
            Thread.sleep(1000L)
            continuation.resume("실행 결과")
        }
    }
    println(result)
}

• 기존의 콜백 방식의 함수를 일시 중단 함수로 바꿀 때 자주 사용