[코틀린 코루틴] 4장. 코루틴의 실제 구현

이 장에서는 코루틴의 내부 구현에 집중하며, 코루틴의 동작 과정에 대해 설명하고자 합니다.

 

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

 

컨티뉴에이션 전달 방식

 

컨티뉴에이션은 함수에서 함수로 인자를 통해 전달되며, 관례상 마지막 파라미터로 전달됩니다.

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

// 자세히 들여다 보면
fun getUser(continuation: Continuation<*>): Any?
fun setUser(user: User, continuation: Continuation<*>): Any
fun checkAvailablility(
    flight: Flight,
    continuation: Continuation<*>
): Any

 

중단 함수 내부를 들여다 보면 원래 선언했던 형태와 반환 타입이 달라졌다는 것을 알 수 있습니다.

Any 또는 Any?로 반환하는 이유는 중단 함수를 실행하는 도중에 중단되면 선언된 타입의 값을 반환하지 않을 수 있기 때문입니다.

선언된 타입의 값을 반환하지 않을 때는, 특별한 마커(marker)인 COROUTINE_SUSPENDED를 반환합니다.

getUser 함수가 User? 또는 COROUTINE_SUSPENDED를 반환할 수 있기 때문에 User?와 Any의 가장 가까운 슈퍼타입인 Any?로 지정됩니다.

 

 

아주 간단한 함수

 

좀 더 살펴보기 위해 다음 함수를 시작해 보겠습니다.

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

myFunction 함수의 시그니처를 다음과 같이 추론할 수 있습니다.

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

이 함수는 상태를 저장하기 위해 자신만의 컨티뉴에이션 객체를 필요로 하며, 이를 MyFunctionContinuation이라 하겠습니다.

본체가 실행될 때 MyFunction은 파라미터인 continuation을 자신만의 컨티뉴에이션인 MyFunctionContinuation으로 포장합니다.

val continuation = MyFunctionContinuation(continuation)

만약 재실행되고 있는 코루틴이라면 이미 래핑되어 있으므로 컨티뉴에이션 객체를 그대로 둬야 합니다.

val continuation =
    if (continuation is MyFunctionContinuation) continuation
    else MyFunctionContinuation(continuation)

다음과 같이 간단히 할 수 있습니다.

val continuation = continuation as? MyFunctionContinuation ?: MyFunctionContinuation

다시 함수 본체로 돌아와, 시작점을 살펴보면 다음과 같습니다.

 

• 함수가 처음 호출될 때
• 컨티뉴에이션이 resume을 호출할 때

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

함수가 처음 시작될 때 이 값은 0으로 설정되며, 이후에는 중단되기 전에 다음과 같이 설정되어 코루틴이 재개될 시점을 알 수 있게 해줍니다.

// 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) {
        println("After")
        return Unit
    }
    error("Impossible")
}

위 코드에서 중요한 부분은, delay에 의해 중단된 경우 myFunction은 COROUTINE_SUSPENDED를 반환한다는 것입니다.

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) {
        println("After")
        return Unit
    }
    error("Impossible")
}

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)
    }
}

 

 

상태를 가진 함수

 

함수가 중단된 후에 다시 사용할 지역 변수나 파라미터와 같은 상태를 가지고 있다면, 이 역시도 컨티뉴에이션 객체에 저장해야 합니다.

다음과 같은 함수를 생각해 봅시다.

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

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

간략화된 중단 함수의 모습은 다음과 같습니다.

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 label = 0
    var counter = 0
    var result: Result<Unit>? = 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)
    }
}

 

 

값을 받아 재개되는 함수

 

중단 함수로부터 값을 받아야 하는 경우는 좀더 복잡합니다.

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

getUserId와 getUserName이라는 두 가지 중단 함수가 있습니다.

파라미터와 반환값 모두 컨티뉴에이션에 저장되어야 하며, 이유는 다음과 같습니다.

 

• token은 상태(label) 0과 1에서 사용됩니다.
• userId는 상태 1과 2에서 사용됩니다.
• Result 타입인 result는 함수가 어떻게 재개되었는지 나타냅니다.

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

함수가 예외로 재개되었다면 결과는 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
    var userName: String? = continuation.userName
    
    if (continuation.label == 0) {
        println("Before")
        continuation.label = 1
        val res = getUserId(token, continuation)
        if (res == COROUTINE_SUSPEND) {
            return COROUTINE_SUSPEND
        }
        result = Result.success(res)
    }
    
    if (continuation.label == 1) {
        userId = result!!.getOrThrow() as String
        println("Got userId: $userId")
        
        continuation.label = 2
        continuation.userId = userId
        val res = getUserName(userId, token, continuation)
        if (res == COROUTINE_SUSPEND) {
            return COROUTINE_SUSPEND
        }
        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<Unit> {
    override val context: CoroutineContext
        get() = completion.context

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

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

 

 

콜 스택

 

콜 스택은 함수 a가 함수 b를 호출했을 때 a의 상태와 b가 끝나면 실행될 지점을 저장합니다.

코루틴에서는 이러한 콜 스택의 역할을 컨티뉴에이션 객체가 대신합니다.

컨티뉴에이션 객체는 중단이 되었을 때의 상태(label)와 함수의 지역 변수와 파라미터, 그리고 중단 함수를 호출한 함수가 재개될 위치 정보를 가지고 있습니다.

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")
}

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

실행이 재개되면 c의 컨티뉴에이션 객체는 c 함수를 재개합니다.

c 함수가 완료되면 c 컨티뉴에이션 객체는 b 함수를 호출하는 b 컨티뉴에이션 객체를 재개합니다.

b 함수가 완료되면 b 컨티뉴에이션 객체는 a 함수를 호출하는 a 컨티뉴에이션 객체를 재개하고 a 함수가 호출되게 됩니다.

// 컨티뉴에이션 의사코드
CContinuation(
    i = 4,
    label = 1,
    completion = BContinuation(
        i = 4,
        label = 1,
        completion = AContinuation(
            label = 2
            user = User@1234,
            completion = ...
        )
    )
)

 

 

 

---

요약

 

• 중단 함수는 상태 머신과 비슷해 함수가 시작될 때와 중단 함수를 호출한 뒤의 상태(label)를 가진다.
• 상태를 나타내는 값과 로컬 데이터는 컨티뉴에이션 객체에 저장된다.
• 호출된 함수의 컨티뉴에이션 객체는 함수가 재개될 때 또는 재개된 함수가 완료될 때 사용되는 콜 스택의 역할을 한다.