Hello there, In our project we are observing an i...

# coroutines

[Anton Krutko]

Hello there,

In our project we are observing an issue with consuming the data from MutableSharedFlow. Under certain conditions we are cancelling the scope in which we produce and collect data. Then we launch data emission and consuming on the new scope, but on the same global instance of MutableSharedFlow.
The issue we face is that after one or several restarts emitted items are no longer delivered to the collector. Below is the minimal example to reproduce the issue. However, reproduction of this issue is greatly affected by the work load performed (or time spent) in the collector.

We managed to reproduce this issue on desktop JVM and on Android. Do you see any obvious mistakes we made or spots we missed?
Note: correctness of counter values themselves is not important in that example. Coroutines version 1.10.2 and 1.8.0.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach

var counter = 0L
val mutableSharedFlow = MutableSharedFlow<Long>()

var cs: CoroutineScope? = null
fun start() {
    cs?.cancel()
    cs = CoroutineScope(Dispatchers.IO + SupervisorJob())

    mutableSharedFlow.onEach {
        delay(50)
        println(it)
    }.launchIn(cs!!)

    cs!!.launch {
        while (currentCoroutineContext().isActive) {
            mutableSharedFlow.emit(counter)
            counter++
        }
    }
}

suspend fun main() {
    while (true) {
        start()
        println("restart, counter = $counter")

        delay(3000)
    }
}

The output we see:

....
47
48
49
50
51
52
53
54
restart, counter = 56
restart, counter = 56
restart, counter = 56
restart, counter = 56
restart, counter = 56
... // proper behavior (emissions happening) is never restored after this point

Any help will be greatly appreciated.

[louiscad]

Have you considered the xxxLatest (e.g. collectLatest) operators for the consumers?

[Anton Krutko]

Just tried collectLatest. It doesn't seem to help. The understanding, however, of the core reason for this issue to happen is very crucial to us. Workaround is not that important.

[Mohammad Fallah]

I know that if you have even one extraBuffer it works as expected, but I don't know why?
It seems this occurs only when no buffer is used and both the collector and emitter are cancelled simultaneously while the collector is suspended.

[louiscad]

If you think about it, a SharedFlow can have multiple collectors.

If a collector is slower than the others to complete processing a value, you'd need to cache the 2nd (or later) emitted value while the slow collector is still not done processing the previous one. That's what replay or extraBufferCapacity are for.

Knowing that, having a special behavior for the case of a single collector would be surprising as that'd be a kind of unintended side effect.

For example, imagine adding a second collector that saves data to a usage statistics/analytics local database (to be flushed later). If the storage saving is done in a suspend or blocking way (i.e. not memory buffered until another coroutine consumes it), it can become slow for external reasons (e.g. system or app update, or file transfer reducing I/O capacity).

Would you want that to affect the behavior in an unpredictable and unchosen way?

I think you wouldn't.

So, to make it explicit, and avoid any weirdness, you need an extra buffer if there's no replay, to be fed for slow collectors, and for fast collectors too, since being fast can never be guaranteed for the library code.

[Anton Krutko]

That is a valid point indeed and offers a good way to deal with slow and fast collectors.

However, in terms of behavior there should be no differences whether MutableSharedFlow has an extra buffer or not. If there is no extra buffer configured cancellation mechanism should still work in an expected way.

[Anton Krutko]

Just updating here. Filed issue https://github.com/Kotlin/kotlinx.coroutines/issues/4583 seems to be confirmed and will be fixed with a new library version.