Fix lost sender during close() call, make close() linearizable

Fixes #359

Author: qwwdfsad

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

@@ -243,23 +243,51 @@ 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()) {
+                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)
         }
     }
 

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

@@ -16,8 +16,7 @@
 
 package kotlinx.coroutines.experimental.internal
 
-import kotlinx.atomicfu.atomic
-import kotlinx.atomicfu.loop
+import kotlinx.atomicfu.*
 
 private typealias Node = LockFreeLinkedListNode
 
@@ -96,8 +95,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

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

@@ -16,23 +16,24 @@
 
 package kotlinx.coroutines.experimental.channels
 
-import com.devexperts.dxlab.lincheck.LinChecker
-import com.devexperts.dxlab.lincheck.annotations.Operation
-import com.devexperts.dxlab.lincheck.annotations.Param
-import com.devexperts.dxlab.lincheck.annotations.Reset
-import com.devexperts.dxlab.lincheck.paramgen.IntGen
+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.Test
+import org.junit.*
+import java.io.*
 
 @Param(name = "value", gen = IntGen::class, conf = "1:3")
 class ChannelLinearizabilityTest : TestBase() {
+
     private val lt = LinTesting()
+    private var capacity = 0
     private lateinit var channel: Channel<Int>
 
     @Reset
     fun reset() {
-        channel = Channel<Int>()
+        channel = Channel(capacity)
     }
 
     @Operation(runOnce = true)
@@ -53,14 +54,32 @@ class ChannelLinearizabilityTest : TestBase() {
     @Operation(runOnce = true)
     fun receive3() = lt.run("receive3") { channel.receive() }
 
-//    @Operation(runOnce = true)
-//    fun close1() = lt.run("close1") { channel.close(IOException("close1")) }
-//
-//    @Operation(runOnce = true)
-//    fun close2() = lt.run("close2") { channel.close(IOException("close2")) }
+    @Operation(runOnce = true)
+    fun close1() = lt.run("close1") { channel.close(IOException("close1")) }
+
+    @Operation(runOnce = true)
+    fun close2() = lt.run("close2") { channel.close(IOException("close2")) }
+
+    @Test
+    fun testRendezvousChannelLinearizability() {
+        runTest(0)
+    }
 
     @Test
-    fun testLinearizability() {
+    fun testArrayChannelLinearizability() {
+        for (i in listOf(1, 2, 16)) {
+            runTest(i)
+        }
+    }
+
+    @Test
+    fun testConflatedChannelLinearizability() = runTest(Channel.CONFLATED)
+
+    @Test
+    fun testUnlimitedChannelLinearizability() = runTest(Channel.UNLIMITED)
+
+    private fun runTest(capacity: Int) {
+        this.capacity = capacity
         val options = StressOptions()
             .iterations(100)
             .invocationsPerIteration(1000 * stressTestMultiplier)
@@ -70,4 +89,4 @@ class ChannelLinearizabilityTest : TestBase() {
             .verifier(LinVerifier::class.java)
         LinChecker.check(ChannelLinearizabilityTest::class.java, options)
     }
-}
+}