[Deep Dives into Coroutines on JVM] (1)

이 게시글의 내용은 아래의 유튜브를 보고 작성한 내용입니다.

무언가를 잘 사용하려면 그 내부를 들여다 보는 것이 중요하다고 생각합니다. 코루틴은 안드로이드 코틀린 개발자라면 사용해야 하는 비동기 프로그래밍 을 지원하는 방식입니다. 그래서 이번 게시글에서 코루틴이 어떻게 비동기 프로그래밍을 지원하는지 알아보고자 합니다.

이제부터 간단한 함수를 가지고 놀아보겠습니다. 이 함수를 [Direct Style] 이라 하겠습니다.

fun postItem(item: Item) {
    val token = requestToken()
    val post = createPost(token, item)
    processPost(post)
}

위의 함수에선 requestToken() 함수로 token을 요청하고 이를 계속해서 사용해 다른 함수의 매개변수로 사용하고 있습니다.

이를 우린 그런 의미를 담아 이런 것들을 Continuation이라고 하겠습니다.

Suspend 함수

코루틴이 나오기 전에는 Callback을 사용해서 비동기 처리를 했습니다. Kotlin에서는 고차 함수를 사용해서 이를 구현헀죠.

Callback을 사용해 위의 함수를 아래와 같이 비동기 처리를 할 수 있습니다.

fun postItem(item: Item) {
    equestToken { token ->
        createPost(token, item) { post -> 
            processPost(post)
        }
    }
}

Callback으로 처리하는 방식도 Continuation을 가지고 있습니다. 그렇기에 Callbacks 을 좀 더 멋있게(?) 말해 보겠습니다.

이런 스타일은 [Continuation-Passing Style, CPS] 입니다.

위와 같은 예시에서는 썩 나쁘지 않아 보이지만 만약 아래와 같다면 어떨까요..? 잘 알아보실 수 있으신가요?

fun main() {
    performTask1 { result1 ->
        if (result1) {
            performTask2 { result2 ->
                if (result2) {
                    performTask3 { result3 ->
                        if (result3) {
                            // ... 계속해서 콜백 함수가 중첩됨
                        } else {
                            // 에러 처리
                        }
                    }
                } else {
                    // 에러 처리
                }
            }
        } else {
            // 에러 처리
        }
    }
}

이런 문제를 JetBrains은 suspend 함수로 해결했습니다. suspend 함수는 함수가 일시 중단될 수 있다는 것을 나타냅니다.

suspend fun postItem(item: Item) {
    val token = requestToken()
    val post = createPost(token, item)
    processPost(post)
}

흠 근데 왜 그렇게 되는지 궁금하지 않나요??

Continuation

조금 더 작게 함수를 나눠서 보겠습니다. 아래의 함수는 위에 있던 함수 호출의 본체입니다.

suspend fun createPost(token: Token, item: Item): Post { … }

이제 코틀린 컴파일러가 이 함수를 어떻게 바꾸는지 볼까요?

// Java/JVM
Object createPost(Token token, Item item, Continuation<Post> cont) { … }

달라진게 거의 없어 보이지만, 추가된 매개변수가 존재합니다. 바로 이전 함수에서 넘어온 Callback인 Continuation입니다.

Continuation은 인터페이스로 매우 단순합니다. 내부에는 Context와 실행을 재개하는 메서드가 있습니다.

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

사실 그냥 Callback입니다. 딱히 다른 것은 없습니다.

만약 멋있어 보이고 싶다면 Contnuation이라 불러보세요! ~~사람들이 이상하게 볼지도 모르지만요~~

이를 통해 알 수 있는 것은 suspend 함수도 결국 우리 눈에는 보이지 않지만 Continuation이라는 이름의 Callback을 사용하고 있다는 점입니다. 그럼 이제 컴파일러가 suspend 함수를 어떻게 CPS로 바꿔서 JVM이 알아듣게 만드는지 보겠습니다.

suspend 함수가 CPS로 바뀌는 과정

Labels

suspend 함수를 CPS로 바꾸기 위해 컴파일러가 가장 먼저 하는 것은 Label을 설정하는 것입니다.

fun postItem(item: Item) {
    // LABEL 0
 ↛    val token = requestToken()
    // LABEL 1
 ↛    val post = createPost(token, item)
    // LABEL 2
    processPost(post)
}

컴파일러는 코드를 하나의 큰 스위치에 넣고 이것들을 중단 지점과 재개 지점으로 구분하기 위해 Label을 설정합니다.

fun postItem(item: Item) {
    switch (label) {
        case 0:
            val token = requestToken()
        case 1:
            val post = createPost(token, item)
        case 2:
            processPost(post)
    }
}

다만 이대로는 문제가 있습니다. 각 함수의 Continuation을 다음으로 전달할 수가 없습니다. 이때 나타나는 것이 바로 State Machine입니다.

suspend 함수는 컴파일시 cont라는 Continuation 파라미터를 받습니다.

fun postItem(item: Item, cont: Continuation) {
    val sm = object : CoroutineImpl { … } // State Machine 추가!
    switch (sm.label) {
        case 0:
            val token = requestToken(sm)
        case 1:
            val post = createPost(token, item, sm)
        case 2:
            processPost(post)
    }
}

이전에 말했듯이 Continuation 파라미터는 Callback입니다. State Machine를 함수의 인수로 이를 전달하는 것은 "야 너 그거 끝나면 끝났다고 나중에 말해줘" 라고 말하는 것과 같습니다.

함수가 Callback을 주면 State Machine은 이를 눈치채고 postItem()을 재시작 해야합니다. 다만 아직 안한것이 있습니다. postItem()을 재시작하기 위해서는 기존의 매개변수였던 item을 다시 전달해야 하고 다음에 수행할 코드의 Label을 넘겨줘야 합니다.

fun postItem(item: Item, cont: Continuation) {
    val sm = cont as? ThisSM ?: object : CorutineImpl {
        fun resume(…) {
            postItem(null, this) // 재시작 할때 새로운 인수 전달
        }
    }
    switch (label) {
        case 0:
            sm.item = item // 기존의 매개변수 item을 상태머신에 저장
            sm.label = 1 // 다음 label 지정
            requestToken(sm)
        case 1:
            val post = createPost(token, item)
        …
    }
}

모든 것을 만족한 코드는 위와 같습니다.

다음 작업을 수행할 때에도 아까와 똑같이 해주면 됩니다. 이전 작업의 결과를 State Machine에서 가져온 다음 Label을 설정하고 인수로 넘겨주면 됩니다.

fun postItem(item: Item, cont: Continuation) {
    val sm = …
    switch (label) {
        case 0:
            sm.item = item
            sm.label = 1
            requestToken(sm)
        case 1:
            val item = sm.item // 기존의 매개변수 item을 상태머신에 저장
            val token = sm.result as Token
            sm.label = 2 // 다음 label 설정
            createPost(token, item, sm)
        …
    }
}

Retrofit2에서

기존의 Retrofit2

다른 것도 한번 볼까요? 코루틴이 나오기 전에는 서버와 통신을 하려면 Retrofit2을 사용해서 아래와 같은 코드로 서버와 통신을 했습니다.

interface Service {
    @GET("api/")
    fun createPost(
        token: Token,
        item: Item
    ): Call<Post>
}

위의 함수에서는 Call을 반환합니다. Call의 내부를 더 자세하게 보면 Call 또한 Callback을 가지고 있다는 것을 알 수 있습니다. 서버가 요청을 받고 데이터를 전송하면 이는 Callback으로 반환됩니다.

public interface Call<T> extends Cloneable {
    Response<T> execute() throws IOException;

    void enqueue(Callback<T> var1);

    boolean isExecuted();

    void cancel();

    boolean isCanceled();

    Call<T> clone();

    Request request();
}

그럼 서버로부터 데이터를 받고 이를 수신할 때는 어떻게 할까요? queue에 들어있는 서버가 보낸 응답의 처리를 우린 지금까지 아래와 같이 해왔습니다.

val retrofit = Retrofit.Builder()
    .baseUrl("https://server-url.com/")
    .addConverterFactory(GsonConverterFactory.create())
    .build()

val service: Service = retrofit.create(Service::class.java)

val call = service.createPost(Token(""), Item(""))

call.enqueue(object : Callback<T> {
    override fun onResponse(p0: Call<T>, p1: Response<T>) {
        TODO("Not yet implemented")
    }

    override fun onFailure(p0: Call<T>, p1: Throwable) {
        TODO("Not yet implemented")
    }
})

왜 이렇게 사용했었는지 이해가 되지 않나요?? 결국 Retrofit2에서 응답을 처리하는 것은 Callback이였습니다.

이제 코루틴이 있으니 이런 과정을 직접하지 않아도 됩니다. 내부적으로 다 해주니까요!!

코루틴을 사용한 Retrofit2

달라진 코드를 볼까요? interface의 메서드에 suspend를 추가해주고 ViewModel에서 아래와 같이 저장소에 호출만 해주면 됩니다.

interface Service {
    @GET("api/")
    suspend fun createPost(
        token: Token,
        item: Item
    ): Post
}

fun create(token: Token, item: Item) = viewModelScope.launch {
    repository.createPost(token, item)
}

코루틴은 아래와 같이 Callback을 대신 처리해줍니다.

suspend fun <T : Any> Call<T>.await(): T = suspendCoroutine { cont ->
    enqueue(object : Callback<T> {
        override fun onResponse(call: Call<T>, response: Response<T>) {
            if (response.isSuccessful) {
                cont.resume(response.body()!!)
            } else {
                cont.resumeWithException(ErrorResponse(response))
            }
        }

        override fun onFailure(call: Call<T>, t: Throwable) {
            cont.resumeWithException(t)
        }
    })
}

다만, 위의 코드에서 주목해야할 점이 있습니다. 서버로부터 받은 Callback은 enqueue에 등록됩니다. 우리는 서버로부터 Callback을 기다릴 때 작업을 중단하고 싶지는 않습니다. 그렇기에 Callback이 호출될 때까지 코루틴의 실행을 멈추고 다른 스레드를 블록하지 않아야 합니다.

이를 위한 것이 suspendCoroutine입니다.

SuspendCoroutine

suspendCoroutine은 말 그대로 코루틴을 중지합니다. 내부를 자세히 보겠습니다.

suspend fun <T> spendCoroutine(block: (Continuation<T>) -> Unit): T {
   ...
}

아까전에 봤던 Continuation이 여기도 있습니다. suspendCoroutine은 모든 것을 Continuation 객체로 표시하고 내부의 코드 블록에 전달하므로 동일한 기능을 수행할 수 있는 것입니다.

마지막으로

지금까지 JetBrains의 유튜브에 올라와 있는 'KotlinConf 2017 - Deep Dive into Coroutines on JVM by Roman Elizarov' 영상을 보고 이를 정리해 봤습니다. 어떻게 코루틴이 Callback을 대체하였는지 알 수 있는 좋은 영상입니다. 이 내용은 영상의 절반 부분이기에 남은 부분은 다음 게시글에서 다뤄보고자 합니다.

[Deep Dives into Coroutines on JVM] (2) - 코루틴의 Context

이 게시글의 내용은 아래의 유튜브를 보고 작성한 내용입니다. 유튜브의 22분부터 마지막까지의 내용을 다루고 있습니다. 코루틴의 Context? 코루틴의 Context에 말해보기에 앞서 Context에 대해 다시

everyday-develop-myself.tistory.com

'Android > Kotlin' 카테고리의 다른 글

Kotlin Flow와 Sequence (0)	2023.10.01
[Deep Dives into Coroutines on JVM] (2) - 코루틴의 Context (0)	2023.09.26
Sealed Class와 Enum Class 뭘 사용해야 하나요? (0)	2023.09.04
Scope Function (범위 지정 함수) with let, run, with, apply, also (0)	2023.06.30
Generics (제네릭) (0)	2023.06.29