[Kotlin/Coroutine] 2. Kotlin에서의 Coroutine: launch와 async

Kotlin에서의 Coroutine

1편인 Coroutine이란 무엇인가 참고

Kotlin은 특정 coroutine은 언어가 지원하는 형태가 아닌, Coroutine을 구현할 수 있는 기본 도구를 언어가 제공하는 형태이다

• Kotlin의 Coroutine 지원 기본 기능들은, kotlin.coroutine 패키지 밑에 있고,
  Kotlin 1.3부터는 Kotlin을 설치하면 별도의 설정없이도 모든 기능을 사용할 수 있다.
• 하지만, Kotlin이 지원하는 기본 기능을 활용해 만든 다양한 형태의 코루틴들은, kotlinx.coroutines 패키지 밑에 있다.

Suspend와 Resume

• Suspend와 Resume은 Callback방식을 대체한 Coroutine만의 메커니즘으로, 다음과 같은 의미가 있다
  • Suspend : 모든 Local 변수를 저장하고, 현 Scope에서의 Coroutine의 실행을 Pause.
    비동기 실행을 위한 중단지점의 의미.
  • Resume : Pause됐던 곳에서, Coroutine을 계속 실행

→ Suspend와 Resume 기능은, Kotlin의 Function에 suspend Keyword를 붙이거나, launch나 async처럼 Coroutine을 시작하는 함수를 사용함으로써 더해지게 된다.

Suspend Function(일시 중단 함수)를 만드는 법

• Kotlin은 Coroutine의 지원을 위해 suspend라는 키워드를 제공한다. 함수 정의 fun 앞에 suspend를 넣으면 일시중단 함수를 만들 수 있는 것이다.

Suspend Function

• Coroutine 안에서는 일반적인 메소드는 호출할 수 없다. Coroutine의 코드는 잠시 실행을 멈추거나(suspend), 다시 실행(resume)될 수 있기 때문이다.
• 따라서 Coroutine의 Scope에서 사용되어지는 Function은 suspend keyword를 붙여주어야 한다.
• 반대로 Coroutine 외부에서 suspend 함수를 호출하려고 하면 에러메시지가 나오게된다.

• coroutine의 가장 큰 특징인 suspend keyword는 Coroutine안에서 사용되면, suspend 함수가 호출 될경우 (1) 이전까지의 코드의 실행이 멈추며
  (2) susepnd 함수가 처리가 완료 된 이후, 멈춰 있던 원래 Scope의
  (3) 다음 코드가 실행된다.

suspend fun subRoutine(){
	for(i in 0..10){
		Log.d("subRoutine", "$i")
	}
}

CoroutineScope(Dispatchers.Main).launch {
	// 선 처리 코드
	subRoutine()
	// 후 처리 코드
}

• 위에서 처럼 suspend 키워드를 사용함에 따라, CoroutinScope 안에서 자동으로 Background Thread처럼 동작하게 된다.
• 가장 큰 특징인 이유는 이 suspend keyword가 붙인 함수가 실행되면서, 호출한 쪽의 코드를 잠시 멈추게 되지만, Thread의 중단이 없기 때문이다.
  • 코틀린이 실행되다가 일시 정지하는 경우(일정 시간 대기 등) Kotlin Runtime은 해당 Coroutine이 실행되던 Thread에 다른 Coroutine을 할당하여 실행되게 한다.
  • 그리고 다시 이전 Coroutine이 재개할 때, 사용 가능한 쓰레드를 Kotlin Runtime이 할당해준다.
    이를 통해 효율적인 쓰레드 활용이 가능해지고, 이러한 메커니즘에 맞게 실행되게 하는 함수가
    suspend function이다.
• 쓰레드에서 아래 코드를 사용했다면, 선처리 코드가 동작하는 Thread를 멈춰야 Subroutine 호출이 가능한데,
  Coroutine은 해당 부모 Routine의 상태를 저장
  → SubRoutine 실행
  → 부모 Routine 복원
  하는 식으로 동작하여 Thread에 영향을 주지 않게 된다.

suspend fun doSomethingUsefulOne(): Int {
        delay(1000L)
        return 13
}

suspend fun doSomethingUsefulTwo(): Int {
        delay(1000L)
        return 29
}

val time = measureTimeMillis {
        val one = doSomethingUsefulOne()
        val two = doSomethingUsefulTwo()
        println("The answer is ${one + two}")
}
println("Completed in $time ms")

Result >
The answer is 42
Completed in 2017 ms

suspend fun doSomethingUsefulOne(): Int {
        delay(1000L)
        return 13
}

suspend fun doSomethingUsefulTwo(): Int {
        delay(1000L)
        return 29
}

val time = measureTimeMillis {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")

Result >
The answer is 42
completed in 1017 ms

동시 처리를 원한다면 async 사용하면 된다

어떤 메서드를 Suspend로 사용하고, 사용하지 말아야 하는가

suspend로 불러야 하는 function이 있는 경우, suspend를 사용하고
suspend로 반드시 불러야 하는 경우가 아니라면, suspend를 사용하지 않는다.

→ 그럼 어느 기능을 최초로 suspend로 불러야 하는가? → I/O 동작 및 동시성을 보장해야할 Function

suspend(일시중단) 함수의 작동

• 예를 들어, 일시중단 함수 안에서 yield()를 해야 하는 경우 어떤 동작이 필요할지 생각해보면 다음과 같다
  1. Coroutine에 진입할 때와 Coroutine에서 나갈 때, Coroutine이 실행중이던 상태를 저장하고, 복구하는 등의 작업을 할 수 있어야 한다.
  2. 현재 실행중이던 위치를 저장하고, 다시 Coroutine이 재개될 때 해당 위치로부터 실행을 재개할 수 있어야 한다
  3. 다음에 어떤 Coroutine을 실행할 지 결정한다
• 위의 3가지 동작 중 마지막 동작(다음에 어떤 Coroutine을 실행할지)은 CoroutineContext에 있는 Dispatcher에 의해 수행된다. 일시 중단 함수를 컴파일하는 컴파일러는 앞의 2가지 작업을 할 수 있는 코드를 생성해 내야 한다.
• 이때 Kotlin은 CPS 변환(Continuation Passing Style)과 함께 상태 기계(State Machine)을 활용해 코드를 생성해낸다.

1편인 Coroutine이란 무엇인가 참고

• Kotlin의 Coroutine은 suspend keyword로 marking된 함수를 CPS(Continuation Passing Style)로 변환하고, 이를 Coroutine Builder를 통해 적절한 Thread상에서 시나리오에 따라 동작하도록 구성된다.
  • CPS 변환은 프로그램의 실행 중 특정 시점 이후에 진행해야 하는 내용을 별도의 함수로 뽑고
    (이런 함수를 Continuation이라 부른다), 그 함수에게 현재 시점까지 실행한 결과를 넘겨서 처리하게 만드는 소스코드 변환 기술이다.
  • CPS를 사용하는 경우 프로그램이 다음에 해야 할일이 항상 Continuation이라는 함수 형태로 전달되므로, 나중에 할일을 명확히 알 수 있고, 그 Continuation에 넘겨야 할 값이 무엇인지도 명확하게 알 수 있기 때문에 프로그램이 실행중이던 특정 시점의 맥락을 잘 저장했다가 필요할 때 다시 재개 할 수 있다.
  • 어떤 면에서 CPS는 Callback 스타일 프로그래밍과도 유사하다
  • CPS를 사용하면 Coroutine을 만들기 위해 필수적인 일시중단 함수를 만드는 문제가 쉽게 해결될 수 있다. 다만, 모든 코드를 전부 CPS로만 변환하면 지나치게 많은 중간 함수들이 생길 수 있으므로, 상태기계(State Machine)을 적절히 사용해 Coroutine이 제어를 다른 함수에 넘겨야하는 시점에만 Cotinuation이 생기도록 만들 수 있다.

• 예를 들어, 다음과 같이 suspend가 붙은 함수가 있다고 가정하면

suspend fun example(v: Int): Int {
	return v*2;
}

• 코틀린 컴파일러는 이 함수를 컴파일하면서, 뒤에 Continuation을 인자로 만들어 붙여준다
  public static final Object example(int v, @NotNull Continuation var1)
• 그리고 이 함수를 호출할 때는, 함수 호출이 끝난 후 수행해야 할 작업을 var1에 Continuation으로 전달하고, 함수 내부에서는 필요한 모든 일을 수행한 다음에 결과를 var1에 넘기는 코드를 추가한다.
  • 예를들어 위 코드에서는 v*2를 인자로 Continuation을 호출하는 코드가 들어간다.

Kotlin에서의 Coroutine 과정 요약

• Suspend function을 callback으로 바꾸어준다.
• code를 state machine 코드로 바꾸어준다.
• stdlib은 Continuation과 CoroutineContext를 가지고 있다.
• 나머지는 library 에 있다. launch/join, async/await, runBlocking 등등은kotlinx.coroutines 라이브러리를 써야 한다.

Kotlin이 kotlinx.coroutines을 통해 제공하는 기본적 Coroutine

kotlinx.coroutines.*

• 아래는 kotlinx.coroutines.core 모듈에 들어있는 코루틴이다.
  정확하게 이야기하자면, 각각은 Coroutine을 만들어주는 Coroutine Builder라고 부른다.
• Kotlin에서는 Coroutine Builder에 원하는 동작을 Lambda로 넘겨서 Coroutine을 만들어 실행하는 방식으로 Coroutine을 활용한다.

kotlinx.coroutines.CoroutineScope.launch

• launch는 Coroutine을 Job으로 반환하며, 만들어진 Coroutine은 기본적으로 즉시 실행된다.
  • 한번 시작된 작업은 예외가 발생하지 않는 한 대기하지 않는다.
• launch가 반환한 Job의 cancel()을 호출하면, Coroutine 실행을 중단(취소)시킬 수도 있다.
• launch가 작동하려면, CoroutineScope 객체가 블록의 this로 지정되어야 하는데,
  (API문서/소스를 보면 launch가 받는 블록의 타입이 suspend CoroutineScope.() → Unit임을 알 수 있음)
  다른 suspend 함수 내부라면, 해당 함수가 사용중인 CoroutineScope가 있지만,
  그렇지 않은 경우엔 GlobalScope를 사용한다.
• 예시 (Kotlin in Action)

package com.enshahar.kot1inStudy

import kotlinx.coroutines.*
import java.time.ZonedDateTime import java.time.temporal.ChronoUnit

fun now() = ZonedDateTime.now().toLocalTime().truncatedTo(ChronoUnit.MILLIS)

fun log(msg:String) = printin(”${now()}:${Thread.currentThread()}: ${msg}”)

fun launchlnGlobalScope() { 
	GlobalScope.launch {
		log("coroutine started.”)
	}
}
fun main() {
	log("main() started.") 
	launchlnGlobalScope() 
	log("launchlnGlobalScope() executed")
	Thread.sleep(5000L)
	log("main() terminated")
}

• 위의 코드를 컴파일하면 다음과 같은 결과를 얻을 수 있다.

23:39:58.200:Thread[main,5,main]: main() started. 
23:39:58.238:Thread[main,5,main]: launchlnGlobalScope() executed 
23:39:58.243:Thread[DefaultDispatcher-worker-2,5,main]: coroutine started. 
23:40:03.239:Thread[main,5,main]: main() terminated

• 여기서 유의할 점은 메인 함수와 GlobalScope.launch가 만들어낸 코루틴이 서로 다른 스레드에서 실행된다는 점이다.
• GlobalScope는 메인 스레드가 실행중인 동안만 코루틴의 동작을 보장해준다.
  • 앞 코드에서 Thread.sleep(5000L)를 삭제하면, 코루틴이 아예 실행되지 않을 것이다.
  • launchlnGlobalScope()가 호출한 launch는 스레드가 생성되고 시작되기 전에, 메인 스레드의 제어를 main()에 돌려주기 때문에, 따로 sleep()하지 않으면 main()이 바로 끝나고, 메인 스레드가 종료되면서 바로 프로그램 전체가 끝나버리기 때문이다.
  • 따라서 GlobalScope를 사용하는 경우에는 조심해서 사용해야 한다.
  • GlobalScope에 대한 자세한 내용은 4편을 참고

[Kotlin/Coroutine] 4. Coroutine의 구성요소: Coroutine의 생성과 활용

• 따라서, 위와같은 상황을 방지하기 위해선 (1) 비동기적으로 launch를 실행하거나, (2) launch가 모두 다 실행될 때까지 기다려야 한다.

// 앞에서의 launchInGlobalScope()를 runBlockingExample()이라는 이름의 함수로 변경
fun runBlockingExample() {
	runBlocking {
		launch {
			log ("GlobalScope.launch started.”)
		}
	}
}

• 위 프로그램을 실행하면 아래와 같다

23:48:11.806:Thread[main,5,main]: main() started. 
23:48:11.851:Thread[main,5,main]: coroutine started. 
23:48:11.852:Thread[main,5,main]: runBlockingExample() executed 
23:48:16.857:Thread[main,5,main]: main() terminated

• 주목할 것은 스레드가 모두 main 스레드라는 점이다.
• Coroutine의 스레드나 다른 비동기 도구와의 다른 장점은 아래의 코드에서 확인할 수 있다.

fun yieldExample () { 
	runBlocking { 
		launch {
			log("l") 
			yield () 
			log("3") 
			yield() 
			log("5")
		}
		log("after first launch") 
		launch {
			log("2") 
			delay(1000L) 
			log("4") 
			delay(1000L) 
			log("6")
		}
		log ("after second launch")
	}
}

• 위 코드를 실행하면 아래와 같다

00:01:34.048:Thread[main,5,main]: main() started. 
00:01:34.092:Thread[main,5,main]: after first launch 
00:01:34.096: Thread [main, 5, main] : after second launch 
00:01:34.098:Thread[main,5,main]: 1 
00:01:34.099:Thread[main,5,main]: 2 
00:01:34.105:Thread[main,5,main]: 3 
00:01:34.105:Thread[main,5,main]: 5 
00:01:35.108:Thread[main,5,main]: 4 
00:01:36.110:Thread[main,5,main]: 6 
00:01:36.Ill:Thread[main,5,main]: after runBlocking 
00:01:36.Ill:Thread[main,5,main]: yieldExample() executed
00:01:41.113:Thread[main,5,main]: main() terminated

• 위 로그를 통해 알 수 있는 launch의 특징은 다음과 같은데
  1. launch는 즉시 실행된다
  2. runBlocking은 내부 코루틴이 모두 끝난 다음에 반환된다.
  3. delay()를 사용한 Coroutine은 그 시간이 지날때 까지 다른 Coroutine에게 실행을 양보한다.
     • 앞 코드에서 delay(1000L) 대신 yield()를 사용했다면, 차례대로 1,2,3,4,5,6이 표시될 것이다.
     • 여기서 주목할점은 1번째 Coroutine이 2번이나 yield()를 했지만, 2번째 Coroutine이 delay() 상태에 있었기 때문에 다시 제어가 1번째 Coroutine으로 돌아왔다는 것이다.

runBlcoking

• 코루틴의 실행이 끝날 때까지 현재 스레드를 Block 시키는 함수로 runBlocking이 있다.
  • 현재 쓰레드를 block하고 실행되는 코드이므로, 메인 쓰레드에서 이용하는 것은 좋은 방법이 아니다.
  • 만약 메인 쓰레드에서 runBlocking을 사용해, 쓰레드를 장시간 점유할 경우, ANR이 발생할 수 있다.
• runBlocking은 CoroutineScope의 확장함수가 아닌, 일반함수이기에 별도의 CoroutineScope객체 없이 사용가능하다.
  • runBlocking() 함수 자체가 점유하고 대기하기 때문에, launch()의 join()이나 async()의 await() 등 다른 함수가 필요 없는 것이다.
• 다만, Coroutine의 취지와는 정반대의 역할을 한다.
  • 실제로, runBlocking에 관한 공식문서를 보면, 이 함수는 Coroutine으로부터 사용되어서는 안된다는 것이 작성되어 있다.
  • 일반적인 Blocking 코드와 Suspend 스타일로 적힌 라이브러리들을 Bridge 해줄 목적으로 설계된 함수이기 때문에, 메인 함수이나 코드 테스트, 라이브러리 통합의 상황에서만 사용되야 한다.
• 예제

GlobalScope.launch { // launch new coroutine in background and continue
	delay(1000L) // non-blocking delay for 1 second (default time unit is ms)
	Log.e("sequence","4") // print after delay
}
Log.e("sequence","1")

runBlocking { // but this expression blocks the main thread
	delay(4000L) // ... while we delay for 2 seconds to keep JVM alive
	Log.e("sequence","3")
}
Log.e("sequence","2")

→ 1 4 3 2 순서로 로그가 찍히게 된다.
→ GlobalScope.~ 에서 delay는 메인 스레드를 막지않고, runBlocking은 메인 스레드를 막는다.

kotlinx.coroutines.CoroutineScope.async

• async는 사실상 launch와 같은 일을 한다.
• 유일한 차이점은, launch가 Job을 반환하는 반면 async는 Deffered를 반환한다는 점뿐이다.

async에서의 Job와 Deffered

• Deffered와 Job의 차이를 살펴보면,
  1. Job은 어떤 Type parameter가 없는데,
     Deffered는 Type parameter가 있는 Generic Type이라는 점
     • Deffered의 Type parameter = Deffered Coroutine이 계산을 하고 돌려주는 값의 타입
       → Job은 Unit을 돌려주는 Deffered<Unit>이라고 생각할 수도 있을 것이다.
  2. Deffered 안에는 await() 함수가 정의되어 있다는 점

async에서의 await

• async 함수는 await() 함수로 완료되기를 기다릴수 있다.
  • task가 종료되는 시점을 기다렸다가, 결과를 받을 수 있다
  • 현재 쓰레드의 blocking없이 먼저 종료되면 결과를 가져올 수 있다.
  • await()를 사용하면 UI를 제외한 Routine만 blocking되므로, UI가 멈추는 경우를 해결할 수 있다.
  • 별도의 await call이 있기전까지는 발생한 Exception을 hold하고 있다
• 특히, suspend 키워드로 시작하는 Coroutine은 해당 함수가 return하는 시점에는 작업이 멈춰있어야 하므로, await()나, awaitAll()을 이용해서 return 전에 결과값을 가지도록 해주어야 한다.
  (return 전에 결과값이 안나오면 Exception이 발생할수 밖에 없기 때문)
  • 즉 return 해야 할 값이 있으므로, return에 있어서 launch 함수를 다룰 때 보다 Exception 처리에 있어 주의가 필요하다.
  • await가 호출되면, await 호출시에 exception을 들고 있다가 다시 rethrow해주는데, await 호출시, 새로운 Coroutine을 실행시키면, Exception이 Drop되어 버릴수 있다.
• 여러개의 async가 존재하는 경우, 1번째 async의 return값만을 return 해준다
• 따라서, async는 코드 블록을 비동기로 실행할 수 있고, async가 반환하는 Deffered의 await를 사용해 Coroutine이 결과값을 내놓을 때 까지 기다렸다가 결과값을 얻어낼 수 있다.
  • 예시(async/await를 이용해 1부터 3까지 수를 더하는 과정)

fun sumAll () { 
	runBlocking {
		val dl = async { delay (1000L); 1 } 
		log("after async(dl)")
		val d2 = async { delay(2000L) ; 2 } 
		log("after async(d2)")
		val d3 = async { delay (3000L) ; 3 } 
		log("after async(d3)”)
		log(”1+2+3 = ${dl.await() + d2.await() + d3.await()}” )
		log (’’after await all & add") }
}

  00:46:45.405:Thread[main,5,main]: after async(dl) 
  00:46:45.409:Thread[main,5,main]: after async(d2) 
  00:46:45.409:Thread[main,5,main]: after async(d3)
  00:46:48.417:Thread[main,5, main]: 1十2+3 = 6 
  00:46:48.418:Thread[main,5,main] : after await all & add

• 코드를 실행한 로그를 살펴보면, d1, d2, d3를 하나하나 순서대로 실행하면 총 6초 이상이 걸려야 하지만, 6이라는 결과가 얻을 때 까지 총 3초가 걸렸음을 알 수 있다.
  • 참고로 하나하나 순서대로 실행하는 경우를 병렬처리에서는 직렬화해 실행한다고 표현한다
• 또한, async로 코드를 실행하는데, 시간이 거의 걸리지 않음을 알 수 있다. 그럼에도, 스레드를 여럿 사용하는 병렬처리와 달리 모든 async 함수들이 메인 스레드 안에서 실행됨을 볼 수 있다.
  • 이부분이 async/await와 스레드를 사용한 병렬 처리의 큰 차이점이다.
• 예시 2 (연산 시간이 오래걸리는 2개의 네트워크 작업)

CoroutineScope(Dispatchers.Default).async {
            val deferred1 = async { 
                delay(500)
                350
            }
            val deferred2 = async { 
                delay(1000)
                200
            }
            Log.d("coroutine", "${deferred1.await() + deferred2.await()}")
        }

• 2개의 작업이 모두 완료되고 나서 이를 처리하려면 await()를 사용할 수 있다. 이때는 async 작업이 모두 완료되고 나서야 await() 호출 줄 코드가 실행된다.

•  

launch와 async 의 공통점과 차이점

이것은 너무너무나도 잘 정리된  launch와 Async 의 공통점과 차이이다

• launch의 내부 구현

public fun CoroutineScope.launch(
	context: Coroutinecontext = EmptyCoroutineContext, 
	start: Coroutinestart = Coroutinestart.DEFAULT, 
	block: suspend CoroutineScope. () -> Unit
) : Job {
	val newContext = newCoroutineContext (context) 
	val coroutine = if (start.isLazy)
		LazyStandaloneCoroutine(newContext, block) else
		StandaloneCoroutine(newContext, active = true) 
	coroutine.start(start, coroutine, block)
	return coroutine
}

 

• async의 내부 구현

public fun <T> CoroutineScope.async(
	context: Coroutinecontext = EmptyCoroutineContext, 
	start: Coroutinestart = Coroutinestart.DEFAULT, 
	block: suspend CoroutineScope. () -> T
) : Deferred<T> {
	val newContext = newCoroutineContext(context) 
	val coroutine = if (start.isLazy)
		LazyDeferredCoroutine(newContext, block) else
		DeferredCoroutine<T>(newContext, active = true) 
	coroutine.start(start, coroutine, block)
	return coroutine
}

• 공통점
  1. 현재의 스레드 중단 없이 Coroutine을 즉시 시작시킨다
• 차이점
  1. 사용환경
     • launch: 일시중단 함수가 아닌 일반 함수에서 일시중단 함수(suspend function)을 호출할 때와, Coroutine의 결과 처리가 필요 없을 때 사용한다
       → 연산이 실패한 경우에만 결과를 통보받기 원하는 Fire and Forget 시나리오
     • async: 병행으로 실행될 필요가 있는 다수의 Coroutine을 사용할 때 사용한다
       → 결국 둘의 가장 큰 차이는 return값이 있느냐 없느냐이며,
       어느 것을 사용하느냐에 따라 사용하는 메서드나 완료 대기에 대한 대응이 조금씩 달라진다.
       또한, 동시처리에 대한 필요성 여부도 마찬가지이다.