Update internal SharedFlow comments by dkhalanskyjb · Pull Request #4591 · Kotlin/kotlinx.coroutines

dkhalanskyjb · 2025-12-12T10:17:44Z

Diff in the JS code (above is the result before the change): diff build?/compileSync/js/test/testDevelopmentExecutable/kotlin/kotlinx-coroutines-core.js 13594c13594,13595 < if (compare(element.index_1, new Long(0, 0)) >= 0 && compare(element.index_1, newMinCollectorIndex) < 0) --- > var containsArg = element.index_1; > if (compare(new Long(0, 0), containsArg) <= 0 ? compare(containsArg, newMinCollectorIndex) < 0 : false)

Fixes #4583 See the `SharedFlow.testCancellingSubscriberAndEmitterWithNoBuffer` test for the problematic scenario, which may arise when the next emitter and the last collector get cancelled simultaneously. After the first iteration of the test, the internal state of the `SharedFlow` was: * buffer = [null, NO_VALUE] * replayIndex = 2 * minCollectorIndex = 1 * bufferSize = 1 * queueSize = 0 * nCollectors = 0 * nextIndex = 0 This is clearly invalid, as `bufferSize` can not be anything but 0 with the default `MutableSharedFlow` configuration. The problem was within the logic that handles unsubscription from a `SharedFlow` without a buffer if the first emitter is also cancelled. There were two separate checks for the cancellation case-- importantly, in that order: - If the last collector was cancelled, the part of the buffer outside the replay buffer was erased, since no one was going to consume it. This was achieved by ordering the start of the buffer to be at *right buffer boundary*. - If the next emitter was cancelled *and* the buffer is disabled, the *right boundary* of the empty buffer was moved past the cancelled emitter. Taken together, these two checks conspired to break the invariants: if *both* conditions are true, then first, the buffer was set to be empty, and then, its right boundary was shifted, resulting in a non-empty buffer when it should be impossible logically.

LouisCAD · 2025-12-14T03:15:50Z

+/**
+ * A common data structure for `StateFlow` and `SharedFlow`.
+ */
+internal abstract class AbstractSharedFlow<This, S : AbstractSharedFlowSlot<This>> : SynchronizedObject() {


Is This a new Kotlin feature to refer to the type of the instance's subclass?

It's only a generic parameter name. Could as well be named T. That said, I was also startled when I saw This. Possibly rename into something less misleading.

Nope, you can read it as T, I just decided to use a more accurate name.

Oh, I just love Github's time travelling machine :)

murfel · 2025-12-12T19:32:46Z

+     * `null` by default, created on demand.
+     * Each cell is also `null` by default, and the specific slot object is [created][createSlot] on demand.
+     * The whole array being `null` or a cell being `null` is equivalent to the cell not being
+     * [*allocated*][AbstractSharedFlowSlot.allocateLocked]--not to be confused with memory allocation, this means


Allocate/free slot was misleading naming, but at least it was consistent.

The current snapshot of all comments uses: register, inhabit, allocate to mean allocate.

Consider picking one term to describe comments. Possibly even replace it in code as well, so there's no need for a lengthy explanation.

Here's some options:
inhabit / free
register / deregister (a bit tricky because you can both register a collector and a slot)
inhabited slot / registered slot / occupied slot / assigned slot

(Another reason for this is because "allocate" used in its true sense at least twice in this file, on L81 and L87)

murfel · 2025-12-12T20:16:04Z

    protected abstract fun createSlotArray(size: Int): Array<S?>

-    @Suppress("UNCHECKED_CAST")
+    /** Register a new collector and return its newly allocated slot. A slot may be [created][createSlot] or reused. */


"may be either"

murfel · 2025-12-12T21:22:58Z

@@ -75,7 +125,8 @@ internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : Sync
            subscriptionCount = _subscriptionCount // retrieve under lock if initialized
            // Reset next index oracle if we have no more active collectors for more predictable behavior next time


nit: you removed the other mention/introduction of the term oracle, so now this comment is less clear

The word oracle was not the best choice, but it was used in the same sense as "hint" in C++ map.emplace_hint. No action, just saying.

murfel · 2025-12-12T21:36:22Z

        // take into account a special case of sync shared flow that can go past 1st queued emitter
        if (bufferCapacity == 0 && queueSize > 0) newMinCollectorIndex++
        forEachSlotLocked { slot ->
-            @Suppress("ConvertTwoComparisonsToRangeCheck") // Bug in JS backend


nit: could you get rid of this now irrelevant suppress in all of the codebase (1 other usage)

murfel · 2025-12-12T21:53:49Z

+    /** A `StateFlow` representing [nCollectors], potentially with some delay. A user-visible API. */
    val subscriptionCount: StateFlow<Int>
        get() = synchronized(this) {
            // allocate under lock in sync with nCollectors variable


allocate -> initialize

murfel · 2025-12-12T22:03:41Z

            }
        }

+    /** Allocate a new implementation-representation of a collector, but do not register it anywhere yet. */


I understand the function's meaning but I can't parse "implementation-representation"

murfel · 2025-12-12T22:06:16Z

            }
        }

+    /** Allocate a new implementation-representation of a collector, but do not register it anywhere yet. */


Avoid "allocate" since it's implementation defined where the implementer gets their slots from. There may not be any allocations. Also, I'm not sure if it is common to refer to memory allocations in the JVM world?

murfel · 2025-12-15T10:01:48Z

+/**
+ * A common data structure for `StateFlow` and `SharedFlow`.
+ */
+internal abstract class AbstractSharedFlow<This, S : AbstractSharedFlowSlot<This>> : SynchronizedObject() {


It's only a generic parameter name. Could as well be named T. That said, I was also startled when I saw This. Possibly rename into something less misleading.

murfel · 2025-12-15T12:11:51Z

+    /**
+     * A good starting index for looking for the next non-*allocated* slot in [slots].
+     *
+     * It is not guaranteed that this slot will not be *allocated*, nor is it guaranteed that it will be the first
+     * non-*allocated* slot.
+     * This is just a heuristic to have a better guess in common scenarios.
+     */
+    private var nextIndex = 0


This comment says that nextIndex is a "good starting index" / "heuristics" / "guess" but doesn't say how this heuristic is used and what this even means. It does provide two options of what it is definitely not, which is useful, but just would be nice to define the object.

The next collector will be assigned to the first free slot at nextIndex or on the cyclical right of nextIndex.

murfel · 2025-12-15T12:39:22Z

@@ -10,19 +10,66 @@ import kotlin.jvm.*
 @JvmField
 internal val EMPTY_RESUMES = arrayOfNulls<Continuation<Unit>?>(0)


First batch of comments. Feel free to address now or wait for the rest to come.

murfel · 2025-12-16T15:55:21Z

+     * These continuations represent suspended emitters that were waiting for the slowest collector to move on
+     * so that the next value can be placed into the buffer.


for the slowest collector (assigned to this slot) to move on

so that the next value -> so that their values

murfel · 2025-12-16T15:55:28Z

 * get values from the buffer without suspending emitters. The buffer space determines how much slow subscribers
 * can lag from the fast ones. When creating a shared flow, additional buffer capacity beyond replay can be reserved
 * using the `extraBufferCapacity` parameter.
- * 


Could you also fix all other IDE warnings/grammar/style in both files?

murfel · 2025-12-16T15:55:30Z


 internal class SharedFlowSlot : AbstractSharedFlowSlot<SharedFlowImpl<*>>() {
+    /**
+     * The first index in the buffer that was not yet consumed by this collector,


nit: that -> which, was not yet -> has not yet been

murfel · 2025-12-16T15:56:14Z

+     * This field is used by the collector each time when it can not acquire the next value and suspends while
+     * waiting for the next one.


Used how? Please formulate more precisely, otherwise it adds more mystery than removes. ("The collector's continuation is saved/written/preserved here" or something)

"cannot", "suspends waiting for the next value to arrive."

murfel · 2025-12-16T16:05:37Z

head + totalSize is off by 1 character, could you move them?

murfel · 2025-12-17T10:34:17Z

+        /* Walk through at most `maxResumeCount` emitters, moving their values to the buffer,
+        * zeroing out the cells the emitters were occupying, and storing their continuations in an array that
+        * will be returned from this function and processed once the lock gets released. */


[IDE hint] Long sentences (41 words here) are harder to read, according to the research; consider splitting

A human confirms. Consider making it into a list, or somehow else rewriting it.

murfel · 2025-12-17T10:59:42Z

make -> mark

murfel · 2025-12-17T11:28:01Z

-) : AbstractSharedFlow<SharedFlowSlot>(), MutableSharedFlow<T>, CancellableFlow<T>, FusibleFlow<T> {
+) : AbstractSharedFlow<SharedFlowImpl<*>, SharedFlowSlot>(), MutableSharedFlow<T>, CancellableFlow<T>, FusibleFlow<T> {
    /*
        Logical structure of the buffer


Suggestion: buffer is a confusing naming and is used to mean 3 different things:

the buffer array of buffered values (including the replay cache) and queued emitters

the buffer of buffered values including the replay cache

the buffer of buffered values excluding the replay cache

2 and 3 is more or less consistent in the kdoc and code and steer towards 2

But the meanings 1 and 2 are used intermittently throughout the code. Could we rename the buffer in this sense into the buffer array? To refer to the code representation.

murfel · 2025-12-17T11:45:13Z

+        /** Represents how many values are stored in the buffer after processing the emitters.
+         * The value may be larger than [bufferCapacity] if [nCollectors] was 0,
+         * but the `minOf(replay)` will clamp this value to a valid one, since `replay <= bufferCapacity`. */


This value is always non-smaller than [bufferCapacity], and could be larger in another case. The head hasn't moved and still points to the original slowest possible collector. The newBufferEndIndex has started out as bufferEndIndex == head + bufferSize (at max that would be head + bufferCapacity) and could only increase.

Also "Represents how many values" -> "The number of values stored in the buffer after processing the emitters."

murfel · 2025-12-17T12:02:19Z

+        /** If [nCollectors] was 0, there are no active collector slots,
+         * and the invariant dictates that [minCollectorIndex] should be [bufferEndIndex] (`replayIndex + bufferSize`).
+         * Ensure this directly. */


The original comment asserted that this invariant can in fact be broken, and thus this assignment is indeed required, and not just some unnecessary duct tape.

I don't understand this point. The invariant can't be broken, we're under the lock. However, to ensure the invariant stays non-broken, this assignment is required.

murfel · 2025-12-17T14:15:23Z

        val head = head
-        var newMinCollectorIndex = head + bufferSize
+        var newMinCollectorIndex = head + bufferSize // = this.bufferEndIndex
        // take into account a special case of sync shared flow that can go past 1st queued emitter


Could you also replace this cryptic "sync shared flow" to "shared flow with no buffer", here and below, to consistently call this situation the same name.

dkhalanskyjb added 3 commits December 10, 2025 13:05

Improve the documentation of SharedFlow internals

0352cca

dkhalanskyjb requested a review from murfel December 12, 2025 10:17

dkhalanskyjb mentioned this pull request Dec 12, 2025

CoroutineScope cancellation affects MutableSharedFlow in unexpected way #4583

Closed

LouisCAD reviewed Dec 14, 2025

View reviewed changes

murfel reviewed Dec 15, 2025

View reviewed changes

murfel requested changes Dec 17, 2025

View reviewed changes

murfel reviewed Dec 17, 2025

View reviewed changes

murfel self-requested a review December 17, 2025 16:23

Address the review

6a559ef

dkhalanskyjb mentioned this pull request Mar 31, 2026

Fix SharedFlow entering an invalid state #4650

Merged

dkhalanskyjb changed the title ~~Fix SharedFlow entering an invalid state~~ Update internal SharedFlow comments Mar 31, 2026

dkhalanskyjb marked this pull request as draft March 31, 2026 11:48

		@@ -75,7 +125,8 @@ internal abstract class AbstractSharedFlow<S : AbstractSharedFlowSlot<*>> : Sync
		subscriptionCount = _subscriptionCount // retrieve under lock if initialized
		// Reset next index oracle if we have no more active collectors for more predictable behavior next time

		@@ -10,19 +10,66 @@ import kotlin.jvm.*
		@JvmField
		internal val EMPTY_RESUMES = arrayOfNulls<Continuation<Unit>?>(0)

		* These continuations represent suspended emitters that were waiting for the slowest collector to move on
		* so that the next value can be placed into the buffer.

		* This field is used by the collector each time when it can not acquire the next value and suspends while
		* waiting for the next one.

Conversation

dkhalanskyjb commented Dec 12, 2025

Uh oh!

LouisCAD Dec 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

LouisCAD Dec 14, 2025 •

edited

Loading