~ Rollback manual dispose of SharedBlock

elizarov · qwwdfsad · commit 33e4ef0350ab · 2020-08-18T15:27:31.000+03:00
diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api
@@ -1139,8 +1139,8 @@ public final class kotlinx/coroutines/selects/SelectBuilderImpl : kotlinx/corout
 	public fun resumeSelectWithException (Ljava/lang/Throwable;)V
 	public fun resumeWith (Ljava/lang/Object;)V
 	public fun toString ()Ljava/lang/String;
-	public fun trySelect (Lkotlin/jvm/functions/Function0;)Z
-	public fun trySelectOther (Lkotlinx/coroutines/internal/LockFreeLinkedListNode$PrepareOp;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object;
+	public fun trySelect ()Z
+	public fun trySelectOther (Lkotlinx/coroutines/internal/LockFreeLinkedListNode$PrepareOp;)Ljava/lang/Object;
 }
 
 public abstract interface class kotlinx/coroutines/selects/SelectClause0 {
@@ -1161,12 +1161,8 @@ 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 resumeSelectWithException (Ljava/lang/Throwable;)V
-	public abstract fun trySelect (Lkotlin/jvm/functions/Function0;)Z
-	public abstract fun trySelectOther (Lkotlinx/coroutines/internal/LockFreeLinkedListNode$PrepareOp;Lkotlin/jvm/functions/Function0;)Ljava/lang/Object;
-}
-
-public final class kotlinx/coroutines/selects/SelectInstance$DefaultImpls {
-	public static synthetic fun trySelect$default (Lkotlinx/coroutines/selects/SelectInstance;Lkotlin/jvm/functions/Function0;ILjava/lang/Object;)Z
+	public abstract fun trySelect ()Z
+	public abstract fun trySelectOther (Lkotlinx/coroutines/internal/LockFreeLinkedListNode$PrepareOp;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/selects/SelectKt {
diff --git a/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt b/kotlinx-coroutines-core/common/src/channels/AbstractChannel.kt
@@ -382,15 +382,11 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
                         select.disposeOnSelect(node)
                         return
                     }
-                    enqueueResult is Closed<*> -> {
-                        node.block.shareableDispose(useIt = true)
-                        throw recoverStackTrace(helpCloseAndGetSendException(enqueueResult))
-                    }
+                    enqueueResult is Closed<*> -> throw recoverStackTrace(helpCloseAndGetSendException(enqueueResult))
                     enqueueResult === ENQUEUE_FAILED -> {} // try to offer
                     enqueueResult is Receive<*> -> {} // try to offer
                     else -> error("enqueueSend returned $enqueueResult ")
                 }
-                node.block.shareableDispose(useIt = true)
             }
             // hm... receiver is waiting or buffer is not full. try to offer
             val offerResult = offerSelectInternal(element, select)
@@ -457,14 +453,13 @@ internal abstract class AbstractSendChannel<E> : SendChannel<E> {
         @JvmField val block: suspend (SendChannel<E>) -> R = block.asShareable()
 
         override fun tryResumeSend(otherOp: PrepareOp?): Symbol? =
-            select.trySelectOther(otherOp, onSelect = block::shareableWillBeUsed) as Symbol? // must return symbol
+            select.trySelectOther(otherOp) as Symbol? // must return symbol
 
         override fun completeResumeSend() {
             startCoroutine(CoroutineStart.ATOMIC, channel, select.completion, block)
         }
 
         override fun dispose() { // invoked on select completion
-            block.shareableDispose(useIt = false)
             remove()
         }
 
@@ -780,11 +775,7 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
     ): Boolean {
         val node = ReceiveSelect(this, select, block, receiveMode)
         val result = enqueueReceive(node)
-        if (result) {
-            select.disposeOnSelect(node)
-        } else {
-            node.block.shareableDispose(useIt = true)
-        }
+        if (result) select.disposeOnSelect(node)
         return result
     }
 
@@ -963,32 +954,27 @@ internal abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E
         @JvmField val block: suspend (Any?) -> R = block.asShareable() // captured variables in this block need screening
 
         override fun tryResumeReceive(value: E, otherOp: PrepareOp?): Symbol? =
-            select.trySelectOther(otherOp, onSelect = block::shareableWillBeUsed) as Symbol?
+            select.trySelectOther(otherOp) as Symbol?
 
         @Suppress("UNCHECKED_CAST")
         override fun completeResumeReceive(value: E) {
             startCoroutine(CoroutineStart.ATOMIC, if (receiveMode == RECEIVE_RESULT) ValueOrClosed.value(value) else value, select.completion, block)
         }
 
         override fun resumeReceiveClosed(closed: Closed<*>) {
-            if (!select.trySelect(onSelect = block::shareableWillBeUsed)) return
+            if (!select.trySelect()) return
             when (receiveMode) {
-                RECEIVE_THROWS_ON_CLOSE -> {
-                    block.shareableDispose(useIt = true)
-                    select.resumeSelectWithException(closed.receiveException)
-                }
+                RECEIVE_THROWS_ON_CLOSE -> select.resumeSelectWithException(closed.receiveException)
                 RECEIVE_RESULT -> startCoroutine(CoroutineStart.ATOMIC, ValueOrClosed.closed<R>(closed.closeCause), select.completion, block)
                 RECEIVE_NULL_ON_CLOSE -> if (closed.closeCause == null) {
                     startCoroutine(CoroutineStart.ATOMIC, null, select.completion, block)
                 } else {
-                    block.shareableDispose(useIt = true)
                     select.resumeSelectWithException(closed.receiveException)
                 }
             }
         }
 
         override fun dispose() { // invoked on select completion
-            block.shareableDispose(useIt = false)
             if (remove())
                 channel.onReceiveDequeued() // notify cancellation of receive
         }
diff --git a/kotlinx-coroutines-core/common/src/internal/Sharing.common.kt b/kotlinx-coroutines-core/common/src/internal/Sharing.common.kt
@@ -23,8 +23,6 @@ internal expect fun disposeContinuation(cont: () -> Continuation<*>)
 internal expect fun <T> CancellableContinuationImpl<T>.shareableResume(delegate: Continuation<T>, useMode: Int)
 
 internal expect fun <T, R> (suspend (T) -> R).asShareable(): suspend (T) -> R
-internal expect fun <T, R> (suspend (T) -> R).shareableDispose(useIt: Boolean)
-internal expect fun <T, R> (suspend (T) -> R).shareableWillBeUsed()
 
 internal expect fun isReuseSupportedInPlatform(): Boolean
 internal expect fun <T> ArrayList<T>.addOrUpdate(element: T, update: (ArrayList<T>) -> Unit)
diff --git a/kotlinx-coroutines-core/common/src/selects/Select.kt b/kotlinx-coroutines-core/common/src/selects/Select.kt
@@ -116,7 +116,7 @@ public interface SelectInstance<in R> {
     /**
      * Tries to select this instance. Returns `true` on success.
      */
-    public fun trySelect(onSelect: () -> Unit = {}): Boolean
+    public fun trySelect(): Boolean
 
     /**
      * Tries to select this instance. Returns:
@@ -130,7 +130,7 @@ public interface SelectInstance<in R> {
      * member is public, but [Symbol] is internal. When [SelectInstance] becomes a `sealed interface`
      * (see KT-222860) we can declare this method as internal.
      */
-    public fun trySelectOther(otherOp: PrepareOp?, onSelect: () -> Unit): Any?
+    public fun trySelectOther(otherOp: PrepareOp?): Any?
 
     /**
      * Performs action atomically with [trySelect].
@@ -282,10 +282,7 @@ internal class SelectBuilderImpl<in R>(
             when {
                 result === UNDECIDED -> {
                     val update = value()
-                    if (_result.compareAndSet(UNDECIDED, update)) {
-                        uCont.shareableDispose() // will return result without calling continuation
-                        return
-                    }
+                    if (_result.compareAndSet(UNDECIDED, update)) return
                 }
                 result === COROUTINE_SUSPENDED -> if (_result.compareAndSet(COROUTINE_SUSPENDED, RESUMED)) {
                     block()
@@ -397,8 +394,8 @@ internal class SelectBuilderImpl<in R>(
         disposeLockFreeLinkedList { this }
     }
 
-    override fun trySelect(onSelect: () -> Unit): Boolean {
-        val result = trySelectOther(null, onSelect)
+    override fun trySelect(): Boolean {
+        val result = trySelectOther(null)
         return when {
             result === RESUME_TOKEN -> true
             result == null -> false
@@ -491,7 +488,7 @@ internal class SelectBuilderImpl<in R>(
 
     // it is just like plain trySelect, but support idempotent start
     // Returns RESUME_TOKEN | RETRY_ATOMIC | null (when already selected)
-    override fun trySelectOther(otherOp: PrepareOp?, onSelect: () -> Unit): Any? {
+    override fun trySelectOther(otherOp: PrepareOp?): Any? {
         _state.loop { state -> // lock-free loop on state
             when {
                 // Found initial state (not selected yet) -- try to make it selected
@@ -506,7 +503,6 @@ internal class SelectBuilderImpl<in R>(
                         val decision = pairSelectOp.perform(this)
                         if (decision !== null) return decision
                     }
-                    onSelect()
                     doAfterSelect()
                     return RESUME_TOKEN
                 }
diff --git a/kotlinx-coroutines-core/js/src/internal/Sharing.kt b/kotlinx-coroutines-core/js/src/internal/Sharing.kt
@@ -58,12 +58,6 @@ internal actual inline fun <T> CancellableContinuationImpl<T>.shareableResume(de
 @Suppress("NOTHING_TO_INLINE") // Should be NOP
 internal actual inline fun <T, R> (suspend (T) -> R).asShareable(): suspend (T) -> R = this
 
-@Suppress("NOTHING_TO_INLINE") // Should be NOP
-internal actual inline fun <T, R> (suspend (T) -> R).shareableDispose(useIt: Boolean) {}
-
-@Suppress("NOTHING_TO_INLINE") // Should be NOP
-internal actual inline fun <T, R> (suspend (T) -> R).shareableWillBeUsed() {}
-
 @Suppress("NOTHING_TO_INLINE") // Save an entry on call stack
 internal actual inline fun isReuseSupportedInPlatform() = true
 
diff --git a/kotlinx-coroutines-core/jvm/src/internal/Sharing.kt b/kotlinx-coroutines-core/jvm/src/internal/Sharing.kt
@@ -75,14 +75,6 @@ internal actual inline fun <T> CancellableContinuationImpl<T>.shareableResume(de
 @Suppress("NOTHING_TO_INLINE") // Should be NOP
 internal actual inline fun <T, R> (suspend (T) -> R).asShareable(): suspend (T) -> R = this
 
-@InlineOnly
-@Suppress("NOTHING_TO_INLINE") // Should be NOP
-internal actual inline fun <T, R> (suspend (T) -> R).shareableDispose(useIt: Boolean) {}
-
-@InlineOnly
-@Suppress("NOTHING_TO_INLINE") // Should be NOP
-internal actual inline fun <T, R> (suspend (T) -> R).shareableWillBeUsed() {}
-
 @InlineOnly
 @Suppress("NOTHING_TO_INLINE") // Save an entry on call stack
 internal actual inline fun isReuseSupportedInPlatform() = true
diff --git a/kotlinx-coroutines-core/native/src/internal/Sharing.kt b/kotlinx-coroutines-core/native/src/internal/Sharing.kt
@@ -86,16 +86,6 @@ internal actual inline fun <T> Continuation<T>.shareableDispose() {
 internal actual fun <T, R> (suspend (T) -> R).asShareable(): suspend (T) -> R =
     ShareableBlock(this)
 
-internal actual fun <T, R> (suspend (T) -> R).shareableDispose(useIt: Boolean) {
-    this as ShareableBlock<*, *> // must have been shared
-    dispose(useIt)
-}
-
-internal actual fun <T, R> (suspend (T) -> R).shareableWillBeUsed() {
-    this as ShareableBlock<*, *> // must have been shared
-    willBeUsed()
-}
-
 @PublishedApi
 internal actual inline fun disposeContinuation(cont: () -> Continuation<*>) {
     (cont() as ShareableContinuation<*>).disposeRef()
@@ -229,27 +219,9 @@ private typealias Fun2<T, R> = Function2<T, Continuation<R>, Any?>
 private class ShareableBlock<T, R>(
     block: Block1<T, R>
 ) : ShareableObject<Block1<T, R>>(block), Block1<T, R>, SuspendFunction<R>, Fun2<T, R> {
-    private val willBeUsed = atomic(false)
-
     override suspend fun invoke(param: T): R = useRef().invoke(param)
 
     @Suppress("UNCHECKED_CAST")
     override fun invoke(param: T, cont: Continuation<R>): Any? =
         (useRef() as Fun2<T, R>).invoke(param, cont)
-
-    fun willBeUsed() {
-        willBeUsed.value = true
-    }
-
-    fun dispose(useIt: Boolean) {
-        if (willBeUsed.value && !useIt) return
-        val thread = ownerThreadOrNull ?: return
-        if (currentThread() == thread) {
-            disposeRef()
-        } else {
-            thread.execute {
-                disposeRef()
-            }
-        }
-    }
 }
