Introduce new select implementation, but leave TODO-s for channels -- they will be replaced with new ones.

Author: ndkoval

gradle.properties

@@ -13,7 +13,7 @@ junit5_version=5.7.0
 atomicfu_version=0.16.3
 knit_version=0.3.0
 html_version=0.7.2
-lincheck_version=2.14
+lincheck_version=2.14.1
 dokka_version=1.5.0
 byte_buddy_version=1.10.9
 reactor_version=3.4.1

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

@@ -96,12 +96,10 @@ public fun <T> CoroutineScope.async(
 private open class DeferredCoroutine<T>(
     parentContext: CoroutineContext,
     active: Boolean
-) : AbstractCoroutine<T>(parentContext, true, active = active), Deferred<T>, SelectClause1<T> {
+) : AbstractCoroutine<T>(parentContext, true, active = active), Deferred<T> {
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T
-    override val onAwait: SelectClause1<T> get() = this
-    override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
-        registerSelectClause1Internal(select, block)
+    override val onAwait: SelectClause1<T> get() = onAwaitInternal as SelectClause1<T>
 }
 
 private class LazyDeferredCoroutine<T>(

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

@@ -79,14 +79,12 @@ public fun <T> CompletableDeferred(value: T): CompletableDeferred<T> = Completab
 @Suppress("UNCHECKED_CAST")
 private class CompletableDeferredImpl<T>(
     parent: Job?
-) : JobSupport(true), CompletableDeferred<T>, SelectClause1<T> {
+) : JobSupport(true), CompletableDeferred<T> {
     init { initParentJob(parent) }
     override val onCancelComplete get() = true
     override fun getCompleted(): T = getCompletedInternal() as T
     override suspend fun await(): T = awaitInternal() as T
-    override val onAwait: SelectClause1<T> get() = this
-    override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
-        registerSelectClause1Internal(select, block)
+    override val onAwait: SelectClause1<T> get() = onAwaitInternal as SelectClause1<T>
 
     override fun complete(value: T): Boolean =
         makeCompleting(value)

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

@@ -7,7 +7,6 @@ package kotlinx.coroutines
 
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.internal.*
-import kotlinx.coroutines.intrinsics.*
 import kotlinx.coroutines.selects.*
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
@@ -25,7 +24,7 @@ import kotlin.native.concurrent.*
  * @suppress **This is unstable API and it is subject to change.**
  */
 @Deprecated(level = DeprecationLevel.ERROR, message = "This is internal API and may be removed in the future releases")
-public open class JobSupport constructor(active: Boolean) : Job, ChildJob, ParentJob, SelectClause0 {
+public open class JobSupport constructor(active: Boolean) : Job, ChildJob, ParentJob {
     final override val key: CoroutineContext.Key<*> get() = Job
 
     /*
@@ -560,26 +559,18 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     }
 
     public final override val onJoin: SelectClause0
-        get() = this
+        get() = SelectClause0Impl(
+            objForSelect = this@JobSupport,
+            regFunc = JobSupport::registerSelectForOnJoin as RegistrationFunction
+        )
 
-    // registerSelectJoin
-    public final override fun <R> registerSelectClause0(select: SelectInstance<R>, block: suspend () -> R) {
-        // fast-path -- check state and select/return if needed
-        loopOnState { state ->
-            if (select.isSelected) return
-            if (state !is Incomplete) {
-                // already complete -- select result
-                if (select.trySelect()) {
-                    block.startCoroutineUnintercepted(select.completion)
-                }
-                return
-            }
-            if (startInternal(state) == 0) {
-                // slow-path -- register waiter for completion
-                select.disposeOnSelect(invokeOnCompletion(handler = SelectJoinOnCompletion(select, block).asHandler))
-                return
-            }
+    private fun registerSelectForOnJoin(select: SelectInstance<*>, ignoredParam: Any?) {
+        if (!joinInternal()) {
+            select.selectInRegPhase(Unit)
+            return
         }
+        val disposableHandle = invokeOnCompletion { select.trySelect(this@JobSupport, Unit) }
+        select.invokeOnCompletion { disposableHandle.dispose() }
     }
 
     /**
@@ -1234,46 +1225,36 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         cont.getResult()
     }
 
-    /**
-     * @suppress **This is unstable API and it is subject to change.**
-     */
-    // registerSelectAwaitInternal
-    @Suppress("UNCHECKED_CAST")
-    internal fun <T, R> registerSelectClause1Internal(select: SelectInstance<R>, block: suspend (T) -> R) {
-        // fast-path -- check state and select/return if needed
-        loopOnState { state ->
-            if (select.isSelected) return
+
+    internal val onAwaitInternal: SelectClause1<*> get() = SelectClause1Impl<Any?>(
+        objForSelect = this,
+        regFunc = JobSupport::onAwaitInternalRegFunc as RegistrationFunction,
+        processResFunc = JobSupport::onAwaitInternalProcessResFunc as ProcessResultFunction
+    )
+
+    private fun onAwaitInternalRegFunc(select: SelectInstance<*>, ignoredParam: Any?) {
+        while (true) {
+            val state = this.state
             if (state !is Incomplete) {
-                // already complete -- select result
-                if (select.trySelect()) {
-                    if (state is CompletedExceptionally) {
-                        select.resumeSelectWithException(state.cause)
-                    }
-                    else {
-                        block.startCoroutineUnintercepted(state.unboxState() as T, select.completion)
-                    }
-                }
+                val result = if (state is CompletedExceptionally) state else state.unboxState()
+                select.selectInRegPhase(result)
                 return
             }
-            if (startInternal(state) == 0) {
-                // slow-path -- register waiter for completion
-                select.disposeOnSelect(invokeOnCompletion(handler = SelectAwaitOnCompletion(select, block).asHandler))
-                return
+            if (startInternal(state) >= 0) break // break unless needs to retry
+        }
+        val disposableHandle = invokeOnCompletion {
+            val state = this.state
+            if (state !is Incomplete) {
+                val result = if (state is CompletedExceptionally) state else state.unboxState()
+                select.trySelect(this, result)
             }
         }
+        select.invokeOnCompletion { disposableHandle.dispose() }
     }
 
-    /**
-     * @suppress **This is unstable API and it is subject to change.**
-     */
-    @Suppress("UNCHECKED_CAST")
-    internal fun <T, R> selectAwaitCompletion(select: SelectInstance<R>, block: suspend (T) -> R) {
-        val state = this.state
-        // Note: await is non-atomic (can be cancelled while dispatched)
-        if (state is CompletedExceptionally)
-            select.resumeSelectWithException(state.cause)
-        else
-            block.startCoroutineCancellable(state.unboxState() as T, select.completion)
+    private fun onAwaitInternalProcessResFunc(ignoredParam: Any?, result: Any?): Any? {
+        if (result is CompletedExceptionally) throw result.cause
+        return result
     }
 }
 
@@ -1420,26 +1401,6 @@ internal class DisposeOnCompletion(
     override fun invoke(cause: Throwable?) = handle.dispose()
 }
 
-private class SelectJoinOnCompletion<R>(
-    private val select: SelectInstance<R>,
-    private val block: suspend () -> R
-) : JobNode() {
-    override fun invoke(cause: Throwable?) {
-        if (select.trySelect())
-            block.startCoroutineCancellable(select.completion)
-    }
-}
-
-private class SelectAwaitOnCompletion<T, R>(
-    private val select: SelectInstance<R>,
-    private val block: suspend (T) -> R
-) : JobNode() {
-    override fun invoke(cause: Throwable?) {
-        if (select.trySelect())
-            job.selectAwaitCompletion(select, block)
-    }
-}
-
 // -------- invokeOnCancellation nodes
 
 /**