How to modify Kotlin StateFlow content in a multi-threaded environment?

Question

I have a MutableStateFlow which holds an immutable data structure.

Contents of this structure have to be changed by appending values (StateFlow.value has to be read, a new object has to be created based on the existing one and then emitted).

In a multi-threaded environment it is possible, that after one thread has read the content of the SharedFlow, but before the modification was written back, the content will be modified by another thread. Then the first thread will override the update.

How can I ensure that modifications are never overwritten in a multi-threaded environment?

So far I see three ways to solve this:

• Use a queue with a worker for writing
• Use a lock
• Use MutableStateFlow.compareAndSet in a loop (keep trying until success)

Are there any other options? Am I missing some extension function which I can use?

Answer 1

Since kotlinx.coroutines 1.5.1 https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.5.1 we can use MutableStateFlow.update. See https://github.com/Kotlin/kotlinx.coroutines/issues/2720 for details.

Answer 2

You can use a fair ReentrantLock, and just wrap your read and modifications inside with lock as in

val lock = ReentrantLock(true) //true makes it fair, keeps the order
lock.withLock{ 
   //all the code written inside this block will be synchronized.
}

I suggest this over other approaches as it is easy to implement, and also with worker queues, you have to continuously listen to the queue to get the next item and looping continuously is an expensive solution for something this simple.

Answer 3

If you are working with it using coroutines, you can use a suspend function with a Mutex lock. It is like a non-reentrant Lock, but suspends instead of blocking when waiting for withLock.

private val _myStateFlow: MutableStateFlow<List<Foo>> = ...
val myStateFlow: StateFlow<List<Foo>> get() = _myStateFlow
private val mutex = Mutex()

suspend fun addValue(foo: Foo) {
    mutex.withLock {
        _myStateFlow.value = _myStateFlow.value + foo
    }
}

If you need to add values from threads instead of coroutines, I'd go with the ReentrantLock from the other answer.