Coroutine 2 : Cancellation and timeouts

김수호·2024년 12월 18일

코틀린

목록 보기

5/5

join()

    fun main5() {
        GlobalScope.launch {
            val job = launch {
                var i = 0
                while (i < 3) {
                    delay(1000L)
                    println("Coroutine running $i")
                    i++
                }
            }

            println("test #1")
            delay(1800L)
            job.join()
            println("test #2")
        }
    }

test #1
Coroutine running 0
Coroutine running 1
Coroutine running 2
test #2

job을 join()으로 실행하였으니 끝날때까지 대기하도록 설정이 된다.

cancel()

    fun main5() {
        GlobalScope.launch {
            val job = launch {
                var i = 0
                while (i < 3) {
                    delay(1000L)
                    println("Coroutine running $i")
                    i++
                }
            }

            println("test #1")
            delay(1800L)
            job.cancel()
            println("test #2")
        }
    }

test #1
Coroutine running 0
test #2

job을 cancel()로 실행하였기 때문에 1을 출력하기전에 취소처리가 된다.

job에 대한 설정이 없는 경우

    fun main5() {
        GlobalScope.launch {
            val job = launch {
                var i = 0
                while (i < 3) {
                    delay(1000L)
                    println("Coroutine running $i")
                    i++
                }
            }

            println("test #1")
            delay(1800L)
            println("test #2")
        }
    }

test #1
Coroutine running 0
test #2
Coroutine running 1
Coroutine running 2

job에 특정영향을 끼치는 내용이 없으므로 그대로 딜레이 만큼 멈추고 실행이 진행된다.

joinAll()

val job1 = launch {
    var i = 0
    while (i < 5) {
    	delay(1000L)
    	i++
    }
}

val job2 = launch {
    var i = 0
    while (i < 5) {
    	delay(1000L)
    	i++
    }
}

// 방법 1 - 각각 코루틴마다 join() 설정
job1.join()
job2.join()

// 방법 2 - joinAll()에 같이 선언하기
joinAll(job1, job2)

Coroutine running1 0
Coroutine running2 0
Coroutine running1 1
Coroutine running2 1
Coroutine running1 2
Coroutine running2 2
Coroutine running1 3
Coroutine running2 3
Coroutine running1 4
Coroutine running2 4

job1과 job2가 동시에 실행된다.

Cancelling coroutine execution

val job = launch {
    repeat(1000) { i ->
        println("job: I'm sleeping $i ...")
        delay(500L)
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancel() // cancels the job
job.join() // waits for job's completion 
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
main: Now I can quit.

cancel이 먼저 동작했기 때문에 이후의 join은 모두 종료된 이후를 리턴하게 되므로 아무것도 리턴하지 않는다.

Cancellation is cooperative

val startTime = System.currentTimeMillis()
val job = launch(Dispatchers.Default) {
    var nextPrintTime = startTime
    var i = 0
    while (i < 5) { // computation loop, just wastes CPU
        // print a message twice a second
        if (System.currentTimeMillis() >= nextPrintTime) {
            println("job: I'm sleeping ${i++} ...")
            nextPrintTime += 500L
        }
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
job: I'm sleeping 3 ...
job: I'm sleeping 4 ...
main: Now I can quit.

여기서는 취소하고 재실행을 하지만 취소에 대해서 체크를 하지 않았기 때문에 join 그대로 진행이 된다.

import kotlinx.coroutines.*

fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
    val job = launch(Dispatchers.Default) {
        var nextPrintTime = startTime
        var i = 0
        while (i < 5) { // computation loop, just wastes CPU
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                println("job: I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancel() // cancels the job and waits for its completion
    println("main: Now I can quit.")    
}

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
main: Now I can quit.
job: I'm sleeping 3 ...
job: I'm sleeping 4 ...

캔슬하더라도 위 설정에서는 런치가 취소되지않는다.

fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
    val job = launch(Dispatchers.Default) {
        var nextPrintTime = startTime
        var i = 0
        while (i < 5) { // computation loop, just wastes CPU
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                yield()
                println("job: I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")    
}

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
job: I'm sleeping 3 ...
main: Now I can quit.

만약 취소를 하고 싶다면 yield()를 사용하여 주기적으로 체크를 해야한다.

val job = launch(Dispatchers.Default) {
    repeat(5) { i ->
        try {
            // print a message twice a second
            println("job: I'm sleeping $i ...")
            delay(500)
        } catch (e: Exception) {
            // log the exception
            println(e)
        }
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job="coroutine#2":StandaloneCoroutine{Cancelling}@476c9f7
job: I'm sleeping 3 ...
kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job="coroutine#2":StandaloneCoroutine{Cancelling}@476c9f7
job: I'm sleeping 4 ...
kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job="coroutine#2":StandaloneCoroutine{Cancelling}@476c9f7
main: Now I can quit.

CancellationException을 잡고 처리하되 다시 던지지 않기 때문에 계속 출력이 됩니다.

수정하고 싶다면 아래 코드를 확인하면 된다.

val job = launch(Dispatchers.Default) {
    repeat(5) { i ->
        try {
            // print a message twice a second
            println("job: I'm sleeping $i ...")
            delay(500)
        } catch (e: CancellationException) {
            println("Cancelled! $e")
            throw e 
        }
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
Cancelled! kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job="coroutine#2":StandaloneCoroutine{Cancelling}@7d8796cb
main: Now I can quit.

이 코드는 안티 패턴이지만 아까와 같이 중복되는 케이스는 막을 수 있었다.

Making computation code cancellable

val startTime = System.currentTimeMillis()
val job = launch(Dispatchers.Default) {
    var nextPrintTime = startTime
    var i = 0
    while (isActive) { // cancellable computation loop
        // print a message twice a second
        if (System.currentTimeMillis() >= nextPrintTime) {
            println("job: I'm sleeping ${i++} ...")
            nextPrintTime += 500L
        }
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
main: Now I can quit.

while문의 검사를 isActive로 체크하면 된다.

isActive는 코루틴이 취소상태인지 체크하는 역할을 하며 kotlinx.coroutines 라이브러리에 포함되어 있다.

실제 구현 코드는 (공개된 소스코드 기준)

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

작동 방식

coroutineContext에서 현재 코루틴의 컨텍스트를 가져온다.
컨텍스트에서 job을 가져온다.
job 객체의 상태가 활성인지 체크한다.
job 객체가 없으면 기본값으로 true를 반환한다.

Closing resources with finally

val job = launch {
    try {
        repeat(1000) { i ->
            println("job: I'm sleeping $i ...")
            delay(500L)
        }
    } finally {
        println("job: I'm running finally")
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
job: I'm running finally
main: Now I can quit.

job.cancelAndJoin()로 인해서 종료를 요청받고 repeat루프 내부에서 CancellationException를 발생시킨다.
코루틴은 이 시점에서 종료가 되며 finally 블록이 실행됩니다.

Run non-cancellable block

val job = launch {
    try {
        repeat(1000) { i ->
            println("job: I'm sleeping $i ...")
            delay(500L)
        }
    } finally {
        withContext(NonCancellable) {
            println("job: I'm running finally")
            delay(1000L)
            println("job: And I've just delayed for 1 sec because I'm non-cancellable")
        }
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
job: I'm running finally
job: And I've just delayed for 1 sec because I'm non-cancellable
main: Now I can quit.

원래는 종료가 된 시점에 코루틴을 제어하는 함수는 사용할 수 없다.
하지만 withContext(NonCancellable)으로 선언한다면 이를 무조건적으로 처리해야한다.
그렇기 때문에 실행결과에 delay가 적용됐음에도 오류가 발생하지 않았다.

val job = launch {
    try {
        repeat(1000) { i ->
            println("job: I'm sleeping $i ...")
            delay(500L)
        }
    } finally {
        try {
            
                println("job: I'm running finally")
                delay(1000L)  // This should not cause CancellationException
                println("job: And I've just delayed for 1 sec because I'm non-cancellable")
            
        } catch (e: CancellationException) {
            println("job: CancellationException caught in finally")
        }
    }
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")

job: I'm sleeping 0 ...
job: I'm sleeping 1 ...
job: I'm sleeping 2 ...
main: I'm tired of waiting!
job: I'm running finally
job: CancellationException caught in finally
main: Now I can quit.

이 결과를 보면 CancellationException이 발생하였을 때 코드가 실행되도록 했던 부분이 실행이 되었다.
이 처럼 강제로 실행하려고 해도 CancellationException이 발생하는 부분을 확인할 수 있다.

Timeout

withTimeout(1300L) {
    repeat(1000) { i ->
        println("I'm sleeping $i ...")
        delay(500L)
    }
}

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1300 ms

오류가 발생하게 된다. withTimeout으로 1300ms까지 실행할 수 있도록 설정을 했기 때문에 넘어가는 시점인 1300ms에서 TimeoutCancellationException오류가 발생하게 된다.

val result = withTimeoutOrNull(1300L) {
    repeat(1000) { i ->
        println("I'm sleeping $i ...")
        delay(500L)
    }
    "Done" // will get cancelled before it produces this result
}
println("Result is $result")

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Result is null

withTimeout -> withTimeoutOrNull 로 변경이 되었다.
이는 비슷하지만 시간초과가 일어난다면 null을 return을 하는 방식으로 변경되어 오류가 나지 않는다.
"Done"의 코드는 이미 타임아웃 전에 실행되지 않으므로 상관없다.

fun main() = runBlocking {
    var result: String? = null
    try {
        result = withTimeout(1300L) {
            repeat(1000) { i ->
                println("I'm sleeping $i ...")
                delay(500L)
            }
            "Done" // 이 코드는 타임아웃 전에 실행되지 않음
        }

    } catch (e: TimeoutCancellationException) {
        println("Caught TimeoutCancellationException: Timeout occurred!")
		result = "timeout"
    }
    println("Result is $result")
}

I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
Caught TimeoutCancellationException: Timeout occurred!
Result is timeout

위 코드와 마찬가지로 타임아웃 오류가 발생했고 이를 캐치하여 result에 timeout이 들어가는 결과로 작동되었다.

Asynchronous timeout and resources

withTimeout 블록 내에서 타임아웃이 발생할 때, 타임아웃 이벤트가 비동기적으로 발생한다는 점을 주의해야 합니다. 이는 타임아웃이 블록 내 코드가 실행 중일 때 언제든지 발생할 수 있음을 의미합니다. 예를 들어, 코드가 withTimeout 블록 내에서 Resource 객체를 생성한 직후, 타임아웃이 발생하면 그 리소스가 제대로 해제되지 않고 남을 수 있습니다.

타임아웃이 발생하고 나서 withTimeout 블록이 종료되면, 그 블록 내에서 실행 중이던 코드가 더 이상 실행되지 않게 되는데, 이로 인해 리소스를 해제할 기회가 사라집니다. 이 문제를 리소스가 제대로 해제되지 않도록 리소스 누수라고 부릅니다.

var acquired = 0

class Resource {
    init { acquired++ } // Acquire the resource
    fun close() { acquired-- } // Release the resource
}

fun main() {
    runBlocking {
        repeat(10_00) { // Launch 10K coroutines
            launch { 
                val resource = withTimeout(60) { // Timeout of 60 ms
                    delay(50) // Delay for 50 ms
                    Resource() // Acquire a resource and return it from withTimeout block     
                }
                resource.close() // Release the resource
            }
        }
    }
    // Outside of runBlocking all coroutines have completed
    println(acquired) // Print the number of resources still acquired
}

객체가 생성되면 Resource의 acquired를 증가시키고, close를 하면 acquired를 감소시킵니다.
1만개의 코루틴을 생성시킵니다.
코루틴의 타임아웃은 60ms이며 딜레이가 50ms만큼 걸린 후 리소스를 생성합니다.
타임아웃 이전에 리소스 해제를 하지 못하면 블록이 종료되어 리소스 누수가 발생합니다.

이를 해결하기 위한 방식으로

runBlocking {
    repeat(10_000) { // Launch 10K coroutines
        launch { 
            var resource: Resource? = null // Not acquired yet
            try {
                withTimeout(60) { // Timeout of 60 ms
                    delay(50) // Delay for 50 ms
                    resource = Resource() // Store a resource to the variable if acquired      
                }
                // We can do something else with the resource here
            } finally {  
                resource?.close() // Release the resource if it was acquired
            }
        }
    }
}
// Outside of runBlocking all coroutines have completed
println(acquired) // Print the number of resources still acquired

withTimeout 블록 내에서 리소스를 반환하는 대신 변수에 리소스를 저장하고, finally 블록에서 리소스를 해제하도록 수정합니다.
finally 블록은 타임아웃 발생 여부와 관계없이 항상 실행되므로, 리소스를 정상적으로 해제합니다.

김수호

정답을 모르지만 답을 찾는 법을 알고, 그 답을 찾아낼 것이다. 그럼 괜찮지 않은가? -크리스 가드너-

이전 포스트

Coroutine 2 : Cancellation and timeouts

코틀린

join()

cancel()

job에 대한 설정이 없는 경우

joinAll()

Cancelling coroutine execution

Cancellation is cooperative

Making computation code cancellable

Closing resources with finally

Run non-cancellable block

Timeout

Asynchronous timeout and resources

Coroutine 1 : 기본

0개의 댓글

Coroutine 2 : Cancellation and timeouts

코틀린

join()

cancel()

job에 대한 설정이 없는 경우

joinAll()

Cancelling coroutine execution

Cancellation is cooperative﻿

Making computation code cancellable

Closing resources with finally

Run non-cancellable block﻿

Timeout

Asynchronous timeout and resources

Coroutine 1 : 기본

0개의 댓글

Cancellation is cooperative

Run non-cancellable block