Kotlin Coroutine - Suspension은 어떻게 동작하는가?

출처: https://kt.academy/article/cc-suspension

How does suspension work in Kotlin coroutines?

 

Suspending function은 Kotlin Coroutine의 hallmark(특징)이다. suspension 기능은 다른 모든 Kotlin Coroutine 개념이 구축되는 가장 기본적인 기능이다. 그렇기 때문에 이 장의 목표는 작동 방식에 대한 확실한 이해를 구축하는 것이다.

Coroutine을 Suspending 하는 것은 도중에 멈춘다는 의미이다. 이것은 비디오 게임을 멈추는 것과 유사하다. 체크포인트에 저장을 하고, 게임을 끈 뒤 컴퓨터와 우리는 다른 일을 할 수 있다. 그러다가 이후 다시 하고 싶을 때게임에 다시 들어가면 저장된 체크포인트 에서 시작하게 된다. 이것은 Coroutine과 유사하다. Coroutine이 suspend 되면 그들은 한 continuation을 리턴한다. 이것은 게임을 저장하는 것과 같다. 우리는 우리가 멈췄던 지점에서 다시 재개하고 싶을 때 continuation을 사용할 수 있다.

저장할 수 없고 오직 block만 시키는 thread와는 아주 다르다는 것에 주목해야 한다. Coroutine은 thread보다 훨씬 더 강력하다. suspend됐을 때, 그것은 어떠한 자원도 소모하지 않는다. Coroutine은 다른 thread에서 재개 될 수 있고(이론상), continuation은 serialized, deserialized가 될 수 있다.

Resume

그럼 실제로 실행해 보자. 이를 위해서는 Coroutine이 필요하다. 우리는 나중에 소개할 Coroutine 빌더(runBlocking 또는 launch와 같은)를 사용하여 Coroutine을 시작한다. 더 간단한 방법도 있지만 suspending main 함수를 사용할 수 있다.

suspend 함수는 coroutine을 중단시킬 수 있는 함수이다. 이것은 Coroutine(또는 다른 suspend 함수)에서 호출되어야 함을 의미한다. 따라서 suspend 시킬 무언가를 가져야 한다. main 함수는 시작점이므로 Kotlin은 실행할 때 Coroutine에서 시작한다.

suspend fun main() {
   println("Before")

   println("After")
}
// Before
// After

“Before”와 “After”를 출력하는 간단한 프로그램이다. 만약 이러한 두 출력 사이에서 suspend를 시킨다면 어떤 일이 일어날까? 이것을 위해 Kotlin 표준 라이브러리에 있는 suspendCoroutine 함수를 사용 할 수 있다.

suspend fun main() {
   println("Before")

   suspendCoroutine<Unit> { }

   println("After")
}
// Before

위의 코드를 실행하면 “After”가 출력되는 것을 볼 수 없으며 main 함수의 실행은 영원히 멈추지 않는다. coroutine은 “Before”이후에 중단되었다. 우리의 게임은 멈췄고 절대로 재개되지 않는 상황이다. 그렇다면, 어떻게 재개할 수 있을까? 앞서 언급한 Continuation은 어디에 있는가?

suspendCoroutine의 호출을 살펴보면 이것의 끝에는 람다식이 있는 것을 볼 수 있다. 인자가 전달되는 이 함수는 중단(suspension)되기 전에 호출 된다. 이 함수는 인자로 continuation을 받는다.

suspend fun main() {
   println("Before")

   suspendCoroutine<Unit> { continuation ->
       println("Before too")
   }

   println("After")
}
// Before
// Before too

이러한 방식의 함수 호출은 새로운 개념이 아니며, let,apply,useLines 등과 유사하다. suspendCoroutine 함수도 같은 방식으로 설계 되어 suspension 직전에 continuation을 사용 할 수 있다. suspendCoroutine 호출 이후에는 너무 늦게 될 것이다. 이러한 람다는 continuation을 어딘가에 저장하거나 해당 Coroutine의 재개 여부를 결정하는데 사용할 수 있다.

Coroutine을 즉시 재개하도록 해보자.

suspend fun main() {
   println("Before")

   suspendCoroutine<Unit> { continuation ->
       continuation.resume(Unit)
   }

   println("After")
}
// Before
// After

suspendCoroutine을 재게 하였기 때문에 “After”가 출력 된 것을 볼 수 있다. 여기서 약간 의문이 들 수 있는데, 람다식은 실제로 중단 이전에 호출되는데 저기서 resume을 호출하면 효과가 없는 것이 아닌가 라는 생각이 들 수 있다. 이 경우 해당 람다에서 재개가 진행될 경우 애초에 중단을 시키지 않는 최적화가 되어 있다고 한다.

Kotlin 1.3부터 Continuation의 정의가 변경되었기 때문에 resume과 resumeWithException 대신 resumeWith를 사용할 수 있다고 한다. 실제로 이후 버전에서 resume과 resumeWithException은 resumeWith를 내부에서 호출하는 형태이다.

inline fun <T> Continuation<T>.resume(value: T): Unit =
   resumeWith(Result.success(value))

inline fun <T> Continuation<T>.resumeWithException(
   exception: Throwable
): Unit = resumeWith(Result.failure(exception))

이번에는 Coroutine 내에서 한 thread를 실행시키고 잠깐 동안 sleep을 한 다음 Coroutine을 재게해보자.

suspend fun main() {
   println("Before")

   suspendCoroutine<Unit> { continuation ->
       thread {
           println("Suspended")
           Thread.sleep(1000)
           continuation.resume(Unit)
           println("Resumed")
       }
   }

   println("After")
}
// Before
// Suspended
// (1 second delay)
// After
// Resumed

이것은 중요한 관찰이다. 우리의 continuation을 정의된 시간 이후 재게 시키는 함수를 만들 수 있다.

fun continueAfterSecond(continuation: Continuation<Unit>) {
   thread {
       Thread.sleep(1000)
       continuation.resume(Unit)
   }
}

suspend fun main() {
   println("Before")

   suspendCoroutine<Unit> { continuation ->
       continueAfterSecond(continuation)
   }

   println("After")
}
// Before
// (1 sec)
// After

이러한 메커니즘은 잘 동작하지만 단 1초의 딜레이 후에만 스레드를 종료하기 위해 불필요하게 스레드를 생성한다. thread를 생성하는 비용은 비싸기 때문에 이것은 낭비일 것이다. 더 나은 방법은 알람 시계를 설정하는 것이다. JVM에서 우리는 이것을 위해 ScheduledExecutorService를 사용할 수 있다. 우리는 이것을 이용하여 정해진 시간 이후에 continuation.resume(unit)을 호출 하도록 설정할 수 있다.

private val executor =
   Executors.newSingleThreadScheduledExecutor {
       Thread(it, "scheduler").apply { isDaemon = true }
   }

suspend fun main() {
   println("Before")

   suspendCoroutine<Unit> { continuation ->
       executor.schedule({
           continuation.resume(Unit)
       }, 1000, TimeUnit.MILLISECONDS)
   }

   println("After")
}
// Before
// (1 second delay)
// After

일정시간 동안 중단을 시키는 것은 유용한 기능이다. 함수로 추출해보자. 우리는 이것을 delay라고 이름 지을 것이다.

private val executor =
   Executors.newSingleThreadScheduledExecutor {
       Thread(it, "scheduler").apply { isDaemon = true }
   }

suspend fun delay(timeMillis: Long): Unit =
   suspendCoroutine { cont ->
       executor.schedule({
           cont.resume(Unit)
       }, timeMillis, TimeUnit.MILLISECONDS)
   }

suspend fun main() {
   println("Before")

   delay(1000)

   println("After")
}
// Before
// (1 second delay)
// After

executor는 여전히 한 thread를 사용하지만 이것은 delay 함수를 사용하는 모든 코루틴에 대한 하나의 thread이다. 이것은 매번 우리가 기다리기를 원할 때 thread를 blocking 시키는 것 보다 훨씬 낫다.

이것이 바로 Kotlin Coroutines 라이브러리의 delay가 실제로 구현된 방식이다. 현재 구현은 주로 테스트의 용도로 더 복잡하지만 핵심 아이디어는 동일하다.

Resuming with a value

지금까지 resume을 호출할 때 Unit을 전달하는 것에 대해 궁금할수 있다. 또한 suspendCoroutine의 type 인자 또한 Unit인 이유가 궁금할 수 있다. 이 둘의 타입이 같은 것은 우연이 아니다. Unit은 함수의 반환형이자 Continuation 인자의 제네릭 타입이다.

val ret: Unit =
   suspendCoroutine<Unit> { cont: Continuation<Unit> ->
       cont.resume(Unit)
   }

suspendCoroutine을 호출할 때 continuation으로 반환될 타입을 지정할 수 있다. resume을 호출할 때 동일한 타입을 사용해야 한다.

suspend fun main() {
   val i: Int = suspendCoroutine<Int> { cont ->
       cont.resume(42)
   }
   println(i) // 42

   val str: String = suspendCoroutine<String> { cont ->
       cont.resume("Some text")
   }
   println(str) // Some text

   val b: Boolean = suspendCoroutine<Boolean> { cont ->
       cont.resume(true)
   }
   println(b) // true
}

이러한 부분은 게임과 비교할 때 잘 맞지는 않는다. 게임을 저장할 때 무언가 넣는 경우는 없기 때문이다(치팅을 쓰지 않는한). 그러나 Coroutine에서는 의미가 있다. 때때로 우리는 특정 데이터를 기다리는 목적(API 호출로부터 네트워크 응답)으로 중단을 시킨다. thread는 데이터가 필요한 지점에 도착할때 까지 business logic을 실행한다. 따라서 netwokr library에게 데이터를 전달해줄 것을 요청한다. Coroutine이 없다면 이 thread는 앉아서 기다려야 한다. thread는 비싸기 때문에 이것은 거대한 낭비이며 특히 안드로이드의 Main Thread와 같은 중요한 thread라면 더욱 문제가 될 수 있다. Coroutine이 있다면 그것은 단지 coroutine을 중단 시키고 library에게 데이터를 얻었으면 resume 함수를 호출하여 데이터 전달과 coroutine의 재개를 요청할 것이다. 따라서 thread는 그동안 다른 일을 할 수 있다. 데이터가 도착하면 coroutine이 중단되었던 지점부터 다시 재게하는데 thread가 사용 될 것이다.

이것이 실제로 작동하는지 보기 위해 일부 데이터를 수신할 때까지 일시 중지하는 방법을 살펴보겠다. 아래 예제에서는 외부에서 구현된 콜백 함수 requestUser를 사용한다.

suspend fun main() {
   println("Before")
   val user = suspendCoroutine<User> { cont ->
       requestUser { user ->
           cont.resume(user)
       }
   }
   println(user)
   println("After")
}
// Before
// (1 second delay)
// User(name=Test)
// After

suspendCoroutine을 직접 호출하는 것은 불편하므로 대신 suspend function을 사용하는 것이 좋다. 이를 추출해보자.

suspend fun requestUser(): User {
   return suspendCoroutine<User> { cont ->
       requestUser { user ->
           cont.resume(user)
       }
   }
}

suspend fun main() {
   println("Before")
   val user = requestUser()
   println(user)
   println("After")
}

현재, suspending function은 많은 유명한 라이브러리(Retrofit, Room…)에서 이미 지원하고 있다. 이것이 suspending function에서 콜백 함수를 거의 사용할 필요가 없는 이유이다. 그러나 만약, 저러한 것이 필요하다면 cancellation chapter에서 설명할 suspendCancellableCoroutine(suspendCoroutine 대신)을 이용하기를 추천한다.

API가 데이터가 아니라 어떤 종류의 문제를 발생시키면 어떻게 되는지 궁금할 것다. 서비스가 죽거나 오류로 응답하면 어떻게 될까? 저러한 경우 데이터를 리턴할 수 없다. 대신 coroutine이 중단된 위치로부터 exception을 던져야 할 것이다.

Resume with an exception

우리가 호출하는 모든 함수는 특정 값을 리턴하거나 exception을 던질 것이다. suspendCoroutine 또한 동일하다. resume이 호출 될때, 인자로 전달된 데이터를 리턴한다. resumeWithException이 호출되면 인자로 전달된 exception은 중단지점으로부터 던져지게 된다.

class MyException : Throwable("Just an exception")

suspend fun main() {
   try {
       suspendCoroutine<Unit> { cont ->
           cont.resumeWithException(MyException())
       }
   } catch (e: MyException) {
       println("Caught!")
   }
}
// Caught!

이러한 메커니즘은 다양한 종류의 문제에 사용되며, 예를 들면 단일 network exception이 있을 수 있다.

suspend fun requestUser(): User {
   return suspendCancellableCoroutine<User> { cont ->
       requestUser { resp ->
           if (resp.isSuccessful) {
               cont.resume(resp.data)
           } else {
               val e = ApiException(
                   resp.code,
                   resp.message
               )
               cont.resumeWithException(e)
           }
       }
   }
}

suspend fun requestNews(): News {
   return suspendCancellableCoroutine<News> { cont ->
       requestNews(
           onSuccess = { news -> cont.resume(news) },
           onError = { e -> cont.resumeWithException(e) }
       )
   }
}

Suspending a coroutine, not a function

강조되어야 하는 것 중 한가지는, coroutine이 suspend되는 것이지 함수가 suspend 되는 것은 아니라는 것이다. suspending function은 coroutine이 아니며 단지 coroutine을 suspend시킬 수 있는 함수이다. 변수에 Continuation을 저장하고 함수 호출 이후에 그것을 재게하려고 시도하는 것을 상상해보자.

// Do not do this
var continuation: Continuation<Unit>? = null

suspend fun suspendAndSetContinuation() {
   suspendCoroutine<Unit> { cont ->
       continuation = cont
   }
}

suspend fun main() {
   println("Before")

   suspendAndSetContinuation()
   continuation?.resume(Unit)

   println("After")
}
// Before

이것은 말이 안된다. 이것은 게임을 멈추고 게임의 그 이후 지점에서 재게하려고 하는 것과 동일하다. resume은 절대로 호출되지 않을 것이다. 오직 “Before”만 출력되는 것을 볼 것이며 또다른 thread나 coroutine에서 재게 하지 않는 한 프로그램은 끝나지 않을 것이다. 이것을 보이기 위해, 몇초 뒤 또다른 coroutine에서 재게를 해보자.

// Do not do this, potential memory leak
var continuation: Continuation<Unit>? = null

suspend fun suspendAndSetContinuation() {
   suspendCoroutine<Unit> { cont ->
       continuation = cont
   }
}

suspend fun main() = coroutineScope {
   println("Before")

   launch {
       delay(1000)
       continuation?.resume(Unit)
   }

   suspendAndSetContinuation()
   println("After")
}
// Before
// (1 second delay)
// After

 

Coroutine의 Suspend 동작에 대한 좋은 글이 있는 것 같아 번역을 해봤습니다. 핵심은 Continuation을 통한 stop/resume 동작을 통해 thread를 blocking 시키지 않을 수 있다는 것입니다. 실제로 함수에 suspend 키워드를 붙이게 되면 java 코드로 디컴파일시 continuation 객체가 인자로 들어오는 것을 확인할 수 있습니다.