3. Kotlin Coroutine - Concurrency issues

https://medium.com/androiddevelopers/coroutines-on-android-part-iii-real-work-2ba8a2ec2f45

 

코루틴은 어떻게 코드를 단순화하고, main-safety 를 제공하고, leaking work 를 피할 수 있는지 등을 살펴봤다. 

 

실제 어떤 작업에 코루틴을 써먹는게 좋냐? 

1. One shot request : 호출 될 때 마다 실행되는 작업. 결과가 준비되면 완료된다. 

2. Streaming requests : 변화를 지속적으로 관찰하고 호출자에 이를 report 한다. 첫 결과가 준비되더라도 끝나지 않는다. 

 

One shot request 의 구체적인 예시를 살펴보자. 

- 정렬된 목록을 사용자에게 보여주는 경우

class ProductsViewModel(val productsRepository: ProductsRepository): ViewModel() {
   private val _sortedProducts = MutableLiveData<List<ProductListing>>()
   val sortedProducts: LiveData<List<ProductListing>> = _sortedProducts

   /**
    * Called by the UI when the user clicks the appropriate sort button
    */
   fun onSortAscending() = sortPricesBy(ascending = true)
   fun onSortDescending() = sortPricesBy(ascending = false)

   private fun sortPricesBy(ascending: Boolean) {
       viewModelScope.launch {
           // suspend and resume make this database request main-safe
           // so our ViewModel doesn't need to worry about threading
           _sortedProducts.value =
                   productsRepository.loadSortedProducts(ascending)
       }
   }
}

 

뷰모델에서 목록을 정렬하고 이를 업데이트 한다 

class ProductsRepository(val productsDao: ProductsDao) {

  /**
    * This is a "regular" suspending function, which means the caller must
    * be in a coroutine. The repository is not responsible for starting or 
    * stopping coroutines since it doesn't have a natural lifecycle to cancel
    * unnecessary work.
    *
    * This *may* be called from Dispatchers.Main and is main-safe because
    * Room will take care of main-safety for us.
    */
   suspend fun loadSortedProducts(ascending: Boolean): List<ProductListing> {
       return if (ascending) {
           productsDao.loadProductsByDateStockedAscending()
       } else {
           productsDao.loadProductsByDateStockedDescending()
       }
   }
}

 

Repository 레이어는 AAC 에서 선택적인 부분인데, 만약 있다면 suspend function 을 노출하도록 하기. 

Repository 는 natural lifecycle 이 없어서, (프레임웍에서 부여하는 라이프사이클 같은게 없는 일반 객체라서) 

작업을 clean up 할 방법이 없다. 결과적으로 repository 에서 시작된 코루틴은 기본적으로 leak 하게 된다. 

 

A repository should prefer to expose regular suspend functions that are main-safe 

 

@Dao
interface ProductsDao {
   // Because this is marked suspend, Room will use it's own dispatcher
   //  to run this query in a main-safe way.
   @Query("select * from ProductListing ORDER BY dateStocked ASC")
   suspend fun loadProductsByDateStockedAscending(): List<ProductListing>

   // Because this is marked suspend, Room will use it's own dispatcher
   //  to run this query in a main-safe way.
   @Query("select * from ProductListing ORDER BY dateStocked DESC")
   suspend fun loadProductsByDateStockedDescending(): List<ProductListing>
}

 

Room 내부적으로 Dispatchers.IO 를 사용하고 있어서, DAO 에 정의된 suspend 함수들은 main-safe 하다. 그래서 loadxxx 들을 main-thread 에서 호출해도 무방한데. 조심할 것은 리스트 불러온 후에 정렬 등 할때 메인 스레드 블럭하지 않도록 해야한다. 

Note: Room uses its own dispatcher to run queries on a background thread. Your code should not use withContext(Dispatchers.IO) to call suspending room queries. It will complicate the code and make your queries run slower.

 

👉 위와 같은 레이어들을 통해서 One shot request 패턴을 구축할 수 있다. 각 레이어의 역할은: 

ViewModel : 코루틴을 시작시키고, 이를 관리한다 (취소 등). expensive 한 연산들은 다른 레이어로 위임한다. UI 로 데이터를 쏴준다. 

Repository : 일반 suspend 함수를 노출해서 데이터 접근 인터페이스 제공. 전형적으로는 자신만의 long lived coroutine 을 시작하지 않는데, 이유는 이것을 취소할 방법이 없기 때문이다. (객체 해제되는 타이밍이 따로 없어서). Repository 에서 expensive 한 연산을 할때는 withContext 를 사용해서 main-safe 한 인터페이스를 제공하자. 

Data layer : Network 또는 Database. 항상 일반 suspend function 을 노출한다. 이 함수들은 main-safe 해야한다. 

 

❗️버그 리포트 

어떤 버튼을 아주 아주 빠르게 눌러서 사용자가 코루틴을 여러개 런칭하여, 의도하지 않은 정렬 리스트를 받게되는 케이스. 

 

It turns out the result shown isn’t actually the “result of the sort,” it’s actually the result of the “last sort to complete.” When the user spams the button — they start multiple sorts at the same time and they can finish in any order!

 

When starting a new coroutine in response to a UI event, consider what happens if the user starts another before this one completes.

 

이것은 동시성 버그이며, coroutine 과는 상관 없다. 콜백, rx, ExecutorService 등을 이용해서 마찬가지로 마주하게 될 버그이다. 

 

해결 방법 

1. Disable the button : 사실상 가장 근본적인 해결책. 

// Solution 0: Disable the sort buttons when any sort is running

class ProductsViewModel(val productsRepository: ProductsRepository): ViewModel() {
   private val _sortedProducts = MutableLiveData<List<ProductListing>>()
   val sortedProducts: LiveData<List<ProductListing>> = _sortedProducts
  
   private val _sortButtonsEnabled = MutableLiveData<Boolean>()
   val sortButtonsEnabled: LiveData<Boolean> = _sortButtonsEnabled
  
   init {
       _sortButtonsEnabled.value = true
   }

   /**
    * Called by the UI when the user clicks the appropriate sort button
    */
   fun onSortAscending() = sortPricesBy(ascending = true)
   fun onSortDescending() = sortPricesBy(ascending = false)

   private fun sortPricesBy(ascending: Boolean) {
       viewModelScope.launch {
           // disable the sort buttons whenever a sort is running
           _sortButtonsEnabled.value = false
           try {
               _sortedProducts.value =
                       productsRepository.loadSortedProducts(ascending)
           } finally {
               // re-enable the sort buttons after the sort is complete
               _sortButtonsEnabled.value = true
           }
       }
   }
}

 

버튼을 꺼버린다. 

 

다른 이야기지만, 위 코드는 메인 스레드에서 코루틴을 시작하는 것에 대한 이점을 잘 보여준다. 버튼 enable, disable 로직을 다른 suspend 함수와 나란히 배치할 수 있음. Context 를 전환하지 않아도 됨. 

 

Important: This code shows a major advantage of starting on main — the buttons disable instantly in response to a click. If you switched dispatchers, a fast-fingered user on a slow phone could send more than one click!

 

버튼을 enable 상태로 두고 문제를 해결할 방법은 없나 ?

 

1. 다른 작업을 더 시작하기 전에 기존 작업을 취소한다 

2. 다음 작업을 큐잉해서 이전 작업이 끝나길 기다린 후 실행한다 

3. 이미 실행중인 작업이 있다면 그 작업에 참여한다

 

1번 : canclePreviousThenRun 

// Solution #1: Cancel previous work

// This is a great solution for tasks like sorting and filtering that
// can be cancelled if a new request comes in.

class ProductsRepository(val productsDao: ProductsDao, val productsApi: ProductsService) {
   var controlledRunner = ControlledRunner<List<ProductListing>>()

   suspend fun loadSortedProducts(ascending: Boolean): List<ProductListing> {
       // cancel the previous sorts before starting a new one
       return controlledRunner.cancelPreviousThenRun {
           if (ascending) {
               productsDao.loadProductsByDateStockedAscending()
           } else {
               productsDao.loadProductsByDateStockedDescending()
           }
       }
   }
}

 

cancelPreviousThenRun 구현을 살펴보면 

// see the complete implementation at
// https://gist.github.com/objcode/7ab4e7b1df8acd88696cb0ccecad16f7
suspend fun cancelPreviousThenRun(block: suspend () -> T): T {
   // If there is an activeTask, cancel it because it's result is no longer needed
   activeTask?.cancelAndJoin()
   
   // ...

 

2번 : Queue the next work 

큐잉을 해서 한번에 하나만 실행되도록 하자. - 항상 작동하는 동시성 버그 해결책 

 

// Solution #2: Add a Mutex

// Note: This is not optimal for the specific use case of sorting
// or filtering but is a good pattern for network saves.

class ProductsRepository(val productsDao: ProductsDao, val productsApi: ProductsService) {
   val singleRunner = SingleRunner()

   suspend fun loadSortedProducts(ascending: Boolean): List<ProductListing> {
       // wait for the previous sort to complete before starting a new one
       return singleRunner.afterPrevious {
           if (ascending) {
               productsDao.loadProductsByDateStockedAscending()
           } else {
               productsDao.loadProductsByDateStockedDescending()
           }
       }
   }
}

 

Mutext 를 사용해서 한번에 하나만 수행되도록 한다 

  suspend fun <T> afterPrevious(block: suspend () -> T): T {
        // Before running the block, ensure that no other blocks are running by taking a lock on the
        // mutex.

        // The mutex will be released automatically when we return.

        // If any other block were already running when we get here, it will wait for it to complete
        // before entering the `withLock` block.
        mutex.withLock {
            return block()
        }
    }

 

Whenever a new sort comes in, it uses a instance of SingleRunner to ensure that only one sort is running at a time.

It uses a Mutex, which is a single ticket (or lock), that a coroutine must get in order to enter the block. If another coroutine tried while one was running, it would suspend itself until all pending coroutines were done with the Mutex.

A Mutex lets you ensure only one coroutine runs at a time — and they will finish in the order they started.

 

3번: Join previous work 

만약 새로운 요청이 이전과 완전히 동일한 요청을 보낼 경우 유요한 솔루션. (매개변수 등 없고..?) 

joinPreviousOrRun 

class ProductsRepository(val productsDao: ProductsDao, val productsApi: ProductsService) {
   var controlledRunner = ControlledRunner<List<ProductListing>>()

   suspend fun fetchProductsFromBackend(): List<ProductListing> {
       // if there's already a request running, return the result from the 
       // existing request. If not, start a new request by running the block.
       return controlledRunner.joinPreviousOrRun {
           val result = productsApi.getProducts()
           productsDao.insertAll(result)
           result
       }
   }
}

 

이미 실행중인 요청이 있는 경우 현재 "진행 중인" 요청의 결과를 기다렸다가 새 요청을 실행하는 대신 이를 반환한다. 

// see the complete implementation at
// https://gist.github.com/objcode/7ab4e7b1df8acd88696cb0ccecad16f7#file-concurrencyhelpers-kt-L124

suspend fun joinPreviousOrRun(block: suspend () -> T): T {
    // if there is an activeTask, return it's result and don't run the block
    activeTask?.let {
        return it.await()
    }

    // use a coroutineScope builder to safely start a new coroutine in a suspend function
    return coroutineScope {
        // create a new task to call the block
        val newTask = async {
            block()
        }
        // when the task completes, reset activeTask to null
        newTask.invokeOnCompletion {
            activeTask = null
        }
        // save newTask to activeTask, then wait for it to complete and return the result
        activeTask = newTask
        newTask.await()
    }
}

 

 

요약) 

코루틴을 사용해서 실제 애플리케이션에서 one shot request 패턴을 많이 사용할텐데 

흔히 마주할 수 있는 동시성 이슈가 있다

- 바로, 여러개의 요청이 한꺼번에 많이 일어날 경우, 의도치 않게 앱 상태를 변경시키는 동시성 이슈. 

이를 해결하려면, 

근본적으로 마구잡이로 요청을 보낼 통로를 막아버리는 방법이 있는데, (버튼 disable) 

그렇지 못할 경우 이를 관리할 Helper 클래스 등 도입할 수 있다. 

 

동시성 패턴 세가지를 기반으로 솔루션을 제시한다 

1. cancel previous work -> 진행중인 작업을 취소 때리고 새로운 작업을 시작한다 

2. queue the next work -> mutex 를 사용해서 한번 요청 들어가면 그 요청 끝나고, 그 다음 요청이 가도록 구조화 한다 

3. join previous work -> 새로운 태스크를 시작하지 않고 진행 중인 애를 쫓아간다