[Kotlin] Continuation Interceptor

비동기 작업에서는 실행 위치나 스레드를 제어할 필요가 자주 생긴다. UI앱의 경우 UI업데이트는 메인 스레드에서만 가능하고, 서버 앱에서는 무거운 연산은 별도 스레드 풀로 분리하는 경우를 예로 들 수 있다. 이를 가능하게 해주는 도구가 바로 ContinuationInterceptor 이다. Coroutine 을 처음 배울때는 Interceptor 의 구현체인 Dispatcher 만 알게 되는데, Dispatcher 가 바로 Interceptor 의 구체클래스인 것을 알면 Continuation 을 "가로채서" 별개의 스레드에 실행하는 역할을 맡은 클래스인것이 좀더 와닿는다. 

 

인터페이스 정의 

interface ContinuationInterceptor : CoroutineContext.Element {
    companion object Key : CoroutineContext.Key<ContinuationInterceptor>

    fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T>
    fun releaseInterceptedContinuation(continuation: Continuation<*>)
}

 

ContinuationInterceptor 는 CoroutineContext.Element 를 구현하며, CoroutineContext 내에서 실행 위치나, 방식을 결정하는 역할을 한다. Dispatchers.Main, Dispatchers.IO, Dispatchers.Default 등은 모두 이 인터페이스를 구현한다. 

 

CoroutineInterceptor 는 어떻게 동작하는가? 

코루틴은 일시 중단되었다가 재개될 때 Continuation 객체를 통해 재개된다. (Coroutine State Machine 참고) 이 재개 시점(resume)에서  ContinuationInterceptor 는 다음과 같이 코루틴 실행을 가로채고 wrapping 한다. 

val intercepted = continuation.context[ContinuationInterceptor]
    ?.interceptContinuation(continuation)
    ?: continuation

코루틴이 재개되는 시점에 코루틴은 자신의 CoroutineContext 에 ContinuationInterceptor 가 있는지 확인한다. 있으면, interceptContinuation() 을 호출하여 래핑된 Continuation 을 사용한다. (interceptContinuation 의 반환 객체) 그 래핑된 Continuation 은 어디서 실행할지를 결정한다. (메인 스레드, IO 스레드 ...) 

 

재사용이 끝난 Continuation 은 다음 메서드를 통해 해제된다. 이 정리는 프레임워크 내부에서 자동으로 수행된다. 

ContinuationInterceptor.releaseInterceptedContinuation(continuation)

 

 

Interceptor의 구현체인 Dispatcher에 대한 추가 정리: 

CoroutineContext 의 Element 중 하나인 CoroutineDispatcher 의 명세

abstract class CoroutineDispatcher : AbstractCoroutineContextElement, ContinuationInterceptor

ContinuationInterceptor는 모든 CoroutineDispatcher 가 구현하는 추상 클래스이다. 

 

공식 문서에 따르면 kotlinx.coroutines 를 사용할 경우, CoroutineDispatcher 가 아닌 다른 형태의 ContinuationInterceptor 구현체는 사용하지 않는 것이 좋다고 한다. 그 이유는 CoroutineDispatcher 는 newCoroutineContext() 함수 내에서 디버깅 기능이 올바르게 작동하도록 보장해주기 때문이라고 한다. 

 

kotlinx.coroutines 는 코루틴의 컨텍스트 설정 시 newCoroutineContext() 라는 내부 함수에서 다음과 같은 작업을 수행한다 - 1) CoroutineName을 컨텍스트에 추가 (디버깅 이름 추적), 2) CoroutineId 부여 (Debug probes 용) 3) 디스패처가 있는 경우 현재 스레드나 컨텍스트에 맞는 위치에서 실행되도록 설정

 

즉, CoroutineDispatcher 는 공식적으로 인정된 인터셉터의 구현체라고 할 수 있다. 따라서 커스텀 Interceptor 구현은 권장되지 않는 것으로 보인다. 

 

ContinuationInterceptor 를 알게된 후 보이는 것

Dispatchers.Unconfined 의 동작이 잘 이해되지 않았었는데, 위 내용을 알고나니 조금 더 이해가 쉬워졌다. Unconfined 디스패처는 실행 흐름이 처음 시작한 스레드에서 시작했다가, 중간에 suspend 가 일어나면 다시 resume 하는 시점에 resume 이 호출된 스레드에서 실행이 이어진다.

예시: 

fun main() = runBlocking {
    withContext(Dispatchers.Unconfined) {
        println("1 - ${Thread.currentThread().name}") // main
        doSomething()
        println("2 - ${Thread.currentThread().name}") // DefaultExecutor
    }
}

suspend fun doSomething() {
    println("doSomething ${Thread.currentThread().name}") // main
    delay(1000)
    println("doSomething ${Thread.currentThread().name}") // DefaultExecutor
}

// 실행결과 
1 - main
doSomething main
doSomething kotlinx.coroutines.DefaultExecutor
2 - kotlinx.coroutines.DefaultExecutor

 

delay 라는 중단 지점에서 중단되었다가 실행이 재개되는 시점에, resume 이 호출되는 스레드는 기본으로 지정되는 Default 스레드임을 알 수 있다. 이 때 doSomething() 이 반환된 이후 실행흐름또한 Default 스레드에서 이어진다.

 

delay 이후 재개되는 시점에 continuation.context[ContinuationInterceptor] 를 맵에서 꺼내서 확인을 하는데, ContinuationInterceptor 가 Unconfined 객체이며 이 객체의 내부 구현체의 isDispatchNeeded 변수가 false 가 되어있어서 dispatch() 가 호출되지 않기 때문이다.

 

 

 

---

출처:

- https://kotlinlang.org/spec/asynchronous-programming-with-coroutines.html#asynchronous-programming-with-coroutines

- https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-dispatcher/