Doc/deferred

.gitignore

@@ -12,4 +12,5 @@ build
 out
 target
 local.properties
+benchmarks.jar
 /kotlin-js-store

README.md

@@ -4,11 +4,11 @@
 [![JetBrains official project](https://jb.gg/badges/official.svg)](https://confluence.jetbrains.com/display/ALL/JetBrains+on+GitHub)
 [![GitHub license](https://img.shields.io/badge/license-Apache%20License%202.0-blue.svg?style=flat)](https://www.apache.org/licenses/LICENSE-2.0)
 [![Download](https://img.shields.io/maven-central/v/org.jetbrains.kotlinx/kotlinx-coroutines-core/1.6.4)](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core/1.6.4/pom)
-[![Kotlin](https://img.shields.io/badge/kotlin-1.6.21-blue.svg?logo=kotlin)](http://kotlinlang.org)
+[![Kotlin](https://img.shields.io/badge/kotlin-1.8.10-blue.svg?logo=kotlin)](http://kotlinlang.org)
 [![Slack channel](https://img.shields.io/badge/chat-slack-green.svg?logo=slack)](https://kotlinlang.slack.com/messages/coroutines/)
 
 Library support for Kotlin coroutines with [multiplatform](#multiplatform) support.
-This is a companion version for the Kotlin `1.6.21` release.
+This is a companion version for the Kotlin `1.8.10` release.
 
 ```kotlin
 suspend fun main() = coroutineScope {
@@ -37,6 +37,7 @@ suspend fun main() = coroutineScope {
 * [core/jvm](kotlinx-coroutines-core/jvm/) — additional core features available on Kotlin/JVM:
   * [Dispatchers.IO] dispatcher for blocking coroutines;
   * [Executor.asCoroutineDispatcher][asCoroutineDispatcher] extension, custom thread pools, and more.
+  * Integrations with `CompletableFuture` and JVM-specific extensions.
 * [core/js](kotlinx-coroutines-core/js/) — additional core features available on Kotlin/JS:
   * Integration with `Promise` via [Promise.await] and [promise] builder;
   * Integration with `Window` via [Window.asCoroutineDispatcher], etc.
@@ -56,7 +57,7 @@ suspend fun main() = coroutineScope {
 * [ui](ui/README.md) — modules that provide coroutine dispatchers for various single-threaded UI libraries:
   * Android, JavaFX, and Swing.
 * [integration](integration/README.md) — modules that provide integration with various asynchronous callback- and future-based libraries:
-  * JDK8 [CompletionStage.await], Guava [ListenableFuture.await], and Google Play Services [Task.await];
+  * Guava [ListenableFuture.await], and Google Play Services [Task.await];
   * SLF4J MDC integration via [MDCContext].
 
 ## Documentation
@@ -92,7 +93,7 @@ And make sure that you use the latest Kotlin version:
 
 ```xml
 <properties>
-    <kotlin.version>1.6.21</kotlin.version>
+    <kotlin.version>1.8.10</kotlin.version>
 </properties>
 ```
 
@@ -111,10 +112,10 @@ And make sure that you use the latest Kotlin version:
 ```kotlin
 plugins {
     // For build.gradle.kts (Kotlin DSL)
-    kotlin("jvm") version "1.6.21"
+    kotlin("jvm") version "1.8.10"
 
     // For build.gradle (Groovy DSL)
-    id "org.jetbrains.kotlin.jvm" version "1.6.21"
+    id "org.jetbrains.kotlin.jvm" version "1.8.10"
 }
 ```
 
@@ -211,7 +212,7 @@ See [Contributing Guidelines](CONTRIBUTING.md).
 [MainScope()]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-main-scope.html
 [SupervisorJob()]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-supervisor-job.html
 [CoroutineExceptionHandler]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-exception-handler/index.html
-[Dispatchers.IO]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-i-o.html
+[Dispatchers.IO]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-i-o.html
 [asCoroutineDispatcher]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/as-coroutine-dispatcher.html
 [Promise.await]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/await.html
 [promise]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/promise.html
@@ -259,9 +260,6 @@ See [Contributing Guidelines](CONTRIBUTING.md).
 
 <!--- MODULE kotlinx-coroutines-jdk8 -->
 <!--- INDEX kotlinx.coroutines.future -->
-
-[CompletionStage.await]: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-jdk8/kotlinx.coroutines.future/await.html
-
 <!--- MODULE kotlinx-coroutines-guava -->
 <!--- INDEX kotlinx.coroutines.guava -->
 

benchmarks/build.gradle.kts

@@ -4,12 +4,12 @@
 
 @file:Suppress("UnstableApiUsage")
 
-import me.champeau.gradle.*
+import me.champeau.jmh.*
 import org.jetbrains.kotlin.gradle.tasks.*
 
 plugins {
     id("com.github.johnrengelman.shadow")
-    id("me.champeau.gradle.jmh") apply false
+    id("me.champeau.jmh")
 }
 
 repositories {
@@ -21,33 +21,19 @@ java {
     targetCompatibility = JavaVersion.VERSION_1_8
 }
 
-apply(plugin="me.champeau.gradle.jmh")
-
 tasks.named<KotlinCompile>("compileJmhKotlin") {
     kotlinOptions {
         jvmTarget = "1.8"
         freeCompilerArgs += "-Xjvm-default=enable"
     }
 }
 
-// It is better to use the following to run benchmarks, otherwise you may get unexpected errors:
-// ./gradlew --no-daemon cleanJmhJar jmh -Pjmh="MyBenchmark"
-extensions.configure<JMHPluginExtension>("jmh") {
-    jmhVersion = "1.26"
-    duplicateClassesStrategy = DuplicatesStrategy.INCLUDE
-    failOnError = true
-    resultFormat = "CSV"
-    project.findProperty("jmh")?.also {
-        include = listOf(".*$it.*")
-    }
-//    includeTests = false
-}
-
 val jmhJarTask = tasks.named<Jar>("jmhJar") {
     archiveBaseName by "benchmarks"
     archiveClassifier by null
     archiveVersion by null
-    destinationDirectory.file("$rootDir")
+    archiveVersion.convention(null as String?)
+    destinationDirectory.set(file("$rootDir"))
 }
 
 tasks {
@@ -63,13 +49,14 @@ tasks {
 }
 
 dependencies {
-    implementation("org.openjdk.jmh:jmh-core:1.26")
+    implementation("org.openjdk.jmh:jmh-core:1.35")
     implementation("io.projectreactor:reactor-core:${version("reactor")}")
     implementation("io.reactivex.rxjava2:rxjava:2.1.9")
     implementation("com.github.akarnokd:rxjava2-extensions:0.20.8")
 
     implementation("com.typesafe.akka:akka-actor_2.12:2.5.0")
     implementation(project(":kotlinx-coroutines-core"))
+    implementation(project(":kotlinx-coroutines-debug"))
     implementation(project(":kotlinx-coroutines-reactive"))
 
     // add jmh dependency on main

benchmarks/src/jmh/kotlin/benchmarks/ChannelProducerConsumerBenchmark.kt

@@ -4,15 +4,14 @@
 
 package benchmarks
 
+import benchmarks.common.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.scheduling.*
 import kotlinx.coroutines.selects.select
 import org.openjdk.jmh.annotations.*
-import org.openjdk.jmh.infra.Blackhole
 import java.lang.Integer.max
-import java.util.concurrent.ForkJoinPool
 import java.util.concurrent.Phaser
-import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeUnit
 
 
@@ -26,14 +25,14 @@ import java.util.concurrent.TimeUnit
  * Please, be patient, this benchmark takes quite a lot of time to complete.
  */
 @Warmup(iterations = 3, time = 500, timeUnit = TimeUnit.MICROSECONDS)
-@Measurement(iterations = 10, time = 500, timeUnit = TimeUnit.MICROSECONDS)
-@Fork(value = 3)
-@BenchmarkMode(Mode.AverageTime)
+@Measurement(iterations = 20, time = 500, timeUnit = TimeUnit.MICROSECONDS)
+@Fork(value = 1)
+@BenchmarkMode(Mode.Throughput)
 @OutputTimeUnit(TimeUnit.MILLISECONDS)
 @State(Scope.Benchmark)
 open class ChannelProducerConsumerBenchmark {
     @Param
-    private var _0_dispatcher: DispatcherCreator = DispatcherCreator.FORK_JOIN
+    private var _0_dispatcher: DispatcherCreator = DispatcherCreator.DEFAULT
 
     @Param
     private var _1_channel: ChannelCreator = ChannelCreator.RENDEZVOUS
@@ -44,12 +43,13 @@ open class ChannelProducerConsumerBenchmark {
     @Param("false", "true")
     private var _3_withSelect: Boolean = false
 
-    @Param("1", "2", "4") // local machine
-//    @Param("1", "2", "4", "8", "12") // local machine
-//    @Param("1", "2", "4", "8", "16", "32", "64", "128", "144") // dasquad
-//    @Param("1", "2", "4", "8", "16", "32", "64", "96") // Google Cloud
+    @Param("1", "2", "4", "8", "16") // local machine
+//    @Param("1", "2", "4", "8", "16", "32", "64", "128") // Server
     private var _4_parallelism: Int = 0
 
+    @Param("50")
+    private var _5_workSize: Int = 0
+
     private lateinit var dispatcher: CoroutineDispatcher
     private lateinit var channel: Channel<Int>
 
@@ -61,13 +61,21 @@ open class ChannelProducerConsumerBenchmark {
     }
 
     @Benchmark
-    fun spmc() {
+    fun mcsp() {
         if (_2_coroutines != 0) return
         val producers = max(1, _4_parallelism - 1)
         val consumers = 1
         run(producers, consumers)
     }
 
+    @Benchmark
+    fun spmc() {
+        if (_2_coroutines != 0) return
+        val producers = 1
+        val consumers = max(1, _4_parallelism - 1)
+        run(producers, consumers)
+    }
+
     @Benchmark
     fun mpmc() {
         val producers = if (_2_coroutines == 0) (_4_parallelism + 1) / 2 else _2_coroutines / 2
@@ -76,7 +84,7 @@ open class ChannelProducerConsumerBenchmark {
     }
 
     private fun run(producers: Int, consumers: Int) {
-        val n = APPROX_BATCH_SIZE / producers * producers
+        val n = (APPROX_BATCH_SIZE / producers * producers) / consumers * consumers
         val phaser = Phaser(producers + consumers + 1)
         // Run producers
         repeat(producers) {
@@ -111,7 +119,7 @@ open class ChannelProducerConsumerBenchmark {
         } else {
             channel.send(element)
         }
-        doWork()
+        doWork(_5_workSize)
     }
 
     private suspend fun consume(dummy: Channel<Int>?) {
@@ -123,28 +131,25 @@ open class ChannelProducerConsumerBenchmark {
         } else {
             channel.receive()
         }
-        doWork()
+        doWork(_5_workSize)
     }
 }
 
 enum class DispatcherCreator(val create: (parallelism: Int) -> CoroutineDispatcher) {
-    FORK_JOIN({ parallelism ->  ForkJoinPool(parallelism).asCoroutineDispatcher() })
+    //FORK_JOIN({ parallelism ->  ForkJoinPool(parallelism).asCoroutineDispatcher() }),
+    @Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
+    DEFAULT({ parallelism -> ExperimentalCoroutineDispatcher(corePoolSize = parallelism, maxPoolSize = parallelism) })
 }
 
 enum class ChannelCreator(private val capacity: Int) {
     RENDEZVOUS(Channel.RENDEZVOUS),
-//    BUFFERED_1(1),
-    BUFFERED_2(2),
-//    BUFFERED_4(4),
-    BUFFERED_32(32),
-    BUFFERED_128(128),
+    BUFFERED_16(16),
+    BUFFERED_64(64),
     BUFFERED_UNLIMITED(Channel.UNLIMITED);
 
     fun create(): Channel<Int> = Channel(capacity)
 }
 
-private fun doWork(): Unit = Blackhole.consumeCPU(ThreadLocalRandom.current().nextLong(WORK_MIN, WORK_MAX))
+private fun doWork(workSize: Int): Unit = doGeomDistrWork(workSize)
 
-private const val WORK_MIN = 50L
-private const val WORK_MAX = 100L
-private const val APPROX_BATCH_SIZE = 100000
+private const val APPROX_BATCH_SIZE = 100_000

benchmarks/src/jmh/kotlin/benchmarks/SemaphoreBenchmark.kt

@@ -7,6 +7,7 @@ package benchmarks
 import benchmarks.common.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.scheduling.*
 import kotlinx.coroutines.sync.*
 import org.openjdk.jmh.annotations.*
 import java.util.concurrent.*
@@ -19,17 +20,16 @@ import java.util.concurrent.*
 @State(Scope.Benchmark)
 open class SemaphoreBenchmark {
     @Param
-    private var _1_dispatcher: SemaphoreBenchDispatcherCreator = SemaphoreBenchDispatcherCreator.FORK_JOIN
+    private var _1_dispatcher: SemaphoreBenchDispatcherCreator = SemaphoreBenchDispatcherCreator.DEFAULT
 
     @Param("0", "1000")
     private var _2_coroutines: Int = 0
 
     @Param("1", "2", "4", "8", "32", "128", "100000")
     private var _3_maxPermits: Int = 0
 
-    @Param("1", "2", "4") // local machine
-//    @Param("1", "2", "4", "8", "16", "32", "64", "128", "144") // dasquad
-//    @Param("1", "2", "4", "8", "16", "32", "64", "96") // Google Cloud
+    @Param("1", "2", "4", "8", "16") // local machine
+//    @Param("1", "2", "4", "8", "16", "32", "64", "128") // Server
     private var _4_parallelism: Int = 0
 
     private lateinit var dispatcher: CoroutineDispatcher
@@ -80,10 +80,11 @@ open class SemaphoreBenchmark {
 }
 
 enum class SemaphoreBenchDispatcherCreator(val create: (parallelism: Int) -> CoroutineDispatcher) {
-    FORK_JOIN({ parallelism -> ForkJoinPool(parallelism).asCoroutineDispatcher() }),
-    EXPERIMENTAL({ parallelism -> Dispatchers.Default }) // TODO doesn't take parallelism into account
+    //    FORK_JOIN({ parallelism -> ForkJoinPool(parallelism).asCoroutineDispatcher() }),
+    @Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
+    DEFAULT({ parallelism -> ExperimentalCoroutineDispatcher(corePoolSize = parallelism, maxPoolSize = parallelism) })
 }
 
-private const val WORK_INSIDE = 80
-private const val WORK_OUTSIDE = 40
-private const val BATCH_SIZE = 1000000
+private const val WORK_INSIDE = 50
+private const val WORK_OUTSIDE = 50
+private const val BATCH_SIZE = 100000