Optimize select expression registration phase (#1445)

* Optimize select expression registration phase

There is no need for multi-word atomic performAtomicIfNotSelected
operation when enqueuing select node for operation. We can simply
enqueue (addLast) xxxSelect node (SendSelect, ReceiveSelect, LockSelect).
If the coroutine that rendezvous with this node finds out that the
select expression was already selected, then it'll try again.

* Removed SelectInstance.performAtomicIfNotSelected function
* Removed Mutex.TryEnqueueLockDesc class, simplified onLock
* Removed AbstractSendChannel.TryEnqueueSendDesc class, simpler onSend
* Removed AbstractChannel.TryEnqueueReceiveDesc class, simpler onReceive
* Simplified SelectInstance.disposeOnSelect. It does not have to
  do atomic addLastIf operation. Can do a simple addLast.

* Fixed unlimited channel select onSend on closed channel

It was hanging not being able to properly see that the channel was
already closed at that send attempt should fail.

Author: elizarov

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -1046,7 +1046,6 @@ public final class kotlinx/coroutines/selects/SelectBuilderImpl : kotlinx/corout
 	public fun invoke (Lkotlinx/coroutines/selects/SelectClause2;Lkotlin/jvm/functions/Function2;)V
 	public fun isSelected ()Z
 	public fun onTimeout (JLkotlin/jvm/functions/Function1;)V
-	public fun performAtomicIfNotSelected (Lkotlinx/coroutines/internal/AtomicDesc;)Ljava/lang/Object;
 	public fun performAtomicTrySelect (Lkotlinx/coroutines/internal/AtomicDesc;)Ljava/lang/Object;
 	public fun resumeSelectCancellableWithException (Ljava/lang/Throwable;)V
 	public fun resumeWith (Ljava/lang/Object;)V
@@ -1069,7 +1068,6 @@ public abstract interface class kotlinx/coroutines/selects/SelectInstance {
 	public abstract fun disposeOnSelect (Lkotlinx/coroutines/DisposableHandle;)V
 	public abstract fun getCompletion ()Lkotlin/coroutines/Continuation;
 	public abstract fun isSelected ()Z
-	public abstract fun performAtomicIfNotSelected (Lkotlinx/coroutines/internal/AtomicDesc;)Ljava/lang/Object;
 	public abstract fun performAtomicTrySelect (Lkotlinx/coroutines/internal/AtomicDesc;)Ljava/lang/Object;
 	public abstract fun resumeSelectCancellableWithException (Ljava/lang/Throwable;)V
 	public abstract fun trySelect (Ljava/lang/Object;)Z

kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.internal
@@ -66,7 +66,7 @@ public expect open class RemoveFirstDesc<T>(queue: LockFreeLinkedListNode): Abst
 public expect abstract class AbstractAtomicDesc : AtomicDesc {
     final override fun prepare(op: AtomicOp<*>): Any?
     final override fun complete(op: AtomicOp<*>, failure: Any?)
-    protected open fun failure(affected: LockFreeLinkedListNode, next: Any): Any?
+    protected open fun failure(affected: LockFreeLinkedListNode): Any?
     protected open fun retry(affected: LockFreeLinkedListNode, next: Any): Boolean
     protected abstract fun onPrepare(affected: LockFreeLinkedListNode, next: LockFreeLinkedListNode): Any? // non-null on failure
     protected abstract fun finishOnSuccess(affected: LockFreeLinkedListNode, next: LockFreeLinkedListNode)

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

@@ -110,11 +110,6 @@ public interface SelectInstance<in R> {
      */
     public fun performAtomicTrySelect(desc: AtomicDesc): Any?
 
-    /**
-     * Performs action atomically when [isSelected] is `false`.
-     */
-    public fun performAtomicIfNotSelected(desc: AtomicDesc): Any?
-
     /**
      * Returns completion continuation of this select instance.
      * This select instance must be _selected_ first.
@@ -129,6 +124,7 @@ public interface SelectInstance<in R> {
 
     /**
      * Disposes the specified handle when this instance is selected.
+     * Note, that [DisposableHandle.dispose] could be called multiple times.
      */
     public fun disposeOnSelect(handle: DisposableHandle)
 }
@@ -329,16 +325,14 @@ internal class SelectBuilderImpl<in R>(
 
     override fun disposeOnSelect(handle: DisposableHandle) {
         val node = DisposeNode(handle)
-        while (true) { // lock-free loop on state
-            val state = this.state
-            if (state === this) {
-                if (addLastIf(node, { this.state === this }))
-                    return
-            } else { // already selected
-                handle.dispose()
-                return
-            }
+        // check-add-check pattern is Ok here since handle.dispose() is safe to be called multiple times
+        if (!isSelected) {
+            addLast(node) // add handle to list
+            // double-check node after adding
+            if (!isSelected) return // all ok - still not selected
         }
+        // already selected
+        handle.dispose()
     }
 
     private fun doAfterSelect() {
@@ -368,12 +362,11 @@ internal class SelectBuilderImpl<in R>(
         }
     }
 
-    override fun performAtomicTrySelect(desc: AtomicDesc): Any? = AtomicSelectOp(desc, true).perform(null)
-    override fun performAtomicIfNotSelected(desc: AtomicDesc): Any? = AtomicSelectOp(desc, false).perform(null)
+    override fun performAtomicTrySelect(desc: AtomicDesc): Any? =
+        AtomicSelectOp(desc).perform(null)
 
     private inner class AtomicSelectOp(
-        @JvmField val desc: AtomicDesc,
-        @JvmField val select: Boolean
+        @JvmField val desc: AtomicDesc
     ) : AtomicOp<Any?>() {
         override fun prepare(affected: Any?): Any? {
             // only originator of operation makes preparation move of installing descriptor into this selector's state
@@ -405,7 +398,7 @@ internal class SelectBuilderImpl<in R>(
         }
 
         private fun completeSelect(failure: Any?) {
-            val selectSuccess = select && failure == null
+            val selectSuccess = failure == null
             val update = if (selectSuccess) null else this@SelectBuilderImpl
             if (_state.compareAndSet(this@AtomicSelectOp, update)) {
                 if (selectSuccess)

kotlinx-coroutines-core/common/src/selects/Select.kt

@@ -246,16 +246,11 @@ internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2<Any?, Mutex> {
                 }
                 is LockedQueue -> {
                     check(state.owner !== owner) { "Already locked by $owner" }
-                    val enqueueOp = TryEnqueueLockDesc(this, owner, state, select, block)
-                    val failure = select.performAtomicIfNotSelected(enqueueOp)
-                    when {
-                        failure == null -> { // successfully enqueued
-                            select.disposeOnSelect(enqueueOp.node)
-                            return
-                        }
-                        failure === ALREADY_SELECTED -> return // already selected -- bail out
-                        failure === ENQUEUE_FAIL -> {} // retry
-                        else -> error("performAtomicIfNotSelected(TryEnqueueLockDesc) returned $failure")
+                    val node = LockSelect(owner, this, select, block)
+                    if (state.addLastIf(node) { _state.value === state }) {
+                        // successfully enqueued
+                        select.disposeOnSelect(node)
+                        return
                     }
                 }
                 is OpDescriptor -> state.perform(this) // help
@@ -291,19 +286,6 @@ internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2<Any?, Mutex> {
         }
     }
 
-    private class TryEnqueueLockDesc<R>(
-        @JvmField val mutex: MutexImpl,
-        owner: Any?,
-        queue: LockedQueue,
-        select: SelectInstance<R>,
-        block: suspend (Mutex) -> R
-    ) : AddLastDesc<LockSelect<R>>(queue, LockSelect(owner, mutex, select, block)) {
-        override fun onPrepare(affected: LockFreeLinkedListNode, next: LockFreeLinkedListNode): Any? {
-            if (mutex._state.value !== queue) return ENQUEUE_FAIL
-            return super.onPrepare(affected, next)
-        }
-    }
-
     public override fun holdsLock(owner: Any) =
             _state.value.let { state ->
                 when (state) {

kotlinx-coroutines-core/common/src/sync/Mutex.kt

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.selects
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
+import kotlin.test.*
+
+class SelectLinkedListChannelTest : TestBase() {
+    @Test
+    fun testSelectSendWhenClosed() = runTest {
+        expect(1)
+        val c = Channel<Int>(Channel.UNLIMITED)
+        c.send(1) // enqueue buffered element
+        c.close() // then close
+        assertFailsWith<ClosedSendChannelException> {
+            // select sender should fail
+            expect(2)
+            select {
+                c.onSend(2) {
+                    expectUnreached()
+                }
+            }
+        }
+        finish(3)
+    }
+}

kotlinx-coroutines-core/common/test/selects/SelectLinkedListChannelTest.kt

@@ -403,6 +403,29 @@ class SelectRendezvousChannelTest : TestBase() {
         finish(10)
     }
 
+    @Test
+    fun testSelectSendWhenClosed() = runTest {
+        expect(1)
+        val c = Channel<Int>(Channel.RENDEZVOUS)
+        val sender = launch(start = CoroutineStart.UNDISPATCHED) {
+            expect(2)
+            c.send(1) // enqueue sender
+            expectUnreached()
+        }
+        c.close() // then close
+        assertFailsWith<ClosedSendChannelException> {
+            // select sender should fail
+            expect(3)
+            select {
+                c.onSend(2) {
+                    expectUnreached()
+                }
+            }
+        }
+        sender.cancel()
+        finish(4)
+    }
+
     // only for debugging
     internal fun <R> SelectBuilder<R>.default(block: suspend () -> R) {
         this as SelectBuilderImpl // type assertion

kotlinx-coroutines-core/common/test/selects/SelectRendezvousChannelTest.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 @file:Suppress("unused")
@@ -129,13 +129,13 @@ public actual abstract class AbstractAtomicDesc : AtomicDesc() {
     actual final override fun prepare(op: AtomicOp<*>): Any? {
         val affected = affectedNode
         val next = affected._next
-        val failure = failure(affected, next)
+        val failure = failure(affected)
         if (failure != null) return failure
         return onPrepare(affected, next)
     }
 
     actual final override fun complete(op: AtomicOp<*>, failure: Any?) = onComplete()
-    protected actual open fun failure(affected: LockFreeLinkedListNode, next: Any): Any? = null // Never fails by default
+    protected actual open fun failure(affected: LockFreeLinkedListNode): Any? = null // Never fails by default
     protected actual open fun retry(affected: LockFreeLinkedListNode, next: Any): Boolean = false // Always succeeds
     protected actual abstract fun finishOnSuccess(affected: LockFreeLinkedListNode, next: LockFreeLinkedListNode)
 }

kotlinx-coroutines-core/js/src/internal/LinkedList.kt

@@ -254,26 +254,6 @@ public actual open class LockFreeLinkedListNode {
         finishRemove(removed.ref)
     }
 
-    public open fun describeRemove() : AtomicDesc? {
-        if (isRemoved) return null // fast path if was already removed
-        return object : AbstractAtomicDesc() {
-            private val _originalNext = atomic<Node?>(null)
-            override val affectedNode: Node? get() = this@LockFreeLinkedListNode
-            override val originalNext get() = _originalNext.value
-            override fun failure(affected: Node, next: Any): Any? =
-                if (next is Removed) ALREADY_REMOVED else null
-            override fun onPrepare(affected: Node, next: Node): Any? {
-                // Note: onPrepare must use CAS to make sure the stale invocation is not
-                // going to overwrite the previous decision on successful preparation.
-                // Result of CAS is irrelevant, but we must ensure that it is set when invoker completes
-                _originalNext.compareAndSet(null, next)
-                return null // always success
-            }
-            override fun updatedNext(affected: Node, next: Node) = next.removed()
-            override fun finishOnSuccess(affected: Node, next: Node) = finishRemove(next)
-        }
-    }
-
     public actual fun removeFirstOrNull(): Node? {
         while (true) { // try to linearize
             val first = next as Node
@@ -376,7 +356,7 @@ public actual open class LockFreeLinkedListNode {
         final override val originalNext: Node? get() = _originalNext.value
 
         // check node predicates here, must signal failure if affect is not of type T
-        protected override fun failure(affected: Node, next: Any): Any? =
+        protected override fun failure(affected: Node): Any? =
                 if (affected === queue) LIST_EMPTY else null
 
         // validate the resulting node (return false if it should be deleted)
@@ -408,7 +388,7 @@ public actual open class LockFreeLinkedListNode {
         protected abstract val affectedNode: Node?
         protected abstract val originalNext: Node?
         protected open fun takeAffectedNode(op: OpDescriptor): Node = affectedNode!!
-        protected open fun failure(affected: Node, next: Any): Any? = null // next: Node | Removed
+        protected open fun failure(affected: Node): Any? = null // next: Node | Removed
         protected open fun retry(affected: Node, next: Any): Boolean = false // next: Node | Removed
         protected abstract fun onPrepare(affected: Node, next: Node): Any? // non-null on failure
         protected abstract fun updatedNext(affected: Node, next: Node): Any
@@ -460,7 +440,7 @@ public actual open class LockFreeLinkedListNode {
                     continue // and retry
                 }
                 // next: Node | Removed
-                val failure = failure(affected, next)
+                val failure = failure(affected)
                 if (failure != null) return failure // signal failure
                 if (retry(affected, next)) continue // retry operation
                 val prepareOp = PrepareOp(next as Node, op as AtomicOp<Node>, this)
@@ -684,8 +664,6 @@ public actual open class LockFreeLinkedListHead : LockFreeLinkedListNode() {
     // just a defensive programming -- makes sure that list head sentinel is never removed
     public actual final override fun remove(): Boolean = throw UnsupportedOperationException()
 
-    public final override fun describeRemove(): Nothing = throw UnsupportedOperationException()
-
     internal fun validate() {
         var prev: Node = this
         var cur: Node = next as Node

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.internal
@@ -47,29 +47,6 @@ class LockFreeLinkedListTest {
         assertContents(list, 1, 3)
     }
 
-    @Test
-    fun testRemoveTwoAtomic() {
-        val list = LockFreeLinkedListHead()
-        val n1 = IntNode(1).apply { list.addLast(this) }
-        val n2 = IntNode(2).apply { list.addLast(this) }
-        assertContents(list, 1, 2)
-        assertFalse(n1.isRemoved)
-        assertFalse(n2.isRemoved)
-        val remove1Desc = n1.describeRemove()!!
-        val remove2Desc = n2.describeRemove()!!
-        val operation = object : AtomicOp<Any?>() {
-            override fun prepare(affected: Any?): Any? = remove1Desc.prepare(this) ?: remove2Desc.prepare(this)
-            override fun complete(affected: Any?, failure: Any?) {
-                remove1Desc.complete(this, failure)
-                remove2Desc.complete(this, failure)
-            }
-        }
-        assertTrue(operation.perform(null) == null)
-        assertTrue(n1.isRemoved)
-        assertTrue(n2.isRemoved)
-        assertContents(list)
-    }
-
     @Test
     fun testAtomicOpsSingle() {
         val list = LockFreeLinkedListHead()
@@ -82,16 +59,6 @@ class LockFreeLinkedListTest {
         assertContents(list, 1, 2, 3)
         val n4 = IntNode(4).also { single(list.describeAddLast(it)) }
         assertContents(list, 1, 2, 3, 4)
-        single(n3.describeRemove()!!)
-        assertContents(list, 1, 2, 4)
-        assertTrue(n3.describeRemove() == null)
-        single(list.describeRemoveFirst())
-        assertContents(list, 2, 4)
-        assertTrue(n1.describeRemove() == null)
-        assertTrue(n2.remove())
-        assertContents(list, 4)
-        assertTrue(n4.remove())
-        assertContents(list)
     }
 
     private fun single(part: AtomicDesc) {