Fixed JobSupport single->list upgrade procedure.

It was hanging at LockFreeLinkedList.helpDelete.

Author: elizarov

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

@@ -528,12 +528,45 @@ 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()
+            // See detailed comment in findHead on why `prev === this` is a special case for which we know that
+            // the prev should have being pointing to the head of list but finishAdd that was supposed
+            // to do that is not complete yet.
+            val removedPrev = (if (prev === this) findHead() else (prev as Node)).removed()
             if (_prev.compareAndSet(prev, removedPrev)) return prev
         }
     }
 
+    /**
+     * Finds the head of the list (implementing [LockFreeLinkedListHead]) by following [next] pointers.
+     *
+     * The code in [kotlinx.coroutines.experimental.JobSupport] performs upgrade of a single node to a list.
+     * It uses [addOneIfEmpty] to add the list head to "empty list of a single node" once.
+     * During upgrade a transient state of the list looks like this:
+     *
+     * ```
+     *                +-----------------+
+     *                |                 |
+     *          node  V       head      |
+     *          +---+---+     +---+---+ |
+     *      +-> | P | N | --> | P | N |-+
+     *      |   +---+---+     +---+---+
+     *      |     |   ^         |
+     *      +---- +   +---------+
+     * ```
+     *
+     * The [prev] pointer in `node` still points to itself when [finishAdd] (invoked inside [addOneIfEmpty])
+     * has not completed yet. If this state is observed, then we know that [prev] should have been pointing
+     * to the list head. This function is looking up the head by following consistent chain of [next] pointers.
+     */
+    private fun findHead(): Node {
+        var cur = this
+        while (true) {
+            if (cur is LockFreeLinkedListHead) return cur
+            cur = cur.next.unwrap()
+            check(cur !== this) { "Cannot loop to this while looking for list head" }
+        }
+    }
+
     // fixes next links to the left of this node
     @PublishedApi
     internal fun helpDelete() {

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

@@ -55,7 +55,7 @@ class JobHandlersUpgradeStressTest : TestBase() {
                 cyclicBarrier.await()
                 val job = job ?: break
                 // burn some time
-                repeat(rnd.nextInt(30)) { sink.incrementAndGet() }
+                repeat(rnd.nextInt(3000)) { sink.incrementAndGet() }
                 // cancel job
                 job.cancel()
                 cyclicBarrier.await()
@@ -72,14 +72,14 @@ class JobHandlersUpgradeStressTest : TestBase() {
                     val job = job ?: break
                     val state = State()
                     // burn some time
-                    repeat(rnd.nextInt(10)) { sink.incrementAndGet() }
+                    repeat(rnd.nextInt(1000)) { 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() }
+                    repeat(rnd.nextInt(1000)) { sink.incrementAndGet() }
                     // dispose
                     handle.dispose()
                     cyclicBarrier.await()