Implement CopyOnWriteList on JS, add reproducer for #412

Fixes #412

Author: qwwdfsad

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

@@ -44,8 +44,14 @@ class ArrayBroadcastChannel<E>(
         require(capacity >= 1) { "ArrayBroadcastChannel capacity must be at least 1, but $capacity was specified" }
     }
 
+    /*
+     *  Writes to buffer are guarded by bufferLock, but reads from buffer are concurrent with writes
+     *    - Write element to buffer then write "tail" (volatile)
+     *    - Read "tail" (volatile), then read element from buffer
+     *  So read/writes to buffer need not be volatile
+     */
     private val bufferLock = ReentrantLock()
-    private val buffer = arrayOfNulls<Any?>(capacity) // guarded by bufferLock
+    private val buffer = arrayOfNulls<Any?>(capacity)
 
     // head & tail are Long (64 bits) and we assume that they never wrap around
     // head, tail, and size are guarded by bufferLock
@@ -56,14 +62,7 @@ class ArrayBroadcastChannel<E>(
     @Volatile
     private var size: Int = 0
 
-    /*
-        Writes to buffer are guarded by bufferLock, but reads from buffer are concurrent with writes
-          - Write element to buffer then write "tail" (volatile)
-          - Read "tail" (volatile), then read element from buffer
-        So read/writes to buffer need not be volatile
-     */
-
-    private val subs = subscriberList<Subscriber<E>>()
+    private val subscribers = subscriberList<Subscriber<E>>()
 
     override val isBufferAlwaysFull: Boolean get() = false
     override val isBufferFull: Boolean get() = size >= capacity
@@ -81,7 +80,7 @@ class ArrayBroadcastChannel<E>(
 
     public override fun cancel(cause: Throwable?): Boolean =
         close(cause).also {
-            for (sub in subs) sub.cancel(cause)
+            for (sub in subscribers) sub.cancel(cause)
         }
 
     // result is `OFFER_SUCCESS | OFFER_FAILED | Closed`
@@ -96,7 +95,7 @@ class ArrayBroadcastChannel<E>(
             this.size = size + 1
             this.tail = tail + 1
         }
-        // if offered successfully, then check subs outside of lock
+        // if offered successfully, then check subscribers outside of lock
         checkSubOffers()
         return OFFER_SUCCESS
     }
@@ -117,7 +116,7 @@ class ArrayBroadcastChannel<E>(
             this.size = size + 1
             this.tail = tail + 1
         }
-        // if offered successfully, then check subs outside of lock
+        // if offered successfully, then check subscribers outside of lock
         checkSubOffers()
         return OFFER_SUCCESS
     }
@@ -126,7 +125,7 @@ class ArrayBroadcastChannel<E>(
         var updated = false
         var hasSubs = false
         @Suppress("LoopToCallChain") // must invoke `checkOffer` on every sub
-        for (sub in subs) {
+        for (sub in subscribers) {
             hasSubs = true
             if (sub.checkOffer()) updated = true
         }
@@ -142,12 +141,12 @@ class ArrayBroadcastChannel<E>(
         bufferLock.withLock {
             if (addSub != null) {
                 addSub.subHead = tail // start from last element
-                val wasEmpty = subs.isEmpty()
-                subs.add(addSub)
+                val wasEmpty = subscribers.isEmpty()
+                subscribers.add(addSub)
                 if (!wasEmpty) return // no need to update when adding second and etc sub
             }
             if (removeSub != null) {
-                subs.remove(removeSub)
+                subscribers.remove(removeSub)
                 if (head != removeSub.subHead) return // no need to update
             }
             val minHead = computeMinHead()
@@ -190,7 +189,7 @@ class ArrayBroadcastChannel<E>(
 
     private fun computeMinHead(): Long {
         var minHead = Long.MAX_VALUE
-        for (sub in subs)
+        for (sub in subscribers)
             minHead = minHead.coerceAtMost(sub.subHead) // volatile (atomic) reads of subHead
         return minHead
     }

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

@@ -21,6 +21,30 @@ import kotlin.coroutines.experimental.*
 import kotlin.test.*
 
 class ArrayBroadcastChannelTest : TestBase() {
+
+    @Test
+    fun testConcurrentModification() = runTest {
+        val channel = ArrayBroadcastChannel<Int>(1)
+        val s1 = channel.openSubscription()
+        val s2 = channel.openSubscription()
+
+        val job1 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(1)
+            s1.receive()
+            s1.cancel()
+        }
+
+        val job2 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            s2.receive()
+        }
+
+        expect(3)
+        channel.send(1)
+        joinAll(job1, job2)
+        finish(4)
+    }
+
     @Test
     fun testBasic() = runTest {
         expect(1)

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

@@ -22,6 +22,29 @@ import kotlin.test.*
 
 class ConflatedBroadcastChannelTest : TestBase() {
 
+    @Test
+    fun testConcurrentModification() = runTest {
+        val channel = ConflatedBroadcastChannel<Int>()
+        val s1 = channel.openSubscription()
+        val s2 = channel.openSubscription()
+
+        val job1 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(1)
+            s1.receive()
+            s1.cancel()
+        }
+
+        val job2 = launch(Unconfined, CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            s2.receive()
+        }
+
+        expect(3)
+        channel.send(1)
+        joinAll(job1, job2)
+        finish(4)
+    }
+
     @Test
     fun testBasicScenario() = runTest {
         expect(1)

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

@@ -9,4 +9,4 @@ internal class NoOpLock {
     fun unlock(): Unit {}
 }
 
-internal actual fun <E> subscriberList(): MutableList<E> = ArrayList()
+internal actual fun <E> subscriberList(): SubscribersList<E> = CopyOnWriteList()

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

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2016-2018 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.internal
+
+/**
+ * Analogue of java.util.concurrent.CopyOnWriteArrayList for JS.
+ * Even though JS has no real concurrency, [CopyOnWriteList] is essential to manage any kinds
+ * of callbacks or continuations.
+ *
+ * Implementation note: most of the methods fallbacks to [AbstractMutableList] (thus inefficient for CoW pattern)
+ * and some methods are unsupported, because currently they are not required for this class consumers.
+ */
+internal class CopyOnWriteList<E>(private var array: Array<E> = emptyArray()) : AbstractMutableList<E>() {
+
+    override val size: Int get() = array.size
+
+    override fun add(element: E): Boolean {
+        val copy = array.asDynamic().slice()
+        copy.push(element)
+        array = copy as Array<E>
+        return true
+    }
+
+    override fun add(index: Int, element: E) {
+        val copy = array.asDynamic().slice()
+        copy.splice(insertionRangeCheck(index), 0, element)
+        array = copy as Array<E>
+    }
+
+    override fun remove(element: E): Boolean {
+        for (index in array.indices) {
+            if (array[index] == element) {
+                val copy = array.asDynamic().slice()
+                copy.splice(index, 1)
+                array = copy as Array<E>
+                return true
+            }
+        }
+
+        return false
+    }
+
+    override fun removeAt(index: Int): E {
+        modCount++
+        val copy = array.asDynamic().slice()
+        val result = if (index == lastIndex) {
+            copy.pop()
+        } else {
+            copy.splice(index, 1)[0]
+        }
+
+        array = copy as Array<E>
+        return result as E
+    }
+
+    override fun iterator(): MutableIterator<E> = IteratorImpl(array)
+
+    override fun listIterator(): MutableListIterator<E> = throw UnsupportedOperationException("Operation is not supported")
+
+    override fun listIterator(index: Int): MutableListIterator<E> = throw UnsupportedOperationException("Operation is not supported")
+
+    override fun isEmpty(): Boolean = size == 0
+
+    override fun set(index: Int, element: E): E = throw UnsupportedOperationException("Operation is not supported")
+
+    override fun get(index: Int): E = array[rangeCheck(index)]
+
+    private class IteratorImpl<E>(private var array: Array<E>) : MutableIterator<E> {
+
+        private var current = 0
+
+        override fun hasNext(): Boolean = current != array.size
+
+        override fun next(): E {
+            if (!hasNext()) {
+                throw NoSuchElementException()
+            }
+
+            return array[current++]
+        }
+
+        override fun remove() = throw UnsupportedOperationException("Operation is not supported")
+    }
+
+    private fun insertionRangeCheck(index: Int) {
+        if (index < 0 || index > size) {
+            throw IndexOutOfBoundsException("index: $index, size: $size")
+        }
+    }
+
+    private fun rangeCheck(index: Int) = index.apply {
+        if (index < 0 || index >= size) {
+            throw IndexOutOfBoundsException("index: $index, size: $size")
+        }
+    }
+}