13-1. 동시성 : 코루틴

13장 - 동시성(concurrent)

 

13-1. 코루틴

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

13-3.  동시성 통신

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

---

내가 듣기로는 악명이 꽤 높은 기술 중 하나가 코루틴에 대한 내용이다.

저번 스터디때 "코틀린의 코루틴과 자바의 RxJava가 서로 대응되는 기술인가요?" 라는 질문을 했었는데,

대응되는 기술은 아니고, 비슷한 기능이다 라는 답변을 받았었다.

사실 코루틴과 RxJava 모두 어려운 기술이라고 들었고,

최근은 아니지만 6개월정도 전에 안드로이드 앱 개발자 취업 공고를 보면

코루틴 사용 가능자, 동시성에 대한 이해 뭐 이런식의 내용들이 적혀있었던 것으로 기억한다.

그래서 어렵긴 하지만 꼭 배워야 할 내용이라고 생각되고,공부할 때도 좀 더 신경써서 해봐야겠다.

 

나는 학교에서 C#을 이용한 유니티 VR 게임 개발을 할 때 코루틴을 사용해 본적이 있어서

단어 자체는 꽤나 익숙하다. 내가 C#에서 코루틴을 사용할 때는 사용자가 어떤 동작을 한 뒤

몇 초 뒤에 코드를 실행하고 싶은 경우 invoke() 또는 코루틴을 사용했다.

C#의 코루틴과 코틀린의 코루틴이 과연 같은 기술인지는 모르겠지만,

내가 알고 있는 코루틴에 대한 내용이 도움이 되길 바래야겠다.

 

이 장에서는 동시성(concurrent) 코드 작성이라는 주제에 대한 내용이다.목표는 코루틴을 이해하는 것 이라고 한다.

---

13-1. 코루틴

 

코틀린 프로그램에서도 자바 동시성 기본 요소를 쉽게 사용해

Thread-safe하게 코드를 작성할 수 있지만 현실적인 이유 때문에

동시성 스레드를 많이 사용하는것은 비실용적이거나 불가능할 수도 있다.

그래서 효율적인 방법인 비동기 프로그래밍을 사용하면 해결할 수 있지만

비동기 프로그래밍의 가장 큰 문제점은 일반적인 명령형 제어 흐름을 사용할 수 없어서

코드 복잡도가 엄청나게 들어난다는 점 이다.

 

코틀린에서는 동기 프로그래밍과 비동기 프로그래밍의 장점을 코루틴을 사용해 둘다 취할 수 있다.

동기 프로그래밍의 명령형 스타일로 코드를 작성하면

컴파일러가 코드를 효율적인 비동기 계산으로 자동 변환해준다.

 

대부분의 코루틴 기능이 별도 라이브러리로 제공되므로 프로젝트 설정에 추가해야 한다고 한다.

나는 책에서 사용하는 버전인 org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.4.3 을 사용한다.

 

1) 코루틴과 일시 중단 함수

코루틴 라이브러리의 기본 요소는 일시 중단 함수이다.

이 함수는 일반적인 함수를 더 일반화해 함수 본문의 원하는 지점에서

함수에 필요한 모든 런타임 문맥을 저장하고 함수 실행을 중단한 다음,

나중에 필요할 때 다시 실행을 계속 진행할 수 있게 한 것 이다.

코틀린에서는 이런 함수에 suspand라는 변경자를 붙인다.

suspend fun foo(){
    println("Task started")
    delay(1000)
    println("Task started")
}

delay() 함수는 코루팀 라이브러리에 정의된 일시 중단 함수로,

Thread.sleep()과 비슷한 일을 한다. 차이점은 delay() 함수는

현재 스레드를 블럭시키지 않고 자신을 호출한 함수를 일시 중단 시키며

스레드를 다른 작업을 수행할 수 있게 풀어준다.

일시 중단 함수(suspand function)는 일시 중단 함수와 일반 함수를 원하는 대로 호출할 수 있다.

일시 중단 함수를 호출하면 해당 호출 지점이 일시 중단 지점이 된다.

코틀린은 일반 함수가 일시 중단 함수를 호출하는 것을 금지한다.

fun main() {
    foo() // ERROR
    // Suspend function 'foo' should be called only from a coroutine or another suspend function
}

일시 중단 함수를 일시 중단 함수에서만 호출할 수 있다면

일반 함수에서 일시 중단 함수를 호출하고 싶은 경우에 대한 의문이 생긴다.

가장 간단한 방법은 main() 함수를 suspand로 변경하는 것 이다.

suspend fun main() {
    foo() // OK
}

하지만 현실적으로 동시성 코드의 동작을 제어하고 싶기 때문에

공통 생명 주기와 문맥이 정해진 몇몇 작업이 정의된 구체적인 영역 에서만 동시성 함수를 호출한다.

코루틴을 사용할 때 사용하는 여러 가지 함수를 코루틴 빌더라고 부르는데,

코루틴 빌더는 CoroutineScope 인스턴스의 확장 함수로 사용된다.

---

2) 코루틴 빌더

자주 사용되는 코루틴 빌더는 launch(), async(), runBlocking() 이라는 코루틴 빌더가 있다.

 

launch() 함수는 코루틴을 시작하고,

코루틴 실행 중인 작업의 상태를 추적하고 변경할 수 있는 Job 객체를 반환한다.

launch() 함수는 CoroutineScope.() -> Unit 타입의 일시 중단 람다를 받는데,이 람다는 새 코루틴의 본문에 해당한다.

import kotlinx.coroutines.*
import java.lang.System.*

fun main() {
    val time = currentTimeMillis()

    GlobalScope.launch {
        delay(100)
        println("Task 1 finished in ${currentTimeMillis() - time} ms")
        // Task 1 finished in 231 ms
    }

    GlobalScope.launch {
        delay(100)
        println("Task 2 finished in ${currentTimeMillis() - time} ms")
        // Task 2 finished in 231 ms
    }

    Thread.sleep(200)
}

위 코드의 결과를 보면 두 작업이 실제로 병력적으로 실행됬다는 점을 알 수 있다.

하지만 실행 순서가 항상 일정하게 보장되지는 않고. 실행 순서를 강제할 수 있는 도구가 존재한다.

 

main() 함수 자체는 Thread.sleep()을 통해 메인 스레드 실행을 잠시 중단한다.

이를 통해 코루틴 스레드가 완료될 수 있도록 충분한 시간을 제공한다.

코루틴을 처리하는 스레드는 데몬 모드(daemon mode)로 실행되므로

main() 스레드가 이 스레드보다 빨리 끝나버리면 자동으로 실행이 종료된다.

동시성 작업의 내부에서는 일시 중단 함수로 sleep() 대신 delay()를 사용해야 한다.

 

데몬 모드가 뭔지 몰라서 찾아봤다.

멀티테스킹 운영 체제에서 데몬은 사용자가 직접적으로 제어하지 않고,

백그라운드에서 돌면서 여러 작업을 하는 프로그램을 말한다고 한다.

 

launch() 빌더는 동시성 작업이 결과를 만들어내지 않는 경우 적합하다.

그래서 이 빌더는 Unit 타입을 반환하는 람다를 인자로 받는다.

반면 결과가 필요한 경우 async() 라는 다른 빌더를 사용해야 한다.

---

async() 함수는 Deferred의 인스턴스를 반환하고, 이 인스턴스는 Job의 하위 타입으로

await() 메서드를 통해 계산 결과에 접근할 수 있게 해준다.

await() 메서드를 호출하면 이 함수는 계산이 끝나거나

계산 작업이 취소될 때 까지 현재 코루틴을 일시 중단시킨다.

만약 작업이 취소되면 await()는 예외를 발생시키면서 실패한다.

suspend fun main() {
    val message = GlobalScope.async {
        delay(100)
        "abc"
    }

    val count = GlobalScope.async {
        delay(100)
        1 + 2
    }

    delay(200)
    val result = message.await().repeat(count.await())
    println(result)
    // abcabcabc
}

위 내용대로, "abc"와 1 + 2의 결과인 3을 이용한 값을 이용해

result가 abcabcabc로 출력되는 것을 확인할 수 있다.

---

launch()와 async() 빌더의 경우 스레드 호출을 블럭시키지는 않지만,

백그라운드 스레드를 공유하는 풀(pool)을 통해 작업을 실행한다.

반대로 runBlocking() 빌더는 디폴트로 현재 스레드에서 실행되는 코루틴을 만들고,

코루틴이 완료될 때 까지 현재 스레드의 실행을 블럭시킨다.

코루틴이 성공적으로 끝나면 람다의 결과가 runBlocking() 호출의 결괏값이 된다.

코루틴이 취소되면 runBlocking()은 예외를 던진다.

반면 블럭된 스레드가 인터럽트되면 runBlocking()에 의해 시작된 코루틴도 취소된다.

fun main() {
    println("test1")
    GlobalScope.launch {
        delay(100)
        println("Background task : ${Thread.currentThread().name}")
    }
    println("test2")
    runBlocking {
        println("Primary task : ${Thread.currentThread().name}")
        delay(200)
    }
    println("test3")
}
//test1
//test2
//Primary task : main
//Background task : DefaultDispatcher-worker-1
//test3

책에는 test 1 2 3를 출력하는 코드가 없지만,

개인적으로 실행 순서가 궁금해서 test 1 2 3을 출력하는 코드를 추가했다.

 

위 코드를 보면 runBlocking() 내부 코루틴은 메인 스레드에서 실행된 반면,

launch()로 시작한 코루틴은 공유 풀에서 백그라운드 스레드를 할당받았음을 알 수 있다.

이러한 동작 때문에 runBlocking()을 다른 코루틴 안에서 사용하면 안된다고 한다.

runBlocking()은 블러킹과 넌블러킹 호출 사이의 다리 역할을 하기 위해 고안된 코루틴 빌더이므로

테스트나 메인 함수에서 최상위 빌더로 사용하는 등의 경우에만 runBlocking()을 써야 한다.

 

위 내용들을 내가 이해한대로 작성해보자면

코루틴에서 자주 사용하는 빌더는 launch(), async(), runBlocking()이 있는데,

launch()는 코루틴을 실행하고 따로 반환할 내용이 필요 없을때 사용

async()는 코루틴을 실행하고 반환받을 내용이 있을 경우에 사용하고, 반환값을 await() 함수를 통해 받음

launch(), async는 백그라운드 스레드 풀을 사용하는 반면,

runBlocking()현재 실행중인 스레드를 일시 중단시키고, 작성한 작업을 현재 스레드로 실행

그래서 다른 코루틴 내부에서 사용할 경우 문제가 생겨 테스트나 메인 함수의 최상위 빌더로만 사용

이정도로 간추릴 수 있을 것 같다.

 

개인적으로 내부 구현도 궁금해서 내부 구현을 찾아봤다

public fun kotlinx.coroutines.CoroutineScope.launch(
	context: kotlin.coroutines.CoroutineContext /* = compiled code */,
    start: kotlinx.coroutines.CoroutineStart /* = compiled code */, 
    block: suspend kotlinx.coroutines.CoroutineScope.() -> kotlin.Unit
): kotlinx.coroutines.Job { /* compiled code */ }
    
public fun <T> kotlinx.coroutines.CoroutineScope.async(
	context: kotlin.coroutines.CoroutineContext /* = compiled code */, 
    start: kotlinx.coroutines.CoroutineStart /* = compiled code */, 
    block: suspend kotlinx.coroutines.CoroutineScope.() -> T
): kotlinx.coroutines.Deferred<T> { /* compiled code */ }

public fun <T> runBlocking(
	context: kotlin.coroutines.CoroutineContext /* = compiled code */, 
    block: suspend kotlinx.coroutines.CoroutineScope.() -> T
): T { contract { /* compiled contract */ }; /* compiled code */ }

launch() 함수와 async() 함수는 반환 타입과 block 부분의 람다 식과 반환 타입이 다른것을 확인할 수 있고,

launch(), async()와 다르게 runBlocking()은 매개변수로 start: CoroutineStart를 안받는데,

CoroutineStart의 내부를 찾아보니 내부에 invoke() 메서드가 있는 것을 확인했다.

개인적인 추측으로는 runBlocking() 함수는 현재 스레드를 할당받아 사용하므로

invoke()함수가 포함된 CoroutineStart가 필요 없는 것으로 생각된다.

---

3) 코루틴 영역과 구조적 동시성

경우에 따라 코루틴이 어떤 연산을 수행하는 도중에만 실행되게 하고 싶을 수 있다.

동시성 작업 사이 부모-자식 관계로 인해 이런 실행 시간 제약이 가능하다.

어떤 코루틴을 다른 코루틴의 문맥에서 실행하면 후자가 전자의 부모가 되고,

이 경우 자식의 실행이 모두 끝나야 부모가 끝날 수 있도록 부모-자식 생명주기가 연관된다.

이러한 기능을 구조적 동시성(structed concurrency)이라고 부른다.

지역 변수 영역 안에서 블럭이나 서브루틴을 사용하는 경우 구조적 동시성을 비교할 수 있다.

fun main() {
    runBlocking {
        println("Parent task started")

        launch {
            println("Task A started")
            delay(200)
            println("Task A finished")
        }

        launch {
            println("Task B started")
            delay(200)
            println("Task B finished")
        }
        delay(100)
        println("Parent task finished")
    }
    println("Shutting down..")
}
//Parent task started
//Task A started
//Task B started
//Parent task finished
//Task A finished
//Task B finished
//Shutting down..

위 결과를 보면 딜레이를 100ms 줬으므로 부모 코루틴의 주 본문이 더 빨리 끝나는 것으로 보이지만,

부모 코루틴 자체는 이 시점에 실행이 끝나지 않고 일시 중단 상태로

두 자식(Task A, Task B)이 모두 끝날 때 까지 기다린다.

이후 runBlocking이 메인 스레드를 블럭하고 있기 때문에 부모 스레드가 끝난 뒤

메인 스레드의 블럭이 풀리고 마지막 메시지가 출력된다.

 

coroutineScope() 호출로 코드 블럭을 감싸면 커스텀 영역을 도입할 수도 있다.

runBlocking()과 유사한 방식으로 동작하지만, 두 함수의 가장 큰 차이는

coroutineScope()는 일시 중단 함수이므로 현재 스레드를 블럭시키지 않는다는 점 이다.

fun main() {
    runBlocking {
        println("Custom scope start")
        coroutineScope {
            launch {
                delay(100)
                println("Task 1 finished")
            }
            launch {
                delay(100)
                println("Task 2 finished")
            }
        }
        println("Custom scope finished")
    }
    println("Shutting down..")
}
//Custom scope start
//Task 1 finished
//Task 2 finished
//Custom scope finished
//Shutting down..

---

4) 코루틴 문맥

코루틴마다 CoroutineContext 인터페이스로 표현되는 문맥(context)이 연관돼 있으며,

코루틴을 감싼 변수 영역의 coroutineContext 프로퍼티를 통해 이 문맥에 접근할 수 있다.

문맥은 키-값 쌍으로 이뤄진 불변 컬렉션이며, 코루틴에서 사용할 수 있는 여러 데이터가 들어있다.

이 데이터중 두 가지 요소들에 대해 작성해보겠다

● 코루틴이 실행 중인 취소 가능한 작업을 표현하는 잡(job) 

● 코루틴과 스레드 연관을 제어하는 디스패처(dispatcher)

 

일반적으로 문맥은 CoroutineContext.Element를 구현하는 아무 데이터나 저장할 수 있다.

특정 원소에 접근하려면 get() 메서드나 인덱스 연산자에 키를 넘겨야 한다.

suspend fun main() {
    GlobalScope.launch {
        println("Task is active: ${coroutineContext[Job.Key]!!.isActive}")
        // 현재 잡을 얻고, "Task is active: true" 출력
    }
    delay(200)
}

새 문맥을 만드려면 두 문맥의 데이터를 합쳐주는 plus() / + 를 사용하거나

주어진 키에 해당하는 원소를 문맥에서 제거해주는 minusKey() 함수를 사용하면 된다.

private fun CoroutineScope.showName(){
    println("Current coroutine: ${coroutineContext[CoroutineName]?.name}")
}
fun main() {
    runBlocking {
        showName() // Current coroutine: null
        launch(coroutineContext + CoroutineName("Worker")){
            showName() // Current coroutine: Worker
        }
    }
}

---

다음글

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