Version 1.3.0-M1

CHANGES.md

@@ -1,5 +1,43 @@
 # Change log for kotlinx.coroutines
 
+## Version 1.3.0-M1
+
+Flow:
+ * Core `Flow` interfaces and operators are graduated from preview status to experimental. 
+ * Context preservation invariant rework (#1210).
+    * `channelFlow` and `callbackFlow` replacements for `flowViaChannel` for concurrent flows or callback-based APIs.
+    * `flow` prohibits emissions from non-scoped coroutines by default and recommends to use `channelFlow` instead to avoid most of the concurrency-related bugs.
+ * Flow cannot be implemented directly 
+    * `AbstractFlow` is introduced for extension (e.g. for managing state) and ensures all context preservation invariants.
+ * Buffer size is decoupled from all operators that imply channel usage (#1233)
+    * `buffer` operator can be used to adjust buffer size of any buffer-dependent operator (e.g. `channelFlow`, `flowOn` and `flatMapMerge`).
+    * `conflate` operator is introduced.
+ * Flow performance is significantly improved.
+ * New operators: `scan`, `scanReduce`, `first`, `emitAll`.
+ * `flowWith` and `flowViaChannel` are deprecated.
+ * `retry` ignores cancellation exceptions from upstream when the flow was externally cancelled (#1122).
+ * `combineLatest` overloads for multiple flows (#1193).
+ * Fixed numerical overflow in `drop` operator.
+
+Channels:
+ * `consumeEach` is promoted to experimental API (#1080).
+ * Conflated channels always deliver the latest value after closing (#332, #1235).
+ * Non-suspending `ChannelIterator.next` to improve iteration performance (#1162).
+ * Channel exception types are consistent with `produce` and are no longer swallowed as cancellation exceptions in case of programmatic errors (#957, #1128).
+ * All operators on channels (that were prone to coroutine leaks) are deprecated in the favor of `Flow`.
+
+General changes:
+ * Kotlin updated to 1.3.31
+ * `Semaphore` implementation (#1088)
+ * Loading of `Dispatchers.Main` is tweaked so the latest version of R8 can completely remove I/O when loading it (#1231).
+ * Performace of all JS dispatchers is significantly improved (#820).
+ * `withContext` checks cancellation status on exit to make reasoning about sequential concurrent code easier (#1177).
+ * Consistent exception handling mechanism for complex hierarchies (#689).
+ * Convenient overload for `CoroutinesTimeout.seconds` (#1184).
+ * Fix cancellation bug in onJoin (#1130).
+ * Prevent internal names clash that caused errors for ProGuard (#1159).
+ * POSIX's `nanosleep` as `delay` in `runBlocking ` in K/N (#1225).
+
 ## Version 1.2.1
 
 Major:

README.md

@@ -2,10 +2,10 @@
 
 [![official JetBrains 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://api.bintray.com/packages/kotlin/kotlinx/kotlinx.coroutines/images/download.svg?version=1.2.1) ](https://bintray.com/kotlin/kotlinx/kotlinx.coroutines/1.2.1)
+[![Download](https://api.bintray.com/packages/kotlin/kotlinx/kotlinx.coroutines/images/download.svg?version=1.3.0-M1) ](https://bintray.com/kotlin/kotlinx/kotlinx.coroutines/1.3.0-M1)
 
 Library support for Kotlin coroutines with [multiplatform](#multiplatform) support.
-This is a companion version for Kotlin `1.3.30` release.
+This is a companion version for Kotlin `1.3.31` release.
 
 ```kotlin
 suspend fun main() = coroutineScope {
@@ -81,15 +81,15 @@ Add dependencies (you can also add other modules that you need):
 <dependency>
     <groupId>org.jetbrains.kotlinx</groupId>
     <artifactId>kotlinx-coroutines-core</artifactId>
-    <version>1.2.1</version>
+    <version>1.3.0-M1</version>
 </dependency>
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```xml
 <properties>
-    <kotlin.version>1.3.30</kotlin.version>
+    <kotlin.version>1.3.31</kotlin.version>
 </properties>
 ```
 
@@ -99,15 +99,15 @@ Add dependencies (you can also add other modules that you need):
 
 ```groovy
 dependencies {
-    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1'
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.0-M1'
 }
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```groovy
 buildscript {
-    ext.kotlin_version = '1.3.30'
+    ext.kotlin_version = '1.3.31'
 }
 ```
 
@@ -125,15 +125,15 @@ Add dependencies (you can also add other modules that you need):
 
 ```groovy
 dependencies {
-    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1")
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.0-M1")
 }
 ```
 
 And make sure that you use the latest Kotlin version:
 
 ```groovy
 plugins {
-    kotlin("jvm") version "1.3.30"
+    kotlin("jvm") version "1.3.31"
 }
 ```
 
@@ -144,7 +144,7 @@ Make sure that you have either `jcenter()` or `mavenCentral()` in the list of re
 Core modules of `kotlinx.coroutines` are also available for 
 [Kotlin/JS](#js) and [Kotlin/Native](#native).
 In common code that should get compiled for different platforms, add dependency to  
-[`kotlinx-coroutines-core-common`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-common/1.2.1/jar)
+[`kotlinx-coroutines-core-common`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-common/1.3.0-M1/jar)
 (follow the link to get the dependency declaration snippet).
 
 ### Android
@@ -153,7 +153,7 @@ Add [`kotlinx-coroutines-android`](ui/kotlinx-coroutines-android)
 module as dependency when using `kotlinx.coroutines` on Android:
 
 ```groovy
-implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.2.1'
+implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.0-M1'
 ```
 
 This gives you access to Android [Dispatchers.Main](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-android/kotlinx.coroutines.android/kotlinx.coroutines.-dispatchers/index.html)
@@ -172,15 +172,15 @@ R8 is a replacement for ProGuard in Android ecosystem, it is enabled by default
 ### JS
 
 [Kotlin/JS](https://kotlinlang.org/docs/reference/js-overview.html) version of `kotlinx.coroutines` is published as 
-[`kotlinx-coroutines-core-js`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-js/1.2.1/jar)
+[`kotlinx-coroutines-core-js`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-js/1.3.0-M1/jar)
 (follow the link to get the dependency declaration snippet).
 
 You can also use [`kotlinx-coroutines-core`](https://www.npmjs.com/package/kotlinx-coroutines-core) package via NPM. 
 
 ### Native
 
 [Kotlin/Native](https://kotlinlang.org/docs/reference/native-overview.html) version of `kotlinx.coroutines` is published as 
-[`kotlinx-coroutines-core-native`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-native/1.2.1/jar)
+[`kotlinx-coroutines-core-native`](https://search.maven.org/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core-native/1.3.0-M1/jar)
 (follow the link to get the dependency declaration snippet).
 
 Only single-threaded code (JS-style) on Kotlin/Native is currently supported. 

benchmarks/build.gradle

@@ -1,6 +1,8 @@
 /*
  * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
+sourceCompatibility = 1.8
+targetCompatibility = 1.8
 
 apply plugin: "net.ltgt.apt"
 apply plugin: "com.github.johnrengelman.shadow"
@@ -10,15 +12,50 @@ repositories {
     maven { url "https://repo.typesafe.com/typesafe/releases/" }
 }
 
-jmh.jmhVersion = '1.21'
+compileJmhKotlin {
+    kotlinOptions {
+        jvmTarget = "1.8"
+        freeCompilerArgs += ['-Xjvm-default=enable']
+    }
+}
+
+/*
+ * Due to a bug in the inliner it sometimes does not remove inlined symbols (that are later renamed) from unused code paths,
+ * and it breaks JMH that tries to post-process these symbols and fails because they are renamed.
+ */
+task removeRedundantFiles(type: Delete) {
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$buildHistoOnScore\$1\$\$special\$\$inlined\$filter\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$nBlanks\$1\$\$special\$\$inlined\$map\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$score2\$1\$\$special\$\$inlined\$map\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$bonusForDoubleLetter\$1\$\$special\$\$inlined\$map\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$nBlanks\$1\$\$special\$\$inlined\$map\$1\$2\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$bonusForDoubleLetter\$1\$\$special\$\$inlined\$map\$1\$2\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$score2\$1\$\$special\$\$inlined\$map\$1\$2\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOptKt\$\$special\$\$inlined\$collect\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOptKt\$\$special\$\$inlined\$collect\$2\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleOpt\$play\$histoOfLetters\$1\$\$special\$\$inlined\$fold\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleBase\$play\$buildHistoOnScore\$1\$\$special\$\$inlined\$filter\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble/FlowPlaysScrabbleBase\$play\$histoOfLetters\$1\$\$special\$\$inlined\$fold\$1\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/scrabble//SaneFlowPlaysScrabble\$play\$buildHistoOnScore\$1\$\$special\$\$inlined\$filter\$1\$1.class"
+
+    // Primes
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/misc/Numbers\$\$special\$\$inlined\$filter\$1\$2\$1.class"
+    delete "$buildDir/classes/kotlin/jmh/benchmarks/flow/misc/Numbers\$\$special\$\$inlined\$filter\$1\$1.class"
+}
+
+jmhRunBytecodeGenerator.dependsOn(removeRedundantFiles)
 
 // It is better to use the following to run benchmarks, otherwise you may get unexpected errors:
-// ../gradlew --no-daemon cleanJmhJar jmh
+// ./gradlew --no-daemon cleanJmhJar jmh -Pjmh="MyBenchmark"
 jmh {
+    jmhVersion = '1.21'
     duplicateClassesStrategy DuplicatesStrategy.INCLUDE
     failOnError = true
     resultFormat = 'CSV'
-//    include = ['.*ChannelProducerConsumer.*']
+    if (project.hasProperty('jmh')) {
+        include = ".*" + project.jmh + ".*"
+    }
+//    includeTests = false
 }
 
 jmhJar {
@@ -29,8 +66,12 @@ jmhJar {
 }
 
 dependencies {
+    compile "org.openjdk.jmh:jmh-core:1.21"
+    compile "io.projectreactor:reactor-core:$reactor_vesion"
+    compile 'io.reactivex.rxjava2:rxjava:2.1.9'
+    compile "com.github.akarnokd:rxjava2-extensions:0.20.8"
+
     compile "org.openjdk.jmh:jmh-core:1.21"
     compile 'com.typesafe.akka:akka-actor_2.12:2.5.0'
     compile project(':kotlinx-coroutines-core')
 }
-

benchmarks/src/jmh/java/benchmarks/flow/scrabble/RxJava2PlaysScrabble.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package benchmarks.flow.scrabble;
+
+import benchmarks.flow.scrabble.IterableSpliterator;
+import benchmarks.flow.scrabble.ShakespearePlaysScrabble;
+import io.reactivex.Flowable;
+import io.reactivex.Maybe;
+import io.reactivex.Single;
+import io.reactivex.functions.Function;
+import org.openjdk.jmh.annotations.*;
+
+import java.util.*;
+import java.util.Map.Entry;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Shakespeare plays Scrabble with RxJava 2 Flowable.
+ * @author José
+ * @author akarnokd
+ */
+@Warmup(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 7, time = 1, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@State(Scope.Benchmark)
+public class RxJava2PlaysScrabble extends ShakespearePlaysScrabble {
+
+    @Benchmark
+    @Override
+    public List<Entry<Integer, List<String>>> play() throws Exception {
+
+        // Function to compute the score of a given word
+        Function<Integer, Flowable<Integer>> scoreOfALetter = letter -> Flowable.just(letterScores[letter - 'a']) ;
+
+        // score of the same letters in a word
+        Function<Entry<Integer, LongWrapper>, Flowable<Integer>> letterScore =
+                entry ->
+                    Flowable.just(
+                        letterScores[entry.getKey() - 'a'] *
+                        Integer.min(
+                                (int)entry.getValue().get(),
+                                scrabbleAvailableLetters[entry.getKey() - 'a']
+                            )
+                    ) ;
+
+        Function<String, Flowable<Integer>> toIntegerFlowable =
+                string -> Flowable.fromIterable(IterableSpliterator.of(string.chars().boxed().spliterator())) ;
+
+        // Histogram of the letters in a given word
+        Function<String, Single<HashMap<Integer, LongWrapper>>> histoOfLetters =
+                word -> toIntegerFlowable.apply(word)
+                            .collect(
+                                () -> new HashMap<>(),
+                                (HashMap<Integer, LongWrapper> map, Integer value) ->
+                                    {
+                                        LongWrapper newValue = map.get(value) ;
+                                        if (newValue == null) {
+                                            newValue = () -> 0L ;
+                                        }
+                                        map.put(value, newValue.incAndSet()) ;
+                                    }
+
+                            ) ;
+
+        // number of blanks for a given letter
+        Function<Entry<Integer, LongWrapper>, Flowable<Long>> blank =
+                entry ->
+                    Flowable.just(
+                        Long.max(
+                            0L,
+                            entry.getValue().get() -
+                            scrabbleAvailableLetters[entry.getKey() - 'a']
+                        )
+                    ) ;
+
+        // number of blanks for a given word
+        Function<String, Maybe<Long>> nBlanks =
+                word -> histoOfLetters.apply(word)
+                            .flatMapPublisher(map -> Flowable.fromIterable(() -> map.entrySet().iterator()))
+                            .flatMap(blank)
+                            .reduce(Long::sum) ;
+
+
+        // can a word be written with 2 blanks?
+        Function<String, Maybe<Boolean>> checkBlanks =
+                word -> nBlanks.apply(word)
+                            .flatMap(l -> Maybe.just(l <= 2L)) ;
+
+        // score taking blanks into account letterScore1
+        Function<String, Maybe<Integer>> score2 =
+                word -> histoOfLetters.apply(word)
+                            .flatMapPublisher(map -> Flowable.fromIterable(() -> map.entrySet().iterator()))
+                            .flatMap(letterScore)
+                            .reduce(Integer::sum) ;
+
+        // Placing the word on the board
+        // Building the streams of first and last letters
+        Function<String, Flowable<Integer>> first3 =
+                word -> Flowable.fromIterable(IterableSpliterator.of(word.chars().boxed().limit(3).spliterator())) ;
+        Function<String, Flowable<Integer>> last3 =
+                word -> Flowable.fromIterable(IterableSpliterator.of(word.chars().boxed().skip(3).spliterator())) ;
+
+
+        // Stream to be maxed
+        Function<String, Flowable<Integer>> toBeMaxed =
+            word -> Flowable.just(first3.apply(word), last3.apply(word))
+                        .flatMap(observable -> observable) ;
+
+        // Bonus for double letter
+        Function<String, Maybe<Integer>> bonusForDoubleLetter =
+            word -> toBeMaxed.apply(word)
+                        .flatMap(scoreOfALetter)
+                        .reduce(Integer::max) ;
+
+        // score of the word put on the board
+        Function<String, Maybe<Integer>> score3 =
+            word ->
+                Maybe.merge(Arrays.asList(
+                        score2.apply(word),
+                        score2.apply(word),
+                        bonusForDoubleLetter.apply(word),
+                        bonusForDoubleLetter.apply(word),
+                        Maybe.just(word.length() == 7 ? 50 : 0)
+                    )
+                )
+                .reduce(Integer::sum) ;
+
+        Function<Function<String, Maybe<Integer>>, Single<TreeMap<Integer, List<String>>>> buildHistoOnScore =
+                score -> Flowable.fromIterable(() -> shakespeareWords.iterator())
+                                .filter(scrabbleWords::contains)
+                                .filter(word -> checkBlanks.apply(word).blockingGet())
+                                .collect(
+                                    () -> new TreeMap<>(Comparator.reverseOrder()),
+                                    (TreeMap<Integer, List<String>> map, String word) -> {
+                                        Integer key = score.apply(word).blockingGet() ;
+                                        List<String> list = map.get(key) ;
+                                        if (list == null) {
+                                            list = new ArrayList<>() ;
+                                            map.put(key, list) ;
+                                        }
+                                        list.add(word) ;
+                                    }
+                                ) ;
+
+        // best key / value pairs
+        List<Entry<Integer, List<String>>> finalList2 =
+                buildHistoOnScore.apply(score3)
+                    .flatMapPublisher(map -> Flowable.fromIterable(() -> map.entrySet().iterator()))
+                    .take(3)
+                    .collect(
+                        () -> new ArrayList<Entry<Integer, List<String>>>(),
+                        (list, entry) -> {
+                            list.add(entry) ;
+                        }
+                    )
+                    .blockingGet() ;
+        return finalList2 ;
+    }
+}