Kotlin Coroutines

🔖 Kotlin

Kotlin Coroutines

beomsic 2024. 12. 28. 00:44

🤔 Kotlin Coroutines

Coroutine 이란?

co(함께) + routine (특정 일을 수행하는 명령의 모음)
코틀린에서 제공하는 기능으로 비동기 작업을 효율적으로 처리하기 위해 설계된 경량화된 동시성 처리 방식

📍 특징

suspend(중단)와 resume(재개)가 가능
비 선점형 멀티태스킹 (Non-preemptive multitasking)
협력형 멀티태스킹 (Cooperative multitasking)

❗ 왜 코루틴을 사용해야 할까요?

📖 비동기 처리

네트워크 요청, DB 작업, 파일 I/O 처럼 오랜 시간이 걸리는 작업을 비동기로 처리하는 것이 중요

코루틴은 Kotlin 에서 동시성을 간단하게 처리할 수 있는 기능 중 하나.

비동기 코드를 동기적으로 작성하는 것처럼 보이게해 코드의 가독성을 높이고 복잡한 콜백이나 다른 동시성 패턴에 비해 쉽게 작성할 수 있게 한다.
코루틴은 중단할 수 있으며, 중단될 때 스레드를 블로킹하지 않고 다른 작업에 양보한다. 따라서 스레드를 다른 작업을 수행시키는데 사용할 수 있다.

🆚 Subroutine

Subroutine	Coroutine
Routine : 특정한 일을 처리하기 위한 명령	일시 중단과 재개가 가능한 작업 단위 - 특정 지점에서 작업을 멈추고, 이후 필요할 때 재개
Subroutine : Routine의 하위에서 실행되는 또 다른 루틴	중단된 곳에서 다시 시작가능 - 여러개의 진입점을 가진다
서브 루틴은 하나의 진입점을 가지며, 한 번 호출되면 작업을 완료할 때까지 멈추지 않고 쭉 실행되어 스레드가 다른 작업을 할 수 없다	작업을 하지 않을 때는 스레드 사용 권한을 양보하며 서로 협력적으로 실행

🏃 동작 방식

Q. 코루틴은 어떻게 중단과 재개를 할까?

⇒ CPS (continuation Passing Style)

CPS란 호출되는 함수에 Continuation을 전달하고 각 함수의 작업이 완료되는 대로 전달받은 Continuation을 호출하는 패러다임이다.

@SinceKotlin("1.3")
public interface Continuation<in T> {
    public val context: CoroutineContext
    public fun resumeWith(result: Result<T>)
}

Continuation : 각 중단 시점마다 코루틴의 현재 상태(실행 지점, 로컬 변수 등)와 다음에 무슨 일을 해야 할지를 기억하는 확장된 콜백 역할을 하는 객체

Continuation 인터페이스에는 크게 context 객체와 resumeWith 함수가 있다.

resumeWith : 특정 함수 a가 suspend 되어야 할 때, 현재 함수에서 a의 결과를 T로 받게 해주는 함수
- 실행을 재개하기 위한 메서드
context: 각 Continuation이 특정 스레드 혹은 스레드 풀에서 실행되는 것을 허용해준다.
- dispatcher, job 등 ConroutineContext를 저장하는 변수

📍 정리

CPS에서 Continuation을 전달하면서 현재 suspend된 부분에서의 resume을 가능하게 해준다.

Continuation은 resume 되었을 때의 동작 관리를 위한 객체

Q. Continuation 객체는 어떻게 생기나?

⇒ 컴파일러가 suspend function을 만나게 되면 내부적으로 Continuation을 활용하도록 변환

🏃 컴파일러의 suspend function 변환 과정

// 예제
suspend fun test() {
		println("Hello")
		delay(1000L) // 코루틴에서 제공하는 일시 중단 함수
		println("World!")
}
suspend fun delay(timeMillis: Long)

1. 컴파일러가 함수 내부의 중단 가능 지점 을 식별

위 코드에서의 중단 가능 지점
- suspend fun test()
- delay(1000L)
- suspend fun delay()

2. 유한 상태 머신(Finite State Machine, FSM) 으로 변환

suspend fun test() {
  when (label) {
    0 -> {
      println("Hello")
      delay(1000L)
    }
    1 -> {
      println("World!")
    }
  }
}

코틀린은 모든 중단 가능 지점을 찾아 when으로 표현
중단 가능 지점을 기준으로 코드 블록을 분리하고 분리된 코드들은 label로 구별
label은 0부터 증가하고 이는 재개했을 때 돌아올 위치를 나타낸다.

3. Continuation 추가

fun test(continuation: Continuation) {
  val continuation = object : ContinuationImpl { ... }
	
  when (continuation.label) {
    0 -> {
      println("Hello")
      delay(1000L)
    }
    1 -> {
      println("World!")
    }
  }
}

suspend 키워드가 사라지고 파라미터로 continuation이 추가
continuation 인터페이스의 구현체인 Continuation object가 생성

4. 실행 후 중단(Suspend)

fun test(continuation: Continuation) {
  val continuation = object : ContinuationImpl { ... }
	
  when (continuation.label) {
    0 -> {
      println("Hello")
      continuation.label = 1 // 중단 이후 라벨 1로 재개
			
      if (delay(1000, continuation) == COROUTINE_SUSPEND) {
        return COROUTINE_SUSPEND
      }
    }
    1 -> {
      println("World!")
      return Unit
    }
  }
}

label 값으로 실행 후 다음 값으로 값을 바꾼다.
suspend function은 COROUTINE_SUSPEND를 반환할 수 있다.
- 이는 현재 코루틴이 중단 상태임을 나타낸다.
- 이를 리턴해줌으로써 함수가 끝난 것 같은 효과를 주어 스레드를 블로킹하지 않을 수 있다

5. 재개 (Resume)

fun test(continuation: Continuation) {
  val continuation = object : ContinuationImpl { 
    override fun resumeWith() {
      test(continuation = this)
    }
  }
	
  when (continuation.label) {
    0 -> {
      println("Hello")
      continuation.label = 1 // 중단 이후 라벨 1로 재개
      delay(1000L)
		}
    1 -> {
      println("World!")
      return Unit
    }
  }
}

Continuation의 resumeWith() 함수를 호출해서 중단된 이후 블록부터 코루틴을 재개

🌟 성능 - Thread vs Coroutine

Coroutine은 light-weight thread 라고 불린다.

Coroutine은 스레드 안에서 실행된다.
같은 스레드에 10, 100개의 Coroutine이 있을 수 있다.
하지만, 한 시점에 여러 Coroutine이 실행된다는 것은 아니다.
Coroutine은 메인 스레드 뿐 아니라 다른 백그라운드 스레드에서도 동작할 수 있다.

❗ 동시성 프로그래밍에서 한 시점에서는 하나의 Coroutine만 실행된다.

즉, Coroutine은 스레드 안에서 실행되고 Coroutine을 활용하여 Main-safe 하게 할 수 있다.

✋ 스레드는 프로세스에 종속이 되지만, Coroutine은 특정 스레드에 종속되지 않는다는 점이 큰 특징이다.

Coroutine은 resume 될 때마다 다른 스레드에서 실행될 수 있다.

💡 작업이 스레드에 종속되지 않는다?

스레드를 blocking 하지 않으면서 작업의 실행을 잠시 중단할 수 있게 한다.

스레드 B가 스레드 A의 작업이 끝날 때 까지 대기해야 하는 작업을 한다면 이를 잠시 중단하고 그동안 다른 작업을 할 수 있게 한다.

이 특징은 스레드를 blocking 하지 않아 더 빠른 연산이 가능하게 하고 메모리 사용량을 줄여 많은 동시성 작업을 수행할 수 있게 한다.

🧑‍💻 성능 차이 코드 예제

fun main() = runBlocking {
	val amount = 10_000 // 실행할 작업 수
	println("✅ Comparing performance with $amount tasks")
	println("🛫 ${Thread.activeCount()} threads active at the start")

	// 코루틴 실행 시간 측정
	val coroutineTime = measureTimeMillis {
		createCoroutines(amount)
	}
	println("⏰ Running time for createCoroutines is $coroutineTime ms")

	// 스레드 실행 시간 측정
	val threadTime = measureTimeMillis {
		createThreads(amount)
	}
	println("⏰ Running time for createThreads is $threadTime ms")
}

suspend fun createCoroutines(amount: Int) = coroutineScope {
	val jobs = ArrayList<Job>()

	for (i in 1..amount) {
		jobs += launch(Dispatchers.Default) {
			delay(1000L)
		}
	}
	println("🛬 ${Thread.activeCount()} threads active after createCoroutines")
	jobs.forEach { it.join() }
}

fun createThreads(amount: Int) {
	val jobs = ArrayList<Thread>()

	for (i in 1..amount) {
		jobs += Thread {
			Thread.sleep(1000L)
		}.also { it.start() }
	}
	println("🛬 ${Thread.activeCount()} threads active after createThreads")
	jobs.forEach { it.join() }
}

1. amount = 100