[Coroutine] 공유 상태로 인한 문제

 💡 코루틴에서 발생할 수 있는 공유 상태로 인한 문제를 확인하고, 이를 동기화 블로킹을 통해 어떻게 해결할 수 있는지 학습하였습니다.

 

공유 상태로 인한 문제

class UserDowloader(
    private val api : NetworkService
) {
    private val users = mutableListOf<User>()

    fun downloaded(): List<User> = users.toList()

    suspend fun fetchUser(id: Int) {
        val newUser = api.fetchUser(id)
        users.add(newUser)
    }
}

• 위 코드는 동시 사용에 대한 대비가 되어 있지 않습니다.
• fetchUser 호출은 users를 변경하는데, 이 경우 같은 시간에 해당 함수가 한 개 스레드에서 시작할 경우에만 정상적으로 작동합니다.
• 만약 동시에 리스트를 변경하면 충돌이 발생할 수 있으며, 아래와 같은 문제가 생길 수 있습니다.
  • 같은 객체에 대하여 1_000_000번 실행할 경우 실제론 기대하는 수보다 작은 호출이 발생하며, 예외가 생길 수도 있습니다.

repeat(1_000_000) {
    launch {
        downloader.fetchUser(it)
    }
}

동기화 블로킹

• 공유 상태로 인한 문제를 해결하기 위해서 자바에서는 synchronized 블록이나 동기화된 컬렉션을 사용합니다.

synchronized(lock) { // 스레드를 블로킹합니다
  counter++
}

• 하지만 위 방식은 아래와 같은 문제점이 있습니다.
  • synchronized 내부에서 중단 함수를 사용할 수 없습니다.
  • synchronized 블록에서 코루틴이 자기 차례를 기다릴 때 스레드를 블로킹합니다.
• 디스패처의 원리를 생각한다면 코루틴이 스레드를 블로킹하는 것을 지양해야 합니다.
• 메인 스레드를 블로킹하거나 제한 된 수의 스레드만 가지고 있는 것은 좋지 않습니다 !!
• 스레드를 낭비하지 않고, 코루틴에 특화된 방법을 사용해야 합니다.

원자성

• 자바에서는 원자값을 활용한 연산을 사용하면 (원자성 연산) 락 없이 로우 레벨로 문제를 해결할 수 있습니다.

private var counter = AtomicInteger() //원자성 연산

fun main() = runBlocking {
    massiveRun {
        counter.incrementAndGet()
    }
}

• 물론 하나의 연산에서 원자성을 가지고 있다고 해서 전체 연산에서 원자성이 보장되는 것은 아닙니다.
• 이는 하나의 프리미티브 변수 또는 하나의 래퍼런스의 안전을 보장하기 위해 사용됩니다.

싱글스레드로 제한된 디스패처

• 공유 상태로 인한 문제를 해결하기 위해서 병렬성을 하나의 스레드로 제한하는 디스패처를 사용할 수 있습니다.

val dispatcher = Dispatchers.IO
    .limitedParallelism(1)
var counter = 0
fun main() = runBlocking {
    massiveRun {
        withContext(dispatcher){
            counter++
        }
    }
}

• 싱글 스레드를 디스패처로 제한하는 방법에는 두가지 방법이 있습니다.
  • 첫번째는 위처럼 withContect로 전체 함수를 래핑하는 코스 그레인드 스레드 한정입니다.
    • 사용하기 쉬우며 충돌을 방지할 수 있지만, 함수 전체에서 멀티스레딩의 이점을 누리지 못하는 문제가 있습니다.
  • 두번째 방법은 파인 그레인드 스레드 한정입니다.
    • 상태를 변경하는 구문들만 래핑하며, 번거롭지만 CPU 집약적인 경우에 더 나은 성능을 제공합니다.

val dispatcher = Dispatchers.IO
    .limitedParallelism(1)

suspend fun downloaded(): List<User> =
    withContext(dispatcher){
        users.toList()
    }

뮤텍스 (Mutex)

• 가장 많이 사용하는 방식이며, 단 하나의 열쇠가 있는 방이라고 표현할 수 있습니다.
• 첫 번째 코루틴이 lock을 호출하면 열쇠를 가지고 중단 없이 작업을 수행합니다.
• 또다른 코루틴이 lock을 호출하면 첫 번째 코루틴이 unlock을 호출할 때까지 중단됩니다.
  • 화장실의 열쇠를 기다리는 작업과 유사 
• 이 상황에서 또 다른 코루틴이 lock을 호출할 경우 작업을 중단한 뒤에 두 번째 코루틴 다음 순서로 큐에 들어가게 됩니다.
• 단 하나의 코루틴만이 lock과 unlock 사이에 존재하는 방법입니다.

suspend fun main() = coroutineScope {
    repeat(5){
        launch {
            delayAndPrint()
        }
    }
}

val mutex = Mutex()

suspend fun delayAndPrint() {
    mutex.lock()
    delay(1000)
    mutex.unlock()
}

• lock과 unlock을 직접 사용하는 것은 예외가 발생하거나 반환이 빠르게 이뤄질 경우 위험한 상황이 발생할 수 있습니다.
  • 다른 코루틴이 lock을 통과할 수 없는 경우 (열쇠를 돌려받지 못함) 데드락 상황이 발생하게 됩니다.
  • [데드락] : 시스템 자원에 대한 요구가 뒤엉킨 상태이며, 점유하고 있는 자원을 서로 기다리는 상황
• 이를 해결하기 위해서 lock으로 시작해 finally 블록에서 unlock을 호출하는 withLock 함수를 사용하면 블록 내에서 어떤 예외가 발생하더라도 성공적으로 해결할 수 있습니다.

val mutex = Mutex()

var counter = 0

fun main() = runBlocking {
    messiveRun {
        mutex.withLock {
            counter++
        }
    }
}

• synchronized 블록과의 차이점은 스레드를 블로킹하는 대신 코루틴을 중단시킨다는 점입니다.
• 성능이 좋지만 사용하기 어려우며, 아래와 같은 문제점이 있습니다.
  • 코루틴이 락을 두 번 통과할 수 없습니다.
  • 코루틴이 중단되었을 때 뮤텍스를 풀 수 없습니다.

suspend fun main() {
    val mutex = Mutex()
    mutex.withLock {
        mutex.withLock {
            // in mutex
        }
    }		// 락을 두번 사용하여 영원히 실행

    coroutineScope {
        launch {
            delay(10000)
            mutex.withLock {
                // in mutex
            }
        }
    } // delay가 소모될 때까지 mutex를 풀 수 없음 
}

세마포어

• 뮤텍스와 비슷하게 작동하지만, 둘 이상이 접근할 수 있다는 특징이 있습니다.
• 세마포어는 여러 개의 접근을 허용하며, acquire, release, withPermit 함수를 가지고 있습니다.

suspend fun main() = coroutineScope {
    val semaphore = Semaphore(2)

    launch {
        semaphore.withPermit {
        }
    }
}

• 세마포어는 공유 상태로 인한 문제를 직접적으로 해결할 수 없지만, 동시 요청을 처리하는 수를 제한할 때 사용할 수 있습니다.
• 이는 처리율 제한 장치를 구현할 때 도움이 됩니다.
  • rate limiter : 클라이언트 또는 서비스가 보내는 트래픽의 처리율을 제어하기 위한 장치

class LimitedNetworkUserRepository(
    private val api: UserApi
) {
    private val semaphore = Semaphore(10)

    suspend fun requestUser(userId: String) =
        semaphore.withPermit {
            api.requstUser(userId)
        }
}

요약

• 공유 상태를 변경할 때 충돌을 피하기 위해 싱글스레드로 제한된 디스패처를 사용해 공유 상태를 변경하는 방법을 주로 활용합니다.
• 파인 그레인드 스레드 한정 : 동기화가 필요한 특정 장소만 래핑
• 코스 그레인드 스레드 한정 : 전체 함수를 래핑
• 이 외에 원자성, 뮤텍스, 세마포어를 활용하는 방법이 있습니다.