Remove DCSS

Author: dkhalanskyjb

kotlinx-coroutines-core/common/src/JobSupport.kt

@@ -159,12 +159,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
      * If final state of the job is [Incomplete], then it is boxed into [IncompleteStateBox]
      * and should be [unboxed][unboxState] before returning to user code.
      */
-    internal val state: Any? get() {
-        _state.loop { state -> // helper loop on state (complete in-progress atomic operations)
-            if (state !is OpDescriptor) return state
-            state.perform(this)
-        }
-    }
+    internal val state: Any? get() = _state.value
 
     /**
      * @suppress **This is unstable API and it is subject to change.**
@@ -324,6 +319,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     private fun notifyCancelling(list: NodeList, cause: Throwable) {
         // first cancel our own children
         onCancelling(cause)
+        list.closeForSome()
         notifyHandlers(list, cause) { it.onCancelling }
         // then cancel parent
         cancelParent(cause) // tentative cancellation -- does not matter if there is no parent
@@ -355,8 +351,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         return parent.childCancelled(cause) || isCancellation
     }
 
-    private fun NodeList.notifyCompletion(cause: Throwable?) =
+    private fun NodeList.notifyCompletion(cause: Throwable?) {
+        close()
         notifyHandlers(this, cause) { true }
+    }
 
     private inline fun notifyHandlers(list: NodeList, cause: Throwable?, predicate: (JobNode) -> Boolean) {
         var exception: Throwable? = null
@@ -488,7 +486,11 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                                 // or we are adding a child to a coroutine that is not completing yet
                                 if (rootCause == null || node is ChildHandleNode && !state.isCompleting) {
                                     // Note: add node the list while holding lock on state (make sure it cannot change)
-                                    if (!addLastAtomic(state, list, node)) return@loopOnState // retry
+                                    if (!list.addLast(
+                                            node,
+                                            allowedAfterPartialClosing = node is ChildHandleNode
+                                        )
+                                    ) return@loopOnState // retry
                                     // just return node if we don't have to invoke handler (not cancelling yet)
                                     if (rootCause == null) return node
                                     // otherwise handler is invoked immediately out of the synchronized section & handle returned
@@ -500,8 +502,24 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                             // Note: attachChild uses invokeImmediately, so it gets invoked when adding to cancelled job
                             if (invokeImmediately) node.invoke(rootCause)
                             return handle
-                        } else {
-                            if (addLastAtomic(state, list, node)) return node
+                        } else if (list.addLast(
+                                node, allowedAfterPartialClosing = !node.onCancelling || node is ChildHandleNode
+                        )) {
+                            if (node is ChildHandleNode) {
+                                /** Handling the following case:
+                                 * - A child requested to be added to the list;
+                                 * - We checked the state and saw that it wasn't `Finishing`;
+                                 * - Then, the job got cancelled and notified everyone about it;
+                                 * - Only then did we add the child to the list
+                                 * - and ended up here.
+                                 */
+                                val latestState = this@JobSupport.state
+                                if (latestState is Finishing) {
+                                    assert { invokeImmediately }
+                                    synchronized(latestState) { latestState.rootCause }?.let { node.invoke(it) }
+                                }
+                            }
+                            return node
                         }
                     }
                 }
@@ -515,9 +533,6 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         }
     }
 
-    private fun addLastAtomic(expect: Any, list: NodeList, node: JobNode) =
-        list.addLastIf(node) { this.state === expect }
-
     private fun promoteEmptyToNodeList(state: Empty) {
         // try to promote it to LIST state with the corresponding state
         val list = NodeList()

kotlinx-coroutines-core/common/src/internal/Atomic.kt

@@ -2,22 +2,18 @@
 
 package kotlinx.coroutines.internal
 
-import kotlinx.coroutines.*
-import kotlin.jvm.*
-
 /** @suppress **This is unstable API and it is subject to change.** */
 public expect open class LockFreeLinkedListNode() {
     public val isRemoved: Boolean
     public val nextNode: LockFreeLinkedListNode
     public val prevNode: LockFreeLinkedListNode
+    public fun addLast(node: LockFreeLinkedListNode, allowedAfterPartialClosing: Boolean): Boolean
     public fun addOneIfEmpty(node: LockFreeLinkedListNode): Boolean
-    public inline fun addLastIf(node: LockFreeLinkedListNode, crossinline condition: () -> Boolean): Boolean
     public open fun remove(): Boolean
-
+    public fun close()
+    public fun closeForSome()
 }
 
-internal fun LockFreeLinkedListNode.addLast(node: LockFreeLinkedListNode) = addLastIf(node) { true }
-
 /** @suppress **This is unstable API and it is subject to change.** */
 public expect open class LockFreeLinkedListHead() : LockFreeLinkedListNode {
     public inline fun forEach(block: (LockFreeLinkedListNode) -> Unit)

kotlinx-coroutines-core/common/src/internal/LockFreeLinkedList.common.kt

@@ -174,6 +174,20 @@ class JobTest : TestBase() {
         finish(4)
     }
 
+    @Test
+    fun testInvokeOnCancellingFiringOnNormalExit() = runTest {
+        val job = launch {
+            expect(2)
+        }
+        job.invokeOnCompletion(onCancelling = true) {
+            assertNull(it)
+            expect(3)
+        }
+        expect(1)
+        job.join()
+        finish(4)
+    }
+
     @Test
     fun testOverriddenParent() = runTest {
         val parent = Job()

kotlinx-coroutines-core/common/test/JobTest.kt

@@ -8,18 +8,6 @@ import kotlin.jvm.*
 
 private typealias Node = LockFreeLinkedListNode
 
-@PublishedApi
-internal const val UNDECIDED: Int = 0
-
-@PublishedApi
-internal const val SUCCESS: Int = 1
-
-@PublishedApi
-internal const val FAILURE: Int = 2
-
-@PublishedApi
-internal val CONDITION_FALSE: Any = Symbol("CONDITION_FALSE")
-
 /**
  * Doubly-linked concurrent list node with remove support.
  * Based on paper
@@ -49,37 +37,10 @@ public actual open class LockFreeLinkedListNode {
     private fun removed(): Removed =
         _removedRef.value ?: Removed(this).also { _removedRef.lazySet(it) }
 
-    @PublishedApi
-    internal abstract class CondAddOp(
-        @JvmField val newNode: Node
-    ) : AtomicOp<Node>() {
-        @JvmField var oldNext: Node? = null
-
-        override fun complete(affected: Node, failure: Any?) {
-            val success = failure == null
-            val update = if (success) newNode else oldNext
-            if (update != null && affected._next.compareAndSet( this, update)) {
-                // only the thread the makes this update actually finishes add operation
-                if (success) newNode.finishAdd(oldNext!!)
-            }
-        }
-    }
-
-    @PublishedApi
-    internal inline fun makeCondAddOp(node: Node, crossinline condition: () -> Boolean): CondAddOp =
-        object : CondAddOp(node) {
-            override fun prepare(affected: Node): Any? = if (condition()) null else CONDITION_FALSE
-        }
-
     public actual open val isRemoved: Boolean get() = next is Removed
 
     // LINEARIZABLE. Returns Node | Removed
-    public val next: Any get() {
-        _next.loop { next ->
-            if (next !is OpDescriptor) return next
-            next.perform(this)
-        }
-    }
+    public val next: Any get() = _next.value
 
     // LINEARIZABLE. Returns next non-removed Node
     public actual val nextNode: Node get() =
@@ -117,20 +78,30 @@ public actual open class LockFreeLinkedListNode {
     // ------ addLastXXX ------
 
     /**
-     * Adds last item to this list atomically if the [condition] is true.
+     * Adds last item to this list. Returns `false` if the list is closed.
      */
-    public actual inline fun addLastIf(node: Node, crossinline condition: () -> Boolean): Boolean {
-        val condAdd = makeCondAddOp(node, condition)
+    public actual fun addLast(node: Node, allowedAfterPartialClosing: Boolean): Boolean {
         while (true) { // lock-free loop on prev.next
-            val prev = prevNode // sentinel node is never removed, so prev is always defined
-            when (prev.tryCondAddNext(node, this, condAdd)) {
-                SUCCESS -> return true
-                FAILURE -> return false
+            val currentPrev = prevNode
+            return when {
+                currentPrev is ListClosedForAll -> false
+                currentPrev is ListClosedForSome ->
+                    allowedAfterPartialClosing && currentPrev.addLast(node, allowedAfterPartialClosing)
+                currentPrev.addNext(node, this) -> true
+                else -> continue
             }
         }
     }
 
-    // ------ addXXX util ------
+    /**
+     * Forbids adding some of the new items to this list.
+     */
+    public actual fun closeForSome() { addLast(ListClosedForSome(), allowedAfterPartialClosing = false) }
+
+    /**
+     * Forbids adding new items to this list.
+     */
+    public actual fun close() { addLast(ListClosedForAll(), allowedAfterPartialClosing = true) }
 
     /**
      * Given:
@@ -165,17 +136,6 @@ public actual open class LockFreeLinkedListNode {
         return true
     }
 
-    // returns UNDECIDED, SUCCESS or FAILURE
-    @PublishedApi
-    internal fun tryCondAddNext(node: Node, next: Node, condAdd: CondAddOp): Int {
-        node._prev.lazySet(this)
-        node._next.lazySet(next)
-        condAdd.oldNext = next
-        if (!_next.compareAndSet(next, condAdd)) return UNDECIDED
-        // added operation successfully (linearized) -- complete it & fixup the list
-        return if (condAdd.perform(this) == null) SUCCESS else FAILURE
-    }
-
     // ------ removeXXX ------
 
     /**
@@ -273,10 +233,6 @@ public actual open class LockFreeLinkedListNode {
                 }
                 // slow path when we need to help remove operations
                 this.isRemoved -> return null // nothing to do, this node was removed, bail out asap to save time
-                prevNext is OpDescriptor -> { // help & retry
-                    prevNext.perform(prev)
-                    return correctPrev() // retry from scratch
-                }
                 prevNext is Removed -> {
                     if (last !== null) {
                         // newly added (prev) node is already removed, correct last.next around it
@@ -332,3 +288,7 @@ public actual open class LockFreeLinkedListHead : LockFreeLinkedListNode() {
     // optimization: because head is never removed, we don't have to read _next.value to check these:
     override val isRemoved: Boolean get() = false
 }
+
+private class ListClosedForSome: LockFreeLinkedListNode()
+
+private class ListClosedForAll: LockFreeLinkedListNode()

kotlinx-coroutines-core/concurrent/src/internal/LockFreeLinkedList.kt

@@ -14,13 +14,24 @@ public actual open class LockFreeLinkedListNode {
     inline actual val prevNode get() = _prev
     inline actual val isRemoved get() = _removed
 
-    @PublishedApi
-    internal fun addLast(node: Node) {
-        val prev = this._prev
-        node._next = this
-        node._prev = prev
-        prev._next = node
-        this._prev = node
+    public actual fun addLast(node: Node, allowedAfterPartialClosing: Boolean): Boolean = when (val prev = this._prev) {
+        is ListClosedForAll -> false
+        is ListClosedForSome -> allowedAfterPartialClosing && prev.addLast(node, allowedAfterPartialClosing)
+        else -> {
+            node._next = this
+            node._prev = prev
+            prev._next = node
+            this._prev = node
+            true
+        }
+    }
+
+    public actual fun closeForSome() {
+        addLast(ListClosedForSome(), allowedAfterPartialClosing = true)
+    }
+
+    public actual fun close() {
+        addLast(ListClosedForAll(), allowedAfterPartialClosing = false)
     }
 
     /*
@@ -30,10 +41,6 @@ public actual open class LockFreeLinkedListNode {
      * invokes handler on remove
      */
     public actual open fun remove(): Boolean {
-        return removeImpl()
-    }
-
-    private fun removeImpl(): Boolean {
         if (_removed) return false
         val prev = this._prev
         val next = this._next
@@ -45,13 +52,7 @@ public actual open class LockFreeLinkedListNode {
 
     public actual fun addOneIfEmpty(node: Node): Boolean {
         if (_next !== this) return false
-        addLast(node)
-        return true
-    }
-
-    public actual inline fun addLastIf(node: Node, crossinline condition: () -> Boolean): Boolean {
-        if (!condition()) return false
-        addLast(node)
+        addLast(node, allowedAfterPartialClosing = false)
         return true
     }
 }
@@ -72,3 +73,7 @@ public actual open class LockFreeLinkedListHead : Node() {
     // just a defensive programming -- makes sure that list head sentinel is never removed
     public actual final override fun remove(): Nothing = throw UnsupportedOperationException()
 }
+
+private class ListClosedForSome: LockFreeLinkedListNode()
+
+private class ListClosedForAll: LockFreeLinkedListNode()