Fix SOE with select onSend/onReceive clauses on the same channel

* Instead of StackOverflowError we throw IllegalStateException and leave
  the channel in the original state.

Fixes #1411

Author: elizarov

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

@@ -325,7 +325,7 @@ internal class SelectBuilderImpl<in R>(
         while (true) { // lock-free loop on state
             val state = this.state
             if (state === this) {
-                if (addLastIf(node, { this.state === this }))
+                if (addLastIf(node) { this.state === this })
                     return
             } else { // already selected
                 handle.dispose()
@@ -344,9 +344,21 @@ internal class SelectBuilderImpl<in R>(
     // it is just like start(), but support idempotent start
     override fun trySelect(idempotent: Any?): Boolean {
         assert { idempotent !is OpDescriptor } // "cannot use OpDescriptor as idempotent marker"
-        while (true) { // lock-free loop on state
-            val state = this.state
+        _state.loop { state -> // lock-free loop on state
             when {
+                state is AtomicSelectOp<*> && state.impl === this -> {
+                    /*
+                     * We cannot do state.perform(this) here and "help" it since it is the same select and
+                     * we'll get StackOverflowError. See https://github.com/Kotlin/kotlinx.coroutines/issues/1411
+                     * "Properly" supporting this case is complicated in the current select architecture, since we
+                     * have to "pull up" body of two clauses that need to be activated and link them
+                     * together so that they are sequentially executed, followed by the continuation of
+                     * the code after the select expression. This requires massive implementation rewrite
+                     * (instead of immediately resuming clause bodies, they have to be stored somewhere first).
+                     */
+                    error("Cannot use different select clauses on the same object")
+                }
+                state is OpDescriptor -> state.perform(this)
                 state === this -> {
                     if (_state.compareAndSet(this, idempotent)) {
                         doAfterSelect()
@@ -361,10 +373,14 @@ 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(this, desc, true).perform(null)
+
+    override fun performAtomicIfNotSelected(desc: AtomicDesc): Any? =
+        AtomicSelectOp(this, desc, false).perform(null)
 
-    private inner class AtomicSelectOp(
+    private class AtomicSelectOp<R>(
+        @JvmField val impl: SelectBuilderImpl<R>,
         @JvmField val desc: AtomicDesc,
         @JvmField val select: Boolean
     ) : AtomicOp<Any?>() {
@@ -375,34 +391,45 @@ internal class SelectBuilderImpl<in R>(
                 // we are originator (affected reference is not null if helping)
                 prepareIfNotSelected()?.let { return it }
             }
-            return desc.prepare(this)
+            try {
+                return desc.prepare(this)
+            } catch (e: Throwable) {
+                // undo prepareIfNotSelected on crash (for example if IllegalStateException is thrown)
+                if (affected == null) undoPrepare()
+                throw e
+            }
         }
 
         override fun complete(affected: Any?, failure: Any?) {
             completeSelect(failure)
             desc.complete(this, failure)
         }
 
-        fun prepareIfNotSelected(): Any? {
-            _state.loop { state ->
+        private fun prepareIfNotSelected(): Any? {
+            impl._state.loop { state ->
                 when {
-                    state === this@AtomicSelectOp -> return null // already in progress
-                    state is OpDescriptor -> state.perform(this@SelectBuilderImpl) // help
-                    state === this@SelectBuilderImpl -> {
-                        if (_state.compareAndSet(this@SelectBuilderImpl, this@AtomicSelectOp))
+                    state === this -> return null // already in progress
+                    state is OpDescriptor -> state.perform(impl) // help
+                    state === impl -> {
+                        if (impl._state.compareAndSet(impl, this))
                             return null // success
                     }
                     else -> return ALREADY_SELECTED
                 }
             }
         }
 
+        // reverts the change done by prepareIfNotSelected
+        private fun undoPrepare() {
+            impl._state.compareAndSet(this, impl)
+        }
+
         private fun completeSelect(failure: Any?) {
             val selectSuccess = select && failure == null
-            val update = if (selectSuccess) null else this@SelectBuilderImpl
-            if (_state.compareAndSet(this@AtomicSelectOp, update)) {
+            val update = if (selectSuccess) null else impl
+            if (impl._state.compareAndSet(this, update)) {
                 if (selectSuccess)
-                    doAfterSelect()
+                    impl.doAfterSelect()
             }
         }
     }

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

@@ -409,4 +409,38 @@ class SelectRendezvousChannelTest : TestBase() {
         if (!trySelect(null)) return
         block.startCoroutineUnintercepted(this)
     }
+
+    @Test
+    fun testSelectSendAndReceive() = runTest {
+        val c = Channel<Int>()
+        assertFailsWith<IllegalStateException> {
+            select<Unit> {
+                c.onSend(1) { expectUnreached() }
+                c.onReceive { expectUnreached() }
+            }
+        }
+        checkNotBroken(c)
+    }
+
+    @Test
+    fun testSelectReceiveAndSend() = runTest {
+        val c = Channel<Int>()
+        assertFailsWith<IllegalStateException> {
+            select<Unit> {
+                c.onReceive { expectUnreached() }
+                c.onSend(1) { expectUnreached() }
+            }
+        }
+        checkNotBroken(c)
+    }
+
+    // makes sure the channel is not broken
+    private suspend fun checkNotBroken(c: Channel<Int>) {
+        coroutineScope {
+            launch {
+                c.send(42)
+            }
+            assertEquals(42, c.receive())
+        }
+    }
 }

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

@@ -466,9 +466,15 @@ public actual open class LockFreeLinkedListNode {
                 val prepareOp = PrepareOp(next as Node, op as AtomicOp<Node>, this)
                 if (affected._next.compareAndSet(next, prepareOp)) {
                     // prepared -- complete preparations
-                    val prepFail = prepareOp.perform(affected)
-                    if (prepFail === REMOVE_PREPARED) continue // retry
-                    return prepFail
+                    try {
+                        val prepFail = prepareOp.perform(affected)
+                        if (prepFail === REMOVE_PREPARED) continue // retry
+                        return prepFail
+                    } catch (e: Throwable) {
+                        // Crashed during preparation (for example IllegalStateExpception) -- undo & rethrow
+                        affected._next.compareAndSet(prepareOp, next)
+                        throw e
+                    }
                 }
             }
         }