Fix SOE with select (WIP, DO NOT MERGE)

* onSend/onReceive clauses on the same channel: Instead of
  StackOverflowError we throw IllegalStateException and leave
  the channel in the original state.
* Fix SOE in select with "opposite channels" stress-test. The fix is
  based on the sequential numbering of atomic select operation.
  Deadlock is detected and the operation with the lower sequential
  number is aborted and restarted (with a larger number).

* TODO:NOTE: SelectDeadlockLFStressTest does not pass and is disabled.
  !!! DO NOT MERGE !!!

Fixes #504
Fixes #1411

Author: elizarov

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

@@ -1049,7 +1049,9 @@ public final class kotlinx/coroutines/selects/SelectBuilderImpl : kotlinx/corout
 	public fun performAtomicTrySelect (Lkotlinx/coroutines/internal/AtomicDesc;)Ljava/lang/Object;
 	public fun resumeSelectCancellableWithException (Ljava/lang/Throwable;)V
 	public fun resumeWith (Ljava/lang/Object;)V
-	public fun trySelect (Ljava/lang/Object;)Z
+	public fun toString ()Ljava/lang/String;
+	public fun trySelect ()Z
+	public fun trySelectIdempotent (Ljava/lang/Object;)Ljava/lang/Object;
 }
 
 public abstract interface class kotlinx/coroutines/selects/SelectClause0 {
@@ -1070,13 +1072,18 @@ public abstract interface class kotlinx/coroutines/selects/SelectInstance {
 	public abstract fun isSelected ()Z
 	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
+	public abstract fun trySelect ()Z
+	public abstract fun trySelectIdempotent (Ljava/lang/Object;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/selects/SelectKt {
 	public static final fun select (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
+public synthetic class kotlinx/coroutines/selects/SelectKtSelectOpSequenceNumberRefVolatile {
+	public fun <init> (J)V
+}
+
 public final class kotlinx/coroutines/selects/SelectUnbiasedKt {
 	public static final fun selectUnbiased (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }

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

@@ -553,7 +553,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             if (select.isSelected) return
             if (state !is Incomplete) {
                 // already complete -- select result
-                if (select.trySelect(null)) {
+                if (select.trySelect()) {
                     block.startCoroutineUnintercepted(select.completion)
                 }
                 return
@@ -1181,7 +1181,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             if (select.isSelected) return
             if (state !is Incomplete) {
                 // already complete -- select result
-                if (select.trySelect(null)) {
+                if (select.trySelect()) {
                     if (state is CompletedExceptionally) {
                         select.resumeSelectCancellableWithException(state.cause)
                     }
@@ -1362,7 +1362,7 @@ private class SelectJoinOnCompletion<R>(
     private val block: suspend () -> R
 ) : JobNode<JobSupport>(job) {
     override fun invoke(cause: Throwable?) {
-        if (select.trySelect(null))
+        if (select.trySelect())
             block.startCoroutineCancellable(select.completion)
     }
     override fun toString(): String = "SelectJoinOnCompletion[$select]"
@@ -1374,7 +1374,7 @@ private class SelectAwaitOnCompletion<T, R>(
     private val block: suspend (T) -> R
 ) : JobNode<JobSupport>(job) {
     override fun invoke(cause: Throwable?) {
-        if (select.trySelect(null))
+        if (select.trySelect())
             job.selectAwaitCompletion(select, block)
     }
     override fun toString(): String = "SelectAwaitOnCompletion[$select]"

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

@@ -56,7 +56,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
 
     /**
      * Tries to add element to buffer or to queued receiver if select statement clause was not selected yet.
-     * Return type is `ALREADY_SELECTED | OFFER_SUCCESS | OFFER_FAILED | Closed`.
+     * Return type is `ALREADY_SELECTED | OFFER_SUCCESS | OFFER_FAILED | RETRY_ATOMIC | Closed`.
      * @suppress **This is unstable API and it is subject to change.**
      */
     protected open fun offerSelectInternal(element: E, select: SelectInstance<*>): Any {
@@ -352,10 +352,11 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
             else -> null
         }
 
-        override fun validatePrepared(node: ReceiveOrClosed<E>): Boolean {
-            val token = node.tryResumeReceive(element, idempotent = this) ?: return false
+        override fun validatePrepared(node: ReceiveOrClosed<E>): Any? {
+            val token = node.tryResumeReceive(element, idempotent = this) ?: return REMOVE_PREPARED
+            if (token === RETRY_ATOMIC) return RETRY_ATOMIC
             resumeToken = token
-            return true
+            return null
         }
     }
 
@@ -388,6 +389,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
             when {
                 offerResult === ALREADY_SELECTED -> return
                 offerResult === OFFER_FAILED -> {} // retry
+                offerResult === RETRY_ATOMIC -> {} // retry
                 offerResult === OFFER_SUCCESS -> {
                     block.startCoroutineUnintercepted(receiver = this, completion = select.completion)
                     return
@@ -438,7 +440,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
         @JvmField val block: suspend (SendChannel<E>) -> R
     ) : Send(), DisposableHandle {
         override fun tryResumeSend(idempotent: Any?): Any? =
-            if (select.trySelect(idempotent)) SELECT_STARTED else null
+            select.trySelectIdempotent(idempotent)
 
         override fun completeResumeSend(token: Any) {
             assert { token === SELECT_STARTED }
@@ -450,7 +452,7 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
         }
 
         override fun resumeSendClosed(closed: Closed<*>) {
-            if (select.trySelect(null))
+            if (select.trySelect())
                 select.resumeSelectCancellableWithException(closed.sendException)
         }
 
@@ -505,7 +507,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
 
     /**
      * Tries to remove element from buffer or from queued sender if select statement clause was not selected yet.
-     * Return type is `ALREADY_SELECTED | E | POLL_FAILED | Closed`
+     * Return type is `ALREADY_SELECTED | E | POLL_FAILED | RETRY_ATOMIC | Closed`
      * @suppress **This is unstable API and it is subject to change.**
      */
     protected open fun pollSelectInternal(select: SelectInstance<*>): Any? {
@@ -655,11 +657,12 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         }
 
         @Suppress("UNCHECKED_CAST")
-        override fun validatePrepared(node: Send): Boolean {
-            val token = node.tryResumeSend(idempotent = this) ?: return false
+        override fun validatePrepared(node: Send): Any? {
+            val token = node.tryResumeSend(idempotent = this) ?: return REMOVE_PREPARED
+            if (token === RETRY_ATOMIC) return RETRY_ATOMIC
             resumeToken = token
             pollResult = node.pollResult as E
-            return true
+            return null
         }
     }
 
@@ -681,6 +684,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                 when {
                     pollResult === ALREADY_SELECTED -> return
                     pollResult === POLL_FAILED -> {} // retry
+                    pollResult === RETRY_ATOMIC -> {}
                     pollResult is Closed<*> -> throw recoverStackTrace(pollResult.receiveException)
                     else -> {
                         block.startCoroutineUnintercepted(pollResult as E, select.completion)
@@ -709,9 +713,10 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                 when {
                     pollResult === ALREADY_SELECTED -> return
                     pollResult === POLL_FAILED -> {} // retry
+                    pollResult === RETRY_ATOMIC -> {} // retry
                     pollResult is Closed<*> -> {
                         if (pollResult.closeCause == null) {
-                            if (select.trySelect(null))
+                            if (select.trySelect())
                                 block.startCoroutineUnintercepted(null, select.completion)
                             return
                         } else {
@@ -746,6 +751,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
                 when {
                     pollResult === ALREADY_SELECTED -> return
                     pollResult === POLL_FAILED -> {} // retry
+                    pollResult === RETRY_ATOMIC -> {} // retry
                     pollResult is Closed<*> -> {
                         block.startCoroutineUnintercepted(ValueOrClosed.closed(pollResult.closeCause), select.completion)
                     }
@@ -928,8 +934,10 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         @JvmField val block: suspend (Any?) -> R,
         @JvmField val receiveMode: Int
     ) : Receive<E>(), DisposableHandle {
-        override fun tryResumeReceive(value: E, idempotent: Any?): Any?  =
-            if (select.trySelect(idempotent)) (value ?: NULL_VALUE) else null
+        override fun tryResumeReceive(value: E, idempotent: Any?): Any? {
+            val result = select.trySelectIdempotent(idempotent)
+            return if (result === SELECT_STARTED) value ?: NULL_VALUE else result
+        }
 
         @Suppress("UNCHECKED_CAST")
         override fun completeResumeReceive(token: Any) {
@@ -938,7 +946,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         }
 
         override fun resumeReceiveClosed(closed: Closed<*>) {
-            if (!select.trySelect(null)) return
+            if (!select.trySelect()) return
             when (receiveMode) {
                 RECEIVE_THROWS_ON_CLOSE -> select.resumeSelectCancellableWithException(closed.receiveException)
                 RECEIVE_RESULT -> block.startCoroutine(ValueOrClosed.closed<R>(closed.closeCause), select.completion)
@@ -985,10 +993,6 @@ internal val POLL_FAILED: Any = Symbol("POLL_FAILED")
 @SharedImmutable
 internal val ENQUEUE_FAILED: Any = Symbol("ENQUEUE_FAILED")
 
-@JvmField
-@SharedImmutable
-internal val SELECT_STARTED: Any = Symbol("SELECT_STARTED")
-
 @JvmField
 @SharedImmutable
 internal val NULL_VALUE: Symbol = Symbol("NULL_VALUE")
@@ -1012,6 +1016,7 @@ internal typealias Handler = (Throwable?) -> Unit
  */
 internal abstract class Send : LockFreeLinkedListNode() {
     abstract val pollResult: Any? // E | Closed
+    // Returns: null - failure, RETRY_ATOMIC for retry (only when idempotent != null), otherwise token for completeResumeSend
     abstract fun tryResumeSend(idempotent: Any?): Any?
     abstract fun completeResumeSend(token: Any)
     abstract fun resumeSendClosed(closed: Closed<*>)
@@ -1022,6 +1027,7 @@ internal abstract class Send : LockFreeLinkedListNode() {
  */
 internal interface ReceiveOrClosed<in E> {
     val offerResult: Any // OFFER_SUCCESS | Closed
+    // Returns: null - failure, RETRY_ATOMIC for retry (only when idempotent != null), otherwise token for completeResumeReceive
     fun tryResumeReceive(value: E, idempotent: Any?): Any?
     fun completeResumeReceive(token: Any)
 }

kotlinx-coroutines-core/common/src/channels/ArrayBroadcastChannel.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.channels
@@ -105,7 +105,7 @@ internal class ArrayBroadcastChannel<E>(
             val size = this.size
             if (size >= capacity) return OFFER_FAILED
             // let's try to select sending this element to buffer
-            if (!select.trySelect(null)) { // :todo: move trySelect completion outside of lock
+            if (!select.trySelect()) { // :todo: move trySelect completion outside of lock
                 return ALREADY_SELECTED
             }
             val tail = this.tail
@@ -299,7 +299,7 @@ internal class ArrayBroadcastChannel<E>(
                     result === POLL_FAILED -> { /* just bail out of lock */ }
                     else -> {
                         // let's try to select receiving this element from buffer
-                        if (!select.trySelect(null)) { // :todo: move trySelect completion outside of lock
+                        if (!select.trySelect()) { // :todo: move trySelect completion outside of lock
                             result = ALREADY_SELECTED
                         } else {
                             // update subHead after retrieiving element from buffer

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

@@ -105,6 +105,7 @@ internal open class ArrayChannel<E>(
                                 return@withLock
                             }
                             failure === OFFER_FAILED -> break@loop // cannot offer -> Ok to queue to buffer
+                            failure === RETRY_ATOMIC -> {} // retry
                             failure === ALREADY_SELECTED || failure is Closed<*> -> {
                                 this.size = size // restore size
                                 return failure
@@ -114,7 +115,7 @@ internal open class ArrayChannel<E>(
                     }
                 }
                 // let's try to select sending this element to buffer
-                if (!select.trySelect(null)) { // :todo: move trySelect completion outside of lock
+                if (!select.trySelect()) { // :todo: move trySelect completion outside of lock
                     this.size = size // restore size
                     return ALREADY_SELECTED
                 }
@@ -206,6 +207,7 @@ internal open class ArrayChannel<E>(
                             break@loop
                         }
                         failure === POLL_FAILED -> break@loop // cannot poll -> Ok to take from buffer
+                        failure === RETRY_ATOMIC -> {} // retry
                         failure === ALREADY_SELECTED -> {
                             this.size = size // restore size
                             buffer[head] = result // restore head
@@ -226,7 +228,7 @@ internal open class ArrayChannel<E>(
                 buffer[(head + size) % buffer.size] = replacement
             } else {
                 // failed to poll or is already closed --> let's try to select receiving this element from buffer
-                if (!select.trySelect(null)) { // :todo: move trySelect completion outside of lock
+                if (!select.trySelect()) { // :todo: move trySelect completion outside of lock
                     this.size = size // restore size
                     buffer[head] = result // restore head
                     return ALREADY_SELECTED

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

@@ -273,7 +273,7 @@ public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
         }
 
     private fun <R> registerSelectSend(select: SelectInstance<R>, element: E, block: suspend (SendChannel<E>) -> R) {
-        if (!select.trySelect(null)) return
+        if (!select.trySelect()) return
         offerInternal(element)?.let {
             select.resumeSelectCancellableWithException(it.sendException)
             return

kotlinx-coroutines-core/common/src/channels/ConflatedChannel.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.channels
@@ -85,6 +85,7 @@ internal open class ConflatedChannel<E> : AbstractChannel<E>() {
                 result === ALREADY_SELECTED -> return ALREADY_SELECTED
                 result === OFFER_SUCCESS -> return OFFER_SUCCESS
                 result === OFFER_FAILED -> {} // retry
+                result === RETRY_ATOMIC -> {} // retry
                 result is Closed<*> -> return result
                 else -> error("Invalid result $result")
             }

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

@@ -4,6 +4,7 @@
 
 package kotlinx.coroutines.channels
 
+import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.selects.*
 
 /**
@@ -51,6 +52,7 @@ internal open class LinkedListChannel<E> : AbstractChannel<E>() {
                 result === ALREADY_SELECTED -> return ALREADY_SELECTED
                 result === OFFER_SUCCESS -> return OFFER_SUCCESS
                 result === OFFER_FAILED -> {} // retry
+                result === RETRY_ATOMIC -> {} // retry
                 result is Closed<*> -> return result
                 else -> error("Invalid result $result")
             }

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

@@ -1,11 +1,12 @@
 /*
- * 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
 
 import kotlinx.atomicfu.atomic
 import kotlinx.coroutines.*
+import kotlin.jvm.*
 
 /**
  * The most abstract operation that can be in process. Other threads observing an instance of this
@@ -19,6 +20,20 @@ public abstract class OpDescriptor {
      * object that indicates the failure reason.
      */
     abstract fun perform(affected: Any?): Any?
+
+    /**
+     * Returns reference to atomic operation that this descriptor is a part of or `null`
+     * if not a part of any [AtomicOp].
+     */
+    abstract val atomicOp: AtomicOp<*>?
+
+    override fun toString(): String = "$classSimpleName@$hexAddress" // debug
+
+    fun isEarlierThan(that: OpDescriptor): Boolean {
+        val thisOp = atomicOp ?: return false
+        val thatOp = that.atomicOp ?: return false
+        return thisOp.opSequence < thatOp.opSequence
+    }
 }
 
 @SharedImmutable
@@ -40,6 +55,19 @@ public abstract class AtomicOp<in T> : OpDescriptor() {
 
     val isDecided: Boolean get() = _consensus.value !== NO_DECISION
 
+    /**
+     * Sequence number of this multi-word operation for deadlock resolution.
+     * An operation with lower number aborts itself with (using [RETRY_ATOMIC] error symbol) if it encounters
+     * the need to help the operation with higher sequence number and then restarts
+     * (using higher `opSequence` to ensure progress).
+     * Simple operations that cannot get into the deadlock always return zero here.
+     *
+     * See https://github.com/Kotlin/kotlinx.coroutines/issues/504
+     */
+    open val opSequence: Long get() = 0L
+
+    override val atomicOp: AtomicOp<*> get() = this
+
     fun tryDecide(decision: Any?): Boolean {
         assert { decision !== NO_DECISION }
         return _consensus.compareAndSet(NO_DECISION, decision)
@@ -59,7 +87,7 @@ public abstract class AtomicOp<in T> : OpDescriptor() {
         if (decision === NO_DECISION) {
             decision = decide(prepare(affected as T))
         }
-
+        // complete operation
         complete(affected as T, decision)
         return decision
     }
@@ -71,6 +99,17 @@ public abstract class AtomicOp<in T> : OpDescriptor() {
  * @suppress **This is unstable API and it is subject to change.**
  */
 public abstract class AtomicDesc {
+    lateinit var atomicOp: AtomicOp<*> // the reference to parent atomicOp, init when AtomicOp is created
     abstract fun prepare(op: AtomicOp<*>): Any? // returns `null` if prepared successfully
     abstract fun complete(op: AtomicOp<*>, failure: Any?) // decision == null if success
+
+    override fun toString(): String = "$classSimpleName@$hexAddress(atomicOp=$atomicOp)" // for debug
 }
+
+/**
+ * It is returned as an error by [AtomicOp] implementations when they detect potential deadlock
+ * using [AtomicOp.opSequence] numbers.
+ */
+@JvmField
+@SharedImmutable
+internal val RETRY_ATOMIC: Any = Symbol("RETRY_ATOMIC")