[코틀린 코루틴] 코틀린의 동시성 모델 - 코루틴

코틀린은 동시성 모델로 쓰레드 활용이 아닌, 코루틴을 권장한다.

코틀린 코루틴 vs 전통적 스레드

코틀린에서는 concurrent.thread 함수를 이용하면, 새 스레드를 간편하게 시작할 수 있다.

하지만 이는 커널 스레드를 직접 만드는 형태이기 때문에 비용이 많이 든다.

• 스레드는 계층 개념이 없기 때문에, 계층적 관리를 하기 까다롭다.
• 스레드는 어떤 작업이 완료되길 기다리는 동안에는 블록된다.

최신 시스템이라도 한번에 몇 천개의 스레드만 효과적으로 관리할 수 있다.

---

기본적으로 스레드는 굉장히 무거운 자원이다.

• 요청 1개를 처리하는 동안 스레드 1개를 할당하는 구조에서,
• DB 호출/외부 HTTP/파일 I/O 등으로 스레드가 블로킹되면 그 스레드는 그 시간 동안 유용한 일을 못 하면서도 계속 점유된다.
• 동시 요청이 늘면 “대기 중인 스레드”가 급격히 늘고, 결국 가용 스레드 풀 고갈이 발생한다.

그 결과, 새 요청은 스레드를 못 받아 큐잉(대기열 증가) → 응답 지연 증가 → 타임아웃으로 전파된다.

 

“그럼 스레드를 더 늘리면 되지 않나?”가 자연스러운 대응인데, 명백한 한계가 있다.

1) 메모리 비용

• JVM 스레드는 보통 스택 메모리 등으로 스레드 수가 늘수록 메모리 사용량이 선형 증가한다.
• 스레드가 수천 단위로 늘면 메모리 압박, GC 압박이 커지게 된다.

2) 스케줄링/컨텍스트 스위칭/캐시 페널티

• runnable 스레드가 많아지면 스케줄링 부담이 커지고, 컨텍스트 스위칭과 캐시 미스가 늘어 처리량이 오히려 떨어지는 구간이 생긴다.

3) 보통 “진짜 병목”은 외부 시스템이다

DB/외부 API가 느려서 블로킹이 길어지는 경우, 스레드를 늘리면 애플리케이션은 더 많은 동시 호출을 만들고, 외부 시스템의 큐/락/커넥션이 터지면서 전체 지연이 더 악화될 수 있다.

 

따라서 새 스레드를 생성할 때는 매우 신중해야 하며, 짧은 시간 동안 잠깐 사용하는 것은 피하는 것이 좋다.

---

코틀린은 스레드에 대한 대안으로 코루틴을 도입했다.

 

코루틴은 일시 중단 가능한 계산을 수행하는 객체를 나타낸다.

• 코루틴은 초경랑 추상화다. 일반적인 노트북에서도 10만개 이상의 코루틴을 쉽게 실행할 수 있다.
• 코루틴은 "시스템 자원"을 블록시키지 않고 실행을 일시 중단할 수 있으며, 나중에 중단된 지점에서 실행을 재개할 수 있다.
• 코루틴은 구조적 동시성(structured concurrency)이라는 개념을 도입해 동시 작업의 구조와 계층을 확립한다.
  • 취소 및 오류 처리를 위한 매커니즘을 제공한다.
  • 동시 계산의 일부가 실패하거나 더 이상 필요하지 않게 됐을 때, 자식으로 시작된 다른 코루틴들도 함께 취소되도록 보장한다.

코루틴은 하나 이상의 JVM 스레드에서 실행된다.

한 프로세스는 기껏해야 수천 개의 스레드를 가질 수 있지만, 코루틴을 사용하면 수백만 개의 동시성 작업을 실행할 수 있다.

 

본격적으로 코틀린의 동시성 프로그래밍 모델인 코틀린 코루틴을 살펴보자.
먼저, 가장 기본적인 구성 요소인 일시 중단 함수부터 시작하자.

---

코틀린 Suspending 함수

코틀린 코루틴이 스레드/리액티브 스트림/콜백과 같은 다른 동시성 접근 방식과 다른 핵심 속성으로는

상당수의 경우 코드 '형태'를 크게 변경할 필요가 없다.

 

라는 점이다.

코드는 여전히 순차적으로 보인다.

 

suspend 함수가 어떻게 이런 방식을 가능하게 하는지 자세히 살펴보자.

 

먼저 테스트를 위해 다음과 같은 샘플 코드를 작성했다.
Thread.sleep은 네트워크 I/O를 가정했다.

data class UserID (val nickname: String)  

data class Credentials(val username: String, val password: String)  

data class UserData(val fullName: String, val email: String)  

fun login(credentials: Credentials): UserID{  
    Thread.sleep(1000)  
    return UserID(credentials.username)  
}  

fun fetchUserData(userID: UserID): UserData {  
    Thread.sleep(1000)  
    return UserData("Full Name of ${userID.nickname}", "sample@sample.net")  
}  

fun showData(data: UserData){  
    println("User Data: $data")  
}

그리고 이제 단순 블로킹 코드를 작성해보자.

계산적인 관점에서 보면 이 코드는 실제로 많은 작업을 소모하진 않는다.
대부분의 작업을 네트워크 작업 결과를 기다리는데 소모하기 때문에, 블로킹된 스레드가 누적되기 쉬워지고, 이는 사용자 경험을 저해시킨다.

 

간단한 예시로, 다음과 같은 코드를 실행해보자.

실제로 이 코드를 실행해보면, 블로킹된 스레드가 누적되는 것을 막기 위해, 일정 갯수 단위로 차례대로 실행되는 것을 확인할 수 있다.

필자의 경우 2만개 언저리씩 동시에 실행되었다.
이는 OS 레벨에서 JVM 프로세스의 최대 스레드 수를 제한했기 때문이다.

이 코드를 완전히 수행하는데, 18초가 걸렸다.

이제 위 기능을 코루틴 + 일시 중단 함수를 이용해 같은 로직을 구현해보자.

suspend fun login(credentials: Credentials): UserID {  
    delay(1000)  
    return UserID(credentials.username)  
}  

suspend fun fetchUserData(userID: UserID): UserData {  
    delay(1000)  
    return UserData("Full Name of ${userID.nickname}", "sample@sample.net")  
}  

fun showData(data: UserData) {  
    println("User Data: $data")  
}  

suspend fun showUserInfo(credentials: Credentials) {  
    val userID = login(credentials)  
    val userData = fetchUserData(userID)  
    showData(userData)  
}

여기서 delay 함수는 kotlinx.coroutines.delay 함수로, 코루틴을 1천ms 딜레이 시키는 역할을 수행한다.
이는 네트워크 I/O를 가정해 스레드를 1초 블로킹하던 Thread.sleep(1000) 을 대체한다.

 

이제 이 코드를 다음과 같이 실행해보자.

스레드보다 훨씬 많은 병렬 실행이 가능한 덕분에, 불과 2.3초만에 모든 작업을 완료할 수 있었다.
실제 I/O 대기시간이 2초임을 감안하면, 경이로운 속도이다.

위 main 함수를 실행할 때 사용했던 launch, runBlocking와 같은 개념들은 곧 알아볼 것이다.

어떻게 이게 가능한걸까?

힌트는 "경량화"에 있다.

일반적인 스레드 모델의 경우, CPU가 사용하는 스레드를 직접 컨텍스트 스위칭하며 가져오게 되고, 이 스레드 하나하나는 지나치게 큰 단위어서 많이 다루기엔 메모리가 부족하고, 따라서 충분한 동시성을 확보하지 못하는 문제가 있었다.

스레드 스택, PC(프로그램 카운터), 스택 포인트, 쓰레드 로컬, 스케줄링 엔티티(TCB) 등등이 "상시 가동하는 애플리케이션"내에는 굳이 필요하지 않을 수 있다.

 

하지만, 코루틴은 이 스레드의 지나친 무거움을 해결하고자, 다음과 같은 접근 방식을 택했다.

스레드 자체를 직접 쓰는 대신, 훨씬 간소화된 Job 객체를 현재까지의 실행 정보를 저장한 객체로 보관하는 것이다.

---

코루틴/콜백/퓨처/리액티브 스트림 비교

그럼 코루틴을 다른 기법들과 비교해보자.

콜백

자바스크립트를 한번이라도 사용해봤다면 Callback Hell이라는 용어로 먼저 접하게되는 Callback 방식부터 알아보자.

Callback Hell은 코드의 가독성을 떨어뜨리는 것으로 유명하다.

Future

Future은 이 Callback Hell 문제를 해결하기 위해, 다른 방식의 복잡성을 택했다.

• thenAccept와 thenCompose같은 새로운 연산자의 의미를 배워야 한다.
• 또한, 반환 타입을 CompletableFuture로 감싸야 한다.

리액티브 스트림

Reactive Stream 방식은 콜백 지옥을 피할 수 있지만, 여전히 함수 시그니처를 변경해야 하며, 반환값을 Single로 감싸야 하고, flatMap, doOnSuccess, subscribe 같은 연산자를 사용해야 한다.

fun login(credentials: Credentials): Single<UserID>
fun fetchUserData(userID: UserID): Single<UserData>
fun showData(data: UserData)

fun showUserInfo(credentials: Credentials){
    login(credentials)
        .flatMap { loadUserData(it) }
        .doOnSuccess { showData(it) }
        .subscribe()
}

---

물론 Future, 리액티브 스트림의 경우에는 적절한 활용 사례가 있다.

• 코틀린에는 deferred 값이라는 자체적인 퓨처 스타일이 있다.
• 코틀린에는 코루틴용 Reactive Stream의 추상화인 Flow가 있다.

이에 대해서는 차근차근 알아볼 예정이다.

---

코루틴 빌더

그러면 suspend 함수는 어떻게 호출해야 하는걸까?

 

기본적으로 suspend함수는 다른 suspend 함수 내부나 코루틴 내부에서 호출해야 한다.

결국 재귀적으로 "코루틴 내부"로 들어가는 입구가 필요하다는 건데, 이에는 두가지 방식이 있다.

• main 함수를 suspend 함수로 만들기
  • 이 경우, main이 외부 프레임워크에 존재할 경우 이를 바꾸기 어렵다.
• 자체적인 코루틴을 Build하기 - Coroutine Builder

코루틴 빌더는 새로운 코루틴 인스턴스를 생성한다.

 

이쯤에서 코루틴에 대한 정의를 다시 한 번 확인해보자.

코루틴은 일시 중단 가능한 계산의 인스턴스다.
이를 다른 코루틴들과 동시에(심지어는 병렬로)실행될 수 있는 코드 블록으로 생각할 수 있다.
스레드와 비슷하지만 코루틴은 함수 실행을 일시 중단하는 데 필요한 매커니즘을 포함하고 있다.

이러한 코루틴을 생성할 때는 코루틴 빌더 함수 중 하나를 사용한다.
'기초'적인 코루틴 빌더 함수는 다음과 같다.

• runBlocking은 블로킹 코드와 일시 중단 함수의 세계를 연결할 때 쓰인다.
• launch는 값을 반환하지 않는 새로운 코루틴을 시작할 때 쓰인다.
• async는 비동기적으로 값을 계산할 때 쓰인다.

이제 각각을 자세히 살펴보자.

runBlocking 함수 - 일반 코드에서 코루틴의 세계로

runBlocking 함수의 경우, 현재 스레드가 현재 열린 코루틴 객체를 처리하는 주체가 된다.

 

현재 스레드는 현재 실행하던 작업을 중단하고, 해당 컨텍스트에 디스패치된 코루틴 재개를 처리하기 위한 내부 루프(이벤트 루프 유사 구조) 를 돌린다.

 

다음 예제는 내장된 delay 함수를 사용해 코루틴을 0.5초동안 일시 중단한 후 텍스트를 출력한다.

해당 runBlocking 함수를 통해, 새로운 코루틴 객체를 생성하고 그 안에서 suspend 함수를 실행할 수 있게 되었다.
하지만 현재 스레드가 실행하던 작업을 중단하기는 원치 않을 수 있다.
이때 launch 함수가 사용된다.

launch 함수 - 발사 후 망각 코루틴 생성

launch 함수는 새로운 자식 코루틴을 시작하는 데 쓰인다.

위에서 코루틴을 처음 사용해봤을 때 사용했던 launch 함수이다.

• 이 함수는 runBlocking과 달리, 현재 실행중인 작업을 중단하지 않는다.
  • 따라서 10만개의 작업을 동시에 코루틴 객체로 생성할 수 있었다.
• 또한, 이 함수는 Job을 반환한다.
  • 이 값은 시작된 코루틴에 대한 핸들(인터럽트, 중단, etc.) 용도로 사용할 수 있다.

runBlocking과 launch는 공통적인 특징도 있다.

• 바로 둘 모두 '발사 후 망각' 형태이다.
• 코루틴을 수행한 뒤의 반환값 개념이 필요 없을 때 사용한다.

그와 반대로, async는 코루틴이 실행되고 난 후 필요한 반환값이 있을 때 사용할 수 있는 유용한 도구이다.

async 빌더 - 대기 가능한 연산

이 함수를 실행하면 다음과 같은 결과를 확인할 수 있다.

즉, 제일 오래 걸리는 시간인 0.4초만에 응답값을 받을 수 잇었다.

 

이는 Deferred 타입에서 사용했던 Future/Promise 의 형태와 유사하다.

즉, 정리하면 다음과 같다.

• 반환값이 필요없을 경우
  • runBlocking - 코루틴 첫 진입 시 사용
  • launch - 코루틴 진입 후 자식 코루틴 생성 시 사용
  • 주로 파일/데이터베이스에 중요하지 않은 쓰기 작업을 실행할 때 사용한다.
• 반환값이 필요할 경우
  • async - 언제든지 사용, deferred 객체 반환.

---

지금까지 코루틴을 시작하기 위한 방법을 살펴보았다.

 

하지만, 지금까지 알아본 방식은 무조건 현재 스레드가 이벤트 루프를 돌며 코루틴 객체를 처리하고 있다.
하지만, 이 비동기 작업을 처리할 스레드를 명시적으로 지정하고, 현재 스레드는 하던 작업을 계속 진행하고 싶을 수 있다.

 

그렇다면 이제 이 코루틴 객체들을 처리하는 스레드는 어떻게 지정할 것인가에 대해서 이해해야 할 차례이다.

디스패처 - 코드를 어디서 실행할 것인가?

코루틴의 디스패처는 코루틴을 실행할 스레드(풀)를 결정한다.
디스패처를 선택함으로써,

• 코루틴을 특정 스레드로 제한하거나
• 코루틴을 스레드 풀에 분산시킬 수 있으며
• 코루틴이 한 스레드에서만 실행될지 여러 스레드에서 실행될지 결정할 수 있다.

 

일반적인 코루틴/Future/Promise과 같은 비동기 프로그래밍 모델의 실행 방식은 다음과 같은 방식이 있다. 

• 현재 스레드에서 실행하기
  • 위에서 알아본 runBlocking 함수의 경우, 현재 스레드가 현재 열린 코루틴 객체를 처리하는 주체가 된다.
  • 현재 스레드는 현재 실행하던 작업을 중단하고, 해당 컨텍스트에 디스패치된 코루틴 재개를 처리하기 위한 내부 루프(이벤트 루프 유사 구조) 를 돌린다.
• 이벤트 루프 방식
  • 프레임워크/라이브러리에 명시된 하나(or n개)의 스레드가 "무한 루프" 형식으로 "준비 완료됨/대기중" 상태 전이 이벤트를 다뤄가며 디멀티플렉싱을 수행하여 상시 Waiting/Running 상태를 유지한다.
  • 자바스크립트(Node.js)와 파이썬의 비동기 처리 방식(asyncio)이 대표적으로 이 방식대로 동작한다.
• 디스패처 기반
  • 작업을 실행할 스레드의 집합을 정의하고, 스케줄링을 정의하는 컨텍스트가 존재한다.
  • 해당 스레드 집합은 코루틴 재개를 처리하기 위한 내부 루프(이벤트 루프 유사 구조)를 갖는다.
  • 각 작업은 지정된 스레드(풀)의 작업 큐로 가서 스레드가 작업을 실행하고, 다시 Continuation으로 돌아가는 과정을 반복한다.

공식 문서에서는 runBlocking 의 경우, 현재 스레드가 블로킹된다라고 정의되어 있는데, 이 부분에 대해서 엄밀한 의미로는 OS 레벨의 블로킹이 아닌, 현재 해당 스레드가 실행하던 작업이 중단된다라는 의미로 보는 것이 타당하다.

공식 문서의 runBlocking 설명

그럼 디스패처는 어떻게 선택해야 할까?
기본적으로 코루틴은 기본적으로 부모 코루틴에서 디스패처를 상속받기 때문에, 모든 코루틴에 대해 명시적으로 디스패처를 지정할 필요는 없다.

 

하지만 선택할 수 있는 유용한 디스패처들이 있다.
이러한 유용한 디스패처에 대해 알아보자.

• Dispatchers.Default
  • 스레드 개수: CPIU 코어 수
  • 일반적인 연산/CPU 집약적인 작업에 사용한다.
• Dispatchers.Main
  • 스레드 개수: 1
  • UI 프레임워크의 맥락에서 주로 사용되는, UI 작업에 사용한다.
• Dispatchers.IO
  • 스레드 개수: 64+CPU 코어 개수(단, 최대 64개만 병렬 실행됨)
  • 블로킹 I/O 작업, 네트워크 작업, 파일 작업에 사용한다.
• Dispatchers.Unconfined
  • 아무 스레드나 급하게 사용한다.
  • 즉시 스케줄링해야 하는 특별한 경우에 사용한다. (일반적인 용도는 아니다.)
• limitedParallelism(n)
  • 스레드 개수: 커스텀(n)
  • 특별한 요구사항이 있을 때 사용한다.

이에 대한 휴리스틱한 선택 기준이 있다.

이렇게 필요할 때 선택해서 사용하자.

---

그렇다면 디스패처는 어떻게 지정할 수 있을까?

 

크게 두가지 방법을 사용할 수 있다.

• 코루틴 빌더에 디스패처 전달
• 코루틴 내에서 디스패처 바꾸기

코루틴 빌더 파라미터로 디스패처 전달

fun main(){
    runBlocking{
        log("Doing some work")
        launch(Dispatchers.Default){
            log("Doing some background work")
        }
    }
}

이와 같이, launch/async의 파라미터로 디스패처를 지정할 수 있다.

withContext() - 코루틴 안에서 디스패처 바꾸기

launch(Dispatchers.Default){
    val result = performBackgroundOperation()
    withContext(Dispatchers.IO){
        callUpdate(result)
    }
}

이와 같이, 코루틴 내에서 특정 작업만을 다른 컨텍스트에서 하고 싶을 때, withContext로 실행되고 있는 디스패처를 바꿀 수 있다.

 

이를 그림으로 표현하면 다음과 같다.

---

코루틴 내에서의 동시성 문제

코루틴도 결국 동시성을 구현하는 도구이기 때문에, 동시성 문제에 대해서 자유롭지 않다.

전형적인 예시지만, 1씩 올리는 카운트를 1만번 반복하는 예제로 진행해보겠다.

fun main() = runBlocking {  
    var counter = 0  
    repeat(10_000) {  
        launch(Dispatchers.Default) {  
            counter++  
        }  
    }    delay(1.seconds)  
    println("Counter = $counter")  
}

10000이 나오질 않는다.

뮤텍스

kotlinx.coroutines.sync.Mutex는 Mutex 잠금을 제공하며, 이를 통해 코드 임계 영역이 한 번에 하나의 코루틴만 실행되게 보장할 수 있다.

fun main() = runBlocking {  
    var counter = 0  
    val mutex = Mutex()  
    repeat(10_000) {  
        launch(Dispatchers.Default) {  
            mutex.withLock {  
                counter++  
            }  
        }    }    delay(1.seconds)  
    println("Counter = $counter")  
}

mutex.withLock() 함수를 이용하자, 정상적으로 1만을 출력하는 것을 확인할 수 있다.

세마포어

이번엔 서버의 부하를 막기 위해, 무료 사용자에 대해서 한번에 3개의 요청만 받는 세마포어가 필요하다고 가정하자.

그러면 다음과 같이 구현할 수 있을 것이다.

이와 같이 Semaphore.withPermit을 이용해 세마포어를 조절할 수 있다.

 

---

지금까지 코틀린의 대표적 동시성 처리 방식인 코루틴에 대해 살펴보았다.

이를 통해 코루틴에 대한 이해 및 코틀린의 디스패처, 그리고 코틀린 코루틴의 사용법(코루틴 빌더)을 익힐 수 있었길 바란다.

 

출처: 코틀린 인 액션

Kotlin in Action 2/e | 세바스티안 아이그너 | 에이콘출판사 - 예스24