Fix flaky shared flow subscription - #4489
Conversation
|
The change makes sense, thanks! However, the failing tests are important. This changes the behavior of other operators non-trivially, and we need to evaluate how big the change is. Could you please classify the tests into those that fail reasonably and those that should ideally still pass? Then, we'll figure out if we need to isolate the change to just |
| fun testBaseline() = | ||
| checkBuffer(-1) { this } |
There was a problem hiding this comment.
Unrelated: Idea gives a warning on this: Flow is constructed but not used. That's a false positive, right? I'll report it if so.
| import kotlinx.coroutines.testing.* | ||
| import kotlinx.coroutines.* | ||
| import kotlinx.coroutines.channels.* | ||
| import kotlinx.coroutines.testing.* |
There was a problem hiding this comment.
Unrelated: L1: Do we need the suppression on Line 1? I don't quite understand how it is relevant here.
…ed-flow-subscription-impl
|
I aimed to fix all tests but these (conflicting with #4493)
|
|
@dkhalanskyjb I don't exactly understand why the build failed, could you take a look? |
|
The test The chain of events is:
|
|
How do you understand which test hangs from this page? https://teamcity.jetbrains.com/buildConfiguration/KotlinTools_KotlinxCoroutines_BuildWindows/5892908 |
|
Click the red exclamation mark here: 
Scroll up paste the thread dump at the moment of termination, to the exact moment the build was cancelled, and unfold the 
|
| val desired = context[ContinuationInterceptor] | ||
| val start = if (desired == null || desired == current) { | ||
| CoroutineStart.UNDISPATCHED | ||
| } else CoroutineStart.ATOMIC |
There was a problem hiding this comment.
Please consider using a consistent code style.
We have some legacy places where one branch is wrapped into braces while another is not, but it's rather confusing than intentional
| override suspend fun collect(collector: FlowCollector<T>): Unit = | ||
| coroutineScope { | ||
| collector.emitAll(produceImpl(this)) | ||
| // If upstream and collect have the same dispatcher, launch the `produce` coroutine undispatched. |
There was a problem hiding this comment.
Previously, produceImpl called markConsumed, but now it is no longer the case. ChannelAsFlow.collect mentions it explicitly.
Probably can be broken by a double collection
There was a problem hiding this comment.
val channel = Channel<Int>(1).also { it.send(1) }
val flow = channel.consumeAsFlow().buffer()
launch { flow.collect() }
channel.close()
flow.collect()
Something like this should fail, but it does not, so I assume this is a regression.
Ditto for collecting non-closing channel, it probably hangs instead of failing fast
| scope.produce(context, produceCapacity, onBufferOverflow, start = CoroutineStart.ATOMIC, block = collectToFun) | ||
| produceImplInternal(scope, CoroutineStart.ATOMIC) | ||
|
|
||
| internal open fun produceImplInternal(scope: CoroutineScope, start: CoroutineStart): ReceiveChannel<T> = |
There was a problem hiding this comment.
Probably shouldn't be open, there is a single implementation of this method
| val start = if (desired == null || desired == current) { | ||
| CoroutineStart.UNDISPATCHED | ||
| } else CoroutineStart.ATOMIC | ||
| collector.emitAll(produceImplInternal(this, start)) |
There was a problem hiding this comment.
Another problem:
produceImpl was overridden in inheritors, produceImplInternal is not, so any behavioural divergence might result in a user-visible regression.
For example, .merge().conflate() chain now behaves differently, depending on whether it is collected or produceIn'd because they handle onBufferOverflow differently.
3 participants
#4488