13-2. 동시성 : 코루틴 흐름 제어와 잡 생명 주기

13장 - 동시성

 

13-1. 코루틴

13-2. 코루틴 흐름 제어와 잡 생명 주기

13-3.  동시성 통신

13-4.  자바 동시성 사용하기

---

13-2. 코루틴 흐름 제어와 잡 생명 주기

 

잡은 동시성 작업의 생명 주기를 표현하는 객체이다.

잡을 사용하면 작업 상태를 추적하고 필요할 때 작업을 취소할 수 있다.

잡이 취할 수 있는 상태들에 대해 알아보고, 각 상태가 무엇을 의미하며

하나의 상태에서 다른 상태로 어떻게 전이되는지 알아보자.

활성화 상태는 작업이 시작됐고 아직 완료나 취소로 끝나지 않았다는 뜻 이다.

활성화 상태가 디폴트 상태로, 잡은 생성되자마자 활성화 상태가 된다.

 

신규 상태의 잡에 대해 start()나 join() 메서드를 호출하면

잡이 시작되면서 활성화 상태가 된다.

fun main() {
    runBlocking {
        val job = launch(start = CoroutineStart.LAZY){
        // CoroutineStart.DEFAULT >> active
        // CoroutineStart.LAZY >> not active(New)
            println("Job started")
        }
        
        delay(100)

        println("Preparing to start..")
        job.start()
    }
}
//Preparing to start..
//Job started

runBlocking {
    val job = launch(start = CoroutineStart.DEFAULT){
        println("Job started")
    }

    delay(100)

    println("Preparing to start..")
    job.start()
}
//Job started
//Preparing to start..

---

활성화 상태에서는 코루틴 장치가 잡을 반복적으로 일시 중단하고 재개시킨다.

잡이 다른 잡을 시작할 수도 있는데, 이 경우 새 잡은 기존 잡의 자식이 된다.

따라서 잡의 부모-자식 관계는 동시성 계산 사이에 트리 형태의 의존 구조를 만든다.

chirdren 프로퍼티를 통해 완료되지 않은 자식 잡들을 얻을 수 있다.

fun main() {
    runBlocking {
        val job = coroutineContext[Job.Key]!!

        launch { println("This is task A") }
        launch { println("This is task B") }
        
        println("${job.children.count()} children running")
        // 2 children running
    }
}

코루틴이 일시 중단 람다 블록의 실행을 끝내면 잡의 상태는 "완료 중" 상태로 바뀐다.

잡은 모든 자식이 완료될 때 까지 이 상태를 유지하고,

모든 자식이 완료되면 "완료 중" 상태에서 "완료됨" 상태로 바뀐다.

 

잡의 join() 메서드를 통해 조인 대상 잡이 완료될 때 까지 현재 코루틴을 일시 중단시킬 수 있다.

fun main() {
    runBlocking {
        val job = coroutineContext[Job.Key]!!

        val jobA = launch { println("This is task A") }
        val jobB = launch { println("This is task B") }

        jobA.join()
        jobB.join()
        
        println("${job.children.count()} children running")
    }
}
//This is task A
//This is task B
//0 children running

위 코드는 루트 코루틴의 메시지가 두 자식 메시지의 실행이 끝난 뒤 출력되도록 보장해준다.

현재 잡 상태를 잡의 isActive, isCancelled, isComplete 프로퍼티를 통해 추적할 수 있다.

잡 상태	isActive	isCompleted	isCancelled
신규	false	false	false
활성화	true	false	false
완료 중	true	false	false
취소 중	false	false	true
취소됨	false	true	true
완료됨	false	true	false

상태가 "완료됨", "취소됨"인 잡의 isComplete가 true인 점에 유의해야 한다.

이 둘은 isCancelled 프로퍼티를 통해 구분할 수 있다.

---

1) 취소

잡의 cancel() 메서드를 호출하면 잡을 취소할 수 있다.

이 메서드는 더 이상 필요 없는 계산을 중단시킬 수 있는 가장 표준적인 방법을 제공해야 한다.

취소 가능한 코루틴이 스스로 취소를 요청했는지 검사해 적절히 반응해줘야 된다(=취소에는 협력해야 한다.)

 

코루틴이 취소를 위해 협력하지 않는 경우가 있을 수 있다.

이를 해결하기 위해 다음 작업을 수행하기 전에 코루틴이 취소됬는지 검사하는 방법이 있다.

suspend fun main() {
    val squarePrinter = GlobalScope.launch(Dispatchers.Default) {
        var i = 1
        while (isActive) {
            println(i++)
        }
    }

    delay(100) 
    // 자식 잡이 어느정도 실행될 시간을 줌
    squarePrinter.cancel()
}

isActive 확장 프로퍼티는 현재 잡이 활성화된 상태인지 검사한다.

 

코루틴 취소 여부를 확인하는 다른 방법은 상태를 검사하는 것 대신에

CancellationException을 발생시키며 취소에 반응할 수 있게 일시 중단 함수를 호출하는 것 이다.

이 예외는 잡을 취소하는 과정이 진행 중이라는 사실을 전달하는 토큰 역할을 하는 예외이다.

코루틴 라이브러리에 정의된 delay(), join() 등의 모든 일시 중단 함수가 이 예외를 발생시켜준다.

 

또 하나의 방법은 yield()를 이용하는 방법이 있다.

이 함수는 실행 중인 잡을 일시 중단시켜서 자신을 실행 중인 스레드를 다른 코루틴에게 양보한다.

suspend fun main() {
    val squarePrinter = GlobalScope.launch(Dispatchers.Default) {
        var i = 1
        while (true) {
            yield()
            println(i++)
        }
    }
    delay(100)  // 자식 잡이 어느정도 실행될 시간을 줌
    squarePrinter.cancel()
}

---

정리하자면 코루틴을 취소하는 방법은 세 가지가 있다.

1. cancel()을 사용해 취소하는 방법

 이 때 코루틴이 협력하지 않는 경우를 대비해 isActive() 함수를 통해 취소 여부를 확인

2. CancellationException을 사용

 delay(), join() 등의 일시 중단 함수가 발생시키는 예외

 잡을 취소하는 과정이 진행 중이라는 사실을 전달하는 토큰 역할을 하는 예외이다.

3. yield()를 사용

실행 중인 잡을 일시 중단시켜서 자신을 실행 중인 스레드를 다른 코루틴에게 양보하는 함수.

---

부모 코루틴이 취소되면 자동으로 모든 자식의 실행을 취소한다.

이 과정은 부모에게 속한 모든 잡 계층이 취소될 때 까지 계속된다.

fun main() {
    runBlocking {
        val parentJob = launch {
            println("Parent started")
            launch {
                println("Child 1 started")
                delay(500)
                println("Child 1 completed")
            }

            launch {
                println("Child 2 started")
                delay(500)
                println("Child 2 completed")
            }

            delay(500)
            println("Parent completed")
        }

        delay(100)
        parentJob.cancel()
    }
}
//Parent started
//Child 1 started
//Child 2 started

위 코드는 1개의 부모 잡과 2개의 자식 잡을 시작하는 코루틴을 실행한다.

이렇게 만들어진 3개의 작업은 모두 완료 메시지를 표시하기 전에 500밀리초를 기다린다.

하지만 부모 잡이 100밀리초만에 취소되므로 세 잡 모두 완료 상태에 도달하지 못한다.

 

---

2) 타임아웃

경우에 따라 작업이 완료되기를 무작정 기다릴 수 없어서 타임아웃을 설정하고 싶을 수 있다.

코루틴 라이브러리에 withTimeout() 함수를 통해 타임아웃을 설정할 수 있다.

fun main() {
    runBlocking {
        val asyncData = async { File("data.txt").readText() }
        try {
            val text = withTimeout(50) { asyncData.await() }
            println("Data loaded: $text")
        } catch (e: Exception) {
            println("Timeout exceeded")
        }
    }
}

위 코드는 파일을 50ms 안에 읽을 수 있다면 결과를 돌려주고,

50ms 안에 읽을 수 없다면 TimeoutCancellationException을 던지고, 파일을 읽는 코루틴을 취소한다.

비슷한 함수로 타임아웃 발생 시 예외를 던지는 대신 널을 반환하는 withTimeoutOrNull() 함수도 있다. 

---

3) 코루틴 디스패치하기

코루틴 라이브러리에는 특정 코루틴을 실행할 때 사용할 스레드를 제어하는 작업을

담당하는 컴포넌트가 있는데, 이 컴포넌트를 코루틴 디스패처(dispatcher)라고 부른다.

디스패처는 코루틴 문맥의 일부이므로 launch()나 runBlocking() 등의 코루틴 빌더 함수에서

이를 지정할 수 있고, 코루틴 빌더에 디스패처를 넘길 수도 있다.

fun main() {
    runBlocking {
        launch(Dispatchers.Default) {
            println(Thread.currentThread().name) // DefaultDispatcher-worker-1
        }
    }
}

---

아래 코드는 실행하는 스레드에 이름을 부여하는

커스텀 스레드 팩토리를 사용하는 풀 기반의 실행기(executor)를 정의한다.

fun main() {
    val id = AtomicInteger(0)

    val executor = ScheduledThreadPoolExecutor(5) { runnable ->
        Thread(
                runnable,
                "WorkerThread-${id.incrementAndGet()}"
        ).also { it.isDaemon = true }
    }

    executor.asCoroutineDispatcher().use { dispatcher ->
        runBlocking {
            for (i in 1..3){
                launch (dispatcher) {
                    println(Thread.currentThread().name)
                    delay(1000)
                }
            }
        }
    }
}
//WorkerThread-1
//WorkerThread-2
//WorkerThread-3

---

4) 예외처리

코루틴 빌더는 예외 처리를 하는데, 두 가지 기본 전략중 하나를 따른다.

1. launch() 빌더가 사용, 예외를 부모 코루틴에게 전달

2. async() 빌더가 사용, 던져진 예외를 저장 > await 호출을 받으면 다시 Throw

 

먼저 첫번째 방법인 예외를 부모 코루틴에게 전달하는 방법에 대해 작성하자면,

예외의 전파 방식은 아래와 같다.

 ● 부모 코루틴이 자식과 같은 오류로 취소되고, 이로 인해 부모의 다른 자식도 취소됨

 ● 자식 코루틴이 모두 취소되고 나면 부모는 예외를 코루틴 윗부분으로 전달

 ● 전역 영역에 코루틴에 도달할 때 까지 위 과정을 반복

 ● 이후 예외가 Coroutine.ExcpetionHandler.Consider에 의해 처리

fun main() {
    runBlocking {
        launch {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch {
            delay(1000)
            println("Task B completed")
        }
        println("Root")
    }
}
// Root
// Exception in thread "main" java.lang.Exception: Error in task A

위 코드가 동작하는 방식을 살펴보면,

최상위 코루틴은 두 개의 내부 작업을 시작하고 그중 첫 번째 코루틴이 예외를 던진다.

이로인해 최상위 작업이 취소되고 자식인 두 작업도 취소된다.

---

개인적으로 결과가중에 Root가 출력되는 것이 이해가 안갔다.

내가 이해한 바로는 자식 코루틴의 예외가 발생한 경우

부모까지 예외가 전파되어 코루틴이 취소된다고 이해했고, 

이 경우 부모의 코루틴은 정상적으로 실행되지 않아야 된다고 생각했다.

그래서 결과는 Root가 출력되지 않고 Task A에 대한 예외만 Throw 되야할 것 같은데,

실제로는 Root가 출력되고나서 Task A에 대한 예외 메시지가 출력됬다.

 

최상위 코루틴의 결과인 Root를 출력하는 이유가

혹시 코드의 실행 속도가 너무 빨라서가 아닐까? 라는 가설을 세워보았고, delay()를 통해 검증해봤다.

여러 곳에 delay()를 넣어본 결과 내가 생각한 가설이 맞는 것 같다. 

 

먼저 최상위 코루틴에 Root를 출력하기 전에 딜레이를 넣어봤다.

fun main() {
    runBlocking {
        launch {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch {
            delay(1000)
            println("Task B completed")
        }
        delay(1000)
        println("Root")
    }
}
//Exception in thread "main" java.lang.Exception: Error in task A

그랬더니 처음에 내가 예상했던 결과가 나왔고, 내가 세운 가설이 맞다고 생각됬다.

그래도 좀 더 검증해 보기 위해 다른 방식으로도 해봤다.

fun main() {
    runBlocking {
        launch {
            delay(1000)
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch {
            println("Task B completed")
        }
        println("Root")
    }
}
//Root
//Task B completed
//Exception in thread "main" java.lang.Exception: Error in task A

결과를 보니 내가 처음 이해한 내용이 맞는 것 같다.

---

CoroutineExceptionHandler는 현재 코루틴 문맥(CoroutineContext)과 던저진 예외를 인자로 받는다.

이를 이용해 핸들러를 만드는 간단한 방법은

인자가 두 개인 람다를 받는 CoroutineExceptionHandler() 함수를 사용하는 것 이다.

suspend fun main() {
    val handler = CoroutineExceptionHandler{_, exception ->
        println("Caught $exception")
    }

    GlobalScope.launch (handler) {
        launch {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        launch {
            delay(1000)
            println("Task B completed")
        }
        println("Root")
    }.join()
}
//Root
//Caught java.lang.Exception: Error in task A

---

두 번째 예외를 처리하는 방법인 async() 빌더에서 사용하는 방법은

던저진 예외를 저장했다가 예외가 발생한 계산에 대해

await() 호출을 받았을 때 다시 예외를 던지는 것 이다.

fun main() {
    runBlocking {
        val deferredA = async {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        val deferredB = async {
            println("Task B completed")
        }
        deferredA.await()
        deferredB.await()
        println("Root")
    }
}
// Exception in thread "main" java.lang.Exception: Error in task A

위 코드에서 Root가 출력되지 않는 이유는, deferredA.await()에서 예외가 다시 던져지므로

프로그램이 println("Root") 문장을 실행하지 못하기 때문이다.

 

코루틴 데이터에 접근할 때 예외를 다시 던지는 방식을 채용하는

async과 유사한 빌더들의 경우CoroutineExceptionHandler를 사용하지 않는다.

따라서 코루틴 문맥에 CoroutineExceptionHandler를 설정했어도

아무 효과가 없이전역 디폴트 핸들러가 호출된다.

---

내포된 코루틴에서 발생한 예외를 전역 핸들러를 통하지 않고

부모 수준에서 처리하고 싶을 경우 try-catchc 블록으로 예외를 처리할 수 없다.

fun main() {
    runBlocking {
        val deferredA = async {
            throw Exception("Error in task A")
            println("Task A completed")
        }

        val deferredB = async {
            println("Task B completed")
        }
        try {
            deferredA.await()
            deferredB.await()
        } catch (e: Exception) {
            println("Caught $e")
        }
        println("Root")
    }
}
//Caught java.lang.Exception: Error in task A
//Root
//Exception in thread "main" java.lang.Exception: Error in task A

catch 블록 내부가 호출되지만 프로그램은 예외와 함께 중단된다.

중단되는 이유는 자식이 실패한 경우 부모를 취소시키기 위해 자동으로 예외를 다시 던지기 때문이다.

try-catch 블록을 이용해 프로그램이 중단되지 않게 하고 싶은 경우 

슈퍼바이저(supervisor) 잡을 이용해야 한다.

 

슈퍼바이저 잡을 이용하면 cancel()이 아래 방향으로만 전달된다.

즉 슈퍼자이저를 취소하면 슈퍼바이저 잡은 자동으로 자신의 모든 자식을 취소하지만,

자식이 취소 된 경우, 부모나 다른 자식은 아무 영향을 받지 않는다

fun main() {
    runBlocking {
        supervisorScope {
            val deferredA = async {
                throw Exception("Error in task A")
                println("Task A completed")
            }

            val deferredB = async {
                println("Task B completed")
            }
            try {
                deferredA.await()
                deferredB.await()
            } catch (e: Exception) {
                println("Caught $e")
            }
            println("Root")
        }
    }
}
//Task B completed
//Caught java.lang.Exception: Error in task A
//Root

예외가 발생했지만, B 작업과 Root 코루틴이 정상동작한것을 확인할 수 있다.

---

다음글

13-3.  동시성 통신