Fixed Channel.close hand on double close with pending send

Author: elizarov

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

@@ -248,7 +248,10 @@ public abstract class AbstractSendChannel<E> : SendChannel<E> {
             val receive = takeFirstReceiveOrPeekClosed()
             if (receive == null) {
                 // queue empty or has only senders -- try add last "Closed" item to the queue
-                if (queue.addLastIfPrev(closed, { it !is ReceiveOrClosed<*> })) {
+                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

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

@@ -0,0 +1,38 @@
+/*
+ * 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.*
+import org.junit.Test
+
+class DoubleChannelCloseStressTest : TestBase() {
+    val nTimes = 1000 * stressTestMultiplier
+
+    @Test
+    fun testDoubleCloseStress() {
+        repeat(nTimes) {
+            val actor = actor<Int>(CommonPool + CoroutineName("actor"), start = CoroutineStart.LAZY) {
+                // empty -- just closes channel
+            }
+            launch(CommonPool + CoroutineName("sender")) {
+                actor.send(1)
+            }
+            Thread.sleep(1)
+            actor.close()
+        }
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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 junit.framework.Assert.assertFalse
+import junit.framework.Assert.assertTrue
+import kotlinx.coroutines.experimental.TestBase
+import kotlinx.coroutines.experimental.runBlocking
+import org.hamcrest.MatcherAssert.assertThat
+import org.hamcrest.core.IsEqual
+import org.hamcrest.core.IsNull
+import org.junit.Test
+
+class LinkedListChannelTest : TestBase() {
+    @Test
+    fun testBasic() = runBlocking {
+        val c = LinkedListChannel<Int>()
+        c.send(1)
+        assertTrue(c.offer(2))
+        c.send(3)
+        assertTrue(c.close())
+        assertFalse(c.close())
+        assertThat(c.receive(), IsEqual(1))
+        assertThat(c.poll(), IsEqual(2))
+        assertThat(c.receiveOrNull(), IsEqual(3))
+        assertThat(c.receiveOrNull(), IsNull())
+    }
+}