Channels concurrency improvement:

Make close, cancel, isClosedForSend, isClosedForReceive and offer linearizable with other operations
Fix bug when send operation can be stuck in channel forever
New concurrency tests

Fixes #359

Author: qwwdfsad

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

@@ -71,13 +71,13 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
      * Returns non-null closed token if it is last in the queue.
      * @suppress **This is unstable API and it is subject to change.**
      */
-    protected val closedForSend: Closed<*>? get() = queue.prevNode as? Closed<*>
+    protected val closedForSend: Closed<*>? get() = (queue.prevNode as? Closed<*>)?.also { helpClose(it) }
 
     /**
      * Returns non-null closed token if it is first in the queue.
      * @suppress **This is unstable API and it is subject to change.**
      */
-    protected val closedForReceive: Closed<*>? get() = queue.nextNode as? Closed<*>
+    protected val closedForReceive: Closed<*>? get() = (queue.nextNode as? Closed<*>)?.also { helpClose(it) }
 
     /**
      * Retrieves first sending waiter from the queue or returns closed token.
@@ -169,7 +169,9 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
         val result = offerInternal(element)
         return when {
             result === OFFER_SUCCESS -> true
-            result === OFFER_FAILED -> false
+            // We should check for closed token on offer as well, otherwise offer won't be linearizable
+            // in the face of concurrent close()
+            result === OFFER_FAILED ->  throw closedForSend?.sendException ?: return false
             result is Closed<*> -> throw result.sendException
             else -> error("offerInternal returned $result")
         }
@@ -231,23 +233,54 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
 
     public override fun close(cause: Throwable?): Boolean {
         val closed = Closed<E>(cause)
+
+        /*
+         * Try to commit close by adding a close token to the end of the queue.
+         * Successful -> we're now responsible for closing receivers
+         * Not successful -> help closing pending receivers to maintain invariant
+         * "if (!close()) next send will throw"
+         */
+        val closeAdded = queue.addLastIfPrev(closed, { it !is Closed<*> })
+        if (!closeAdded) {
+            helpClose(queue.prevNode as Closed<*>)
+            return false
+        }
+
+        helpClose(closed)
+        onClosed(closed)
+        afterClose(cause)
+        return true
+    }
+
+    private fun helpClose(closed: Closed<*>) {
+        /*
+         * It's important to traverse list from right to left to avoid races with sender.
+         * Consider channel state
+         * head sentinel -> [receiver 1] -> [receiver 2] -> head sentinel
+         * T1 invokes receive()
+         * T2 invokes close()
+         * T3 invokes close() + send(value)
+         *
+         * If both will traverse list from left to right, following non-linearizable history is possible:
+         * [close -> false], [send -> transferred 'value' to receiver]
+         */
         while (true) {
-            val receive = takeFirstReceiveOrPeekClosed()
-            if (receive == null) {
-                // queue empty or has only senders -- try add last "Closed" item to the queue
-                if (queue.addLastIfPrev(closed, { prev ->
-                    if (prev is Closed<*>) return false // already closed
-                    prev !is ReceiveOrClosed<*> // only add close if no waiting receive
-                })) {
-                    onClosed(closed)
-                    afterClose(cause)
-                    return true
-                }
-                continue // retry on failure
+            val previous = closed.prevNode
+            // Channel is empty or has no receivers
+            if (previous is LockFreeLinkedListHead || previous !is Receive<*>) {
+                break
+            }
+
+            if (!previous.remove()) {
+                // failed to remove the node (due to race) -- retry finding non-removed prevNode
+                // NOTE: remove() DOES NOT help pending remove operation (that marked next pointer)
+                previous.helpRemove() // make sure remove is complete before continuing
+                continue
             }
-            if (receive is Closed<*>) return false // already marked as closed -- nothing to do
-            receive as Receive<E> // type assertion
-            receive.resumeReceiveClosed(closed)
+
+            @Suppress("UNCHECKED_CAST")
+            previous as Receive<E> // type assertion
+            previous.resumeReceiveClosed(closed)
         }
     }
 

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

@@ -103,7 +103,7 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
 
     @Suppress("UNCHECKED_CAST")
     public override fun openSubscription(): ReceiveChannel<E> {
-        val subscriber = Subscriber<E>(this)
+        val subscriber = Subscriber(this)
         _state.loop { state ->
             when (state) {
                 is Closed -> {
@@ -238,6 +238,7 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
         block.startCoroutineUndispatched(receiver = this, completion = select.completion)
     }
 
+    @Suppress("DEPRECATION")
     private class Subscriber<E>(
         private val broadcastChannel: ConflatedBroadcastChannel<E>
     ) : ConflatedChannel<E>(), ReceiveChannel<E>, SubscriptionReceiveChannel<E> {

common/kotlinx-coroutines-core-common/src/internal/LockFreeLinkedList.common.kt

@@ -24,6 +24,13 @@ public expect open class LockFreeLinkedListNode() {
     ): Boolean
 
     public open fun remove(): Boolean
+
+    /**
+     * Helps fully finish [remove] operation, must be invoked after [remove] if needed.
+     * Ensures that traversing the list via prev pointers sees this node as removed.
+     * No-op on JS
+     */
+    public fun helpRemove()
     public fun removeFirstOrNull(): LockFreeLinkedListNode?
     public inline fun <reified T> removeFirstIfIsInstanceOfOrPeekIf(predicate: (T) -> Boolean): T?
 }

common/kotlinx-coroutines-core-common/test/channels/BasicOperationsTest.kt

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+import kotlinx.coroutines.experimental.*
+import kotlin.coroutines.experimental.*
+import kotlin.test.*
+
+class BasicOperationsTest : TestBase() {
+
+    @Test
+    fun testSimpleSendReceive() = runTest {
+        // Parametrized common test :(
+        TestChannelKind.values().forEach { kind -> testSendReceive(kind, 100) }
+    }
+
+    @Test
+    fun testOfferAfterClose() = runTest {
+        TestChannelKind.values().forEach { kind -> testOffer(kind) }
+    }
+
+    @Test
+    fun testReceiveOrNullAfterClose() = runTest {
+        TestChannelKind.values().forEach { kind -> testReceiveOrNull(kind) }
+    }
+
+    @Test
+    fun testReceiveOrNullAfterCloseWithException() = runTest {
+        TestChannelKind.values().forEach { kind -> testReceiveOrNullException(kind) }
+    }
+
+    private suspend fun testReceiveOrNull(kind: TestChannelKind) {
+        val channel = kind.create()
+        val d = async(coroutineContext) {
+            channel.receive()
+        }
+
+        yield()
+        channel.close()
+        assertTrue(channel.isClosedForReceive)
+
+        assertNull(channel.receiveOrNull())
+        assertNull(channel.poll())
+
+        d.join()
+        assertTrue(d.getCancellationException().cause is ClosedReceiveChannelException)
+    }
+
+    private suspend fun testReceiveOrNullException(kind: TestChannelKind) {
+        val channel = kind.create()
+        val d = async(coroutineContext) {
+            channel.receive()
+        }
+
+        yield()
+        channel.close(IndexOutOfBoundsException())
+        assertTrue(channel.isClosedForReceive)
+
+        assertFailsWith<IndexOutOfBoundsException> { channel.poll() }
+        try {
+            channel.receiveOrNull()
+            fail()
+        } catch (e: IndexOutOfBoundsException) {
+            // Expected
+        }
+
+        d.join()
+        assertTrue(d.getCancellationException().cause is IndexOutOfBoundsException)
+    }
+
+
+    private suspend fun testOffer(kind: TestChannelKind) {
+        val channel = kind.create()
+        val d = async(coroutineContext) { channel.send(42) }
+        yield()
+        channel.close()
+
+        assertTrue(channel.isClosedForSend)
+        try {
+            channel.offer(2)
+            fail()
+        } catch (e: ClosedSendChannelException) {
+            if (!kind.isConflated) {
+                assertEquals(42, channel.receive())
+            }
+        }
+
+        d.await()
+    }
+
+    private suspend fun testSendReceive(kind: TestChannelKind, iterations: Int) {
+        val channel = kind.create()
+
+        launch(coroutineContext) {
+            repeat(iterations) { channel.send(it) }
+            channel.close()
+        }
+        var expected = 0
+        for (x in channel) {
+            if (!kind.isConflated) {
+                assertEquals(expected++, x)
+            } else {
+                assertTrue(x >= expected)
+                expected = x + 1
+            }
+        }
+        if (!kind.isConflated) {
+            assertEquals(iterations, expected)
+        }
+    }
+}

common/kotlinx-coroutines-core-common/test/channels/SimpleSendReceiveTest.kt

@@ -4,8 +4,7 @@
 
 package kotlinx.coroutines.experimental.internal
 
-import kotlinx.atomicfu.atomic
-import kotlinx.atomicfu.loop
+import kotlinx.atomicfu.*
 
 private typealias Node = LockFreeLinkedListNode
 
@@ -84,8 +83,6 @@ public actual open class LockFreeLinkedListNode {
             override fun prepare(affected: Node): Any? = if (condition()) null else CONDITION_FALSE
         }
 
-    public val isFresh: Boolean get() = _next.value === this
-
     public actual val isRemoved: Boolean get() = next is Removed
 
     // LINEARIZABLE. Returns Node | Removed
@@ -231,6 +228,10 @@ public actual open class LockFreeLinkedListNode {
     /**
      * Removes this node from the list. Returns `true` when removed successfully, or `false` if the node was already
      * removed or if it was not added to any list in the first place.
+     *
+     * **Note**: Invocation of this operation does not guarantee that remove was actually complete if result was `false`.
+     * In particular, invoking [nextNode].[prevNode] might still return this node even though it is "already removed".
+     * Invoke [helpRemove] to make sure that remove was completed.
      */
     public actual open fun remove(): Boolean {
         while (true) { // lock-free loop on next
@@ -246,6 +247,11 @@ public actual open class LockFreeLinkedListNode {
         }
     }
 
+    public actual fun helpRemove() {
+        val removed = this.next as? Removed ?: error("Must be invoked on a removed node")
+        finishRemove(removed.ref)
+    }
+
     public open fun describeRemove() : AtomicDesc? {
         if (isRemoved) return null // fast path if was already removed
         return object : AbstractAtomicDesc() {

core/kotlinx-coroutines-core/src/internal/LockFreeLinkedList.kt

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental.channels
+
+import com.devexperts.dxlab.lincheck.*
+import com.devexperts.dxlab.lincheck.annotations.*
+import com.devexperts.dxlab.lincheck.paramgen.*
+import com.devexperts.dxlab.lincheck.stress.*
+import kotlinx.coroutines.experimental.*
+import org.junit.*
+import java.io.*
+
+@Param(name = "value", gen = IntGen::class, conf = "1:3")
+class ChannelIsClosedLinearizabilityTest : TestBase() {
+
+    private val lt = LinTesting()
+    private lateinit var channel: Channel<Int>
+
+    @Reset
+    fun reset() {
+        channel = Channel()
+    }
+
+    @Operation(runOnce = true)
+    fun send1(@Param(name = "value") value: Int) = lt.run("send1") { channel.send(value) }
+
+    @Operation(runOnce = true)
+    fun send2(@Param(name = "value") value: Int) = lt.run("send2") { channel.send(value) }
+
+    @Operation(runOnce = true)
+    fun receive1() = lt.run("receive1") { channel.receive() }
+
+    @Operation(runOnce = true)
+    fun receive2() = lt.run("receive2") { channel.receive() }
+
+    @Operation(runOnce = true)
+    fun close1() = lt.run("close1") { channel.close(IOException("close1")) }
+
+    @Operation(runOnce = true)
+    fun isClosedForReceive() = lt.run("isClosedForReceive") { channel.isClosedForReceive }
+
+    @Operation(runOnce = true)
+    fun isClosedForSend() = lt.run("isClosedForSend") { channel.isClosedForSend }
+
+    @Test
+    fun testLinearizability() {
+        val options = StressOptions()
+            .iterations(100)
+            .invocationsPerIteration(1000 * stressTestMultiplier)
+            .addThread(1, 3)
+            .addThread(1, 3)
+            .addThread(1, 3)
+            .verifier(LinVerifier::class.java)
+        LinChecker.check(ChannelIsClosedLinearizabilityTest::class.java, options)
+    }
+}