ConflatedChannel

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/AbstractChannel.kt

@@ -143,12 +143,25 @@ public abstract class AbstractChannel<E> : Channel<E> {
     protected fun sendBuffered(element: E): Boolean =
         queue.addLastIfPrev(SendBuffered(element), { it !is ReceiveOrClosed<*> })
 
+    /**
+     * @suppress **This is unstable API and it is subject to change.**
+     */
+    protected fun sendConflated(element: E): Boolean {
+        val node = SendBuffered(element)
+        if (!queue.addLastIfPrev(node, { it !is ReceiveOrClosed<*> })) return false
+        // remove previous SendBuffered
+        val prev = node.prev
+        if (prev is SendBuffered<*>)
+            prev.remove()
+        return true
+    }
+
     /**
      * @suppress **This is unstable API and it is subject to change.**
      */
     protected fun describeSendBuffered(element: E): AddLastDesc<*> = SendBufferedDesc(queue, element)
 
-    private class SendBufferedDesc<out E>(
+    private open class SendBufferedDesc<out E>(
         queue: LockFreeLinkedListHead,
         element: E
     ) : AddLastDesc<SendBuffered<E>>(queue, SendBuffered(element)) {
@@ -158,6 +171,23 @@ public abstract class AbstractChannel<E> : Channel<E> {
         }
     }
 
+    /**
+     * @suppress **This is unstable API and it is subject to change.**
+     */
+    protected fun describeSendConflated(element: E): AddLastDesc<*> = SendConflatedDesc(queue, element)
+
+    private class SendConflatedDesc<out E>(
+        queue: LockFreeLinkedListHead,
+        element: E
+    ) : SendBufferedDesc<E>(queue, element) {
+        override fun finishOnSuccess(affected: LockFreeLinkedListNode, next: LockFreeLinkedListNode) {
+            super.finishOnSuccess(affected, next)
+            // remove previous SendBuffered
+            if (affected is SendBuffered<*>)
+                affected.remove()
+        }
+    }
+
     // ------ SendChannel ------
 
     public final override val isClosedForSend: Boolean get() = closedForSend != null

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/Channel.kt

@@ -224,23 +224,31 @@ public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> {
      */
     public companion object Factory {
         /**
-         * Requests channel with unlimited capacity buffer in [Channel()][invoke] factory function.
+         * Requests channel with unlimited capacity buffer in [Channel()][invoke] factory function --
+         * the [LinkedListChannel] gets created.
          */
         public const val UNLIMITED = Int.MAX_VALUE
 
+        /**
+         * Requests conflated channel in [Channel()][invoke] factory function --
+         * the [ConflatedChannel] gets created.
+         */
+        public const val CONFLATED = -1
+
         /**
          * Creates a channel with specified buffer capacity (or without a buffer by default).
          *
          * The resulting channel type depends on the specified [capacity] parameter:
          * * when `capacity` is 0 -- creates [RendezvousChannel];
          * * when `capacity` is [UNLIMITED] -- creates [LinkedListChannel];
+         * * when `capacity` is [CONFLATED] -- creates [ConflatedChannel];
          * * otherwise -- creates [ArrayChannel].
          */
         public operator fun <E> invoke(capacity: Int = 0): Channel<E> {
-            check(capacity >= 0) { "Channel capacity cannot be negative, but $capacity was specified" }
             return when (capacity) {
                 0 -> RendezvousChannel()
                 UNLIMITED -> LinkedListChannel()
+                CONFLATED -> ConflatedChannel()
                 else -> ArrayChannel(capacity)
             }
         }

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ConflatedChannel.kt

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+import kotlinx.coroutines.experimental.ALREADY_SELECTED
+import kotlinx.coroutines.experimental.selects.SelectInstance
+
+/**
+ * Channel that buffers at most one element and conflates all subsequent `send` and `offer` invocations,
+ * so that the receiver always gets the most recently sent element.
+ * Back-to-send sent elements are _conflated_ -- only the the most recently sent element is received,
+ * while previously sent elements **are lost**.
+ * Sender to this channel never suspends and [offer] always returns `true`.
+ *
+ * This implementation is fully lock-free.
+ */
+public open class ConflatedChannel<E> : AbstractChannel<E>() {
+    protected final override val isBufferAlwaysEmpty: Boolean get() = true
+    protected final override val isBufferEmpty: Boolean get() = true
+    protected final override val isBufferAlwaysFull: Boolean get() = false
+    protected final override val isBufferFull: Boolean get() = false
+
+    // result is always `OFFER_SUCCESS | Closed`
+    protected override fun offerInternal(element: E): Any {
+        while (true) {
+            val result = super.offerInternal(element)
+            when {
+                result === OFFER_SUCCESS -> return OFFER_SUCCESS
+                result === OFFER_FAILED -> { // try to buffer
+                    if (sendConflated(element))
+                        return OFFER_SUCCESS
+                }
+                result is Closed<*> -> return result
+                else -> error("Invalid offerInternal result $result")
+            }
+        }
+    }
+
+    // result is always `ALREADY_SELECTED | OFFER_SUCCESS | Closed`.
+    protected override fun offerSelectInternal(element: E, select: SelectInstance<*>): Any {
+        while (true) {
+            val result = if (hasReceiveOrClosed)
+                super.offerSelectInternal(element, select) else
+                (select.performAtomicTrySelect(describeSendConflated(element)) ?: OFFER_SUCCESS)
+            when {
+                result === ALREADY_SELECTED -> return ALREADY_SELECTED
+                result === OFFER_SUCCESS -> return OFFER_SUCCESS
+                result === OFFER_FAILED -> {} // retry
+                result is Closed<*> -> return result
+                else -> error("Invalid result $result")
+            }
+        }
+    }
+}
+

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedList.kt

@@ -68,9 +68,9 @@ public typealias AddLastDesc<T> = LockFreeLinkedListNode.AddLastDesc<T>
 @Suppress("LeakingThis")
 public open class LockFreeLinkedListNode {
     @Volatile
-    private var _next: Any = this // DoubleLinkedNode | Removed | OpDescriptor
+    private var _next: Any = this // Node | Removed | OpDescriptor
     @Volatile
-    private var _prev: Any = this // DoubleLinkedNode | Removed
+    private var _prev: Any = this // Node | Removed
     @Volatile
     private var _removedRef: Removed? = null // lazily cached removed ref to this
 
@@ -114,7 +114,7 @@ public open class LockFreeLinkedListNode {
 
     public val isRemoved: Boolean get() = next is Removed
 
-    // LINEARIZABLE.
+    // LINEARIZABLE. Returns Node | Removed
     public val next: Any get() {
         while (true) { // operation helper loop on _next
             val next = this._next
@@ -123,10 +123,12 @@ public open class LockFreeLinkedListNode {
         }
     }
 
-    // LINEARIZABLE. Note: use it on sentinel (never removed) node only
-    public val prev: Node get() {
-        while (true) {
-            val prev = this._prev as Node // this sentinel node is never removed
+    // LINEARIZABLE. Returns Node | Removed
+    public val prev: Any get() {
+        while (true) { // insert helper loop on _prev
+            val prev = this._prev
+            if (prev is Removed) return prev
+            prev as Node // otherwise, it can be only node otherwise
             if (prev.next === this) return prev
             helpInsert(prev, null)
         }
@@ -155,7 +157,7 @@ public open class LockFreeLinkedListNode {
      */
     public fun addLast(node: Node) {
         while (true) { // lock-free loop on prev.next
-            val prev = prev
+            val prev = prev as Node // sentinel node is never removed, so prev is always defined
             if (prev.addNext(node, this)) return
         }
     }
@@ -168,7 +170,7 @@ public open class LockFreeLinkedListNode {
     public inline fun addLastIf(node: Node, crossinline condition: () -> Boolean): Boolean {
         val condAdd = makeCondAddOp(node, condition)
         while (true) { // lock-free loop on prev.next
-            val prev = prev
+            val prev = prev as Node // sentinel node is never removed, so prev is always defined
             when (prev.tryCondAddNext(node, this, condAdd)) {
                 SUCCESS -> return true
                 FAILURE -> return false
@@ -178,7 +180,7 @@ public open class LockFreeLinkedListNode {
 
     public inline fun addLastIfPrev(node: Node, predicate: (Node) -> Boolean): Boolean {
         while (true) { // lock-free loop on prev.next
-            val prev = prev
+            val prev = prev as Node // sentinel node is never removed, so prev is always defined
             if (!predicate(prev)) return false
             if (prev.addNext(node, this)) return true
         }
@@ -191,7 +193,7 @@ public open class LockFreeLinkedListNode {
     ): Boolean {
         val condAdd = makeCondAddOp(node, condition)
         while (true) { // lock-free loop on prev.next
-            val prev = prev
+            val prev = prev as Node // sentinel node is never removed, so prev is always defined
             if (!predicate(prev)) return false
             when (prev.tryCondAddNext(node, this, condAdd)) {
                 SUCCESS -> return true

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ChannelAtomicCancelStressTest.kt

@@ -98,9 +98,11 @@ class ChannelAtomicCancelStressTest(val kind: TestChannelKind) : TestBase() {
         println("          Missed $missedCnt ints")
         println("      Duplicated $dupCnt ints")
         failed.get()?.let { throw it }
-        assertEquals(0, missedCnt)
         assertEquals(0, dupCnt)
-        assertEquals(lastSent, lastReceived)
+        if (kind != TestChannelKind.CONFLATED) {
+            assertEquals(0, missedCnt)
+            assertEquals(lastSent, lastReceived)
+        }
     }
 
     fun launchSender() {

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ChannelSendReceiveStressTest.kt

@@ -81,6 +81,14 @@ class ChannelSendReceiveStressTest(
                 sendersCompleted.incrementAndGet()
             }
         }
+        // print progress
+        val progressJob = launch(context) {
+            var seconds = 0
+            while (true) {
+                delay(1000)
+                println("${++seconds}: Sent ${sentTotal.get()}, received ${receivedTotal.get()}")
+            }
+        }
         try {
             withTimeout(timeLimit) {
                 senders.forEach { it.join() }
@@ -90,6 +98,7 @@ class ChannelSendReceiveStressTest(
         } catch (e: CancellationException) {
             println("!!! Test timed out $e")
         }
+        progressJob.cancel()
         println("Tested $kind with nSenders=$nSenders, nReceivers=$nReceivers")
         println("Completed successfully ${sendersCompleted.get()} sender coroutines")
         println("Completed successfully ${receiversCompleted.get()} receiver coroutines")
@@ -103,16 +112,18 @@ class ChannelSendReceiveStressTest(
         assertEquals(nReceivers, receiversCompleted.get())
         assertEquals(0, dupes.get())
         assertEquals(nEvents, sentTotal.get())
-        assertEquals(nEvents, receivedTotal.get())
+        if (kind != TestChannelKind.CONFLATED) assertEquals(nEvents, receivedTotal.get())
         repeat(nReceivers) { receiveIndex ->
             assertTrue("Each receiver should have received something", receivedBy[receiveIndex] > 0)
         }
     }
 
     private suspend fun doSent() {
         sentTotal.incrementAndGet()
-        while (sentTotal.get() > receivedTotal.get() + maxBuffer)
-            yield() // throttle fast senders to prevent OOM with LinkedListChannel
+        if (kind != TestChannelKind.CONFLATED) {
+            while (sentTotal.get() > receivedTotal.get() + maxBuffer)
+                yield() // throttle fast senders to prevent OOM with LinkedListChannel
+        }
     }
 
     private suspend fun doSend(senderIndex: Int) {

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ConflatedChannelTest.kt

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+import kotlinx.coroutines.experimental.TestBase
+import kotlinx.coroutines.experimental.launch
+import kotlinx.coroutines.experimental.runBlocking
+import kotlinx.coroutines.experimental.yield
+import org.hamcrest.core.IsEqual
+import org.hamcrest.core.IsNull
+import org.junit.Assert.assertThat
+import org.junit.Test
+
+class ConflatedChannelTest : TestBase() {
+    @Test
+    fun testBasicConflationOfferPoll() {
+        val q = ConflatedChannel<Int>()
+        assertThat(q.poll(), IsNull())
+        assertThat(q.offer(1), IsEqual(true))
+        assertThat(q.offer(2), IsEqual(true))
+        assertThat(q.offer(3), IsEqual(true))
+        assertThat(q.poll(), IsEqual(3))
+        assertThat(q.poll(), IsNull())
+    }
+
+    @Test
+    fun testConflationSendReceive() = runBlocking<Unit> {
+        val q = ConflatedChannel<Int>()
+        expect(1)
+        launch(context) { // receiver coroutine
+            expect(4)
+            assertThat(q.receive(), IsEqual(2))
+            expect(5)
+            assertThat(q.receive(), IsEqual(3)) // this receive suspends
+            expect(8)
+            assertThat(q.receive(), IsEqual(6)) // last conflated value
+            expect(9)
+        }
+        expect(2)
+        q.send(1)
+        q.send(2) // shall conflate
+        expect(3)
+        yield() // to receiver
+        expect(6)
+        q.send(3) // send to the waiting receiver
+        q.send(4) // buffer
+        q.send(5) // conflate
+        q.send(6) // conflate again
+        expect(7)
+        yield() // to receiver
+        finish(10)
+    }
+}

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/SimpleSendReceiveTest.kt

@@ -19,10 +19,12 @@ package kotlinx.coroutines.experimental.channels
 import kotlinx.coroutines.experimental.CommonPool
 import kotlinx.coroutines.experimental.launch
 import kotlinx.coroutines.experimental.runBlocking
+import org.hamcrest.core.IsEqual
+import org.junit.Assert.assertThat
+import org.junit.Assert.assertTrue
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.Parameterized
-import org.junit.Assert.*
 
 @RunWith(Parameterized::class)
 class SimpleSendReceiveTest(
@@ -47,10 +49,17 @@ class SimpleSendReceiveTest(
             repeat(n) { channel.send(it) }
             channel.close()
         }
-        var received = 0
+        var expected = 0
         for (x in channel) {
-            assertEquals(received++, x)
+            if (kind != TestChannelKind.CONFLATED) {
+                assertThat(x, IsEqual(expected++))
+            } else {
+                assertTrue(x >= expected)
+                expected = x + 1
+            }
+        }
+        if (kind != TestChannelKind.CONFLATED) {
+            assertThat(expected, IsEqual(n))
         }
-        assertEquals(n, received)
     }
 }