Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/channels/BufferedChannel.kt (Kotlin)

Code before:
     * side, [receiveResult] stores the element retrieved by [hasNext]
     * (or a special [CHANNEL_CLOSED] token if the channel is closed).
     *
     * The [invoke] function is a [CancelHandler] implementation,
     * which requires knowing the [segment] and the [index] in it
     * that specify the location of the stored iterator.
     *
     * To resume the suspended [hasNext] call, a special [tryResumeHasNext]
     * function should be used in a way similar to [CancellableContinuation.tryResume]
     * and [SelectInstance.trySelect]. When the channel becomes closed,
     * [tryResumeHasNextOnClosedChannel] should be used instead.
     */
    private inner class BufferedChannelIterator : ChannelIterator<E>, Waiter {
        /**
         * Stores the element retrieved by [hasNext] or
         * a special [CHANNEL_CLOSED] token if this channel is closed.
         * If [hasNext] has not been invoked yet, [NO_RECEIVE_RESULT] is stored.
         */
        private var receiveResult: Any? = NO_RECEIVE_RESULT

        /**
         * When [hasNext] suspends, this field stores the corresponding
         * continuation. The [tryResumeHasNext] and [tryResumeHasNextOnClosedChannel]
         * function resume this continuation when the [hasNext] invocation should complete.
         */
        @BenignDataRace // See its uses for the explanation of the race
        private var continuation: CancellableContinuationImpl<Boolean>? = null

        // `hasNext()` is just a special receive operation.
        override suspend fun hasNext(): Boolean =
            receiveImpl( // <-- this is an inline function
                // Do not create a continuation until it is required;
                // it is created later via [onNoWaiterSuspend], if needed.
                waiter = null,
                // Store the received element in `receiveResult` on successful
                // retrieval from the buffer or rendezvous with a suspended sender.
                // Also, inform the `BufferedChannel` extensions that
                // the synchronization of this receive operation is completed.
                onElementRetrieved = { element ->
                    this.receiveResult = element
                    true
                },
                // As no waiter is provided, suspension is impossible.
                onSuspend = { _, _, _ -> error("unreachable") },
                // Return `false` or throw an exception if the channel is already closed.
                onClosed = { onClosedHasNext() },
                // If `hasNext()` decides to suspend, the corresponding
                // `suspend` function that creates a continuation is called.
                // The tail-call optimization is applied here.
                onNoWaiterSuspend = { segm, i, r -> return hasNextOnNoWaiterSuspend(segm, i, r) }
            )

Review comment (suggestion):
I think we should explicitly list the races right next to the annotation. This way, if a new race is discovered, we won't look at the annotation and go, "oh yeah, we know about this race, it's benign, no worries."

Code after:
     * side, [receiveResult] stores the element retrieved by [hasNext]
     * (or a special [CHANNEL_CLOSED] token if the channel is closed).
     *
     * The [invoke] function is a [CancelHandler] implementation,
     * which requires knowing the [segment] and the [index] in it
     * that specify the location of the stored iterator.
     *
     * To resume the suspended [hasNext] call, a special [tryResumeHasNext]
     * function should be used in a way similar to [CancellableContinuation.tryResume]
     * and [SelectInstance.trySelect]. When the channel becomes closed,
     * [tryResumeHasNextOnClosedChannel] should be used instead.
     */
    private inner class BufferedChannelIterator : ChannelIterator<E>, Waiter {
        /**
         * Stores the element retrieved by [hasNext] or
         * a special [CHANNEL_CLOSED] token if this channel is closed.
         * If [hasNext] has not been invoked yet, [NO_RECEIVE_RESULT] is stored.
         */
        private var receiveResult: Any? = NO_RECEIVE_RESULT

        /**
         * When [hasNext] suspends, this field stores the corresponding
         * continuation. The [tryResumeHasNext] and [tryResumeHasNextOnClosedChannel]
         * function resume this continuation when the [hasNext] invocation should complete.
         *
         * This property is the subject to bening data race:
         * It is nulled-out on both completion and cancellation paths that
         * could happen concurrently.
         */
        @BenignDataRace
        private var continuation: CancellableContinuationImpl<Boolean>? = null

        // `hasNext()` is just a special receive operation.
        override suspend fun hasNext(): Boolean =
            receiveImpl( // <-- this is an inline function
                // Do not create a continuation until it is required;
                // it is created later via [onNoWaiterSuspend], if needed.
                waiter = null,
                // Store the received element in `receiveResult` on successful
                // retrieval from the buffer or rendezvous with a suspended sender.
                // Also, inform the `BufferedChannel` extensions that
                // the synchronization of this receive operation is completed.
                onElementRetrieved = { element ->
                    this.receiveResult = element
                    true
                },
                // As no waiter is provided, suspension is impossible.
                onSuspend = { _, _, _ -> error("unreachable") },
                // Return `false` or throw an exception if the channel is already closed.
                onClosed = { onClosedHasNext() },
                // If `hasNext()` decides to suspend, the corresponding