비동기와 병렬성은 다루는 방법: 코틀린 코루틴(Kotlin Coroutine) 1

 

병렬성(parallelism)의 문제

https://software.rajivprab.com/2018/04/29/myths-programmers-believe-about-cpu-caches/

 

 

1) CPU마다 갖는 캐시를 서로 다르기 때문에 동기화가 어렵다는 문제가 있음(서로 로컬 캐시 이기 때문)

2) L3 캐시를 사용하더라도 사로 다른 코어가 동시에 접근할 수 없음(lock)

    - 보통의 경우 먼저 선점한 코어가 빠져나오기까지 block 됨.

    - 이럴 거면 멀티 코어를 쓰는 의미가 없어짐

 

위와 같이 비동기(순서가 서로 다른), 병렬성에 따른 문제가 있었고 이를 해결하기 위해 프로그래밍 언어 레벨에서 해결하려는 노력들(RxJava, 콜백)이 있었다.

 

이러한 노력에도 단점들이 있는데, 콜백을 사용할 경우 콜백 정의부가 메소드 내부 인자로 들어감에 따라 코드 depth가 커지고 가독성이 나빠질 수 있으며 RxJava를 사용할 때엔 항상 스트림을 사용해야 한다는 단점이 있다.

 

또한 콜백, RxJava 모두 동기, 비동기(병렬) 형태로 모두 프로그래밍이 가능하나, 각각의 경우에 따라 코드가 달라진다는 문제가 있다.

 

 

Why 코루틴

동시성이냐 동시성이 아니냐(코어레벨 비동기 병렬성)에 따라 코드가 달라지지 않으며 같은 형태의 코드로 모든 곳에 적용할 수 있다.

 

단, 스트림이 적합한 코드가 있는 것이고 코루틴이 적합한 코드가 있는 것이기 때문에 어떠한 것이 더 좋다고 말할 순 없다.

 

- '코루틴'은 비동기와 병렬성을 순차적으로 프로그래밍할 수 있는 방법을 제공하고 

- '플로우'는 비동기와 병렬성을 스트림 형태로 풀 수 있도록 한다.

 

이 둘을 사용해서 순차적으로 짤 수 있는 부분에는 코루틴을 사용하고

스트림이 더 맞는 부분에선 플로우를 사용해야 한다.

 

 

코루틴 빌더

- 코루틴 빌더(Coroutine Builder)는 코틀린에서 코루틴을 생성하고 시작하는 데 사용되는 함수

- 코루틴의 결과나 상태관리를 담당

- 주요 빌더: launch, async, runBlocking, withContext

* runBlocking 외 코루틴 빌더는 반드시 다른 코루틴 빌더 내에서 실행해야 함

 

1. runBlocking 코루틴 빌더

fun main(): Unit = runBlocking<Unit> {
    println(Thread.currentThread().name)
    println(this)
}

// 출력결과
// main
// BlockingCoroutine{Active}@6a6824be

 

- 자바에서 this를 출력하면 현재 오브젝트의 클래스 이름과 JVM 메모리 주소를 출력하는 것과 마찬가지로, 코루틴 안에서 this를 출력할 경우 BlockingCoroutine가 출력됨을 볼 수 있다. BlockingCoroutine 객체는 CoroutineScope의 자식이며 runBlocking을 사용할 경우 쓰레드 풀에서 다른 쓰레드를 사용하지 않고 현재 main 쓰레드를 사용한다.

- runBlocking 코루틴 빌더는 현재 쓰레드를 block 하고 코루틴을 실행한다. 물론 해당 코루틴을 현재 쓰레드가 실행할 수도 있다.

- runBlocking 코루틴 빌더는 반환값을 지정할 수 있으며 반환값이 없는 경우 보통 <Unit>이 생략된다. (컴파일러가 알 수 있는 경우 생략 가능)

 

 

cf. CoroutineScope

- 모든 코루틴은 CoroutineScope로 시작한다.

- 즉, 위 코드에서 runBlocking과 같은 방식으로 코루틴을 정의한 구간에서는 CoroutineScope이 갖는 필드를 참조할 수 있다.

- CoroutineScope에는 코루틴에 대한 정보인 coroutineContext를 가진다.

 

fun main() = runBlocking {
    println(coroutineContext)
    println(Thread.currentThread().name)
    println(this)
}

// 출력결과
// [BlockingCoroutine{Active}@2c13da15, BlockingEventLoop@77556fd]
// main
// BlockingCoroutine{Active}@2c13da15

 

 

2. launch 코루틴 빌더

- runBlocking 코루틴 빌더가 block 되어 실행되었다면 launch 코루틴 빌더는 새로운 코루틴을 생성해서 실행된다.

- launch 코루틴 빌더는 가능하다면 병렬적으로 코루틴이 실행되도록 한다.

- runBlocking, coroutineScope 빌더 안에만 선언될 수 있다.

 

fun main() = runBlocking {
    runBlocking {
        println(coroutineContext)
        println("runBlocking1: ${Thread.currentThread().name}")
        println(this)
    }
    launch {
        println(coroutineContext)
        println("launch1: ${Thread.currentThread().name}")
        println(this)
    }
    println(coroutineContext)
    println("runBlocking2: ${Thread.currentThread().name}")
    println(this)
}

// [BlockingCoroutine{Active}@77556fd, BlockingEventLoop@368239c8]
// runBlocking1: main
// BlockingCoroutine{Active}@77556fd
// [BlockingCoroutine{Active}@7b3300e5, BlockingEventLoop@368239c8]
// runBlocking2: main
// BlockingCoroutine{Active}@7b3300e5
// [StandaloneCoroutine{Active}@7a46a697, BlockingEventLoop@368239c8]
// launch1: main
// StandaloneCoroutine{Active}@7a46a697

 

- 위와 같이 launch는 main 블록이 먼저 실행된 후에 실행되도록 설계되어 있다.

- 또한 코루틴 빌더에 디스패처를 등록하지 않았기 때문에 Default 디스패처가 사용되어 모두 같은 쓰레드에서 실행되었다.

 

 

delay 함수

- suspension point(중단점) 지정을 통한 코루틴 간 비동기 처리

- delay 함수는 코루틴이나 suspend 함수에서만 호출 가능

fun main() {
    println("[0] main: ${Thread.currentThread().name}")
    runBlocking {
        launch {
            println("[2] launch1: ${Thread.currentThread().name}")
            delay(1000L)
            println("[5] launch1 end")
        }
        launch {
            println("[3] launch2: ${Thread.currentThread().name}")
        }
        println("[1] runBlocking: ${Thread.currentThread().name}")
        delay(500L)
        println("[4] runBlocking end") // 부모 코루틴은 자식 코루틴이 끝나기를 기다림
    }
    println("[6] main end")
}

// [0] main: main
// [1] runBlocking: main
// [2] launch1: main
// [3] launch2: main
// [4] runBlocking end
// [5] launch1 end
// [6] main end

 

- delay 함수는 코루틴이 사용 중인 쓰레드를 잠시 놓아주고 다른 코루틴이 사용할 수 있도록 한다(non-blocking)

- sleep은 현재 쓰레드를 잡아둔 채로 실행을 중지하지만 delay는 쓰레드를 놓아주기 때문에 비동기 작업 설계 시 사용될 수 있다.

- 다른 언어에서 코루틴은 계층적이지 않은 경우가 많은데, 코틀린에서의 코루틴들은 계층적인 구조를 가진다. 따라서 부모 코루틴이 취소되면 자식 코루틴도 모두 취소된다. 또한 Orphan 코루틴이 생기지 않도록 부모 코루틴은 항상 자식 코루틴이 끝나기를 기다린다. ([4], [5] 출력 결과 참고)

 

 

suspend 함수

- suspend 함수는 중단 가능한 함수이다.

- 따라서 suspend 함수는 다르 코루틴이나 suspend 함수를 호출할 수 있다.

- suspend 함수는 코루틴이 아니다. 잠시 중단될 수 있는 함수임을 정의한 것뿐이다.

 

suspend fun do1() {
    println("[2] launch1: ${Thread.currentThread().name}")
    delay(1000L)
    println("[5] launch1 end")
}

fun do2() {
    println("[3] launch2: ${Thread.currentThread().name}")
}

suspend fun do3() {
    println("[1] runBlocking: ${Thread.currentThread().name}")
    delay(500L)
    println("[4] runBlocking end")
}

fun main() {
    println("[0] main: ${Thread.currentThread().name}")
    runBlocking<Unit> {
        launch {
            do1()
        }
        launch {
            do2()
        }
        do3()
    }
    println("[6] main end")
}

// [0] main: main
// [1] runBlocking: main
// [2] launch1: main
// [3] launch2: main
// [4] runBlocking end
// [5] launch1 end
// [6] main end

 

 

 

코루틴 스코프 빌더

// 에러 코드 //
suspend fun example() {
    launch {  // Unresolved reference: launch
        println("${Thread.currentThread().name}")
        delay(500L)
    }
}

fun main(): Unit = runBlocking<Unit> {
    example()
}

 

위 코드는 컴파일 에러가 발생하는 코드이다. runBlocking을 제외한 모든 코루틴은 다른 코루틴 안에서 실행되어야 하기 때문이다. 따라서 launch 빌더로 코루틴을 생성할 수 없게 된다.

 

위와 같은 상황에서 코루틴을 생성하기 위해 코루틴 스코프 빌더가 적용된다.

suspend fun example() = coroutineScope {
    this.launch {
        println("${Thread.currentThread().name}")
        delay(500L)
    }
}

fun main(): Unit = runBlocking<Unit> {
    example()
}

 

즉, suspend 함수와 같이 코루틴이 정의될 수 있는 함수에 코루틴 빌더를 사용하고 싶을 경우 coroutineScope를 사용하면 된다. coroutineScope 빌더 역시 하나의 코루틴을 생성하기 때문에 내부에 선언된 모든 코루틴이 끝난 뒤에 coroutineScope로 선언한 코루틴 빌더가 모두 완료하게 된다.

 

runBlocking 코루틴 빌더는 현재 쓰레드를 block 하고 코루틴을 실행한다면 coroutineScope 빌더는 launch와 비슷하게 현재 쓰레드가 다른 일을 할 수 있도록 설계되어있다. (사실 runBlocking, withContext 코루틴 빌더 외 대부분의 코루틴 빌더는 모두 non-block이다.)

 

 

 

 

cf. 코루틴 상태정보 : Job

- launch와 같이 비동기로 실행된 코루틴에 대한 상태 정보들을 Job 객체로 참조하여 각종 조작을 할 수 있다.

- 아래는 Job 객체에 대해 suspension point를 지정한 join 함수이다.

- join 함수는 해당 코루틴이 끝나기를 기다린다.

suspend fun example() = coroutineScope {
    var job = this.launch {
        println("[1] ${Thread.currentThread().name}")
        delay(500L)
    }
    job.join() // suspension point

    this.launch {
        println("[3] ${Thread.currentThread().name}")
        delay(500L)
    }

    println("[2]")
}

fun main(): Unit = runBlocking<Unit> {
    example()
    println("[4]")
}

// [1] main
// [2]
// [3] main
// [4]

 

 

 

 

3. async 코루틴 빌더

- launch 코루틴 빌더와 마찬가지로 새로운 코루틴을 만들고 실행된다.

- launch는 Job을 통해 수행결과를 받아올 수 있지만 async는 await 키워드를 통해 수행결과를 받아올 수 있다.

- 코루틴 결과를 받아야 할 경우 async, 결과를 받지 않을 경우 launch를 사용한다.

 

suspend fun example(): Int {
    delay(100L)

    return Random.nextInt(0, 10)
}

fun main() = runBlocking {
    var res = async { example() }  // this: 코루틴
    println("${res.await()}")      // job의 join + return 값과 같음
}

 

- await이 호출되는 지점에선 실행 중단을 잠시 멈추고 결과가 반환되기를 기다리는 suspension point이다.

- async와 await은 항상 같이 사용된다.

- async 키워드는 작업 내용을 큐에 넣는 행위이며 실제 실행 시점은 알 수 없다.

 

 

 

코루틴 예외 전파 **

- 코루틴은 서로 계층 구조로 구성되기 때문에 코루틴에서 예외가 발생하면 부모 및 형제 코루틴에게도 모두 전파된다.

- 전파를 받은 부모 및 형제 코루틴들은 모두 작업을 cancel한다.

 

suspend fun example1(): Int {
    try {
        delay(100L)

        return Random.nextInt(0, 10)
    } finally { // 현재 함수에서 예외가 발생한게 아니기 때문에 catch로 잡히지 않음
        println(">>> example1 canceled")
    }
}

suspend fun example2(): Int {
    try {
        delay(100L)
        throw RuntimeException()
    } catch (e: Exception) {
        println(">>> example2 canceled")
        throw e
    }
}

suspend fun exampleParent() = coroutineScope {  // 부모 코루틴
    var res1 = async { example1() }  // 자식 코루틴1
    var res2 = async { example2() }  // 자식 코루틴2

    try {
        println("${res1.await() + res2.await()}")
    } catch (e: Exception) {
        println(">>> exampleParent canceled")
    }
}

fun main() = runBlocking {
    try {
        exampleParent()
    } catch (e: Exception) {
        println("catch in main")
    }
}

// >>> example1 canceled
// >>> example2 canceled
// >>> exampleParent canceled
// catch in main

 

- async 키워드는 작업 내용을 큐에 넣는 행위이기 때문에 예외는 await에서 받아올 수 있다.

- 위와 같이 하나의 all-or-nothing 성향의 작업을 여러 코루틴으로 나누고 작업한다면 특정 코루틴에서 실패시 전체 코루틴이 취소되도록할 수 있다.

 

 

 

---

 

 

정리

- 코루틴 블록 안에서는 CoroutineScope이라는 객체로 정의된다.

- 코루틴 블록 안에서의 this는 CoroutineScope 객체이다.

- CoroutineScope가 가지는 필드 중 코루틴에 대한 정보인 coroutineContext가 있다.

- 코루틴을 만드는 방법

    - 코루틴 빌더 사용: launch, async, runBlocking, withContext

    - 코루틴 스코프 빌더 사용: coroutineScope(suspend 함수 내에서 코루틴 정의 시)

- 현재 쓰레드를 block 하고 코루틴을 실행하는 빌더: runBlocking, withContext

- 코틀린에서의 코루틴들은 계층적인 구조로 구성(클래스 상속구조와 비슷한 개념)

- 코틀린 코루틴의 부모 코루틴이 취소되면 자식 코루틴도 모두 취소됨

- Orphan 코루틴이 생기지 않도록 부모 코루틴은 항상 자식 코루틴이 끝나기를 기다림

- suspend 함수: delay 등으로 중단될 수 있는 코루틴이 정의된 함수

- 코루틴 결과를 받아야 할 경우 async, 결과를 받지 않을 경우 launch를 사용

- 코루틴은 계층 구조를 가지며 특정 코루틴에서 예외 발생시 부모 형제 코루틴 역시 cancel된다.

 

 

 

 

Reference

- Coroutine Dispatcher, 넌 대체 뭐야?, https://todaycode.tistory.com/182

- [F-Lab 미니컨] 비동기로부터 우리를 구원해줄 코루틴, https://youtu.be/w_kRlfhNb3c?si=oc9S9mZfe5DHd9O2

- Kotlinc docs, https://kotlinlang.org/docs/coroutines-guide.html