[Kotlin Coroutines] 공유 상태로 인한 문제

동기화 블로킹

멀티 스레딩, 멀티 프로세싱에서 중요한 것은 공유 상태 및 경쟁 상태이다. 

다른 스레드에서 하나의 값을 수정, 조회할 때 서로 다른 값이 출력된다. 그래서 다음과 같은 코드에서는 예상하는  값이 나오지 않는다.

suspend fun main(): Unit {
   var i = 0
   coroutineScope {
      repeat(1_000_000) {
         launch {
            i++
         }
      }
   }
   println(i) // ~998242
}

 

자바에서 사용되는 `synchronized` 블록이나 동기화된 컬렉션을 사용해 해결할 수 있다.

 

var i = 0
fun main() = runBlocking {
   val lock = Any()
   repeat(1_000_000) {
      synchronized(lock) {
         i ++
      }
   }

   println(i) //1000000
}

 

이 코드의 문제점은

1. `synchronized` 블록 내부에서 중단 함수를 사용할 수 없다.

2. `synchronized` 블록에서 코루틴이 자기 차례를 기다릴 때 스레드를 블로킹한다.

 

디스패처의 원리를 생각해 보면 코루틴이 스레드를 블로킹하는 건 지양해야 한다. 그래서 블로킹 없이 중단하거나 충돌을 회피하는 방법을 사용해야 한다. 

 

원자성

자바의 경우 다양한 원자값을 가지고 있다. 원자값을 활용한 연산은 빠르며 '스레드 안전'을 모장한다. 이러한 연산을 원자성 연산이라고 한다.

원자성 연산은 락 없이 로우 레벨로 구현되어 효율적이고 사용하기 쉽다.

다음 예제는 `AtomicInteger`를 사용한 예제이다.

var i = AtomicInteger()
fun main() = runBlocking {
   repeat(1_000_000) {
         i.incrementAndGet()
   }

   println(i.get())
}

 

원자값은 의도대로 완벽하게 작동하지만 사용성이 제한되기 때문에 조심히 다뤄야 한다. (하나의 연산에서 원자성을 가지고 있다고 해서 전체 연산에서 원자성이 보장되는 것이 아니기 때문)

단순한 변수의 원자성을 보장하기 위해 사용되지만 더 복잡한 경우에는 다른 방법을 사용해야 한다.

 

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

싱글 스레드 디스패처를 사용하는 것이 공유 상태와 관련된 대부분의 문제를 해결하는 가장 쉬운 방법이다.

val dispatcher = Dispatchers.IO.limitedParallelism(1)

 

두 가지 방법으로 디스패처를 사용할 수 있는데, 각각은 다음과 같은 특징을 가지고 있다.

1) 코스그레인드 스레드 한정 (coarse-grained thread confinement)

   - 디스패처를 싱글스레드로 제한한 `withContext`로 전체 함수를 래핑 하는 방법

   - 사용하기 쉽고 충돌을 방지할 수 있지만, 함수 전체에서 멀티스레딩의 이점을 누리지 못하는 문제가 발생

2) 파인 그레인드 스레드 한정 (fine-grained thread confinement)

   - 상태를 변경하는 구문들만 래핑

   - 번거롭지만 크리티컬 섹션이 아닌 부분이 블로킹되거나 CPU 집약적인 경우에 더 나은 성능을 제공

   - 일반적인 중단 함수에 적용하는 경우에는 성능에 큰 차이가 없음

 

대부분의 경우, 표준 디스패처가 같은 스레드 풀을 사용하기 때문에 싱글스레드를 가진 디스패처를 사용하는 건 쉽고 효율적이다.

 

 

뮤텍스

가장 인기 있는 방식은 Mutex를 사용하는 것이다. 뮤텍스를 단 하나의 열쇠가 있는 방이라고 생각할 수 있다.

뮤텍스의 가장 중요한 기능은 `lock`이다. 코루틴이 `lock`을 호출하면 그 코루틴이 `unlock`을 호출할 때까지 중단된다. 즉, 단 하나의 코루틴 만이 `lock`과 `unlock`사이에 있을 수 있다.

 

suspend fun main() = coroutineScope {
   repeat(5) {
      launch {
         delayAndPrint()
      }
   }
}
val mutex = Mutex()
suspend fun delayAndPrint() {
   mutex.lock()
   delay(1_000)
   println("완료")
   mutex.unlock()
}

// 1초 후
완료
// 1초 후
완료
// 1초 후
완료
// 1초 후
완료
// 1초 후
완료

 

`lock` 과 `unlock`을 직접 사용하는 것은 위험한데, 두 함수 사이에서 예외가 발생할 경우 `unlock`이 호출되지 않으며 그 결과 다른 코루틴이 `lock`을 통과할 수 없게 된다. 즉, 데드락이 발생한다.

 

`synchronized`와 비슷하지만 다른 점은, 스레드를 블로킹하는 대신 코루틴을 중단시킨다는 것이다. 좀 더 안전하고 가벼운 방식이다.

병렬 실행이 싱글스레드로 제한된 디스패처를 사용하는 것과 비교하면 뮤텍스가 더 가벼우며 나은 성능을 가진다. 하지만 적절히 사용하는 것이 더 어렵다.

뮤텍스를 사용할 때 맞닥뜨리는 위험한 경우는

1) 코루틴이 락을 두 번 통과할 수 없다

2) 코루틴이 중단되었을 때 뮤텍스를 풀 수 없다

는 것이다.

 

따라서 전체 함수를 뮤텍스로 래핑 하는 건 지양해야 한다. (코스 그레인드 방식)

뮤텍스를 사용하기로 했다면 락을 두 번 걸지 않고 중단 함수를 호출하지 않도록 신경 써야 한다.

파인 그레인드 방식을 적절히 잘 사용해야 한다.

 

세마포어

Mutex와 비슷한 방식으로 작동하지만 둘 이상이 접근할 수 있고 사용법이 다른 세마포어도 알아야 한다.

Mutex는 하나의 접근만 허용하므로 `lock` , `unlock`, `withLock` 함수를 가지고 있다.

세마포어는 여러 개의 접근을 허용하므로 `acquire`, `release`, `withPermit` 함수를 가지고 있다.

 

suspend fun main() = coroutineScope {
   val semaphore = Semaphore(2)
   repeat(5) {
      launch {
         semaphore.withPermit {
            delay(1000)
            print(it)
         }
      }
   }
}

// 01 
// 1초 후
// 23
// 1초 후
// 4

 

세마포어는 동시 요청을 처리하는 수를 제한할 때 사용할 수 있어 처리율 제한 장치(rate limiter)를 구현할 때 도움이 된다.

 

 

사진 출처: https://scrutinybykhimaanshu.blogspot.com/2019/08/all-about-java-semaphore.html

뮤텍스와 세마포어

정리

공유 상태를 변경할 때 발생할 수 있는 충돌을 피하기 위해 코루틴을 다루는 방법은 다양하다.

가장 많이 쓰이는 방법은 싱글스레드로 제한된 디스패처를 사용해 공유 상태를 변경하는 것이다.

동기화가 필요한 특정 장소만 래핑 하는 파인 그레인드 스레드 한정이나 전체 함수를 래핑하는 코스 그레인드 스레드 한정을 활용할 수 있다. 

두 번째 방법이 더 쉽지만 성능은 떨어진다. 

원자값이나 뮤텍스를 사용하는 방법도 있다.

 

참고자료 

마르친 모스카와(2023), 코틀린 코루틴 Kotlin Coroutines : Deep Dive - 안드로이드 및 백엔드 개발자를 위한 비동기 프로그래밍, 인사이트

 

https://kotlinlang.org/docs/shared-mutable-state-and-concurrency.html

Shared mutable state and concurrency | Kotlin

 

https://scrutinybykhimaanshu.blogspot.com/2019/08/all-about-java-semaphore.html

Part 9: Java Thread Interview Questions & Answers (All About Java Semaphore and Mutex)