2.7 코틀린 코루틴 라이브러리 - 디스패처

 

 

코루틴에서는 디스패처를 통해 코루틴이 실행되어야 할 스레드(또는 스레드 풀)를 결정한다.

영어 사전에서 디스패처는 '사람이나 긴급 차량을 필요한 곳에 보내는 사람'이라 정의한다.

 

기본 디스패처

디스패처를 설정하지 않으면 Dispatchers.Default가 기본적으로 설정된다.

Dispatchers.Default는 프로그램이 실행되는 컴퓨터의 CPU 개수와 동일한 수(최소 두 개 이상)의 스레드 풀을 가지고 있다.

이는 스레드를 효율적으로 사용한다고 가정했을 때, 이론적으로 가장 최적의 스레드 수다.

따라서 Dispatchers.Default는 CPU 집약적인 연산을 수행하기에 적합하다.

 

suspend fun main() = coroutineScope {
    repeat(1000) {
        launch { // launch(Dispatchers.Default)
            List(1000) { Random.nextLong() }.maxOrNull()

            val threadName = Thread.currentThread().name
            println("Running on thread: $threadName")
        }
    }
}

Running on thread: DefaultDispatcher-worker-1
Running on thread: DefaultDispatcher-worker-5
Running on thread: DefaultDispatcher-worker-7
Running on thread: DefaultDispatcher-worker-6
Running on thread: DefaultDispatcher-worker-11
Running on thread: DefaultDispatcher-worker-2
Running on thread: DefaultDispatcher-worker-10
Running on thread: DefaultDispatcher-worker-4
...

위 코드를 실행한 컴퓨터가 12개의 CPU를 가지고 있다면, 스레드 풀의 스레드 개수는 12개가 된다.

 

fun main() = runBlocking {
    repeat(1000) {
        launch { // launch(Dispatchers.Default)
            List(1000) { Random.nextLong() }.maxOrNull()

            val threadName = Thread.currentThread().name
            println("Running on thread: $threadName")
        }
    }
}

Running on thread: main
Running on thread: main
Running on thread: main
...

runBlocking은 디스패처가 설정되어 있지 않으면 Dispatchers.Default이 아닌 자신만의 디스패처를 사용한다.

위 예제에서 coroutineScope 대신 runBlocking을 사용하면 모든 코루틴은 'main'에서 실행된다.

 

 

기본 디스패처를 제한하기

고비용 작업이 Dispatchers.Default의 스레드를 다 써버려서 Dispatchers.Default를 사용하는 다른 코루틴의 실행이 제한되는 상황을 가정해 보자.

private val dispatcher = Dispatchers.Default
    .limitedParallelism(5)

limitedParallelism을 사용하면 같은 시간에 특정 수 이상 스레드를 사용하지 못하도록 제한할 수 있다.

 

 

메인 디스패처

안드로이드와 같은 애플리케이션 프레임워크는 일반적으로 메인 스레드(UI 스레드)를 가지고 있다.

메인 스레드는 자주 사용되는 만큼 조심스럽게 다뤄야 한다. 메인 스레드가 블로킹되면 애플리케이션 전체가 멈춰 버리기 때문이다.

메인 스레드에서 코루틴을 실행하려면 Dispatchers.Main을 사용한다.

스레드를 블로킹하는 대신 중단 함수를 사용하고, 간단한 연산이라면 Disptatchers.Main만으로도 충분하다.

 

class SomeTest {
    private val dispatcher = Executors
        .newSingleThreadExecutor()
        .asCoroutineDispatcher()

    @Before
    fun setup() {
        Dispatchers.setMain(dispatcher)
    }

    @After
    fun tearDown() {
        Dispatchers.resetMain()
        dispatcher.close()
    }

    @Test
    fun test() = runBlocking {
        launch(Dispatchers.Main) {
            // ...
        }
    }
}

단위 테스트에서 메인 디스패처를 사용하고 싶다면 Dispatchers.setMain(dispatcher)로 디스패처를 설정해야 한다.

단위 테스트가 끝나면 Dispatchers.resetMain()으로 메인 디스패처를 원래의 디스패처를 되돌린다.

 

 

IO 디스패처

스레드를 블로킹해야 하는 경우 어떤 디스패처를 사용할까?

메인 스레드를 블로킹하면 애플리케이션이 멈출 것이다.

그렇다고 기본 디스패처를 블로킹하면 스레드 풀에 있는 모든 스레드가 블로킹되어 아무런 연산도 수행할 수 없다.

이런 경우 Dispatchers.IO를 사용하면 된다. 

 

Dispatchers.IO는 I/O 연산으로 스레드를 블로킹할 때 사용한다.

예를 들어 아래와 같은 경우에 사용할 수 있다.

• 파일을 읽고 쓰는 작업
• 안드로이드의 SharedPreferences 사용
• 블로킹 함수 호출

 

suspend fun main() {
    val time = measureTimeMillis {
        coroutineScope {
            repeat(50) {
                launch(Dispatchers.IO) {
                    Thread.sleep(1000)
                }
            }
        }
    }
    println(time) // 〜1000
}

위 코드는 1초밖에 걸리지 않는다. 50개의 코루틴이 동시에 실행되었기 때문이다.

Dispatchers.IO는 동시에 64개(64개보다 더 많은 코어가 있다면 해당 코어의 수)의 스레드를 사용할 수 있다.

 

suspend fun main() = coroutineScope {
    repeat(1000) {
        launch(Dispatchers.IO) {
            Thread.sleep(200)
            val threadName = Thread.currentThread().name
            println("Running on thread: $threadName")
        }
    }
}

Running on thread: DefaultDispatcher-worker-1
...
Running on thread: DefaultDispatcher-worker-35
Running on thread: DefaultDispatcher-worker-64

Dispatchers.Default의 예제에서도 스레드 이름이 "DefaultDispatcher-worker-XX"였다.

즉, Dispatchers.Default와 Dispatchers.IO는 같은 스레드 풀을 공유한다.

스레드 풀을 공유하면 스레드를 재사용할 수 있고, 스레드가 다시 배분되지 않아 효율적이다. 

 

suspend fun main(): Unit = coroutineScope {
    launch(Dispatchers.Default) {
        println(Thread.currentThread().name)
        withContext(Dispatchers.IO) {
            println(Thread.currentThread().name)
        }
    }
}

DefaultDispatcher-worker-2
DefaultDispatcher-worker-2

위 코드에서 launch 코루틴과 withContext 코루틴은 같은 스레드로 실행이 된다. 스레드가 재사용되기 때문이다.

 

Dispatchers.Default와 Dispatchers.IO 모두를 최대치로 사용한다면, 활성화된 스레드의 수는 스레드 한도 전부를 합친 것과 동일하다.

예를 들어 Dispatchers.IO에서 64개의 스레드까지 사용할 수 있고, 8개의 코어를 가지고 있다면 공유 스레드 풀에서 활성화된 스레드는 72개일 것이다.

 

하지만 스레드의 한도는 독립적이기 때문에 다른 디스패처의 스레드를 고갈시키는 경우는 없다. 

예를 들어 Dispatchers.IO 작업이 매우 많아져서 64개의 스레드를 모두 사용하더라도, Dispatchers.Default의 스레드를 고갈시키지 않는다.

 

class DiscUserRepository(
    private val discReader: DiscReader
) : UserRepository {
    override suspend fun getUser(): UserData =
        withContext(Dispatchers.IO) {
            UserData(discReader.read("userName"))
        }
}

블로킹 함수를 호출해야 하는 경우 withContext(Dispatchers.IO)로 감싸 중단 함수로 만드는 것이 좋다.

이러한 함수는 다른 중단 함수와 다르지 않기 때문에 간단하게 사용할 수 있다.

 

 

커스텀 스레드 풀을 사용하는 IO 디스패처

withContext(Dispatchers.IO)로 감싼 함수가 너무 많은 스레드를 블로킹하면 문제가 발생할 수 있다.

64개보다 훨씬 많은 스레드를 블로킹하는 코드가 있다면, 나머지 스레드는 자기 차례가 돌아올 때까지 스레드 전부를 기다려야 한다.

이런 경우 limitedParallelism을 활용한다.

 

Dispatchers.IO에서 limitedParallelism() 함수를 사용하면 독립적인 스레드 풀을 가진 새로운 디스패처를 만든다.

원하는 만큼 스레드 수를 설정할 수 있으므로, 새로운 스레드 풀의 스레드 수는 64개로 제한되지 않는다.

suspend fun main(): Unit = coroutineScope {
    launch {
        printCoroutinesTime(Dispatchers.IO) // Dispatchers.IO took: 2074
    }
    launch {
        val dispatcher = Dispatchers.IO
            .limitedParallelism(100)
        printCoroutinesTime(dispatcher) // LimitedDispatcher@XXX took: 1082
    }
}

suspend fun printCoroutinesTime(
    dispatcher: CoroutineDispatcher,
) {
    val test = measureTimeMillis {
        coroutineScope {
            repeat(100) {
                launch(dispatcher) {
                    Thread.sleep(1000)
                }
            }
        }
    }
    println("$dispatcher took: $test")
}

100개의 코루틴이 각각 스레드를 1초씩 블로킹하는 위 코드를 보자.

Dispatchers.IO에서 실행하면 2초가 걸린다. (64개 동시에 실행(1초) + 나머지 36개 동시에 실행(1초))

limitedParallelism으로 100개의 스레드를 사용하는 Dispatchers.IO에서 실행하면 1초가 걸린다. (100개가 동시에 실행 (1초))

 

• Dispatchers.Default의 limitedParallelism은 디스패처에 스레드 수 제한을 추가한다.
• Dispatchers.IO의 limitedParallelism은 Dispatchers.IO와 독립적인 새로운 디스패처를 만든다.
• 모든 디스패처는 스레드 수가 무제한인 스레드 풀을 함께 공유한다.

 

class DiscUserRepository(
    private val discReader: DiscReader,
) : UserRepository {
    private val dispatcher = Dispatchers.IO
        .limitedParallelism(5)

    override suspend fun getUser(): UserData =
        withContext(dispatcher) {
            UserData(discReader.read("userName"))
        }
}

스레드를 자주 블로킹하는 클래스에서 limitedParallism를 사용해 적절한 스레드 수를 가진 커스텀 디스패처를 정의할 수 있다.

적절한 스레드 수에 정답은 없다.

너무 많은 스레드는 자원을 비효율적으로 사용한다.

너무 적은 스레드는 성능 상 좋지 않다. 스레드를 사용할 수 있을 때까지 기다리게 되기 때문이다.

 

 

정해진 수의 스레드 풀을 가진 디스패처

val NUMBER_OF_THREADS = 20
val dispatcher = Executors
    .newFixedThreadPool(NUMBER_OF_THREADS)
    .asCoroutineDispatcher()

개발자들이 직접 스레드 풀을 관리하기 위해 자바의 Executors를 사용할 수 있다.

asCoroutineDispatcher() 함수를 사용해 코루틴 디스패처로 변환할 수도 있다.

 

이렇게 만들어진 디스패처의 문제점은 아래와 같다.

• close() 함수를 꼭 호출해야 한다. 개발자가 close() 함수 호출을 깜빡한다면 스레드 누수가 발생한다.
• 정해진 수의 스레드 풀을 만들면 스레드가 효율적으로 사용되지 않는다. 사용하지 않는 스레드가 다른 디스패처와 공유되지 않고 살아있기 때문이다.

 

 

싱글 스레드로 제한된 디스패처

var i = 0

suspend fun main(): Unit = coroutineScope {
    repeat(10_000) {
        launch {
            i++
        }
    }
    delay(1000)
    println(i) // ~9930
}

2개 이상의 스레드를 사용하는 디스패처에서는 공유 상태로 인한 문제가 발생한다.

위 코드에서 10,000개의 코루틴이 i를 1씩 증가시키고 있지만, 10,000보다 작은 값이 출력된다.

동시에 다수의 스레드가 공유 상태 i를 변경했기 때문이다.

 

val dispatcher = Executors.newSingleThreadExecutor()
    .asCoroutineDispatcher()

var i = 0

suspend fun main(): Unit = coroutineScope {
    val dispatcher = Dispatchers.Default
        .limitedParallelism(1)

    repeat(10_000) {
        launch(dispatcher) {
            i++
        }
    }
    delay(1000)
    println(i) // 10000
}

이 문제를 해결하는 방법 중 하나는 싱글 스레드를 가진 디스패처를 사용하는 것이다.

스레드의 수를 1개로 제한한 Dispatchers.Default나 (스레드를 블로킹한다면) Dispatchers.IO를 주로 사용한다.

 

suspend fun main(): Unit = coroutineScope {
    val dispatcher = Dispatchers.Default
        .limitedParallelism(1)

    val job = Job()
    repeat(5) {
        launch(dispatcher + job) {
            Thread.sleep(1000)
        }
    }
    job.complete()
    val time = measureTimeMillis { job.join() }
    println("Took $time") // ~5000
}

단 하나의 스레드가 블로킹되면 작업이 순차적으로 처리된다는 문제가 있다.

 

 

프로젝트 룸의 가상 스레드 사용하기

JVM 플랫폼의 프로젝트 룸은 일반적인 스레드보다 훨씬 가벼운 가상 스레드를 사용한다.

프로젝트 룸이 코루틴보다 유용한 경우는 Dispatchers.IO 대신 가상 스레드를 사용할 때다.

Dispatchers.IO는 스레드를 블로킹할 수밖에 없다.

일반적인 스레드를 블로킹하는 것보다 가상 스레드를 블로킹하는 것이 비용이 훨씬 적게 든다.

 

가상 스레드를 사용하는 룸 디스패처를 만들기 위한 코드는 아래와 같다.

val LoomDispatcher = Executors
    .newVirtualThreadPerTaskExecutor()
    .asCoroutineDispatcher()

object LoomDispatcher : ExecutorCoroutineDispatcher() {
    override val executor: Executor = Executor { command ->
        Thread.startVirtualThread(command)
    }

    override fun dispatch(
        context: Coroutinecontext,
        block: Runnable,
    ) {
        executor.execute(block)
    }

    override fun close() {
        error("Cannot be invoked on Dispatchers.LOOM")
    }
}

 

val Dispatchers.Loom: CoroutineDispatcher
    get() = LoomDispatcher

룸 디스패처를 쉽게 찾고, 다른 디스패처와 비슷하게 사용하기 위해 Dispatchers의 확장 프로퍼티를 정의했다.

 

이제 Dispatchers.Loom와 Dispatchers.IO를 비교해 보자.

가상 스레드를 블로킹하면, 메모리 사용량도 적고 시간도 적게 걸릴 것이다.

 

예시 상황은 100,000개의 코루틴에서 각각 1초 동안 스레드를 블로킹하는 것이다. 

Dispatchers.Loom에서는 2초가 걸렸다.

suspend fun main() = measureTimeMillis {
    coroutineScope {
        repeat(100_000) {
            launch(Dispatchers.Loom) {
                Thread.sleep(1000)
            }
        }
    }
}.let(::println) // 2273

 

Dispatchers.IO에서는 Dispatchers.Loom의 10배인 23초가 걸렸다.

(Dispatchers.Loom과 동일한 비교를 위해 Dispatchers.IO의 스레드 제한을 코루틴 수만큼 증가시켰다)

suspend fun main() = measureTimeMillis {
    val dispatcher = Dispatchers.IO
        .limitedParallelism(100_100)

    coroutineScope {
        repeat(100_000) {
            launch(dispatcher) {
                Thread.sleep(1000)
            }
        }
    }
}.let(::println) // 23803

 

프로젝트 룸은 아직 시작 단계라 실제로 사용하기엔 어렵지만 Dispatchers.IO를 대체할 수 있는 경쟁자라 생각한다.

프로젝트 룸이 안정화되면 코루틴에서 가상 스레드를 기본으로 사용할 수 있다고 했기 때문에 나중에는 룸 디스패처가 필요하지 않을 수도 있다.

 

 

제한받지 않는 디스패처

suspend fun main(): Unit =
    withContext(newSingleThreadContext("Thread1")) {
        var continuation: Continuation<Unit>? = null

        launch(newSingleThreadContext("Thread2")) {
            delay(1000)
            continuation?.resume(Unit)
        }

        launch(Dispatchers.Unconfined) {
            println(Thread.currentThread().name) // Thread1

            suspendCancellableCoroutine<Unit> {
                continuation = it
            }

            println(Thread.currentThread().name) // Thread2 
            delay(1000)
            println(Thread.currentThread().name) // kotlinx.coroutines.DefaultExecutor (delay에서 사용한 스레드)
        }
    }

Dispatchers.Unconfined는 스레드를 바꾸지 않는 디스패처다.

코루틴을 시작한 스레드에서 실행된다. 재개되었을 때는 재개한 스레드에서 실행된다.

 

Dispatchers.Unconfined는 단위 테스트할 때 유용하다.

launch를 호출하는 함수를 테스트하는 상황을 가정해 보자. launch가 비동기적으로 실행된다면, 테스트할 함수가 끝날 시기를 제어하는 것은 쉽지 않다.

이런 경우 Dispatchers.Unconfined로 다른 디스패처를 대체하면 된다. (추후 살펴볼 runTest를 사용할 수도 있다)

모든 스코프에서 Dispatchers.Unconfined를 사용하면 모든 작업이 동일한 스레드에서 실행되기 때문에, 연산의 순서를 훨씬 쉽게 통제할 수 있다.

 

스레드 스위칭을 일으키기 않기 때문에 비용도 가장 저렴하다.

어떤 스레드에서 코루틴을 실행할지를 신경쓰지 않아도 되는 경우에 사용할 수 있다.

하지만 현업에서 사용하기엔 적절하지 않다. 실수로 메인 스레드에서 블로킹 연산을 실행한다면 애플리케이션 전체가 멈추게 될 것이다. 

 

 

메인 디스패처로 즉시 옮기기

코루틴을 스레드에 배정하는 것도 비용이 든다.

이미 실행되고 있는 스레드에 새로운 코루틴이 배정되면 필요없는 연산을 하게 된다.

 

suspend fun showUser(user: User) =
    withContext(Dispatchers.Main) {
        userNameElement.text = user.name
        // ...
    }

위 함수가 이미 메인 디스패처에서 호출이 되었다면 다시 메인 디스패처에 배정하기 위한 필요없는 비용이 발생했을 것이다.

 

또한 메인 스레드를 기다리는 큐에 코루틴이 쌓여있었다면 withContext 때문에 약간의 지연 후 실행된다.

withContext가 호출되면 코루틴은 중단되고 스레드를 기다리는 큐에서 기다리다가 재개된다.

(큐에 있는 다른 코루틴의 실행이 끝날 때까지 기다려야 하기 때문에)

 

suspend fun showUser(user: User) =
    withContext(Dispatchers.Main.immediate) {
        userNameElement.text = user.name
        // ...
    }

이런 문제를 방지하기 위해 Dispatchers.Main.immediate를 사용할 수 있다.

메인 디스패처에서 위 함수를 호출 하면 디스패처 배정 없이 즉시 실행된다.

 

현재는 메인 디스패처만 immediate 옵션을 지원하고 있다.

 

 

컨티뉴에이션 인터셉터

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

    fun <T> interceptContinuation(
        continuation: Continuation<T>
    ): Continuation<T>

    fun releaseInterceptedContinuation(
        continuation: Continuation<*>
    ) {
    }

    // ... 
}

디스패칭은 ContinuationInterceptor라는 컨텍스트를 기반으로 작동한다.

• interceptContiunation(): 코루틴이 중단되었을 때 호출되어 Continuation 객체를 수정/래핑한다.
• releaselnterceptedContinuation(): Continuation이 종료되었을 때 호출된다.

DispatchedContinuation은 디스패처가 작동하기 위한 핵심 요소이다.

디스패처는 Continuation 객체를 DispatchedContinuation으로 감싸서, 어떤 스레드에서 코루틴을 실행할지 제어한다.

이 과정에서 ContinuationInterceptor의 interceptContiunation() 함수를 사용하는 것이다.

 

 

작업의 종류에 따른 각 디스패처의 성능 비교

작업의 종류에 따라 각 디스패처의 성능을 비교하기 위해 벤치마크 테스트를 수행했다.

다음 표는 평균 실행 시간을 밀리초(millisecond) 단위로 나타낸 것이다.

 

	중단	블로킹	CPU 집약적인 연산	메모리 집약적인 연산
싱글스레드	1,002	100,003	39,103	94,358
디폴트 디스패처(스레드 8개)	1,002	13,003	8,473	21,461
10 디스패처(스레드 64개)	1,002	2,003	9,893	20,776
스레드 100개	1,002	1,003	16,379	21,004

 

표의 주요 사항은 다음과 같다.

1. 단순히 중단만 하는 경우에는 스레드 수가 실행 시간에 영향을 끼치지 않는다.
2. 스레드를 블로킹하는 경우에는 스레드 수가 많을수록 실행 시간이 빠르다.
3. CPU 집약적인 연산에서는 Dispatchers.Default가 가장 좋다.
   • 스레드를 더 많이 사용할수록 프로세서는 스레드 사이를 스위칭하는 데 쓰는 시간이 더 늘어나 의미있는 연산을 하는 시간은 줄어든다. 그렇기 때문에 스레드 100개를 사용하는 것보다 Dispatchers.Default가 더 효율적인 것이다.
   • CPU 집약적인 연산에서 Dispatchers.IO을 사용하면 안 된다. Dispatchers.IO는 블로킹 연산을 처리하기 위한 용도이기 때문에 다른 작업이 스레드 전체를 블로킹할 수 있다.
4. 메모리 집약적인 연산에서는 스레드를 많이 사용하는 것이 좋지만, 큰 차이가 나지는 않는다.

 

 

요약

디스패처는 코루틴이 실행될 스레드나 스레드 풀을 결정한다.

• Dispatchers.Default는 CPU 집약적인 연산에 사용한다.
• Dispatchers.Main은 안드로이드의 메인 스레드에 접근할 때 사용한다.
• Dispatchers.Main.immediate는 Dispatchers.Main를 사용하는 스레드에서 실행되지만 꼭 필요할 때만 재배정된다.
• Dispatchers.IO는 스레드가 블로킹되는 연산을 할 때 사용한다.
• 스레드 수를 제한한 Dispatchers.IO와 특정 스레드 풀을 사용하는 커스텀 디스패처는 블로킹하는 함수를 많이 호출할 때 사용한다.
• 공유 상태 동기화 문제를 방지하기 위해 싱글 스레드 디스패처를 사용한다.
  • 스레드 수가 1개로 제한된 Dispatchers.Default
  • 스레드 수가 1개로 제한된 Dispatchers.IO
  • 싱글 스레드를 사용하는 커스텀 디스패처
• Dispatchers.Unconfined는 코루틴이 실행될 스레드에 대해 신경 쓸 필요가 없을 때 사용한다.

 

 

 

출처

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