Revert "Introduce reusable cancellable continuations for hot loops with channels (#1534)"

This reverts commit 946e578

Author: elizarov

benchmarks/src/jmh/kotlin/benchmarks/CancellableContinuationBenchmark.kt

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package benchmarks
+
+import kotlinx.coroutines.*
+import org.openjdk.jmh.annotations.*
+import java.util.concurrent.*
+import kotlin.coroutines.*
+import kotlin.coroutines.intrinsics.*
+
+@Warmup(iterations = 5)
+@Measurement(iterations = 10)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Benchmark)
+@Fork(2)
+open class CancellableContinuationBenchmark {
+
+    @Benchmark
+    fun awaitWithSuspension(): Int {
+        val deferred = CompletableDeferred<Int>()
+        return run(allowSuspend = true) { deferred.await() }
+    }
+
+    @Benchmark
+    fun awaitNoSuspension(): Int {
+        val deferred = CompletableDeferred(1)
+        return run { deferred.await() }
+    }
+
+    private fun run(allowSuspend: Boolean = false, block: suspend () -> Int): Int {
+        val value = block.startCoroutineUninterceptedOrReturn(EmptyContinuation)
+        if (value === COROUTINE_SUSPENDED) {
+            if (!allowSuspend) {
+                throw IllegalStateException("Unexpected suspend")
+            } else {
+                return -1
+            }
+        }
+
+        return value as Int
+    }
+
+    object EmptyContinuation : Continuation<Int> {
+        override val context: CoroutineContext
+            get() = EmptyCoroutineContext
+
+        override fun resumeWith(result: Result<Int>) {
+        }
+    }
+}

benchmarks/src/jmh/kotlin/benchmarks/flow/NumbersBenchmark.kt

@@ -13,6 +13,22 @@ import kotlinx.coroutines.flow.*
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.*
 
+/*
+ * Results:
+ *
+ * // Throw FlowAborted overhead
+ * Numbers.primes            avgt    7  3039.185 ± 25.598  us/op
+ * Numbers.primesRx          avgt    7  2677.937 ± 17.720  us/op
+ *
+ * // On par
+ * Numbers.transformations   avgt    7    16.207 ±  0.133  us/op
+ * Numbers.transformationsRx avgt    7    19.626 ±  0.135  us/op
+ *
+ * // Channels overhead
+ * Numbers.zip               avgt    7   434.160 ±  7.014  us/op
+ * Numbers.zipRx             avgt    7    87.898 ±  5.007  us/op
+ *
+ */
 @Warmup(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
 @Measurement(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
 @Fork(value = 1)
@@ -23,11 +39,11 @@ open class NumbersBenchmark {
 
     companion object {
         private const val primes = 100
-        private const val natural = 1000L
+        private const val natural = 1000
     }
 
-    private fun numbers(limit: Long = Long.MAX_VALUE) = flow {
-        for (i in 2L..limit) emit(i)
+    private fun numbers() = flow {
+        for (i in 2L..Long.MAX_VALUE) emit(i)
     }
 
     private fun primesFlow(): Flow<Long> = flow {
@@ -64,7 +80,7 @@ open class NumbersBenchmark {
 
     @Benchmark
     fun zip() = runBlocking {
-        val numbers = numbers(natural)
+        val numbers = numbers().take(natural)
         val first = numbers
             .filter { it % 2L != 0L }
             .map { it * it }
@@ -89,7 +105,8 @@ open class NumbersBenchmark {
 
     @Benchmark
     fun transformations(): Int = runBlocking {
-        numbers(natural)
+        numbers()
+            .take(natural)
             .filter { it % 2L != 0L }
             .map { it * it }
             .filter { (it + 1) % 3 == 0L }.count()
@@ -103,4 +120,4 @@ open class NumbersBenchmark {
             .filter { (it + 1) % 3 == 0L }.count()
             .blockingGet()
     }
-}
+}

benchmarks/src/jmh/kotlin/benchmarks/tailcall/SimpleChannel.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
@@ -223,40 +223,6 @@ public suspend inline fun <T> suspendAtomicCancellableCoroutine(
         cancellable.getResult()
     }
 
-/**
- *  Suspends coroutine similar to [suspendAtomicCancellableCoroutine], but an instance of [CancellableContinuationImpl] is reused if possible.
- */
-internal suspend inline fun <T> suspendAtomicCancellableCoroutineReusable(
-    crossinline block: (CancellableContinuation<T>) -> Unit
-): T = suspendCoroutineUninterceptedOrReturn { uCont ->
-        val cancellable = getOrCreateCancellableContinuation(uCont.intercepted())
-        block(cancellable)
-        cancellable.getResult()
-    }
-
-internal fun <T> getOrCreateCancellableContinuation(delegate: Continuation<T>): CancellableContinuationImpl<T> {
-    // If used outside of our dispatcher
-    if (delegate !is DispatchedContinuation<T>) {
-        return CancellableContinuationImpl(delegate, resumeMode = MODE_ATOMIC_DEFAULT)
-    }
-    /*
-     * Attempt to claim reusable instance.
-     *
-     * suspendAtomicCancellableCoroutineReusable { // <- claimed
-     *     // Any asynchronous cancellation is "postponed" while this block
-     *     // is being executed
-     * } // postponed cancellation is checked here.
-     *
-     * Claim can fail for the following reasons:
-     * 1) Someone tried to make idempotent resume.
-     *    Idempotent resume is internal (used only by us) and is used only in `select`,
-     *    thus leaking CC instance for indefinite time.
-     * 2) Continuation was cancelled. Then we should prevent any further reuse and bail out.
-     */
-    return delegate.claimReusableCancellableContinuation()?.takeIf { it.resetState() }
-        ?: return CancellableContinuationImpl(delegate, MODE_ATOMIC_DEFAULT)
-}
-
 /**
  * @suppress **Deprecated**
  */