Make ConflatedChannel linearizable

  * Rename onClosed to onClosedIdempotent
  * Use onClosedIdempotent with helpClose on concurrent close
  * Get rid of afterClose and replace it with onClosedIdempotent
  * Help interleaving operations with conflation

Author: qwwdfsad

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

@@ -119,12 +119,18 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
         return null
     }
 
-    /**
-     * @suppress **This is unstable API and it is subject to change.**
-     */
     protected fun conflatePreviousSendBuffered(node: LockFreeLinkedListNode) {
-        val prev = node.prevNode
-        (prev as? SendBuffered<*>)?.remove()
+        /*
+         * Conflate all previous SendBuffered,
+         * helping other sends to coflate
+         */
+        var prev = node.prevNode
+        while (prev is SendBuffered<*>) {
+            if (!prev.remove()) {
+                prev.helpRemove()
+            }
+            prev = prev.prevNode
+        }
     }
 
     /**
@@ -249,15 +255,15 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
          */
         val closeAdded = queue.addLastIfPrev(closed, { it !is Closed<*> })
         if (!closeAdded) {
-            helpClose(queue.prevNode as Closed<*>)
+            val actualClosed = queue.prevNode as Closed<*>
+            helpClose(actualClosed)
+            onClosedIdempotent(actualClosed)
             return false
         }
 
         helpClose(closed)
         invokeOnCloseHandler(cause)
-        // TODO We can get rid of afterClose
-        onClosed(closed)
-        afterClose(cause)
+        onClosedIdempotent(closed)
         return true
     }
 
@@ -322,15 +328,10 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
     }
 
     /**
-     * Invoked when [Closed] element was just added.
-     * @suppress **This is unstable API and it is subject to change.**
-     */
-    protected open fun onClosed(closed: Closed<E>) {}
-
-    /**
-     * Invoked after successful [close].
+     * Invoked when channel is closed as the last action of [close] invocation.
+     * This method should be idempotent and can be called multiple times.
      */
-    protected open fun afterClose(cause: Throwable?) {}
+    protected open fun onClosedIdempotent(closed: LockFreeLinkedListNode) {}
 
     /**
      * Retrieves first receiving waiter from the queue or returns closed token.

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

@@ -5,7 +5,7 @@
 package kotlinx.coroutines.channels
 
 import kotlinx.coroutines.selects.*
-import kotlinx.coroutines.*
+import kotlinx.coroutines.internal.*
 
 /**
  * Channel that buffers at most one element and conflates all subsequent `send` and `offer` invocations,
@@ -24,11 +24,7 @@ internal open class ConflatedChannel<E> : AbstractChannel<E>() {
     protected final override val isBufferAlwaysFull: Boolean get() = false
     protected final override val isBufferFull: Boolean get() = false
 
-    /**
-     * This implementation conflates last sent item when channel is closed.
-     * @suppress **This is unstable API and it is subject to change.**
-     */
-    override fun onClosed(closed: Closed<E>) {
+    override fun onClosedIdempotent(closed: LockFreeLinkedListNode) {
         conflatePreviousSendBuffered(closed)
     }
 

core/kotlinx-coroutines-core/test/linearizability/ChannelLinearizabilityTest.kt

@@ -25,7 +25,6 @@ class ChannelLinearizabilityTest : TestBase() {
     private val lt = LinTesting()
     private var channel: Channel<Int> = Channel(capacity)
 
-
     @Operation(runOnce = true)
     fun send1(@Param(name = "value") value: Int) = lt.run("send1") { channel.send(value) }
 

reactive/kotlinx-coroutines-reactive/src/Channel.kt

@@ -7,6 +7,7 @@ package kotlinx.coroutines.reactive
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.internal.*
 import org.reactivestreams.*
 
 /**
@@ -74,7 +75,7 @@ private class SubscriptionChannel<T>(
     }
 
     @Suppress("CANNOT_OVERRIDE_INVISIBLE_MEMBER")
-    override fun afterClose(cause: Throwable?) {
+    override fun onClosedIdempotent(closed: LockFreeLinkedListNode) {
         subscription?.cancel()
     }
 

reactive/kotlinx-coroutines-rx2/src/RxChannel.kt

@@ -8,6 +8,7 @@ import io.reactivex.*
 import io.reactivex.disposables.*
 import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.internal.*
 
 /**
  * Subscribes to this [MaybeSource] and returns a channel to receive elements emitted by it.
@@ -64,9 +65,8 @@ private class SubscriptionChannel<T> :
     @Volatile
     var subscription: Disposable? = null
 
-    // AbstractChannel overrides
     @Suppress("CANNOT_OVERRIDE_INVISIBLE_MEMBER")
-    override fun afterClose(cause: Throwable?) {
+    override fun onClosedIdempotent(closed: LockFreeLinkedListNode) {
         subscription?.dispose()
     }