Almost non-blocking ConflatedBufferedChannel

Signed-off-by: Nikita Koval <[email protected]>

Author: ndkoval

kotlinx-coroutines-core/common/src/channels/BufferedChannel.kt

@@ -68,7 +68,7 @@ internal open class BufferedChannel<E>(
 
     internal val sendersCounter: Long get() = sendersAndCloseStatus.value.sendersCounter
     internal val receiversCounter: Long get() = receivers.value
-    internal val bufferEndCounter: Long get() = bufferEnd.value
+    private val bufferEndCounter: Long get() = bufferEnd.value
 
     /*
       Additionally to the counters above, we need an extra one that
@@ -813,17 +813,11 @@ internal open class BufferedChannel<E>(
     }
 
     // TODO: method name, documentation, implementation.
-    protected fun dropFirstElementsIfNeeded(s: Long) {
+    protected fun dropFirstElementIfNeeded(s: Long) {
         // Read the segment reference before the counter increment;
         // it is crucial to be able to find the required segment later.
         var segment = receiveSegment.value
         while (true) {
-            // Similar to the `send(e)` operation, `receive()` first checks
-            // whether the channel is already closed for receiving.
-            if (isClosedForReceiveImpl) {
-                if (receiversCounter < sendersCounter) segment.cleanPrev()
-                return
-            }
             // Atomically increments the `receivers` counter
             // and obtain the value right before the increment.
             val r = this.receivers.value
@@ -1534,7 +1528,8 @@ internal open class BufferedChannel<E>(
             onCancellationConstructor = onUndeliveredElementReceiveCancellationConstructor
         )
 
-    protected open fun registerSelectForReceive(select: SelectInstance<*>, ignoredParam: Any?) =
+    @Suppress("UNUSED_PARAMETER")
+    private fun registerSelectForReceive(select: SelectInstance<*>, ignoredParam: Any?) =
         receiveImpl( // <-- this is an inline function
             waiter = select,
             onElementRetrieved = { elem -> select.selectInRegistrationPhase(elem) },
@@ -1606,7 +1601,7 @@ internal open class BufferedChannel<E>(
      * and [SelectInstance.trySelect]. When the channel becomes closed,
      * [tryResumeHasNextOnClosedChannel] should be used instead.
      */
-    protected open inner class BufferedChannelIterator : ChannelIterator<E>, BeforeResumeCancelHandler(), Waiter {
+    private inner class BufferedChannelIterator : ChannelIterator<E>, BeforeResumeCancelHandler(), Waiter {
         /**
          * Stores the element retrieved by [hasNext] or
          * a special [CHANNEL_CLOSED] token if this channel is closed.
@@ -1775,7 +1770,7 @@ internal open class BufferedChannel<E>(
      */
     private val _closeCause = atomic<Any?>(NO_CLOSE_CAUSE)
     // Should be called only if this channel is closed or cancelled.
-    protected val closeCause get() = _closeCause.value as Throwable?
+    internal val closeCause get() = _closeCause.value as Throwable?
 
     /** Returns the closing cause if it is non-null, or [ClosedSendChannelException] otherwise. */
     protected val sendException get() = closeCause ?: ClosedSendChannelException(DEFAULT_CLOSE_MESSAGE)
@@ -1953,17 +1948,21 @@ internal open class BufferedChannel<E>(
     /**
      * Completes the started [close] or [cancel] procedure.
      */
-    protected open fun completeCloseOrCancel() {
+    private fun completeCloseOrCancel() {
         isClosedForSendImpl // must finish the started close/cancel if one is detected.
     }
 
+    protected open val isConflatedDropOldest get() = false
+
     /**
      * Completes the channel closing procedure.
      */
-    private fun completeClose(sendersCur: Long): ChannelSegment<E> {
+    protected open fun completeClose(sendersCur: Long): ChannelSegment<E> {
         // Close the linked list for further segment addition,
         // obtaining the last segment in the data structure.
         val lastSegment = closeLinkedList()
+        // TODO
+        if (isConflatedDropOldest) xxx(lastSegment)
         // Resume waiting `receive()` requests,
         // informing them that the channel is closed.
         cancelSuspendedReceiveRequests(lastSegment, sendersCur)
@@ -1972,6 +1971,31 @@ internal open class BufferedChannel<E>(
         return lastSegment
     }
 
+    private fun xxx(lastSegment: ChannelSegment<E>) {
+        var segment = lastSegment
+        traverse@while (true) {
+            for (index in SEGMENT_SIZE - 1 downTo 0) {
+                cell_update@while (true) {
+                    val state = segment.getState(index)
+                    when {
+                        state === null || state === IN_BUFFER -> {
+                            if (segment.casState(index, state, CHANNEL_CLOSED)) {
+                                segment.onSlotCleaned()
+                                break@cell_update
+                            }
+                        }
+                        state === BUFFERED -> {
+                            dropFirstElementIfNeeded(segment.id * SEGMENT_SIZE + index)
+                            break@traverse
+                        }
+                        else -> break@cell_update
+                    }
+                }
+            }
+            segment = segment.prev ?: break
+        }
+    }
+
     /**
      * Completes the channel cancellation procedure.
      */

kotlinx-coroutines-core/common/src/channels/ConflatedBufferedChannel.kt

@@ -37,6 +37,9 @@ internal open class ConflatedBufferedChannel<E>(
         }
     }
 
+    override val isConflatedDropOldest: Boolean
+        get() = onBufferOverflow == DROP_OLDEST
+
     override suspend fun send(element: E) {
         // Should never suspend, implement via `trySend(..)`.
         trySend(element).onClosed { // fails only when this channel is closed.
@@ -69,17 +72,7 @@ internal open class ConflatedBufferedChannel<E>(
         return success(Unit)
     }
 
-    private val closeStarted = atomic(false)
-    private val trySendStarted = atomic(1L)
-    private val trySendCompleted = atomic(1L)
-
-    private fun trySendDropOldest(element: E): ChannelResult<Unit> = try {
-        trySendStarted.incrementAndGet()
-        if (closeStarted.value) {
-            trySendCompleted.incrementAndGet()
-            while (!isClosedForSend) {}
-            trySendStarted.incrementAndGet()
-        }
+    private fun trySendDropOldest(element: E): ChannelResult<Unit> =
         sendImpl( // <-- this is an inline function
             element = element,
             // Put the element into the logical buffer in any case,
@@ -94,15 +87,12 @@ internal open class ConflatedBufferedChannel<E>(
             // overflowed, the first (oldest) element has to be extracted.
             // After that, the operation finishes.
             onSuspend = { segm, i ->
-                dropFirstElementsIfNeeded(segm.id * SEGMENT_SIZE + i)
+                dropFirstElementIfNeeded(segm.id * SEGMENT_SIZE + i)
                 success(Unit)
             },
             // If the channel is closed, return the corresponding result.
             onClosed = { closed(sendException) }
         )
-    } finally {
-        trySendCompleted.incrementAndGet()
-    }
 
     @Suppress("UNCHECKED_CAST")
     override fun registerSelectForSend(select: SelectInstance<*>, element: Any?) {
@@ -122,26 +112,4 @@ internal open class ConflatedBufferedChannel<E>(
     }
 
     override fun shouldSendSuspend() = false // never suspends
-
-    override fun closeOrCancelImpl(cause: Throwable?, cancel: Boolean): Boolean {
-        // Inform `trySend(..)` the the channel closing procedure has been started.
-        // All `trySend(..)`s that start after setting  this flag, should wait
-        // until the channel is properly closed. Otherwise, a non-linearizable
-        // behaviour can be observed.
-        closeStarted.value = true
-        // It is critical to wait until all the started `trySend(..)`s
-        // complete their operation.
-        waitUntilStartedTrySendsComplete()
-        // Finally, close/cancel this channel.
-        return super.closeOrCancelImpl(cause, cancel)
-    }
-
-    private fun waitUntilStartedTrySendsComplete() {
-        while (true) {
-            val started = trySendStarted.value
-            val completed = trySendCompleted.value
-            if (started != trySendStarted.value) continue
-            if (started == completed) break
-        }
-    }
 }