Coroutine 공유 상태와 동시성 3편 - Mutex 사용하기, Actors 사용하기

Mutex 사용하기

문제에 대한 상호 배제 솔루션은 모든 모든 공유 상태에 대한 변경을 절대로 동시에 실행되지 않는 critical section*1으로 만들어 보호하는 것이다. 블로킹을 수행하기 위해서는 일반적으로 synchronized나 ReentrantLock을 사용한다. Coroutine의 대체제는 Mutex*2라 불린다. 이는 critical section을 구분하기 위한 lock과 unlock 함수를 가진다. 중요한 차이점은 Mutext.lock()이 일시중단 함수라는 것이다. 이는 Thread를 블록하지 않는다.

mutex.lock(); try { ... } finally { mutex.unlock() } 패턴을 나타내는 withLock 확장함수 또한 있다.

val mutex = Mutex()
var counter = 0

fun main() = runBlocking {
    withContext(Dispatchers.Default) {
        massiveRun {
            // protect each increment with lock
            mutex.withLock {
                counter++
            }
        }
    }
    println("Counter = $counter")
}

📌 전체 코드는 이곳에서 확인할 수 있습니다.

이 예에서 Lock을 거는 것은 세밀해서 비용이 든다. 하지만, 이는 주기적으로 공유 상태를 수정해야 하지만 상태를 제한시킬 수 있는 자연 스레드*3가 없는 일부 상황에서 좋은 선택이 된다. 하지만, 

📖 아래 내용은 독자의 이해를 위해 번역자가 추가한 글입니다.

*1. critical section을 한 번에 하나의 스레드만 접근할 수 있는 코드 블록(섹션)을 뜻한다.

*2. Mutex는 단순한 Lock이다. 

*3. natural thread라서 자연 스레드로 번역했지만, 해당 lock을 수행하는 외부에 선언된 스레드가 없는 경우를 뜻하는 것으로 보인다.

이 글은 Coroutines 공식 문서를 번역한 글입니다.

원문 : Shared mutable state and concurrency - Mutual exclusion

원문 최종 수정 :  2022년 6월 27일

Actors 사용하기

actor는 Coroutine의 조합으로 만들어진 엔티티이다. actor의 상태는 제한되고 Coroutine 속에 캡슐화되며, Channel을 통해 다른 Coroutine과 통신한다. 간단한 actor는 함수로 작성될 수 있지만, 복잡한 상태를 가진 actor는 class로 작성되는 것이 적합하다.

간단한 actor Coroutine 빌더를 통해서 actor의 수신 Channel을  Scope에 편리하게 결합해서 메세지를 수신할수 있다. 또한 발신 Channel을 결과 Job 객체에 결합함으로써 actor에 대한 단일 참조를 통해 제어할 수 있다.

actor을 사용하는 첫 단계는 actor가 처리할 메세지의 class를 정의하는 것이다. Kotlin의 sealed class는 이 목적으로 아주 적합하다. 우리는 sealed class로 CounterMsg를 정의하고, IncCounter 메세지를  counter을 증가시키는데 사용하고 GetCounter 메세지를 값을 가져오는데 사용한다. 후자(GetCounter)는 응답을 보내는 것이 필요하다. 미래에 알려질(통신될) 단일 값을 나타내는 통신 원시값(primitive)인 CompletableDeffered이 이를 위해 사용된다.

// Message types for counterActor
sealed class CounterMsg
object IncCounter : CounterMsg() // one-way message to increment counter
class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() // a request with reply

그 다음 actor Coroutine 빌더를 사용해 actor을 실행시키는 함수를 정의한다.

// This function launches a new counter actor
fun CoroutineScope.counterActor() = actor<CounterMsg> {
    var counter = 0 // actor state
    for (msg in channel) { // iterate over incoming messages
        when (msg) {
            is IncCounter -> counter++
            is GetCounter -> msg.response.complete(counter)
        }
    }
}

전체 코드는 다음과 같다 : 

fun main() = runBlocking<Unit> {
    val counter = counterActor() // create the actor
    withContext(Dispatchers.Default) {
        massiveRun {
            counter.send(IncCounter)
        }
    }
    // send a message to get a counter value from an actor
    val response = CompletableDeferred<Int>()
    counter.send(GetCounter(response))
    println("Counter = ${response.await()}")
    counter.close() // shutdown the actor
}

📌 전체 코드는 이곳에서 확인할 수 있습니다.

어떤 Context에서 actor가 실행되는지는 상관 없다(정확성을 위해). actor는 Coroutine이고 Coroutine은 순차적으로 실행된다. 따라서 특정한 Coroutine에 대한 상태의 제한은 공유된 가변 상태에 대한 해결책으로 동작한다. 실제로 actor는 그들의 비공개 상태를 수정할 수 있지만, 메세지를 통해서만 서로에게 영향을 줄 수 있다.(lock이 필요하지 않도록)

Actor는 로드 시 Lock보다 효율적이다. 항상 해야 할 작업이 있으며, 다른 Context로의 전환이 전혀 필요없기 때문이다.

📖  actor Coroutine 빌더는 produce Coroutine 빌더 두개라는 것을 명심하자. actor는 메세지를 수신하는 Channel과 연관되어 있고, producer는 메세지를 보내는 Channel과 연관되어 있다.*1

📖 아래 내용은 독자의 이해를 위해 번역자가 추가한 글입니다.

*1. Actor는 단일 채널을 통해 메세지를 보내고 수신한다. 또한 Pub-Sub 모델에서 하듯이, 단일 채널이 메세지를 받아 처리하는 Mediator 역할을 하는 것으로 보인다. 

이 글은 Coroutines 공식 문서를 번역한 글입니다.

원문 : Shared mutable state and concurrency - Actors

원문 최종 수정 :  2022년 6월 27일