[Android] Coroutine - suspend와 Continuation

이동건·2024년 2월 27일

android

목록 보기

3/3

기존의 RxJava와 같은 코드 스타일은 콜백을 사용하여 백그라운드 스레드의 작업을 처리하였다. 작업 완료 여부를 메인 스레드가 busy-waiting하고 있지 않고, 완료 되었을 때 콜백 코드가 호출되어 Main-Safe하게 작업을 처리할 수 있다.

Main-Safe하다는 것은 메인 스레드를 블락시키지 않아, 메인 스레드에서 실행되어도 안전하다는 것을 의미한다.

// RxJava Callback 방식
fun createUser(id: String, password: String){
		repository.checkUser(id, password)
			.observeOn(scheduler)
			.toObservable
			.subscribe{ user -> 
				repository.signIn(user)
					.doOnSuccess{
						// 로그인 성공
						Timber.d("login success")
					}
			}.let(compositeDisposable::add)
}

하지만 콜백 요청이 많아지고 중첩으로 하게된다면 코드의 가독성을 해치게 된다. 코루틴은 이러한 콜백 기반 코드를 Sequential Code(동기 코드 같이)로 작성할 수 있게 해준다.

// Coroutine 방식
suspend fun createUser(id: String, password: String){
		val user = repository.checkUser(id, password)
		repository.signIn(user)
		// 로그인 성공
		Timber.d("login success")
}

다음과 같이 코루틴을 사용하면 비동기 동작을 동기 코드처럼 표현함으로써 코드의 가독성이 향상되었다. 더욱 직관적이고 추론하기 쉬운 코드는 예측 가능한 프로그램을 만들며 이는 에러 핸들링 및 디버깅 작업에 이점을 준다.

위의 예시에서 함수 앞에 suspend키워드가 붙은 것을 볼 수 있다. suspend 키워드가 붙은 함수는 Kotlin 컴파일러가 어떤 작업을 해주길래 동기 코드 처럼 작성할 수 있는지 알아보자.

Continuation

https://tech.devsisters.com/posts/crunchy-concurrency-kotlin/

코루틴은 Continuation Passing Style(CPS) 형태로 동작한다.

https://kotlinlang.org/spec/asynchronous-programming-with-coroutines.html

CPS는 호출되는 함수에 Continuation을 전달하고, 각 함수의 작업이 완료되는 대로 전달받은 Continuation을 호출하는 방식을 말한다.

Continuation 인터페이스에는 context와 resumeWith가 있다.

context : 현재 Continuation과 관련된 CoroutineContext이다. CoroutineContext는 코루틴이 실행되는 환경 및 문맥을 말하며, 현재 실행되는 코루틴의 상태 정보를 담고 있다.
resumeWith : 특정 함수가 일시 중단(suspend)되어야 할 때, 결과값을 T로 받게 해주고 코루틴의 실행을 재개해주는 함수이다.

즉, CPS에선 Continuation을 전달하면서, 현재 일시 중단된 부분에서 재개가 가능하게 한다. Continuation은 재개되었을 때의 동작 관리를 위한 객체로, 연속적인 상태 간의 communicator라고 볼 수 있다.

Suspend 내부 동작

컴파일에서 suspend

함수 앞에 suspend 키워드를 붙이면 Kotlin 컴파일러가 suspend 키워드를 지우고 Continuation 객체를 파라미터로 붙인다.

// 기존 코틀린 코드
private suspend fun registerPostWithImage(uriList: List<String>){ }

// 디컴파일된 자바 코드
ResultViewModel$registerPostWithImage$$inlined$with$lambda$1$1(ResultViewModel$registerPostWithImage$$inlined$with$lambda$1 var1, Continuation var2) {
      super(2, var2);
      this.this$0 = var1;
   }
@Nullable
   public final Object invokeSuspend(@NotNull Object var1) {
      Object var2 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
      switch (this.label) {
         case 0:
            ResultKt.throwOnFailure(var1);
            return Unit.INSTANCE;
         default:
            throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
      }
   }

   @NotNull
   public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
      Intrinsics.checkNotNullParameter(completion, "completion");
      ResultViewModel$registerPostWithImage$$inlined$with$lambda$1$1 var3 = new ResultViewModel$registerPostWithImage$$inlined$with$lambda$1$1(this.this$0, completion);
      return var3;
   }

   public final Object invoke(Object var1, Object var2) {
      return ((ResultViewModel$registerPostWithImage$$inlined$with$lambda$1$1)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
   }

suspend 함수의 작업이 완료되면 Continuation 객체로 결과값을 호출한 코루틴에 전달한다.

중단 지점

여기부터 설명을 쉽게 하기 위해 Kotlin 코드로 먼저 설명을 하고난 후에 자바 디컴파일 코드에서 어떻게 구현되어 있는지 알아보겠다.

fun createUser(id: String, password: String, completion: Continuation<Any?>){
		val user = repository.checkUser(id, password)

		repository.signIn(user)
		
		completion.resume(Unit)
}

fun createUser(id: String, password: String, completion: Continuation<Any?>){
		when(label){
			0 -> { // Label 0 -> first execution
					repository.checkUser(id, password)
			}
			1 -> { // Label 1 -> resumes from repository
					repository.signIn(user)
			}
			2 -> { // Label 2 -> resumes from repository
					completion.resume(Unit)
			}
			else -> throw IllegalStateException(/* ... */)
		}
}

Kotlin은 suspend 함수 내에 모든 중단 가능 지점을 찾아 when으로 표현한다. 각 중단 가능 지점에 대해 label 값으로 표현이 되며 모든 지점은 유한 상태 기계로 표현된다.

각 중단 지점에 대해 라벨링을 하고 코드를 명확히 구분하여, 중단 가능 지점에서 suspend되고 작업이 완료되면 코루틴이 다시 재개 되어 다음 블록을 실행할 수 있도록 한다.

State Machine

fun createUser(id: String, password: String, completion: Continuation<Any?>){
		class CreateUserStateMachine(completionL Continuation<Any?>): CoroutineImpl(completion) {
        var user: User? = null

        var result: Any? = null
        var label: Int = 0

        override fun invokeSuspend(result: Any?) {
            this.result = result
            createUser(id, password, thist)
        }
}

Kotlin 컴파일러는 함수 내부에 CreateUserStateMachine이라는 클래스를 생성한다 이 클래스는 CorutineImpl클래스를 구현하고 있다.

CoroutineImpl 클래스의 내부 코드는 다음을 참고하자. https://github.com/JetBrains/kotlin/blob/master/libraries/stdlib/wasm/src/kotlin/coroutines/CoroutineImpl.kt

해당 StateMachine 클래스는 다음과 같이 사용이 된다.

fun createUser(id: String, password: String, completion: Continuation<Any?>){
		val continuation = completion as? CreateUserStateMachine ?: CreateUserStateMachine(completion)
		when(continuation.label) {
        0 -> {
            // Checks for failures
            throwOnFailure(continuation.result)
            // Next time this continuation is called, it should go to state 1
            continuation.label = 1
            // The continuation object is passed to logUserIn to resume
            // this state machine's execution when it finishes
            repository.checkUser(id, password, continuation)
        }
        1 -> {
            // Checks for failures
            throwOnFailure(continuation.result)
            // Gets the result of the previous state
            continuation.user = continuation.result as User
            // Next time this continuation is called, it should go to state 2
            continuation.label = 2
            // The continuation object is passed to logUserIn to resume
            // this state machine's execution when it finishes
            repository.signIn(continuation.user, continuation)
        }
          /* ... leaving out the last state on purpose */
    }
}

컴파일러가 생성한 코드에는 다음 사항들을 포함하고 있다.

새로운 state가 진행될 때마다, 함수가 일시 중지 되었을 때 오류가 발생했는지 확인한다.
각 label 분기에선 label 값을 증가시켜 다음 실행지점을 가리킨다.
state machine 내부에서 다른 suspend 함수를 호출할 때, continuation 객체를 파라미터로 넘긴다.

Kotlin 컴파일러는 suspend 키워드를 만나면 CPS 패러다임을 구현하여, Continuation이라는 일종의 콜백을 주고 받도록 코드를 변환해준다. 코루틴이 재개될때 필요한 정보들이 Continuation 에 저장됨으로써 스레드를 블럭시키지 않고도 일시 중단이 가능하다.

자바 코드에서 확인해보기

중단 지점이 두 개 있다는 것을 보여주기 위해 사진 캡쳐를 하였다.

private final void registerPost() {
   BuildersKt.launch$default(ViewModelKt.getViewModelScope(this), (CoroutineContext)this.exceptionHandler, (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
      int label;

      @Nullable
      public final Object invokeSuspend(@NotNull Object $result) {
         Object var7 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
         Object var10000;
         boolean var4;
         switch (this.label) {
            case 0:
               ResultKt.throwOnFailure($result);
               ResultViewModel.access$updateState(ResultViewModel.this, (ResultContract.ResultReduce)(new ResultContract.ResultReduce.UpdateRegisterState((ResultContract.ResultRegisterState)RegisterLoading.INSTANCE)));
               Object var2 = ResultViewModel.this.getViewState().getValue();
               ResultContract.ResultState $this$with = (ResultContract.ResultState)var2;
               var4 = false;
               ImageData var8 = $this$with.getCaptureData();
               List tempList = (var8 != null ? var8.getByteArray() : null) != null ? CollectionsKt.listOf($this$with.getCaptureData().getByteArray()) : $this$with.getImageList();
               Flow var9 = ResultViewModel.this.uploadImagesUseCase.invoke($this$with.getPostId(), tempList);
               this.label = 1;
               var10000 = FlowKt.first(var9, this);
               if (var10000 == var7) {
                  return var7;
               }
               break;
            case 1:
               var4 = false;
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break;
            case 2:
               var4 = false;
               ResultKt.throwOnFailure($result);
               return Unit.INSTANCE;
            default:
               throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
         }

         List res = (List)var10000;
         ResultViewModel var10 = ResultViewModel.this;
         this.label = 2;
         if (var10.registerPostWithImage(res, this) == var7) {
            return var7;
         } else {
            return Unit.INSTANCE;
         }
      }

      @NotNull
      public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
         Intrinsics.checkNotNullParameter(completion, "completion");
         Function2 var3 = new <anonymous constructor>(completion);
         return var3;
      }

      public final Object invoke(Object var1, Object var2) {
         return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
      }
   }), 2, (Object)null);

디컴파일된 자바 코드를 보면 switch 문으로 각 중단 가능 지점에 대해 라벨링을 해놓은 것을 볼 수 있다. ResultKt.throwOnFailure($result 를 통해 오류가 발생했는지 확인하고, this.label = 1; 과 같이 상태값을 증가시켜 다음 실행 지점을 가리키는 것을 볼 수 있다.

정리

컴파일러는 함수 내부에 State Machine 클래스를 만들고, 각 중단 지점에 대해 라벨링을 하여 유한 상태를 나타낸다. 함수는 switch문을 사용해 각각의 상태값에 따라 실행될 지점을 분리한다. 코루틴이 재개될때 필요한 정보들이 Continuation 에 저장됨으로써 스레드를 블럭시키지 않고도 일시 중단이 가능하다.