좌충우돌 Coroutine 적용기

최근 담당하고 있는 서비스에 Coroutine을 적용할 기회가 있었다. 이에 Coroutine를 적용하면서 알게 된 내용을 정리해보고자 한다. 참고로 코루틴 개념을 처음 접한 건 3년 전 어드민 서비스를 만들 때 Generator함수를 사용해 보면서 였다. 

 

Coroutine이란 무엇일까?

사람들마다 정의가 조금씩 다를 순 있겠지만 나는 코루틴을 비동기 프로그래밍을 하기 위한 일시 정지 가능한 경량화된 스레드(light-weight thread)라고 정의를 내린다.

 

그럼 경량화된 스레드란 무엇일까?

이것에 앞서 프로세스와 스레드의 개념부터 우선 정리해 보고자 한다. 흔히 프로세스는 실행되고 있는 프로그램(disk에 있는 program이 memory에 올라간 상태) 또는 OS로부터 시스템 자원을 할당받는 작업 단위라고 말한다. 또한, 프로세스는 기본적으로 1개의 main 스레드를 가지고 있으며 상황에 따라 여러 개의 스레드를 가질 수도 있다.

 

스레드란 한 개의 프로세스 내에서 동작되는 실행 흐름 또는 프로세스가 할당받은 자원을 이용하는 실행 단위를 말한다. 이 때문에 스레드를 경량화된 프로세스(light-weight process)라고 부르기도 한다. 스레드는 프로세스 내에서 별도의 Stack 영역을 가지며 Code, Data, Heap 영역은 프로세스 내에 스레드끼리 공유하고 있다. 이로 인해 스레드 간 Context Switching 비용은 단순하게 Stack 영역만 교체하면 되기 때문에 프로세스 간 Context Switching 비용에 비해 상대적으로 오버헤드가 적다.

 

그럼 다시 본론으로 돌아와서 코루틴이 경량화된 스레드란 것은 무엇을 의미하는 것일까?

위에서 스레드를 경량화된 프로세스라고 부를 수 있었던 이유는 스레드가 한 개의 프로세스 내에서 작은 프로세스 역할을 수행했기 때문이었다. 이와 유사하게 코루틴을 경량화된 스레드라고 부른 이유는 코루틴이 한 개의 스레드 내에서 작은 스레드 역할을 수행하기 때문이다. 

 

CoroutineContext 

코루틴 실행 환경(다양한 Element들의 집합)을 나타내는 역할.

Element로는 Job, CoroutineDispatcher, ContinuationInterceptor, CoroutineExceptionHandler 등이 있다.

 

CoroutineScope  

코루틴을 제어(실행)할 수 있는 범위를 나타내는 역할. 이로 인해, 코루틴은 코루틴 스코프 안에서만 제어(실행)할 수 있다.

 

Dispatcher

사용자가 만든 코루틴을 스레드 또는 스레드 풀에게 전달하는 중간자 역할. 즉, 사용자가 코루틴을 만들어서 Dispatcher에게 전달하면 Disparcher은 자신이 관리하고 있는 스레드 풀내 스레드에게 코루틴 작업을 전달하게 된다. Dispatcher에게 코루틴을 전달하는 작업은 코루틴 빌더인 launch, async, runBlocking, produce, actor 메소드를 통해 가능하다. 

// CoroutineScope에 정의된 CoroutineContext 환경에서 코루틴이 실행된다.
CoroutineScope(Dispatchers.IO).launch {
  ...
}

참고로 코루틴 빌더인 launch와 async의 차이는 launch는 Job을 반환하고 async는 Deferred(미루다, 연기하다)를 반환한다는 점이다. Deffered는 Job을 상속한 클래스기 때문에 launch대신 async를 사용해도 전혀 문제는 없다. 참고로 Job과 Deferred의 차이를 살펴보면 Job과 달리 Deferred는 제네릭 타입이라는 점과 Deferred 안에는 await 메소드가 있다는 점이다.

 

간단한 개념들을 살펴봤으니 이젠 내가 코루틴을 적용했던 기능과 유사한 샘플 코드 2개를 가지고 비교 분석을 진행해 보겠다. 

 

아래 코드는 어떤 의미일까?

Case 1. structured concurrency 원칙을 따르고 있지 않다.

fun coroutineTest( ... ) {
    // 1번.
    val parentJob = CoroutineScope(Dispatchers.IO).launch {
        numbers.chunked(50) { chunkedNumbers ->
            chunkedNumbers.forEach { number ->
                launch { 
                    doNetworkRequest(number, changeInfo)
                }
            }
        }
    }

    // 2번.
    runBlocking {
       parentJob.join()
    }
}

private suspend fun doNetworkRequest( ... ) {
   ...
}

[1번] 코루틴으로 네트워크 I/O 작업을 수행해야 했기 때문에 이것에 최적화돼 있는 IO Dispatcher를 사용했다. IO Dispatcher에서 사용하는 스레드 풀의 최대 스레드 개수는 "kotlinx.coroutines.io.parallelism" 시스템 프로퍼티의 값에 따라 결정되며 기본적으로 64개로 설정돼 있다. 만약 코어의 수가 64보다 크다면 코어의 수로 설정된다.

[2번] 코루틴(네트워크 I/O) 작업이 끝날 때까지 기다렸다가 결과를 반환해야 했기 때문에 runBlocking 코루틴 빌더를 이용하여 parentJob을 대기하는 코드를 추가했다.

 

사실 이렇게 구현해도 기능상으로 이상은 없다. 하지만, parentJob.join 메소드를 누락할 경우 예상과 다르게 동작할 수 있기 때문에 structured concurrency 원칙을 따르고 있지 않다고 생각했다. structured concurrency란 새로운 코루틴을 만들 때는 코루틴의 lifetime을 제한하는 특정 코루틴 스코프에서만 만들어야 한다는 개념이다. 쉽게 설명하면, 구조적인 형태를 이용해서 코루틴을 관리하라는 의미이다.

 

Case 2. structured concurrency 원칙을 따르고 있다.

fun coroutineTest( ... ) {
    numbers.chunked(50) { chunkedNumbers ->
        runBlocking(Dispatchers.IO) {
            chunkedNumbers.forEach { number ->
                launch { 
                    doNetworkRequest(number, changeInfo)
                }
            }
        }
    }
}

private suspend fun doNetworkRequest( ... ) {
   ...
}

Case 2 방식은 Case 1 방식과 달리 parentJob.join 메소드를 호출할 필요가 없다. 왜냐하면, 외부 scope가 모든 children 코루틴이 완료될 때까지 기다려주기 때문이다. 이 때문에 해당 방식은 structured concurrency 원칙을 따르고 있다고 생각했다.

 

 

 

참고 자료

• https://kotlinlang.org/docs/coroutines-guide.html
• https://myungpyo.medium.com/reading-coroutine-official-guide-thoroughly-part-1-7ebb70a51910
• https://kotlinworld.com/139