[Kotlin] Coroutine - 2. CoroutineScope & Context & Dispathcer 을 파헤쳐보자

Coroutine

이전 시간에는 코루틴의 동작원리와 동시성 과 병렬성 의 차이에 대해 알아보았습니다. 

https://huisam.tistory.com/entry/coroutine1

[Kotlin] Coroutine - 1. 코루틴에서 동시성이란?

그런데..

아직도 코루틴의 동작 방식이 그렇게 와닿는 것 같지는 않습니다.

 

도대체 어떻게 동시에 처리하는 것처럼 보인다는 것인지

코루틴이라는 것은 어떻게 쓰레드를 왔다갔다 하면서 작업을 처리하는 것인지

 

마법같은 동시처리를 어떻게 코루틴은 처리하고 있는 걸까요?

 

우리는 이에 대한 이해를 하기 위해서 코드를 하나씩 살펴볼건데요.

살펴보기전에 코루틴이 동작하기 위한 요소들을 하나씩 볼려고 합니다.

 

코루틴의 지정요소

지정요소가 어떻게 되어 있는지 살펴보기 위해 간단한 코루틴을 하나 생성해보도록 하겠습니다.

CoroutineScope(Dispatchers.Default).launch {
        println("Starting in ${Thread.currentThread().name}")
        delay(500)
}.join()

public fun CoroutineScope(context: CoroutineContext): CoroutineScope =
    ContextScope(if (context[Job] != null) context else context + Job())

우리는 코루틴을 생성할 때 

CorotineScope, Dispathcer 를 활용하여 CoroutineContext 를 지정하는 것을 볼 수 있습니다.

 

이 용어들에 대한 정의와 인터페이스가 어떻게 되어 있는지 살펴보겠습니다.

CoroutineScope

CoroutineScope 은 코루틴의 범위를 의미하며, 코루틴의 생명주기를 함께하는 전체적인 Scope 라고 이해하시면 될 것 같습니다.

public interface CoroutineScope {
    /**
     * The context of this scope.
     * Context is encapsulated by the scope and used for implementation of coroutine builders that are extensions on the scope.
     * Accessing this property in general code is not recommended for any purposes except accessing the [Job] instance for advanced usages.
     *
     * By convention, should contain an instance of a [job][Job] to enforce structured concurrency.
     */
    public val coroutineContext: CoroutineContext
}

그렇기 때문에 인터페이스에 대한 변수는 CoroutineContext 라는 필드밖에 존재하지 않네요.

즉, 코루틴을 관리하는 요소중에서 제일 큰 범위의 인터페이스 라고 보시면 될 것 같아요.

CoroutineScope 은 제일 큰 범위의 인터페이스다

 

CoroutineContext

CoroutineContext 는 쉽게 설명드리면, 코루틴을 구성하는 집합이라고 보시면 될 것 같습니다.

public interface CoroutineContext {

    public operator fun <E : Element> get(key: Key<E>): E?

    public fun <R> fold(initial: R, operation: (R, Element) -> R): R

    public operator fun plus(context: CoroutineContext): CoroutineContext =
        if (context === EmptyCoroutineContext) this else // fast path -- avoid lambda creation
            context.fold(this) { acc, element ->
                val removed = acc.minusKey(element.key)
                if (removed === EmptyCoroutineContext) element else {
                    // make sure interceptor is always last in the context (and thus is fast to get when present)
                    val interceptor = removed[ContinuationInterceptor]
                    if (interceptor == null) CombinedContext(removed, element) else {
                        val left = removed.minusKey(ContinuationInterceptor)
                        if (left === EmptyCoroutineContext) CombinedContext(element, interceptor) else
                            CombinedContext(CombinedContext(left, element), interceptor)
                    }
                }
            }

    public fun minusKey(key: Key<*>): CoroutineContext

    public interface Key<E : Element>
}

인터페이스가 제공하는 함수들을 보시면 CoroutineContext 들에 대하여 Element 로 구성되어 있고,

각각의 Element get 혹은 fold 를 통하여 추가하거나 꺼내올 수 있습니다.

 

더군다나 다른 CoroutineContext 를 plus 하거나 minus 하는 연산을 통하여 특정한 Element 를 합치거나 삭제할 수도 있군요.

 

즉, CoroutineContext 는 Coroutine 에 대한 영구 컨텍스트라고 이해하시면 될 것 같아요.

 

이러한 Context 를 구성하는 요소 중의 Dispathcer 라는 구현체가 있습니다.

Dispatcher

Dispathcer 는 코루틴에 대한 task 수행 분배를 어떻게 할 것인지 결정하는 역할입니다.

정확히는 Interceptor 에 의해서 Coroutine 은 동작에 대한 중지 혹은 동작에 대한 재개를 실행하게 되는데, 이는 나중에 설명드릴게요

public abstract class CoroutineDispatcher :
    AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {

    /** @suppress */
    @ExperimentalStdlibApi
    public companion object Key : AbstractCoroutineContextKey<ContinuationInterceptor, CoroutineDispatcher>(
        ContinuationInterceptor,
        { it as? CoroutineDispatcher })

    public open fun isDispatchNeeded(context: CoroutineContext): Boolean = true

    public abstract fun dispatch(context: CoroutineContext, block: Runnable)

    @InternalCoroutinesApi
    public open fun dispatchYield(context: CoroutineContext, block: Runnable): Unit = dispatch(context, block)

    public final override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
        DispatchedContinuation(this, continuation)

    public final override fun releaseInterceptedContinuation(continuation: Continuation<*>) {
        val dispatched = continuation as DispatchedContinuation<*>
        dispatched.release()
    }
}

dispatch 라는 메서드를 통해서 실행가능한(Runnable) 한 task를 구현체에서 구현된 방식을 토대로 task를 던짐으로써 

코루틴에 대한 동작을 실행하게 되는 것이죠.

 

Task 를 관리하는 구현체중의 하나인 Event Loop 코드를 보시게 되면,

public final override fun dispatch(context: CoroutineContext, block: Runnable) = enqueue(block)

public fun enqueue(task: Runnable) {
    if (enqueueImpl(task)) {
        // todo: we should unpark only when this delayed task became first in the queue
        unpark()
    } else {
        DefaultExecutor.enqueue(task)
    }
}

Dispatcher 는 정말 말 그대로 task 를 EventLoop queue 에 넣고 끝내는 작업만 진행하게 됩니다.

다른 작업이 아무것도 없네요.

	override fun processNextEvent(): Long {
        // unconfined events take priority
        if (processUnconfinedEvent()) return 0
        // queue all delayed tasks that are due to be executed
        val delayed = _delayed.value
        if (delayed != null && !delayed.isEmpty) {
            val now = nanoTime()
            while (true) {
                // make sure that moving from delayed to queue removes from delayed only after it is added to queue
                // to make sure that 'isEmpty' and `nextTime` that check both of them
                // do not transiently report that both delayed and queue are empty during move
                delayed.removeFirstIf {
                    if (it.timeToExecute(now)) {
                        enqueueImpl(it)
                    } else
                        false
                } ?: break // quit loop when nothing more to remove or enqueueImpl returns false on "isComplete"
            }
        }
        // then process one event from queue
        val task = dequeue()
        if (task != null) {
            task.run()
            return 0
        }
        return nextTime
    }

그러면 Event Loop 를 관리하는 쓰레드에서 하나씩 deque 를 진행하여 task 를 꺼내와서 실행하게 되는 것이죠

 

즉, Dispatcher 는 코루틴이 동작하는 방식하고 연관이 있는 것이 아닌 Task 에 대한 분배에 대한 책임만 있습니다.

 

코루틴은 어떻게 흘러가니?

너무 어렵죠 

 

지금까지 배운 것을 토대로 한번 더 깔끔하게 정리해보겠습니다.

이해를 돕기 위하여 샘플 예제 코드와 함께 더 진행해볼게요.

fun main() {
    println("Starting a coroutine block...")
    runBlocking {
        println(" Coroutine block started")
        launch {
            println("  1/ First coroutine start")
            delay(100)
            println("  1/ First coroutine end")
        }
        launch {
            println("  2/ Second coroutine start")
            delay(50)
            println("  2/ Second coroutine end")
        }
        println(" Two coroutines have been launched")
    }
    println("Back from the coroutine block")
}

runBlocking 을 통해 EmptyCoroutineContext 를 만들고, 해당 Scope 안에서 여러 코루틴들을 만들어서 실행해보는 코드입니다.

코루틴은 저번에 말씀드린 것처럼 동시에 실행되는 것처럼 보인다고 했죠.? 과연 어떻게 출력이 될까요

실행결과

예상했던대로 1번 코루틴과 2번 코루틴이 동시에 시작하는 것을 볼 수 있습니다.

내부 동작은 어떻게 될까요?

동작과정

Coroutine1 이 생성되고 Default Dispathcer 에 의해서 task 를 던지고 즉각적으로 수행되는 모습입니다.

마찬가지로 Coroutine2 가 생성되고 Default Dispathcer 에 의해서 task 를 던지고 즉각적으로 수행하였네요.

따라서 전체 소요된시간은 최대 delay 시간인 100ms 근처의 시간으로 동작한 것을 확인했습니다!

 

하지만 여기서 의문이 들수가 있는게

Coroutine 1 은 delay 를 하였는데, 어떻게 Coroutine2 가 즉각적으로 실행되는 걸까요?

Dispatcher 는 그냥 task 를 던지기만 했을 뿐인데, 어떻게 blocking 에 대한 구간을 해결할 수 있었던 걸까요?

 

코루틴이 정말 어떻게 중단 포인트를 찾고, 어떻게 중단포인트를 마킹하고, 어떻게 중단포인트를 알고 다시 실행하는지에 대해 더 알려드리도록 할게요

이 과정에 대해서는 아래 게시글을 참고해주세요 ㅎㅎ

https://huisam.tistory.com/entry/coroutine6

[Kotlin] Coroutine - 6. Suspend 함수에 대해 deep dive 를 해보자

 

요약

CoroutineScope 은 코루틴의 범위를 의미하며, 코루틴의 생명주기를 함께하는 전체적인 Scope 다.

CoroutineContext 는 코루틴을 구성하는 집합

Dispathcer 는 코루틴에 대한 task 수행 분배를 어떻게 할 것인지 결정

 

참고

https://kotlinlang.org/docs/coroutine-context-and-dispatchers.html

https://silica.io/understanding-kotlin-coroutines/5/