1.4 코틀린 코루틴 이해하기 - 코루틴의 실제 구현

 

Continuation 전달 방식

Continuation은 함수에서 함수로 인자를 통해 전달된다. Continuation은 마지막 파라미터로 전달된다.

suspend fun getUser(): User?
suspend fun setUser(user: User)
suspend fun checkAvailability(flight: Flight): Boolean

위 중단 함수의 실제 함수 시그니처는 아래와 같다.

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

반환 타입이 Any 또는 Any?로 바뀌었다는 것을 알 수 있다.

중단 함수가 실행 도중 중단되면 COROUTINE_SUSPENDED를 반환하기 때문이다.

 

 

아주 간단한 함수

suspend fun myFunction() {
    println("Before")
    delay(1000)
    println("After")
}

myFunction() 함수의 시그니처는 아래와 같다.

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

 

 

 

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

var continuation = MyFunctionContinuation(continuation)

본체가 시작될 때 MyFunction은 파라미터인 continuation을 자신만의 Continuation인 MyFunctionContinuation으로 래핑한다.

(실제로 Continuation은 익명 클래스로 구현되어있지만 쉽게 설명하기 위해 이름을 붙였다)

 

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

클래스에 래핑이 없는 경우에만 클래스를 래핑해야 한다.

만약 코루틴이 중단 후 재개된 적이 있으면 Continuation 객체는 이미 래핑되어 있을 것이다.

이 경우 Continuation 객체를 래핑하지 않고 그대로 둔다.

 

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

다시 중단 함수의 본체를 보자.

함수가 시작되는 지점은 두 곳이다.

• 함수가 처음 호출될 때
• delay(1000)를 만나 중단 후 재개될 때

현재 상태를 저장하려면 label이라는 필드를 사용한다.

함수가 처음 시작될 때 이 값은 0으로 설정되고, 중단되기 직전에 1씩 증가된다.

이 label 값을 통해 코루틴이 재개될 시점을 알 수 있다.

 

myFunction()의 세부 구현을 간단하게 표현하면 다옴과 같다.

fun myFunction(continuation: Continuation<Unit>): Any {
    val continuation = continuation as? MyFunctionContinuation
        ?: MyFunctionContinuation(continuation)

    if (continuation.label == 0) {
        println("Before")
        continuation.label = 1
        if (delay(1000, continuation) == COROUTINE_SUSPENDED) {
            return COROUTINE_SUSPENDED
        }
    }
    if (continuation.label == 1) {
        printIn("After")
        return Unit
    }
    error("Impossible")
}

delay()에 의해 중단된 경우 delay()는 COROUTINE_SUSPENDED를 반환하고, myFunction() 또한 COROUTINE_SUSPENDED를 반환한다.

myFunction()을 호출한 함수는 물론이고 콜 스택에 있는 모든 함수도 동일한 과정을 거친다.

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

즉, 코루틴을 실행하던 스레드는 다른 할일을 할 수 있는 것이다.

 

 

이제 Continuation 객체를 알아보자.

가독성을 위해 익명 클래스로 구현된 Continuation 객체를 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
        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)
    }
}

중단 함수를 자세히 알아보고 싶으면 자바로 디컴파일된 코드를 확인하면 된다.

 

 

상태를 가진 함수

재개된 후 다시 사용할 지역 변수나 파라미터와 같은 상태를 가지고 있는 함수라면, 함수의 Continuation 객체에 상태를 저장해야 한다.

다음과 같은 함수를 생각해 보자.

suspend fun myFunction() {
    println("Before")
    var counter = 0
    delay(1000) 
    counter++
    println("Counter: $counter")
    println("After")
}

여기서 counter는 재개된 이후에도 사용되므로 Continuation 객체를 통해 저장해야 한다.

지역 변수나 파라미터 같이 함수 내에서 사용되던 값들은 중단되기 직전에 저장되고, 함수가 재개될 때 복구된다.

 

간략화된 중단 함수와 Continuation의 코드는 다음과 같다.

fun myFunction(continuation: Continuation<Unit>): Any {
    val continuation = continuation as? MyFunctionContinuation
        ?: MyFunctionContinuation(continuation)

    var counter = continuation.counter
    
    if (continuation.label == 0) {
        println("Before")
        counter = 0
        continuation.counter = counter
        continuation.label = 1
        if (delay(1000, continuation) == COROUTINE_SUSPENDED) {
            return COROUTINE_SUSPENDED
        }
    }
    if (continuation.label == 1) {
        counter = (counter as Int) + 1
        println("Counter: $counter")
        println("After")
        return Unit
    }
    error("Impossible")
}

class MyFunctionContinuation(
    val completion: Continuation<Unit>
) : Continuation<Unit> {
    override val context: Coroutinecontext
        get() = completion.context

    var result: Result<Unit>? = null
    var label = 0
    var counter = 0

    override fun resumeWith(result: Result<Unit>) {
        this.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)
    }
}

 

 

값을 받아 재개되는 함수

suspend fun printUser(token: String) {
    println("Before")
    val userId = getUserId(token) // 중단 함수
    println("Got userid: $userId")
    val userName = getUserName(userId, token) // 중단 함수
    println(User(userId, userName))
    println("After")
}

두 가지 중단 함수 getUserld()와 getUserName()이 있다.

token이라는 파라미터를 받으면 중단 함수는 특정 값을 반환한다.

파라미터와 반환값 모두 Continuation 객체에 저장되어야 하는 이유는 다음과 같다.

• token은 label이 0과 1일 때 사용된다.
• userId는 label이 1과 2일 때 사용된다.
• Result 타입인 result는 함수가 어떻게 재개되었는지 나타낸다.

 

함수가 값으로 재개되었다면 결과는 Result.Success(value)가 되고, 이 result 값을 얻어 사용할 수 있다.

함수가 예외로 재개되었다면 결과는 Result.Failure(exception)이 되고, 이때는 예외를 던진다.

fun printUser(
    token: String,
    continuation: Continuation<*>,
): Any {
    val continuation = continuation as? PrintUserContinuation
        ?: PrintUserContinuation(
            continuation as Continuation<Unit>,
            token,
        )

    var result: Result<Any>? = continuation.result
    var userId: String? = continuation.userId
    val userName: String

    if (continuation.label == 0) {
        println("Before")
        continuation.label = 1
        val res = getUserId(token, continuation)
        if (res == COROUTINE_SUSPENDED) {
            return COROUTINE_SUSPENDED
        }
        result = Result.success(res)
    }
    if (continuation.label == 1) {
        userld = result!!.getOrThrow() as String
        println("Got userId: $userId")
        continuation.label = 2
        continuation.userId = userId
        val res = getUserName(userId, token, continuation)
        if (res == COROUTINE_SUSPENDED) {
            return COROUTINE_SUSPENDED
        }
        result = Result.success(res)
    }
    if (continuation.label == 2) {
        userName = result!!.getOrThrow() as String
        println(User(userId as String, userName))
        println("After")
        return Unit
    }
    error("Impossible")
}

class PrintUserContinuation(
    val completion: Continuation<Unit>,
    val token: String,
) : Continuation<String> {
    override val context: Coroutinecontext
        get() = completion.context

    var label = 0
    var result: Result<Any>? = null
    var userId: String? = null

    override fun resumeWith(result: Result<String>) {
        this.result = result
        val res = try {
            val r = printUser(token, this)
            if (r == COROUTINE_SUSPENDED) return
            Result.success(r as Unit)
        } catch (e: Throwable) {
            Result.failure(e)
        }
        completion.resumeWith(res)
    }
}

 

 

콜 스택

함수 a가 함수 b를 호출하면 가상 머신은 a의 상태와 b가 끝나면 실행이 될 지점을 어딘가에 저장해야 한다.

이런 정보들은 모두 콜 스택이라는 자료 구조에 저장된다.

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

따라서 코루틴을 재개할 때 콜 스택을 사용할 수는 없지만, 대신 Continuation 객체가 콜 스택의 역할을 대신한다.

 

Continuation 객체는 중단이 되었을 때의 상태를 가지고 있다. 이때 상태란 아래 정보를 포함한다.

• label
• 함수의 지역 변수
• 함수의 파라미터
• 중단 함수를 호출한 함수가 재개될 위치 정보

하나의 Continuation 객체가 다른 하나를 참조하고, 참조된 객체가 또 다른 Continuation 객체를 참조한다.

이러한 이유로 Continuation 객체는 거대한 양파와 같으며, 일반적으로 콜 스택에 저장되는 정보를 모두 가지고 있다.

 

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

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

suspend fun c(i: Int) {
    delay(i * 100L)
    println("Tick")
}

위 코드가 사용하는 Continuation 객체를 다음과 같이 나타낼 수 있다.

 

CContinuation(
    i = 4,
    label = 1,
    completion = BContinuation(
        i = 4,
        label = 1,
        completion = AContinuation(
            label = 2,
            user = User@1234,
            completion = ...
        )
    )
)

 

Continuation 객체가 재개될 때 각 Continuation 객체는 자신이 담당하는 함수를 먼저 호출한다.

함수의 실행이 끝나면 자신을 호출한 함수의 Continuation을 재개한다.

재개된 Continuation 객체도 자신이 담당하는 함수를 호출하며, 스택의 끝에 다다를 때까지 이 과정을 반복한다.

override fun resumeWith(result: Result<String>) {
    this.result = result
    val res = try {
        val r = printUser(token, this)
        if (r == COROUTINE_SUSPENDED) return
        Result.success(r as Unit)
    } catch (e: Throwable) {
        Result.failure(e)
    }
    completion.resumeWith(res)
}

 

함수 a가 함수 b를 호출하고, 함수 b는 함수 c를 호출하며, 함수 c에서 중단된 상황을 예로 들어 보자.

1. 실행이 재개된다.
2. c의 Continuation 객체는 c() 함수를 재개한다.
3. c() 함수가 완료된다.
4. c의 Continuation 객체는 b의 Continuation 객체를 재개한다.
5. b() 함수가 완료된다.
6. b의 Continuation 객체는 a의 Continuation 객체를 재개한다.
7. a() 함수가 완료된다.

 

 

override fun resumeWith(result: Result<String>) {
    this.result = result
    val res = try {
        val r = printUser(token, this)
        if (r == COROUTINE_SUSPENDED) return
        Result.success(r as Unit)
    } catch (e: Throwable) {
        Result.failure(e)
    }
    completion.resumeWith(res)
}

예외를 던질 때도 비슷하다.

처리되지 못한 예외가 resumeWith() 함수에서 잡히면 Result.failure(e)로 래핑된다.

예외를 던진 함수를 호출한 함수는 래핑된 결과를 받게 된다.

 

 

실제 코드

Continuation 객체와 중단 함수를 컴파일한 실제 코드는 최적화되어 있다.

아래와 같은 처리 과정이 더 포함되어 있어 더 복잡하다.

• 예외가 발생했을 때 더 나은 스택 트레이스 생성
• 코루틴 중단 인터셉션
• 다양한 단계에서의 최적화 (사용하지 않는 변수 제거, 테일콜 최적화(tail-call optimization) 등)

 

 

중단 함수의 성능

코루틴 내부 구현을 본 후 대부분의 사람들은 비용이 클 거라 생각하지만, 실제로는 그렇지 않다.

함수를 상태로 나누는 것은 숫자를 비교하는 것만큼 쉽고, 실행 지점이 변하는 비용도 거의 들지 않는다.

Continuation 객체에 상태를 저장하는 것도 간단하다. 지역 변수를 복사하지 않고 새로운 변수가 메모리 내 특정 값을 가리킨다.

Continuation 객체를 생성하는 비용이 들지만, 절대 크지 않다.

 

 

요약

• 중단 함수는 함수가 시작할 때와 중단 함수가 호출되었을 때의 상태를 가진다는 점에서 상태 머신과 비슷하다.
• Continuation 객체는 상태를 나타내는 숫자 값과 로컬 데이터를 가지고 있다.
• 호출된 함수의 Continuation 객체는 호출한 함수의 Continuation을 장식한다. 따라서, 모든 Continuation 객체는 함수를 재개하거나 재개된 함수가 완료될 때 사용되는 콜 스택 역할을 한다.

 

 

 

출처

https://www.yes24.com/product/goods/123034354