Develop/Reactor+Coroutines

데드락을 유도하는 코루틴 사용법

연로그 2025. 1. 11. 18:19
반응형

서론

 

이 글은 코루틴을 사용하다 겪었던 장애를 소개하는 글이다. 장애 원인이 밝혀지기까지 복잡한 과정이 있었으나, 모두 생략하고 코루틴에 대한 내용만 골라내 설명한다. 장애 원인은 무엇이었는지, 그 원인을 유도한 상황은 무엇인지, 어떻게 해결했는지 등을 정리해 보도록 하겠다.

 

코루틴에서 발생할 수 있는 데드락 (feat. runBlocking)

 

데드락이 발생할 수 있는 예제를 가져와보았다. 이 예제에는 두 가지 가정이 존재한다.

  1. dispatcher는 스레드를 2개까지 할당할 수 있는 디스패처
  2. async(dispatcher) { ... } 의 ...은 동기 호출이라 중단될 일이 없다

🔻 실행 가능한 예제코드

더보기

아래 코드를 실행하면 애플리케이션이 종료되지 않는다. 그 이유는 하단에서 자세히 살펴보겠다.

fun main() {
    val thread1 = Thread {
        println("1번 호출 ${death(1)}")
    }
    val thread2 = Thread {
        println("2번 호출 ${death(2)}")
    }

    thread1.start()
    thread2.start()

    thread1.join()
    thread2.join()
}

val dispatcher = Dispatchers.IO.limitedParallelism(2)
private fun death(index: Int) = runBlocking {
    return@runBlocking async(dispatcher) {
        return@async runBlocking {
            println("$index -> async 내부의 runBlocking 호출 ${Thread.currentThread().name}")
            val r1 = async(dispatcher) {
                println("$index -> async 1번 호출 sleep before ${Thread.currentThread().name}")
                Thread.sleep(1000L)
                println("$index -> async 1번 호출 sleep after ${Thread.currentThread().name}")
                return@async "1"
            }
            val r2 = async(dispatcher) {
                println("$index -> async 2번 호출 sleep before ${Thread.currentThread().name}")
                Thread.sleep(1000L)
                println("$index -> async 2번 호출 sleep after ${Thread.currentThread().name}")
                return@async "2"
            }
            val result = r1.await() + r2.await()
            return@runBlocking result
        }
    }.await()
}
출력 결과 - 애플리케이션이 종료되지 않았다

 

API 첫 번째 호출이 시작된다.

 

API 첫 번째 호출)

  • 로직의 흐름에 따라, 2번 라인의 async(dispatcher) 수행을 위해 첫 번째 스레드를 할당받는다.

 

API 첫 번째 호출)

  • runBlocking은 현재 스레드(= 첫 번째 스레드, 2번 라인의 async(dispatcher)를 실행하던 스레드)를 블록 한다.
  • runBlocking 내의 모든 자식 코루틴들이 완료될 때까지 스레드 블로킹이 유지된다.

 

이때 API 두 번째 호출이 시작된다.

 

API 두 번째 호출)

  • 2번 라인의 async(dispatcher)를 실행하기 위해 두 번째 스레드를 할당받는다.


API 첫 번째 호출)

  • 첫 번째 호출에서의 로직도 동시에 수행되고 있음을 잊지 말자. 
  • 4번 라인의 async(dispatcher)를 수행하기 위해 스레드 할당을 대기한다.

 

API 두 번째 호출)

  • runBlocking은 현재 스레드(= 두 번째 스레드, 2번 라인의 async(dispatcher)를 실행하던 스레드)를 블록 한다.
  • runBlocking 내의 모든 자식 코루틴들이 완료될 때까지 스레드 블로킹이 유지된다.

 

API 첫 번째 호출)

  • 4번 라인의 async(dispatcher)를 수행하기 위해 스레드 할당을 대기 중이다.

 

API 두 번째 호출)

  • 4번 라인의 async(dispatcher)를 수행하기 위해 스레드 할당을 대기한다.

 

 

후일담

 

왜 runBlocking을 중첩하여 사용했는가?

runBlocking이 중첩되는 상황은 설계 단계에서 예상하지 못한, 비의도적인 상황이다. 최초에 controller에서만 호출되던, runBlocking을 적용된 클래스가 있었다.

 

여기서 여러 과제가 진행하고 다양한 개발자가 작업을 하면서 '내가 필요한 데이터가 이미 만들어져 있네? 이 메서드를 호출해서 조회해야겠다.'는 상황이 발생하며 아래와 같은 구조가 생겼다.

 

 

이는 다양한 문제가 혼합되어 있기 때문이라고 생각한다.

  1. runBlocking을 쓴 클래스가 가장 바깥(controller) 외에서도 사용될 수 있다
  2. 코루틴을 복잡하게 사용하고 있다
  3. 모듈 및 패키지의 의존성이 과하게 열려 있다

 

개선한 점

1. 일단은 runBlocking을 사용한 클래스가 다양한 곳에서 호출될 수 있다는 것이 가장 큰 문제라고 생각했다. 그래서 runBlocking 호출은 모두 controller로 옮겼다. 앞으로도 runBlocking은 무조건 컨트롤러에서만 호출할 수 있게끔 팀의 컨벤션으로 가져갔다.

 

2. 코루틴을 복잡하게 사용하고 있다. 이에 대해서는 전에 Spring MVC + Coroutines에 대한 고민에서 언급한 적 있다. (저 글을 쓸 때 당시와 지금의 생각이 많이 달라졌다. 코루틴 적용을 위해 참고하기보다는 그냥 이렇게도 돌아가는구나 정도로 참고하길 권장한다.) 지금의 코루틴을 사용하는 구조가 너무 복잡하다고 판단, 이를 간소화해 이해하기 쉬운 구조로 변경하는 작업을 진행하고 있다. 이에 대해서는 별도 글로 다뤄보도록 하겠다.

 

3. 모듈 및 패키지의 의존성이 과하게 열려 있다. 이 부분에 대해서는 어떻게 개선할 수 있을지 논의 중에 있다.

 

 

생각해 볼 점 - runBlocking vs coroutineScope

위에서 설명했던 예제에서 중첩된 runBlocking 부분을 coroutineScope로 변경하면 데드락이 더 이상 발생하지 않는다. 둘 다 자식 코루틴이 모두 완료할 때까지 대기할 텐데, 무엇이 다른 걸까?

 

🔻 예제 코드 및 실행 결과

더보기 
fun main() {
    val thread1 = Thread {
        println("1번 호출 ${death(1)}")
    }
    val thread2 = Thread {
        println("2번 호출 ${death(2)}")
    }

    thread1.start()
    thread2.start()

    thread1.join()
    thread2.join()
}

val dispatcher = Dispatchers.IO.limitedParallelism(2)
private fun death(index: Int) = runBlocking {
    return@runBlocking async(dispatcher) {
        return@async coroutineScope { // 여기를 runBlocking -> coroutineScope로 변경했다
            println("$index -> async 내부의 coroutineScope 호출 ${Thread.currentThread().name}")
            val r1 = async(dispatcher) {
                println("$index -> async 1번 호출 sleep before ${Thread.currentThread().name}")
                Thread.sleep(1000L)
                println("$index -> async 1번 호출 sleep after ${Thread.currentThread().name}")
                return@async "1"
            }
            val r2 = async(dispatcher) {
                println("$index -> async 2번 호출 sleep before ${Thread.currentThread().name}")
                Thread.sleep(1000L)
                println("$index -> async 2번 호출 sleep after ${Thread.currentThread().name}")
                return@async "2"
            }
            val result = r1.await() + r2.await()
            return@coroutineScope result
        }
    }.await()
}
결과가 모두 출력되었고, 애플리케이션도 잘 종료되었다

 

코루틴 공식 문서에서 runBlocking에 대한 설명을 살펴보자.

Runs a new coroutine and blocks the current thread until its completion.

 

runBlocking은 코루틴이 완료될 때까지 현재 스레드를 블록 한다. 하지만 coroutineScope는 다르다. suspend가 붙어있는 중단 함수다.

public suspend fun <R> coroutineScope(block: suspend CoroutineScope.() -> R): R {
    contract {
        callsInPlace(block, InvocationKind.EXACTLY_ONCE)
    }
    return suspendCoroutineUninterceptedOrReturn { uCont ->
        val coroutine = ScopeCoroutine(uCont.context, uCont)
        coroutine.startUndispatchedOrReturn(coroutine, block)
    }
}

 

단순 추측으로는 자식 코루틴들을 수행하는 동안 현재 코루틴은 중단되고, 이후 자식 코루틴들이 완료되면 현재 코루틴을 재개하는 형식으로 만들어진 게 아닐까 싶다. 정확한 동작은 디버깅 등을 통해 살펴봐야 하나, 이 글에서는 생략한다. coroutineScope는 스레드를 계속 블록 하지는 않는다는 것을 기억해두면 좋을 것 같다.

반응형