JobHandlersUpgradeStressTest introduced

Author: elizarov

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

@@ -189,6 +189,29 @@ public open class LockFreeLinkedListNode {
 
     // ------ addXXX util ------
 
+    /**
+     * Given:
+     * ```
+     *                +-----------------------+
+     *          this  |         node          V  next
+     *          +---+---+     +---+---+     +---+---+
+     *  ... <-- | P | N |     | P | N |     | P | N | --> ....
+     *          +---+---+     +---+---+     +---+---+
+     *                ^                       |
+     *                +-----------------------+
+     * ```
+     * Produces:
+     * ```
+     *          this            node             next
+     *          +---+---+     +---+---+     +---+---+
+     *  ... <-- | P | N | ==> | P | N | --> | P | N | --> ....
+     *          +---+---+     +---+---+     +---+---+
+     *                ^         |   ^         |
+     *                +---------+   +---------+
+     * ```
+     *  Where `==>` denotes linearization point.
+     *  Returns `false` if `next` was not following `this` node.
+     */
     @PublishedApi
     internal fun addNext(node: Node, next: Node): Boolean {
         node._prev.lazySet(this)
@@ -462,6 +485,28 @@ public open class LockFreeLinkedListNode {
 
     // ------ other helpers ------
 
+    /**
+     * Given:
+     * ```
+     *
+     *          prev            this             next
+     *          +---+---+     +---+---+     +---+---+
+     *  ... <-- | P | N | --> | P | N | --> | P | N | --> ....
+     *          +---+---+     +---+---+     +---+---+
+     *              ^ ^         |             |
+     *              | +---------+             |
+     *              +-------------------------+
+     * ```
+     * Produces:
+     * ```
+     *          prev            this             next
+     *          +---+---+     +---+---+     +---+---+
+     *  ... <-- | P | N | --> | P | N | --> | P | N | --> ....
+     *          +---+---+     +---+---+     +---+---+
+     *                ^         |   ^         |
+     *                +---------+   +---------+
+     * ```
+     */
     private fun finishAdd(next: Node) {
         next._prev.loop { nextPrev ->
             if (nextPrev is Removed || this.next !== next) return // next was removed, remover fixes up links
@@ -483,6 +528,7 @@ public open class LockFreeLinkedListNode {
     private fun markPrev(): Node {
         _prev.loop { prev ->
             if (prev is Removed) return prev.ref
+            check(prev !== this) { "Cannot remove node that was not fully added" }
             val removedPrev = (prev as Node).removed()
             if (_prev.compareAndSet(prev, removedPrev)) return prev
         }

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/JobHandlersUpgradeStressTest.kt

@@ -0,0 +1,109 @@
+/*
+ * 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
+
+import kotlinx.atomicfu.*
+import java.util.*
+import java.util.concurrent.*
+import kotlin.concurrent.*
+import kotlin.test.*
+
+class JobHandlersUpgradeStressTest : TestBase() {
+    private val nSeconds = 3 * stressTestMultiplier
+    private val nThreads = 4
+
+    private val cyclicBarrier = CyclicBarrier(1 + nThreads)
+    private val threads = mutableListOf<Thread>()
+
+    private val inters = atomic(0)
+    private val removed = atomic(0)
+    private val fired = atomic(0)
+
+    private val sink = atomic(0)
+
+    @Volatile
+    private var done = false
+
+    @Volatile
+    private var job: Job? = null
+
+    class State {
+        val state = atomic(0)
+    }
+
+    @Test
+    fun testStress() {
+        println("--- JobHandlersUpgradeStressTest")
+        threads += thread(name = "creator", start = false) {
+            val rnd = Random()
+            while (true) {
+                job = if (done) null else Job()
+                cyclicBarrier.await()
+                val job = job ?: break
+                // burn some time
+                repeat(rnd.nextInt(30)) { sink.incrementAndGet() }
+                // cancel job
+                job.cancel()
+                cyclicBarrier.await()
+                inters.incrementAndGet()
+            }
+        }
+        threads += List(nThreads) { threadId ->
+            thread(name = "handler-$threadId", start = false) {
+                val rnd = Random()
+                while (true) {
+                    val onCancelling = rnd.nextBoolean()
+                    val invokeImmediately: Boolean = rnd.nextBoolean()
+                    cyclicBarrier.await()
+                    val job = job ?: break
+                    val state = State()
+                    // burn some time
+                    repeat(rnd.nextInt(10)) { sink.incrementAndGet() }
+                    val handle =
+                        job.invokeOnCompletion(onCancelling = onCancelling, invokeImmediately = invokeImmediately) {
+                            if (!state.state.compareAndSet(0, 1))
+                                error("Fired more than once or too late: state=${state.state.value}")
+                        }
+                    // burn some time
+                    repeat(rnd.nextInt(10)) { sink.incrementAndGet() }
+                    // dispose
+                    handle.dispose()
+                    cyclicBarrier.await()
+                    val resultingState = state.state.value
+                    when (resultingState) {
+                        0 -> removed.incrementAndGet()
+                        1 -> fired.incrementAndGet()
+                        else -> error("Cannot happen")
+                    }
+                    if (!state.state.compareAndSet(resultingState, 2))
+                        error("Cannot fire late: resultingState=$resultingState")
+                }
+            }
+        }
+        threads.forEach { it.start() }
+        repeat(nSeconds) { second ->
+            Thread.sleep(1000)
+            println("${second + 1}: ${inters.value} iterations")
+        }
+        done = true
+        threads.forEach { it.join() }
+        println("        Completed ${inters.value} iterations")
+        println("  Removed handler ${removed.value} times")
+        println("    Fired handler ${fired.value} times")
+
+    }
+}