Kotlin Coroutine (4) : Coroutine의 실제 구현

0. 들어가며

이번 글은 마르친 모스카와의 Kotlin Coroutine 4장을 기반으로 Coroutine의 실제 구현을 다룰 것이다. 너무 딥한데? 싶다면 뒤로 가기를 눌러도 좋다.

나를 포함한 대부분의 운전 면허 소지자는 자동차를 어떻게 운전하는 지에만 관심을 둘 뿐, 기계 공학의 정수인 자동차의 정밀한 기어 변속기가 어떻게 작동하는 지는 알 필요가 없는 것과 같은 이유다.

1. Continuation 전달 방식

앞선 글에서 소개한 중단함수는 여러 언어에서 다양한 방식으로 구현되었다. Kotlin 팀은 그 중 Continuation 전달 방식을 채택했을 뿐이다.

Continuation은 함수 to 함수로 인자를 통해 전달되는데, 관례 상 함수의 마지막 파라미터로 전달한다.

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

...

/*실제 구현*/
fun getUser(continuation : Continuation<*>) : Any?
fun setUser(user : Userm continuation : Continuation<*>) : Any

중단 함수 내부를 잘 들여다보면, 내가 원래 선언한 구현과 달라졌음을 알 수 있다. 형태와 반환 타입 모두 바뀌었는데, 특히 User의 경우 User?가 Any?로 바뀌었음을 알 수 있다.

그 이유는 간단하다. 중단 함수 실행 도중 중단된다면 개발자가 선언한 값인 User를 반환하지 않을 수 있기 때문이다. 후에 살펴보겠으나, 중단될 경우 중단 함수는 특별한 마커인 COROUTINE_SUSPEND을 반환한다.

즉, getUser() 함수가 반환할 수 있는 것은 User? 또는 COROUTINE_SUSPEND인 것이다.

따라서, 반환 타입이 User?와 Any의 가장 가까운 Supertype인 Any?로 지정된 것이다.

2. 아주 간단한 함수

suspend fun myFunction() {
	println("A")
    delay(1000)
    println("B")
}

위와 같은 함수를 가정하자. 앞서 본 예시에 따라, 해당 함수의 시그니처를 아래와 같이 추론할 수 있다.

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

해당 함수는 상태를 저장하기 위해 자신만의 Continuation 객체가 필요하며, 해당 객체를 myFunctionContinuation이라고 하자.
단, 클래스에 포장이 없는 경우에만 아래처럼 continuation을 포장해야 한다.

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

당장은 이해하기 어려울 수 있다.
일단 앞의 함수 본체를 다시 보자.

suspend fun myFunction() {
	println("A")
    delay(1000)
    println("B")
}

여기서 시작점은 두 곳이다.

• 함수의 시작점
• 중단 이후 재개 시점

중단 상태를 저장할 때는 label이라는 필드를 사용한다.
함수가 처음 시작될 땐 0, 이후 중단 전 다음 상태로 설정되어 Coroutine이 재개될 시점임을 알 수 있게 해준다.

fun myFunction(continuation : Continuation<Unit>) : Any {
	val continuation = Continuation as? myFunctionContinuation(continuation)
    
    if(continuation.level == 0) {
    	...
    	continuation.level = 1
    	if (delay(1000, continuation) == COROUTINE_SUSPEND) {
    		return COROUTINE_SUSPEND
    	}
    }
    
    if (continuation.level == 1) {
    	...
        return Unit
    }
    
    error("Impossible")    
}

상기 예시에서, delay에 의해 중단된 경우 COROUTINE_SUSPEND가 반환됨을 알 수 있다.

따라서, 중단이 발생하면 myFuncion을 호출한 Caller부터 모든 Call Stack의 함수들이 줄줄이 종료되며,현재 중단된 Coroutine을 실행하던 스레드를 실행 가능한 다른 코드가 사용할 수 있게 된다.

3. 상태를 가진 함수

그렇다면, 상태를 가진 함수가 중단되면 어떨까?

suspend fun myFunction() {
	println("A")
    val counter = 0
    delay(1000) /* 중단 함수 */
    counter++
    println("B")
}

위 counter의 상태를 중단 직전 저장하고, 함수가 재개될 때 복구되도록 하려면 아래와 같이 중단함수가 작성되어야 한다.

fun myFunction(continuation : Continuation<Unit>) : Any {
	val continuation = Continuation as? myFunctionContinuation(continuation)
    
    if(continuation.level == 0) {
    	...
        counter = 0
        continuation.counter = counter
    	continuation.level = 1
    	if (delay(1000, continuation) == COROUTINE_SUSPEND) {
    		return COROUTINE_SUSPEND
    	}
    }
    
    if (continuation.level == 1) {
    	...
        counter = (counter as Int) + 1
        return Unit
    }
    
    error("Impossible")    
}

4. 값을 받아 재개되는 함수

중단 함수로부터 값을 받아야 하는 경우는 상태를 저장하는 것보다 조금 더 복잡하다.

suspend fun myFunction(token : String) {
	println("A")
    val userId = getUserId(token) /* 중단 함수 */
    val userName = getUserName(userId, token) /* 중단 함수 */
    println("B")
}

fun myFunction(
	token : String,
	continuation : Continuation<*>
) : Any {
	val continuation = Continuation as? myFunctionContinuation ?:
    myFunctionContinuation(continuation as Continuation<Unit>, token)
    
    val result : Result<Any>? = continuation.result
    val userId : String? = continuation.userId
    val userName : String
    
    if(continuation.level == 0) {
    	...
        counter = 0
    	continuation.level = 1
        val res = getUserId(token, continuation)
    	if (res == COROUTINE_SUSPEND) {
    		return COROUTINE_SUSPEND
    	}
        result = Result.success(res)
    }
    
    if (continuation.level == 1) {
    	...
        userId = result!!.getOrThrow() as String
        continuation.level = 2
        continuation.userId = userId
        val res = getUserName(userId, token, continuation)
        if (res == COROUTINE_SUSPEND) {
    		return COROUTINE_SUSPEND
    	}
        result = Result.success(res)
    }
    
    if (continuation.level == 2) {
    	...
        userName = result!!.getOrThrow() as String
        return Unit
    }
    
    error("Impossible")    
}

이처럼 level이 0, 1, 2 ... 단조 증가하며 설정되는 것을 확인할 수 있다.

5. Call Stack

Coroutine을 중단하면 스레드를 반환하므로, Call Stack에 있는 정보가 모두 사라질 것이다. 그럼 Coroutine의 정보를 어디에 저장할까? 앞서 말했듯, Continuation 객체가 Call Stack을 대신한다.

Continuation 객체는 중단되었을 당시의 상태(label)와 함수의 지역변수, 파라미터, 중단함수를 호출한 함수가 재개될 위치 정보를 갖고 있다. 이를 표현하면 아래와 같다.

Ccontinuation(
	i = 4,
    label = 1,
    completion = Bcontinuation(
    	i = 4,
        label = 1,
        completion = Acontinuation(
        	label = 2,
            user = User@1234,
            completion = ...
        )
    )
    
)

Continuation 객체가 재개되면, 각 객체는 자신이 담당하는 함수를 먼저 호출한다. 해당 함수의 실행이 끝나면, 자신을 호출한 함수의 Continuation을 재개한다. 쉽게 설명하자면 아래와 같다.

함수 a -> 함수 b -> 함수 c를 호출하는 구조일 때,
함수 c에서 중단 발생!

1. C의 Continuation 객체에서 C 함수 재개
2. C 함수 완료 시 C C의 Continuation 객체에서 B 함수를 호출하는 
   B의 Continuation 객체를 재개
3. B 함수 완료 시 ......

예외를 던질 때도 이와 비슷한 과정으로 전개된다.
실제 코드는 더 복잡하고, 최적화되어 있으나 본 글에서는 지문 상 다루지 않는다. 궁금할 시 Kotlin Coroutine 4장, 실제 코드 파트를 참고하면 된다.

간단하게만 언급하자면, 아래 과정들이 추가되어 있다고 보면 된다.

• 예외 발생 시 StackTrace 생성
• Coroutine 중단 인터셉션
• 미사용 변수 제거 및 Tail Call 최적화 수행

또한 재귀 대신 반복문 형태로 최적화되어 있다.

6. 중단 함수의 성능

일반적인 함수 대신 중단함수를 사용한다고 해서 비용이 특별히 더 들지 않는다.

• 실행점 옮기기
• 함수를 상태로 나누기
• Continuation 객체 생성하기

이 모든 건 간단하고 쉬운 일이다.

또한 Continuation 객체에 상태 저장하는 것도 지역 변수를 복사하는 것이 아닌, 새 변수가 메모리 내 특정 주소를 가리키게 하는 방식으로 해결한다.

7. 마치며

실제 구현은 소개한 것보다 훨씬 복잡한 과정으로 이뤄진다. 다만, 이번 글을 읽으며 Coroutine 내부가 대략적으로 어떻게 구현되어있는지만 알아가도 성공이라고 할 수 있겠다.

본 글을 간단하게 요약하자면 아래와 같다. 어려웠더라도 잘 읽어보았기를 바란다.

• 중단함수는 State Machine과 비슷하게 동작하며, 함수가 시작될 때와 중단함수를 호출할 때의 상태를 가진다.
• 상태를 나타내는 값과 로컬 데이터는 Continuation 객체에 저장한다.
• Callee의 Continuation 객체는 Caller의 Continuation 객체를 장식한다. 따라서, Call Stack의 역할을 대신한다.