2.4 코틀린 코루틴 라이브러리 - 취소

 

코루틴에서 아주 중요한 기능 중 하나는 바로 취소다.

작업을 취소하기 위해 단순히 스레드를 죽이는 것은 최악의 방법이다. 연결을 닫고 자원을 해제할 방법이 없기 때문이다.

개발자들이 상태가 Active한지 계속해서 확인하는 방법도 불편하다.

코루틴의 취소 방식은 아주 간단하고 편리하며, 안전하다.

 

 

기본적인 취소

Job 인터페이스는 잡을 취소하는 cancel() 함수를 가지고 있다. cancel() 함수를 호출하면 다음과 같은 효과가 일어난다.

• cancel()을 호출한 코루틴은 첫 번째 중단점에서 잡을 끝낸다.
• 잡의 자식들도 취소된다. 하지만 잡의 부모는 영향을 받지 않는다.
• 취소된 잡은 새로운 코루틴의 부모로 사용될 수 없다. 취소된 잡은 Cancelling 상태가 되었다가 Cancelled 상태가 된다.

 

cancel() 함수에 예외를 인자로 넣으면 취소된 원인을 명확히 할 수 있다.

코루틴을 취소하는 예외는 CancellationException이어야 한다.

따라서 인자로 사용되는 예외도 반드시 CancellationException의 서브 타입이어야 한다.

 

suspend fun main() = coroutineScope {
    val job = launch {
        repeat(1_000) { i ->
            delay(100)
            Thread.sleep(100) // 오래 걸리는 연산이라 가정
            printIn("Printing $i")
        }
    }
    delay(1000)
    job.cancel()
    println("CanceUed successfully")
}

Printing 0
Printing 1
Printing 2
Printing 3
Cancelled successfully
Printing 4

일반적으로 cancel()을 호출하고 다음 작업을 진행하기 전에 취소 과정이 완료되는 걸 기다리기 위해 join()을 호출한다.

join()을 호출하지 않으면 경쟁 상태(race condition)가 발생할 수 있다.

위 코드에서 join()을 호출하지 않았기 때문에 'Cancelled successfully' 뒤에 'Printing 4'가 출력되는 것을 확인할 수 있다.

 

suspend fun main() = coroutineScope {
    val job = launch {
        repeat(1_000) { i ->
            delay(100)
            Thread.sleep(100) // 오래 걸리는 연산이라 가정
            printIn("Printing $i")
        }
    }
    delay(1000)
    job.cancel()
    job.join()
    println("CanceUed successfully")
}

Printing 0
Printing 1
Printing 2
Printing 3
Printing 4
Cancelled successfully

cancel() 뒤에 join()을 추가하면 코루틴이 취소를 마칠 때까지 중단되기 때문에 경쟁 상태가 발생하지 않는다.

 

suspend fun Job.cancelAndJoin() {
    cancel()
    return join()
}

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        repeat(1_000) { i ->
            delay(200)
            println("Printing $i")
        }
    }
    delay(1100)
    job.cancelAndJoin()
    println("Cancelled successfully")
}

코루틴 라이브러리는 call과 join을 함께 호출할 수 있는 cancelAndJoin() 확장 함수를 제공한다.

 

class ProfileViewModel : ViewModel() {
    private val scope =
        CoroutineScope(Dispatchers.Main + SupervisorJob())

    fun onCreate() {
        scope.launch { loadUserData() }
    }

    override fun onCleared() {
        scope.coroutineContext.cancelChildren()
    }

    // ... 
}

cancelChildren()으로 자식 잡을 한꺼번에 취소할 수 있다.

어떤 플랫폼이든 동시에 수행되는 작업 전체를 취소시켜야 할 때가 있다.

안드로이드를 예로 들면 사용자가 뷰를 나갔을 때 해당 뷰에서 시작된 모든 코루틴을 취소해야 한다.

 

 

취소는 어떻게 작동하는가?

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        try {
            repeat(1_000) { i ->
                delay(200)
                println("Printing $i")
            }
        } catch (e: CancellationException) {
            println(e)
            throw e
        }
    }
    delay(1100)
    job.cancelAndJoin()
    println("Cancelled successfully")
    delay(1000)
}

Printing 0
Printing 1
Printing 2
Printing 3
Printing 4
JobCancellationException...
Cancelled successfully

잡이 취소되면 Cancelling 상태로 바뀐다.

상태가 바뀐 뒤 첫 번째 중단점에서 CancellationException 예외를 던진다.

따라서 try-catch문으로 예외를 잡을 수도 있지만, 다시 던지는 것이 좋다.

 

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        try {
            delay(Random.nextLong(2000))
            println("Done")
        } finally {
            print("Will always be printed")
        }
    }
    delay(1000)
    job.cancelAndJoin()
}

(랜덤으로 생성된 값이 1000 초과인 경우)
Will always be printed
(랜덤으로 생성된 값이 1000 이하인 경우)
Done
Will always be printed

취소된 코루틴은 단순히 멈추는 것이 아니라 내부적으로 CancellationException을 사용해 취소된다.

따라서 finally 블록 안에서 모든 것을 정리할 수 있다.

예를 들어 finally 블록에서 파일이나 데이터베이스 연결을 닫을 수 있다.

 

 

취소 중 코루틴을 한 번 더 호출하기

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        try {
            delay(2000)
            println("Job is done")
        } finally {
            println("Finally")
            launch { // 무시됨
                println("Will not be printed")
            }
            delay(1000) // 예외 발생
            println("Will not be printed")
        }
    }
    delay(1000)
    job.cancelAndJoin()
    println("Cancel done")
}

(1초 후)
Finally
Cancel done

코루틴은 정리할 자원이 있다면 계속해서 실행될 수 있다.

하지만 정리 과정 중에 중단할 수는 없다.

Job은 이미 Cancelling 상태가 되었기 때문에 중단되거나 다른 코루틴을 시작할 수는 없다.

다른 코루틴을 시작하려고 하면 무시해 버리고, 중단하려고 하면 CancellationException을 던진다.

 

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        try {
            delay(200)
            println("Coroutine finished")
        } finally {
            println("Finally")
            withContext(NonCancellable) {
                delay(1000L)
                println("Cleanup done")
            }
        }
    }
    delay(100)
    job.cancelAndJoin()
    println("Done")
}

Finally
Cleanup done 
Done

코루틴이 취소된 후 중단 함수를 반드시 호출해야 하는 경우도 있다. 예를 들어 데이터베이스의 변경 사항을 롤백하는 경우다.

일반적으로 호출할 함수를 withContext(NonCanceUable)로 감싸서 해결한다.

코드 블록의 컨텍스트를 취소될 수 없는 Job인 NonCancellable 객체로 바꾸는 것이다.

따라서 withContext 블록 내부의 잡은 Active 상태를 유지하고, 중단 함수를 호출할 수 있다.

 

 

invokeOnCompletion

suspend fun main() Unit = coroutineScope {
    val job = launch {
        delay(1000)
    }
    job.invokeOnCompletion { exception: Throwable? ->
        println("Finished")
    }
    delay(400)
    job.cancelAndJoin()
}

Finished

자원을 해제하는 데 자주 사용되는 또 다른 방법은 Job의 invokeOnCompletion() 함수를 호출하는 것이다.

invokeOnCompletion() 함수는 잡이 Completed나 Cancelled와 같은 마지막 상태에 도달했을 때 호출될 핸들러를 지정한다.

 

abstract fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle

typealias CompletionHandler = (cause: Throwable?) -> Unit

invokeOnCompletion()의 파라미터는 CompletionHandler이다.

CompletionHandler는 (Throwable?) -> Unit 람다 타입의 별칭이다.

이 람다의 파라미터인 예외는 아래 중 하나가 된다.

• 잡이 예외 없이 끝나면 null
• 코루틴이 취소되었으면 CancellationException
• 코루틴을 종료시킨 예외

 

suspend fun main(): Unit = coroutineScope {
    val job = launch {
        delay(Random.nextLong(2400))
        println("Finished")
    }
    delay(800)
    job.invokeOnCompletion { exception: Throwable? ->
        println("Will always be printed")
        println("The exception was: $exception")
    }
    delay(800)
    job.cancelAndJoin()
}

(랜덤으로 생성된 값이 800 초과인 경우)
Will always be printed
The exception was: kotlinx.coroutines.JobCancellationException
(랜덤으로 생성된 값이 800 이하인 경우)
Finished
Will always be printed 
The exception was null

잡이 invokeOnCompletion이 호출되기 전에 완료되었으면 핸들러는 즉시 호출된다.

invokeOnCompletion은 취소 중에 동기적으로 호출되고, 어떤 스레드에서 실행할지 결정할 수는 없다.

 

onCancelling과 invokelmmediately 파라미터를 사용하면 핸들러의 동작 방식을 변경할 있다.

• onCancelling: 이 값을 true로 설정하면 Cancelled 상태 전인 Cancelling 상태 핸들러가 호출된다. 디폴트 값은 false다.
• invokelmmediately: 잡이 이미 Completed나 Cancelled와 같은 마지막 상태일 때 핸들러를 즉시 호출할지 여부를 결정한다. 디폴트 값은 true다.

 

 

중단될 수 없는 걸 중단하기

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        repeat(1_000) { i ->
            Thread.sleep(200) // 파일을 읽는 등의 복잡한 작업이 있다고 가정
            println("Printing $i")
        }
    }
    delay(1000)
    job.cancelAndJoin()
    println("Cancelled successfully")
    delay(1000)
}

Printing 0
Printing 1 
Printing 2 
...
Printing 1000

취소는 중단점에서 일어나기 때문에 중단점이 없으면 취소를 할 수 없다.

예시를 위해 delay 대신 Thread.sleep을 사용했다. Thread.sleep은 좋지 않은 방식이므로 현업에선 절대 사용하면 안 된다.

위 코드는 1초 뒤 실행이 취소되어야 하지만 실제론 3분이 넘게 걸린다.

 

이런 상황에 대처하는 몇 가지 방법을 알아보자.

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        repeat(1_000) { i ->
            Thread.sleep(200)
            yield()
            println("Printing $i")
        }
    }
    delay(1100)
    job.cancelAndJoin()
    println("Cancelled successfully")
    delay(1000)
}

Printing 0
Printing 1 
Printing 2
Printing 3
Printing 4 
Cancelled successfully

첫 번째 방법은 yield()를 주기적으로 호출하는 것이다.

yield()는 코루틴을 중단하고 즉시 재실행한다.

중단점이 생겼기 때문에 취소(또는 디스패처 변경)와 중단(또는 재실행)을 할 수 있는 기회가 주어진다.

 

suspend fun cpuIntensiveOperations() =
    withContext(Dispatchers.Default) {
        cpuIntensiveOperation1()
        yield()
        cpulntensive0peration2()
        yield()
        cpulntensiveOperation3()
    }

중단 가능하지 않으면서 CPU나 시간을 많이 필요로 하는 연산이 중단 함수에 있다면, 각 연산들 사이에 yield()를 사용하는 것이 좋다.

 

val CoroutineScope.isActive: Boolean
    get() = coroutinecontext[Job]?.isActive ?: true

두 번째 방법은 잡의 상태를 확인하는 것이다.

코루틴 빌더 내부 람다의 리시버는 CoroutineScope이다.

CoroutineScope는 CoroutineContext를 참조할 수 있기 때문에, 잡에 접근해 잡의 현재 상태를 확인할 수 있다.

코루틴 라이브러리는 잡의 Active 상태를 확인하는 isActive 프로퍼티를 제공한다.

 

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        do {
            Thread.sleep(200)
            println("Printing")
        } while (isActive)
    }
    delay(1100)
    job.cancelAndJoin()
    println("Cancelled successfully")
}

Printing 
Printing 
Printing 
Printing
Printing 
Printing
Cancelled successfully

isActive 프로퍼티로 잡이 액티브한지 확인한다. 액티브하지 않을 때는 연산을 중단할 수 있다.

 

suspend fun main(): Unit = coroutineScope {
    val job = Job()
    launch(job) {
        repeat(1000) { num ->
            Thread.sleep(200)
            ensureActive()
            println("Printing $num")
        }
    }
    delay(1100)
    job.cancelAndJoin()
    println("Cancelled successfully")
}

Printing 0
Printing 1 
Printing 2
Printing 3
Printing 4 
Cancelled successfully

세 번째 방법은 잡이 액티브 상태가 아니면 CancellationException을 던지는 ensureActive() 함수를 사용하는 것이다.

 

 

yield()와 ensureActive() 함수 모두 다른 코루틴이 실행할 수 있는 기회를 준다는 점에서 결과는 비슷하지만, 둘은 매우 다르다.

• ensureActive()
  • CoroutineScope(또는 Coroutinecontext나 Job)에서 호출되어야 한다.
  • 잡이 액티브 상태가 아니면 CancellationException을 던진다.
  • 일반적으로 ensureActive()가 좀 더 가벼워 선호된다.
• yield()
  • 전형적인 최상위 중단 함수다.
  • 스코프가 필요하지 않기 때문에 일반적인 중단 함수에서도 사용할 수 있다.
  • 중단하고 재개하기 때문에 스레드 풀을 가진 디스패처를 사용하면 스레드가 바뀔 수 있다.
  • CPU 사용량이 크거나 스레드를 블로킹하는 중단 함수에서 자주 사용된다.

 

 

suspendCancellableCoroutine

suspend fun someTask() = suspendCancellableCoroutine { cont ->
    cont.invokeOnCancellation {
        // 정리 작업을 수행
    }
    // 나머지 구현 부분
}

suspendCancellableCoroutine() 함수는 suspendCoroutine() 함수와 비슷하지만, Continuation 객체를 CancellableContinuation<T>로 래핑한다.

CancellableContinuation는 Continuation에 몇 가지 함수가 추가된 객체다.

 

가장 중요한 함수는 코루틴이 취소되었을 때 행동을 정의하는 데 사용하는 invokeOnCancellation() 함수다.

이 함수는 라이브러리의 실행을 취소하거나 자원을 해제할 때 주로 사용된다.

 

아래 코드에서 Retrofit Call 객체를 중단 함수로 래핑한 것을 확인할 수 있다.

suspend fun getOrganizationRepos(
    organization: String
): List<Repo> =
    suspendCancellableCoroutine { continuation ->
        val orgReposCall = apiService
            .getOrganizationRepos(organization)
        orgReposCall.enqueue(object : Callback<List<Repo>> {
            override fun onResponse(
                call: Call<List<Repo>>,
                response: Response<List<Repo>>,
            ) {
                if (response.isSuccessful) {
                    val body = response.body()
                    if (body != null) {
                        continuation.resume(body)
                    } else {
                        continuation.resumeWithException(
                            ResponseWithEmptyBody
                        )
                    }
                } else {
                    continuation.resumeWithException(
                        ApiException(
                            response.code(),
                            response.message(),
                        )
                    )
                }
            }

            override fun onFailure(
                call: Call<List<Repo>>,
                t: Throwable,
            ) {
                continuation.resumeWithException(t)
            }
        })
        continuation.invokeOnCancellation {
            orgReposCall.cancel()
        }
    }

 

class GithubApi {
    @GET("orgs/{organization}/repos?per_page=100")
    suspend fun getOrganizationRepos(
        @Path("organization") organization: String,
    ): List<Repo>
}

Retrofit도 중단 함수를 지원한다.

 

CanceUableContinuation<T>에서도 잡의 상태를 확인할 수 있다. (isActive, isCompleted, isCancelled 프로퍼티 사용)

또한 Continuation을 취소할 때 취소의 원인을 추가적으로 제공할 수 있다.

 

 

 

출처

https://www.yes24.com/product/goods/123034354