Kotlin Coroutines Deep Dive - 코루틴의 실제 구현

이 글은 Kotlin Coroutines : Deep Dive 책 내용을 다루고 있습니다.

 

자동차를 탈 때 내부 구조를 알 필요는 없다. 다만 이해도를 높여줄 수 있다.

한 번에 이해하기에 내용이 어려워서 그냥 이런게 있구나 하고 가볍게 넘어가려고 한다.

 

들어가기 전에

이 책에서는 코루틴 실제 구현에 대해 다음과 같이 중요한 점을 얘기한다.

• 중단 함수(suspend function)가 시작할 때, 호출되었을 때, 상태를 가진다는 점에서 상태 머신과 비슷하다.
• Continuation 객체는 상태를 나타내는 숫자와 로컬 데이터를 가진다.
• 함수의 Continuation 객체가 이 함수를 부르는 다른 함수의 Continuation 객체를 장식(decorate)한다. 그 결과 모든 Continuation 객체는 실행 재개하거나 재개된 함수를 완료할 때 사용되는 콜 스택으로 사용된다.

여기서 Continuation 객체가 뭔지 알고 넘어가자.

 

Continuation

게임의 체크포인트처럼 중단가능한 함수(suspend function) 이 중단되면 Continuation 객체를 반환한다. 이 객체를 이용하면 멈췄던 곳에서 다시 코루틴을 실행할 수 있으며 직렬화, 역직렬화가 가능하다.

 

이 Continuation 객체는 함수에서 함수로 인자를 통해 전달된다. (관례상 마지막 파라미터)

 

suspend fun getUser(): User?
suspend fun setUser(user: User)

 

이 중단함수를 자세히 들여다 보면

 

fun getUser(continuation: Continuation<*>): Any?
fun setUser(user: User, continuation: Continuation<*>): Any

이렇게 Continuation 객체를 파라미터로 받으면서 반환 타입이 Any 나 Any? 로 달라진 것을 알 수 있다.

 

이는 도중에 중단되면 선언된 타입의 값을 반환하지 않을 수 있기 때문이다.

위의 getUser() 함수는 User? 혹은 COROUTINE_SUSPENDED 를 반환할 수 있다.

 

간단한 함수

간단한 함수로 지연이 일어나기 전과 후 어떻게 변하는지 알아보자

 

suspend fun myFunction() {
   println("Before")
   delay(1000) // 중단 함수
   println("After")
}

 

그럼 이 함수는 앞에서 알아본 것과 같이 변할 것이다.

 

fun myFunction(continuation: Continuation<*>): Any

이 함수의 상태를 저장하기 위해서는 이 함수만의 Continuation 객체가 필요하다.

실제 구현과 다르지만 이해하기 쉬운 구조로 설명하자면 다음과 같다.

 

val continuation = continuation as? MyFunctionContinuation 
   ?: MyContinuation(continuation)

반환된 continuation 이 이미 포장된 객체라면 그대로 사용하고 그렇지 않은 경우에는 포장이 필요하다.

 

함수가 시작되는 지점은 함수의 처음 호출과 함수의 재개 시점 두 곳인데

label 이라는 필드로 이 상태를 저장하는데 처음 호출될 때는 0 으로 설정되며 나중에 재개된 시점을 알 수 있게 도와준다.

 

fun myFunction(continuation: Continuation<*>): Any {
  val continuation = continuation as? MyFunctionContinuation 
   ?: MyContinuation(continuation)
   
   if(continuation.label == 0) { // 처음 호출되었다면
     println("Before")
     continuation.label = 1
     
     // delay에 의해 중단된 경우 COROUTINE_SUSPENDED 를 반환
     if(delay(1000, continuation)) == COROUTINE_SUSPENDED) {
        return COROUTINE_SUSPENDED
     }
   }
   if(continuation.label == 1) { // 재개되었다면
     println("After")
     return Unit
   }
   error("Impossible")
}

 

myFunction 을 호출한 함수부터 콜 스택에 있는 모든 함수는 중단된 경우 COROUTINE_SUSPENDED 를 반환한다. 

 

중단이 일어나면 콜 스택에 쌓여있는 모든 함수가 종료된다.

그럼 이 함수를 실행하던 스레드를 다른 실행 가능한 코드가 사용하게 된다.

 

 

앞에서 말한 MyFunctionContinuation wrapper class는 사실 익명 class 로 구현되어 있다.

쉽게 이해하기 위해서 MyFunctionContinuation 로 나타내었는데 결국 다음과 같다.

 

class MyFunctionContinuation(
       val completion: Continuation<Unit>
    ): Continuation<Unit> {
      // 생성자로 받은 컨티뉴에이션 객체의 코루틴 컨택스트를 사용
      override val context: CoroutineContext
         get() = completion.context 
         
      var label = 0
      var result: Result<Any>? = null
      
      override fun resumeWith(result: Result<Unit>) { // 재개하는 함수
      	this.result = result // result 를 저장한다.
        val res = try {
           val r = myFunction(this) // 함수를 실행한다.
           if(r == COROUTINE_SUSPENDED) return // 실행한 결과가 중단이라면 재개하지 않는다.
           Result.success(r as Unit)
        } catch (e: Throwable){
           Result.failure(e)
        }
        completion.resumeWith(res) // 이 결과로 호출한 함수의 재개 함수를 실행한다
      }
}

 

위 코드를 해석해보자.

함수 실행 결과가 중단일 경우 재개하지 않는다.

중단이 아닐 때 Result.success() 를, Result.failure() 를 인자로 이 중단 함수를 호출한 함수를 호출한다.

 

위 함수는 상태도 없고 반환도 없는 간단한 함수이다.

 

만약 그렇지 않다면 어떻게 해야할까?

 

상태를 가진 함수나 값을 받아 재개하는 함수는 파라미터, 반환값을 함수의 Continuation 객체에 저장해야한다.

 

이 부분의 코드는 복잡하기도 하고 필자도 제대로 이해하지 못해서 다루지 않고 넘어가겠다. 책을 참고해주었으면 좋겠다.

 

콜 스택

 

함수 a 가 함수 b 를 호출하면 a 의 상태와 b 가 끝나고 실행될 지점을 저장해야한다.

이 부분이 저장되는 곳이 콜 스택(call stack) 이다.

 

코루틴을 중단하면 스레드를 반환해 콜 스택에 있는 정보가 사라진다.

그럼 재개할 때 사용할 수 없게된다.

 

이때 Continuation 객체가 이 콜 스택 역할을 대신한다!

 

이 객체는 중단 되었을 때 상태(label) 과 함수의 지역 변수, 필드, 중단 함수를 호출한 함수가 재개될 위치 정보를 가지고 있다.

 

컨티뉴에이션 객체가 다른 Continuation 객체를 계속해서 참조하는 구조이다.

 

책에서는 이 Continuation 객체가 거대한 양파와 같다고 표현한다. 

 

예를 들면 다음과 같은 구조가 있다.

 

suspend fun a() {
    val user = readUser()
    b()
    b()
    b()
}

suspend fun b() {
    // ...
    for(i in 1..10){
       c(i)
    }
}

suspend fun c(i: Int) {
    delay(1000)
    // ...
}

어떤 시점에 중단된다면 다음과 같은 구조의 Continuation 객체가 만들어진다.

 

CContinuation(
   i = 4, // 인자 i
   label = 1,
   completion = BContinuation(
      i = 4, // 상태 i
      label = 1,
      completion = AContinuation(
        label = 2, // 어디까지 수행했는가
        user = User@1234, // 상태 user
        completion = ...
      )
   )
)

 

Continuation 객체는 재개될 때 자신이 담당하는 함수를 먼저 호출하는데

이 함수의 실행이 끝나면 자신이 호출한 함수의 Continuation 을 재개한다. 이 과정을 계속 반복한다.

 

위와 같이 a 가 b 를 실행하고 b 가 c 를 호출하는 구조라고 했을 때

 

일반적인 콜 스택은 다음과 같이 실행된다. 

a, b, c 모두 중단 함수고 c 에서 중단이 발생했다면 다음과 같이 실행된다.

 

실행 재개 시 c 의 Continuation 객체가 c 함수를 먼저 재개한다.  함수가 완료되면 c Continuation 객체는 b 함수를 호출하는 b Continuation 을 재개한다. b 함수가 완료되면 b Continuation 객체는 a Continuation 을 재개하고 a 함수가 호출되게 된다.

실제 코드

실제 코드는 최적화 및 몇가지 처리 과정 때문에 더 복잡하다.

 

• 예외 발생 시 더 나은 스택 트레이스 생성
• 코루틴 중단 인터셉션
• 사용하지 않는 변수를 제거 및 테일 콜 최적화 등의 최적화

 

중단 함수의 성능

코루틴 내부 구현을 보고나서 일반 함수에 비해 비용이 클거라 예상된다.

중단 함수의 비용은 어떻게 될까?

 

실제로는 비용이 크지 않다.

 

함수를 상태로 나누는 것, 실행점이 변하는 것의 비용은 거의 들지 않는다. 상태 저장 또한 간단하다.

Continuation 객체 생성 비용이 어느 정도 들지만 마찬가지로 크지 않다.

 

결론 및 요약

1. 중단 함수는 상태 머신과 비슷해서 함수가 시작될 때와 중단 함수를 호출한 뒤의 상태를 가진다.

2. 상태를 나타내는 값과 로컬 데이터는 Continuation 객체에 저장된다.

3. 호출된 함수의 Continuation 객체는 호출한 함수의 Continuation 를 decorate(장식)한다. 그 결과, 모든 Continuation 객체는 함수가 재개될 때 또는 함수가 완료될 때 사용되는 콜 스택 역할을 한다.

 

마무리

실제 구현은 훨씬 복잡할 것이고 간단하게 알아본 것인데도 코루틴 구현... 이해하기 어렵다. 이정도만 하고 넘어가자. 

 

끗