Introduce InlineList to simplify helpClose logic, reverse helpClose resume order
diff --git a/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt b/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt
index b5dfd95..be18942 100644
--- a/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt
+++ b/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt
@@ -304,8 +304,7 @@
* Now if T2's close resumes T1's receive_2 then it's receive gets "closed for receive" exception, but
* its subsequent attempt to send successfully rendezvous with receive_1, producing non-linearizable execution.
*/
- var closedNode: Receive<E>? = null // used when one node was closed to avoid extra memory allocation
- var closedList: ArrayList<Receive<E>>? = null // used when more nodes were closed
+ var closedList = InlineList<Receive<E>>()
while (true) {
// Break when channel is empty or has no receivers
@Suppress("UNCHECKED_CAST")
@@ -316,19 +315,14 @@
previous.helpRemove() // make sure remove is complete before continuing
continue
}
- // add removed nodes to a separate list
- if (closedNode == null) {
- closedNode = previous
- } else {
- val list = closedList ?: ArrayList<Receive<E>>().also { closedList = it }
- list += previous
- }
+ // add removed nodes to a separate list
+ closedList += previous
}
- // now notify all removed nodes that the channel was closed
- if (closedNode != null) {
- closedNode.resumeReceiveClosed(closed)
- closedList?.forEach { it.resumeReceiveClosed(closed) }
- }
+ /*
+ * Now notify all removed nodes that the channel was closed
+ * in the order they were added to the channel
+ */
+ closedList.forEachReversed { it.resumeReceiveClosed(closed) }
// and do other post-processing
onClosedIdempotent(closed)
}
diff --git a/kotlinx-coroutines-core/common/src/internal/InlineList.kt b/kotlinx-coroutines-core/common/src/internal/InlineList.kt
new file mode 100644
index 0000000..062a910
--- /dev/null
+++ b/kotlinx-coroutines-core/common/src/internal/InlineList.kt
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+@file:Suppress("UNCHECKED_CAST")
+
+package kotlinx.coroutines.internal
+
+import kotlinx.coroutines.assert
+
+/*
+ * Inline class that represents a mutable list, but does not allocate an underlying storage
+ * for zero and one elements.
+ * Cannot be parametrized with `List<*>`.
+ */
+internal inline class InlineList<E>(private val holder: Any? = null) {
+ public operator fun plus(element: E): InlineList<E> {
+ assert { element !is List<*> } // Lists are prohibited
+ return when (holder) {
+ null -> InlineList(element)
+ is ArrayList<*> -> {
+ (holder as ArrayList<E>).add(element)
+ InlineList(holder)
+ }
+ else -> {
+ val list = ArrayList<E>(4)
+ list.add(holder as E)
+ list.add(element)
+ InlineList(list)
+ }
+ }
+ }
+
+ public inline fun forEachReversed(action: (E) -> Unit) {
+ when (holder) {
+ null -> return
+ !is ArrayList<*> -> action(holder as E)
+ else -> {
+ val list = holder as ArrayList<E>
+ for (i in (list.size - 1) downTo 0) {
+ action(list[i])
+ }
+ }
+ }
+ }
+}
diff --git a/kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt b/kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt
index 983f353..42cc855 100644
--- a/kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt
+++ b/kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt
@@ -20,6 +20,37 @@
}
@Test
+ fun testCloseWithMultipleWaiters() = runTest {
+ val channel = Channel<Int>()
+ launch {
+ try {
+ expect(2)
+ channel.receive()
+ expectUnreached()
+ } catch (e: ClosedReceiveChannelException) {
+ expect(5)
+ }
+ }
+
+ launch {
+ try {
+ expect(3)
+ channel.receive()
+ expectUnreached()
+ } catch (e: ClosedReceiveChannelException) {
+ expect(6)
+ }
+ }
+
+ expect(1)
+ yield()
+ expect(4)
+ channel.close()
+ yield()
+ finish(7)
+ }
+
+ @Test
fun testAssociate() = runTest {
assertEquals(testList.associate { it * 2 to it * 3 },
testList.asReceiveChannel().associate { it * 2 to it * 3 }.toMap())