구조화된 동시성 - 1

이윤설·2024년 10월 22일
코루틴코틀린

코루틴 연구소

목록 보기
6/10

섹션 소개

구조화된 동시성의 원칙이란, 비동기 작업을 구조화하여 비동기 프로그래밍을 보다 안정적이고, 예측 가능하게 만드는 원칙이다.

코루틴은 구조화된 동시성의 원칙을 사용하여, 비동기 작업인 코루틴을 부모-자식 관계로 구조화해 코루틴이 보다 안전하게 관리되고 제어될 수 있도록 한다.

코루틴을 부모-자식 관계로 구조화 하기 위해서는 코루틴 빌더 함수의 람다식 안에서 새로운 코루틴 빌더 함수를 호출하기만 하면 된다.

1. 실행 환경 상속

부모 코루틴의 실행 환경 상속

코루틴에다가 부모-자식 관계를 적용할 수 있다.
부모 코루틴은 자식 코루틴에게 실행 환경을 상속한다.
만약 부모 코루틴이 자식 코루틴을 생성하면, 부모 코루틴의 CoroutineContext가 자식 코루틴에게 전달된다.

정말 그런지 확인해보자.
부모 코루틴을 만들 때 Coroutine Dispatcher를 MyThread, CoroutineName을 CoroutineA로 지정했다.
자식 코루틴의 name을 확인해보니 정말 동일하다!

실행 환경 덮어씌우기

만약 자식 코루틴을 생성하는 코루틴 빌더 함수에 새로운 CoroutineContext 객체가 전달되면,
부모 코루틴에서 전달 받은 CoroutineContext의 구성 요소들은 자식 코루틴 빌더 함수로 전달된 CoroutineContext 객체의 구성 요소들로 덮어 씌워진다.

상속되지 않는 Job

바로 위에서 만약 자식 코루틴을 생성하는 코루틴 빌더 함수에 새로운 CoroutineContext 객체가 전달되면,부모 코루틴에서 전달 받은 CoroutineContext의 구성 요소들은 자식 코루틴 빌더 함수로 전달된 CoroutineContext 객체의 구성 요소들로 덮어 씌워진다. 라고 했다.

하지만 예외가 있다.
Job은 상속되지 않는다.

launch나 async를 포함한 모든 코루틴 빌더 함수는 호출 시마다 코루틴 추상체인 Job 객체를 새롭게 생성한다. 당연히 객체의 상태(Job)라는 것은, 때에 따라 달라지는 속성을 갖고있기 때문에 이를 상속하는 것은 무의미한 행동일 것이다.

실제 세계에서도 부모로부터 나은 자식의 이름, 주소, 재산 등은 물려받아도 부모의 상태를 자식에게 물려줄 수는 없는 것이다.

그리고 코루틴 제어에 Job 객체가 필요하기 때문에, Job 객체를 부모 코루틴으로부터 상속 받지 않는다. 즉, 코루틴 빌더를 통해 생성된 코루틴들은 서로 다른 Job을 가진다.

fun main() = runBlocking<Unit> { // 부모 코루틴 생성
    val runBlockingJob = coroutineContext[Job] // 부모 코루틴의 Job 추출
    launch { // 자식 코루틴 생성
        val launchJob = coroutineContext[Job] // 자식 코루틴의 Job 추출
        if (runBlockingJob === launchJob) {
            println("runBlocking으로 생성된 Job과 launch로 생성된 Job이 동일합니다")
        } else {
            println("runBlocking으로 생성된 Job과 launch로 생성된 Job이 다릅니다")
        }
    }
}

[]를 통해 Job을 get해서 코드를 실행해보면, 결과는 다음과 같다.

구조화에 사용되는 Job

부모 코루틴으로부터 전달 받은 Job 객체는 코루틴을 구조화 하는데 사용된다.
코루틴의 구조화는 다음을 의미한다.

  • 부모 Job이 취소되면 모든 자식들도 취소된다.
  • 자식 코루틴이 실패하면 부모와 다른 자식들도 취소됨
  • 부모는 모든 자식이 완료될 때까지 완료되지 않음

public interface Job : CoroutineContext.Element {
    public val parent: Job?
    public val children: Sequence<Job>
    // ...
}

  • parent는 Job? 타입의 속성(property)이다.

    • null이 될 수 있는 단일 Job 객체를 참조
    • 최상위 코루틴의 경우 null

  • children은 Sequence 타입의 속성이다.

    • Job 객체들의 시퀀스를 반환
    • 현재 Job의 모든 활성 자식 코루틴들을 포함
  1. parent는 단일 객체 참조다 (한 코루틴은 하나의 부모만 가질 수 있음)
  2. children은 여러 Job을 포함할 수 있는 Sequence다 (한 코루틴은 여러 자식을 가질 수 있음)
  3. 둘 다 속성(property)이며, Job 인터페이스에 정의되어 있다.

정리하면,
Job은 두 개의 속성, parent와 children이 있다.
부모 Job의 children == 자식 Job의 Parent이며,
최상위 Parent는 parent가 없고, 최하위 Child는 children이 없다.

cf. Sequence란?

Kotlin의 Sequence는 지연(lazy) 계산을 지원하는 컬렉션의 일종으로, 데이터를 하나씩 처리할 수 있도록 도와준다. Sequence는 일반적인 컬렉션과 달리 모든 요소를 한 번에 메모리에 로드하지 않고, 필요할 때마다 요소를 생성한다. 이로 인해 대량의 데이터 처리 시 메모리 사용량을 줄이고 성능을 향상시킬 수 있다.

주요 특징

  1. 지연 계산: Sequence의 요소는 실제로 필요할 때까지 계산되지 않는다. 예를 들어, 연산을 여러 개 연속으로 사용할 경우, 모든 요소를 미리 계산하지 않고 마지막에 한 번에 계산한다.

  2. 무한 시퀀스 지원: Sequence는 무한한 요소를 생성할 수 있다. 예를 들어, 무한한 피보나치 수열을 생성하는 시퀀스를 만들 수 있다.

요약

Sequence는 메모리 효율성을 높이고 성능을 개선하기 위해 지연 계산을 활용하는 컬렉션이다.
대량의 데이터를 처리하거나 무한 시퀀스를 생성할 때 유용하게 사용할 수 있다.

2. 코루틴의 구조화와 작업 제어

코루틴의 구조화

코루틴의 구조화는 하나의 큰 비동기 작업을 작은 비동기 작업으로 나눌 때 발생한다.
예를 들어 세 개의 서버로부터 데이터를 다운로드 하고, 그 후에 합쳐진 데이터를 변환하는 비동기 작업이 있다고 하면 다음과 같이 표현될 수 있다.

구조화된 코루틴의 특성 - 1. 취소의 전파

코루틴으로 취소가 요청되면, 자식 코루틴에 취소가 전파된다.

구조화된 코루틴의 특성 - 2. 부모 코루틴의 자식 코루틴에 대한 완료 의존성

부모 코루틴은 모든 자식 코루틴이 실행 완료되어야만 완료될 수 있다.
코틀린의 구조화는 큰 작업을 연관된 여러 작업으로 나누는 방식으로 일어나기 때문에, 당연히 작은 작업이 완료돼야 큰 작업이 완료될 수 있기 때문이다.

만약 여러 여버 서버로부터 3GB짜리 데이터를 내려 받는 작업을 3개의 자식 코루틴으로 1GB씩 내려받는다고 가정하자. 자식 코루틴들이 작업이 완료되지 않으면 당연히 부모 코루틴이 완료될 수가 없다.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    val startTime = System.currentTimeMillis()
    val parentJob = launch { // 부모 코루틴 실행
        launch { // 자식 코루틴 실행
            delay(1000L) // 1초간 대기
            println("[${getElapsedTime(startTime)}] 자식 코루틴 실행 완료")
        }
        println("[${getElapsedTime(startTime)}] 부모 코루틴이 실행하는 마지막 코드")
    }
    parentJob.invokeOnCompletion { // 부모 코루틴이 완료될 시 호출되는 콜백 등록
        println("[${getElapsedTime(startTime)}] 부모 코루틴 실행 완료")
    }
}

private fun getElapsedTime(startTime: Long): String = "지난 시간: ${System.currentTimeMillis() - startTime}ms"

코드의 실행 순서는 다음과 같다.
헷갈리는 개념이므로 자주 보자!!

  1. 부모 코루틴이 시작되고
  2. 부모 코루틴 내에서 자식 코루틴을 launch
  3. 자식 코루틴은 비동기적으로 실행되기 시작
  4. 부모 코루틴은 자식 코루틴 실행을 기다리지 않고 다음 라인("마지막 코드")을 즉시 실행
  5. 자식 코루틴은 1초간 delay를 수행
  6. 자식 코루틴이 완료되면, 부모 코루틴도 완료 (모든 자식의 완료를 기다림)
  7. 마지막으로 completion callback이 실행

그런데 여기에서 의문점이 하나 생긴다.
부모 코루틴이 자식 코루틴의 완료를 기다리는 상태를 무엇이라고 할까?

‘실행 완료 중(Completing)’ 상태란 부모 코루틴의 모든 코드가 실행됐지만, 자식 코루틴이 실행 중인 경우 부모 코루틴이 갖는 상태이다.

부모 코루틴은 더이상 실행할 코드가 없더라도 자식 코루틴들이 모두 완료될 때까지 기다리는데 이때, ‘실행 완료 중‘ 상태에 머무른다.

‘실행 완료 중’ 상태의 부모 코루틴은 자식 코루틴이 모두 완료되면 자동으로 ‘실행 완료‘ 상태로 바뀐다.

코루틴 상태표 총정리

3. CoroutineScope 사용해 코루틴 관리하기

CoroutineScope이란?

CoroutineScope와 CoroutineContext는 Kotlin 코루틴에서 서로 밀접한 관련이 있지만 서로 다른 역할을 한다.

1. CoroutineScope

  • 정의: CoroutineScope는 코루틴을 시작하고 관리하는 범위를 정의한다. 각 CoroutineScope는 특정 CoroutineContext를 가지고 있으며, 그 범위 내에서 생성된 모든 코루틴은 해당 스코프의 생명 주기를 따른다.
    참고로 CoroutineScope는 CoroutineContext의 상위 인터페이스다.

  • 주요 역할:

    • 코루틴을 시작할 수 있는 방법을 제공한다. 예를 들어, launch, async, coroutineScope 등의 함수를 통해 코루틴을 생성한다.
    • 범위가 종료되면 해당 범위 내에서 생성된 모든 코루틴이 자동으로 취소된다.

  • 예시:

    val scope = CoroutineScope(Dispatchers.Main)

scope.launch {
    // 여기서 코루틴 작업을 수행
}

2. CoroutineContext

  • 정의: CoroutineContext는 코루틴의 실행 환경을 정의하는 여러 요소(컨텍스트 요소)들의 집합이다. 이 요소들은 코루틴의 디스패처, Job, 및 다른 설정을 포함한다.

  • 주요 역할:

    • 코루틴이 실행될 스레드나 디스패처를 정의한다. (예: Dispatchers.Main, Dispatchers.IO 등)
    • 코루틴의 상태를 관리하는 Job 객체를 포함하여, 생명 주기를 관리한다.

  • 예시:

    val context: CoroutineContext = Dispatchers.IO + Job()

요약

  • CoroutineScope: 코루틴을 생성하고 관리하는 범위를 정의하며, 그 범위 내의 코루틴들은 생명 주기를 관리받는다.
  • CoroutineContext: 코루틴의 실행 환경을 정의하는 여러 요소들의 집합으로, 스레드, Job, 기타 설정이 포함된다.

즉, CoroutineScope는 코루틴을 실행하는 범위이고, CoroutineContext는 그 범위 내에서 코루틴이 실행되는 환경이다.

CoroutineScope 생성하기

CoroutineScope 인터페이스를 구현한 구체적인 클래스(Concrete Class)를 사용하면 CoroutineScope 객체를 생성할 수 있다.

예를 들어 newSingleThreadContext(“CustomScopeThread”)와 Job 객체를 가진 CoroutineScope은
다음과 같이 만들 수 있다.

import kotlinx.coroutines.*
import kotlin.coroutines.CoroutineContext

class CustomCoroutineScope : CoroutineScope {
  override val coroutineContext: CoroutineContext = Job() +
      newSingleThreadContext("CustomScopeThread")
}

fun main() {
  val coroutineScope = CustomCoroutineScope() // CustomCoroutineScope 인스턴스화
  coroutineScope.launch {
    delay(100L) // 100밀리초 대기
    println("[${Thread.currentThread().name}] 코루틴 실행 완료")
  }
  Thread.sleep(1000L) // 코드 종료 방지
}

CoroutineScope 생성하기 - 2

그러나 위 방법은 약간 복잡하다.
코틀린은 CoroutineScope 객체를 한번에 생성할 수 있는 함수를 제공한다.

이름도 CoroutineScope()다.
CoroutineScope 생성 함수를 사용해 CoroutineScope 객체를 생성할 수 있다.

fun main() {
    val coroutineScope = CoroutineScope(Dispatchers.IO)
    coroutineScope.launch {
        delay(100L)
        println("[${Thread.currentThread().name}] 코루틴 실행 완료")
    }
    Thread.sleep(1000L)
}

  1. 코루틴 스코프 생성: val coroutineScope = CoroutineScope(Dispatchers.IO)에서 새로운 CoroutineScope가 생성되고, 이 스코프는 I/O 디스패처를 사용한다.

  2. 코루틴 실행: coroutineScope.launch { ... }에서 새로운 코루틴이 생성되어 비동기적으로 실행된다. 이 코루틴은 100ms 동안 대기한 후, 현재 스레드의 이름과 함께 "코루틴 실행 완료"라는 메시지를 출력한다.

  3. 메인 스레드 대기: Thread.sleep(1000L)로 메인 스레드가 1초 동안 대기합니다. 이 시간 동안 코루틴이 실행될 수 있도록 한다.

CoroutineScope 객체가 코루틴에게 실행 환경을 제공하는 방식

launch, async 함수는 다음 과정을 통해 CoroutineScope 객체로부터 실행 환경을 제공 받아 실행 환경을 설정한다.

  1. 수신 객체인 CoroutineScope으로부터 CoroutineContext 객체를 제공 받는다.
  2. 제공 받은 CoroutineContext 객체에 launch 함수의 context 인자로 넘어온 CoroutineContext를 더한다.
  3. 생성된 CoroutineContext에 새로운 Job 객체를 생성해 더한다.
    -> 이때 CoroutineContext를 통해 전달되는 Job 객체는 새로 생성되는 Job 객체의 부모 Job이 된다.

말이 어려우니 차근차근 알아보자..

1. 수신 객체인 CoroutineScope으로부터 CoroutineContext 객체를 제공 받는다.

CoroutineScope으로부터 CoroutineContext 제공받기

val newScope = CoroutineScope(CoroutineName("MyCoroutine") + Dispatchers.IO)
  • newScope의 CoroutineContext는 다음 요소들을 포함:
    • CoroutineName("MyCoroutine")
    • Dispatchers.IO
    • Job() (CoroutineScope 생성 시 자동으로 추가됨)

2. 제공 받은 CoroutineContext 객체에 launch 함수의 context 인자로 넘어온 CoroutineContext를 더한다.

2단계: launch의 context 파라미터가 기존 Context와 병합

newScope.launch(context = CoroutineName("LaunchCoroutine"))
  • 기존 Context + 새로운 Context 병합 시:
    • CoroutineName: "LaunchCoroutine"으로 덮어씌워짐 (같은 키값은 새것이 우선)
    • Dispatchers.IO: 유지 (겹치지 않는 키값은 보존)
    • Job: 유지 (아직 새 Job 생성 전)

3. 생성된 CoroutineContext에 새로운 Job 객체를 생성해 더한다. 이때 CoroutineContext를 통해 전달되는 Job 객체는 새로 생성되는 Job 객체의 부모 Job이 된다.

3단계: 새로운 Job 생성 및 부모-자식 관계 설정

// launch 내부에서 자동으로 수행
val newJob = Job(parent = coroutineContext[Job])
val newContext = coroutineContext + newJob

결과물

이 계층 구조로 인해:

  • 부모 Job이 취소되면 자식 Job도 취소됨
  • 자식 Job이 실패하면 부모에게 전파됨
  • 부모 Job은 모든 자식 Job이 완료될 때까지 완료되지 않음

수신객체(Receiver Object)

수신 객체는 Kotlin의 람다 또는 확장 함수에서 사용되는 개념이다.
간단히 말해, 함수나 람다식이 어떤 객체에 속한 것처럼 해당 객체의 속성과 메서드에 직접 접근할 수 있도록 해주는 메커니즘이다. 수신 객체는 그 함수나 람다 내부에서 this로 참조할 수 있다.

예시: 확장 함수에서의 수신 객체

fun String.customPrint() {
    println(this) // 여기서 'this'는 String 객체 (수신 객체)
}

"Hello".customPrint() // "Hello"가 수신 객체로 동작함
// 출력: Hello

이 경우 "Hello"가 수신 객체다. 함수 내부에서 this를 통해 그 객체에 접근할 수 있다.

실행 환경 상속의 원리 (중요)

코루틴의 CoroutineContext는 코루틴 빌더 람다식의 수신객체로 제공된다.

  1. CoroutineContext: 코루틴이 실행되는 환경을 정의하는 객체다. 이 환경에는 코루틴을 어떤 스레드에서 실행할지(예: Dispatchers.IO), 이름(예: CoroutineName), 코루틴의 상태를 관리하는 Job 등이 포함된다.
  2. 수신 객체: 함수나 람다식 내부에서 그 함수가 특정 객체에 속한 것처럼 해당 객체의 속성이나 메서드를 호출할 수 있는 기능이다. 이때 this가 그 수신 객체를 가리킨다.

이 코드에서의 역할 설명

fun main() = runBlocking<Unit> {
  this.launch { // 'this'는 'runBlocking'에서 제공하는 CoroutineScope
    this.async { // 'this'는 현재 코루틴이 속한 CoroutineScope
      // async 내부에서 비동기 작업 수행
    }
  }
}

  • runBlocking<Unit> {}: 이 함수는 코루틴을 블로킹 방식으로 실행하는 스코프를 만들어준다.
    이때, this는 runBlocking 스코프의 수신 객체다. 즉, runBlocking이 실행되고 있는 CoroutineScope를 가리킨다.

  • this.launch {}: launch는 코루틴을 실행하는 빌더다. 이때 this는 CoroutineScope로서 runBlocking이 제공하는 스코프를 사용하게 된다.

  • this.async {}: launch 내부에서 다시 async 코루틴을 실행한다. 이때도 this는 여전히 CoroutineScope다. 즉, 내부적으로 코루틴이 실행되는 스코프는 같은 컨텍스트를 공유한다.

"코루틴의 CoroutineContext는 코루틴 빌더 람다식의 수신 객체로 제공된다"의 의미

이 문장은 코루틴을 만들 때 사용하는 함수(launch, async 등)의 람다 블록에서 this가 코루틴의 실행 환경(CoroutineContext)을 포함한 CoroutineScope를 가리킨다는 뜻이다. 그래서 this를 사용하면 해당 코루틴의 실행 환경에 바로 접근할 수 있다.

하지만 this는 생략 가능하다는 말은, this를 명시적으로 적지 않아도 기본적으로 해당 코루틴 스코프의 컨텍스트를 사용한다는 것을 의미한다. 즉, this 없이도 다음과 같은 방식으로 코드를 작성할 수 있다.

fun main() = runBlocking<Unit> {
  launch {
    async {
      // 비동기 작업
    }
  }
}

요약

  • this는 코루틴 스코프에서 현재 실행 중인 코루틴의 CoroutineScope를 가리킨다.
  • 수신 객체로 제공되는 CoroutineScope는 코루틴 컨텍스트(실행 환경) 정보를 담고 있으며, 이를 통해 현재 코루틴의 실행 환경을 제어할 수 있다.
  • this는 생략 가능하며, 생략해도 동일한 코루틴 스코프가 적용된다.

CoroutineScope에 속한 코루틴의 범위

CoroutineScope 객체는 특정 범위의 코루틴을 제어하는 역할을 한다고 했었다.
범위는 CoroutineScope의 Job에 해당하는 코루틴과 자식, 자손 되는 모든 코루틴이다.


runBlocking 람다식의 CoroutineScope 범위는 위 그림과 같다.
그런데 특정 코루틴을 runBlocking 범위에서 뺄수는 없을까?
가능하다.

특정 코루틴을 CoroutineScope 범위에서 벗어나게 하기

CoroutineScope 범위에서 뺄 코루틴을 CoroutineScope으로 재작성하면 새로운 scope가 생성된다.
Coroutine4는 root job을 부모로 갖게 된다.
하지만 이러한 행위는 코루틴 전체의 구조를 불안정하게 만들어서 지양하는 것이 좋다.

4. CoroutineScope 취소하기 (중요)

CoroutineScope 인터페이스는 확장함수로 cancel 함수를 가지고, 이 cancel 함수는 CoroutineScope 범위에 속한 모든 코루틴을 취소한다.

이 때, 취소의 전파 특성 사용하여, 취소를 위해 CoroutineScope의 coroutineContext의 Job을 취소하는 방법을 사용한다.


Job runBlocking은 Coroutine1,2를 자식으로 둔다.
Coroutine1은 3,4를 자식으로 두고 Coroutine2는 자식이 없다.
따라서 Coroutine1의 CoroutineScope에다가 cancel 요청하면, 오직 Coroutine3,4의 Job이 취소된다.
당연히 Coroutine2의 Job은 아무런 변화가 없다.

CoroutineScope 활성화 상태 확인하기

CoroutineScope 객체는 현재 CoroutineScope이 실행 중, 즉 Job이 Active인지 확인하는 isActive 프로퍼티를 제공한다.
isActive 프로퍼티는 coroutineContext로부터 Job 객체를 가져와 isActive를 확인한다.

isActive 프로퍼티는 코루틴이 아직 활성화되어 있는지 확인할 수 있게 해주는 중요한 도구입니다. 코루틴이 취소되었거나 완료되었을 때, 코루틴을 중단하거나 취소된 상태에서 더 이상 불필요한 작업을 수행하지 않도록 도와줍니다.

예시 코드

import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        for (i in 1..5) {
            if (!isActive) { // 현재 코루틴이 활성화되어 있는지 확인
                println("코루틴이 취소됨. 작업을 중단합니다.")
                return@launch
            }
            println("작업 중... $i")
            delay(500L) // 0.5초 대기
        }
        println("작업 완료")
    }

    delay(1000L) // 1초 대기 후
    println("코루틴 취소")
    job.cancel() // 코루틴을 취소
}

실행 결과

작업 중... 1
작업 중... 2
코루틴 취소
코루틴이 취소됨. 작업을 중단합니다.

언제 쓰면 좋은가?

  1. 취소 가능 작업: 긴 작업을 수행하는 중간에 코루틴이 취소될 수 있는 경우, isActive를 사용하여 더 이상 불필요한 연산을 중단할 수 있다.
    • 예: 파일 다운로드, 데이터베이스 연산, 복잡한 계산 등
  2. 주기적인 체크: 반복문이나 긴 작업 중, 코루틴의 상태를 주기적으로 확인하면서 취소 여부를 감지하여 빠르게 작업을 종료할 때 유용하다.
  3. 자원 낭비 방지: 취소된 코루틴이 계속해서 작업을 진행하지 않도록 방지하여, 불필요한 자원 소비를 막는다.

isActive는 코루틴의 취소를 부드럽게 처리하고 불필요한 작업을 방지하는 중요한 역할을 수행한다.

강의자료 출처 및 참고한 강의:

코틀린 코루틴 완전 정복 by 조세영님
https://www.inflearn.com/course/%EC%BD%94%ED%8B%80%EB%A6%B0-%EC%BD%94%EB%A3%A8%ED%8B%B4-%EC%99%84%EC%A0%84-%EC%A0%95%EB%B3%B5/dashboard

profile
이윤설
화려한 외면이 아닌 단단한 내면
이전 포스트

CoroutineContext

다음 포스트

구조화된 동시성 - 2

0개의 댓글