New channel algorithms

Author: ndkoval

Diff for: benchmarks/src/jmh/kotlin/benchmarks/ParametrizedDispatcherBase.kt

@@ -25,7 +25,7 @@ abstract class ParametrizedDispatcherBase : CoroutineScope {
     private var closeable: Closeable? = null
 
     @Setup
-    @UseExperimental(InternalCoroutinesApi::class)
+    @OptIn(InternalCoroutinesApi::class)
     open fun setup() {
         coroutineContext = when {
             dispatcher == "fjp" -> ForkJoinPool.commonPool().asCoroutineDispatcher()

Diff for: gradle.properties

@@ -5,7 +5,7 @@
 # Kotlin
 version=1.6.0-SNAPSHOT
 group=org.jetbrains.kotlinx
-kotlin_version=1.6.0
+kotlin_version=1.7.0-dev-1132
 
 # Dependencies
 junit_version=4.12

Diff for: gradle/test-mocha-js.gradle

@@ -29,7 +29,7 @@ task testMochaNode(type: NodeTask, dependsOn: [compileTestJsLegacy, installDepen
 
 def jsLegacyTestTask = project.tasks.findByName('jsLegacyTest') ? jsLegacyTest : jsTest
 
-jsLegacyTestTask.dependsOn testMochaNode
+//jsLegacyTestTask.dependsOn testMochaNode
 
 // -- Testing with Mocha under headless Chrome
 
@@ -99,5 +99,5 @@ task testMochaJsdom(type: NodeTask, dependsOn: [compileTestJsLegacy, installDepe
     if (project.hasProperty("teamcity")) args += ['--reporter', 'mocha-teamcity-reporter']
 }
 
-jsLegacyTestTask.dependsOn testMochaJsdom
+//jsLegacyTestTask.dependsOn testMochaJsdom
 

Diff for: kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -51,7 +51,7 @@ public final class kotlinx/coroutines/CancellableContinuation$DefaultImpls {
 	public static synthetic fun tryResume$default (Lkotlinx/coroutines/CancellableContinuation;Ljava/lang/Object;Ljava/lang/Object;ILjava/lang/Object;)Ljava/lang/Object;
 }
 
-public class kotlinx/coroutines/CancellableContinuationImpl : kotlin/coroutines/jvm/internal/CoroutineStackFrame, kotlinx/coroutines/CancellableContinuation {
+public class kotlinx/coroutines/CancellableContinuationImpl : kotlin/coroutines/jvm/internal/CoroutineStackFrame, kotlinx/coroutines/CancellableContinuation, kotlinx/coroutines/channels/Waiter {
 	public fun <init> (Lkotlin/coroutines/Continuation;I)V
 	public final fun callCancelHandler (Lkotlinx/coroutines/CancelHandler;Ljava/lang/Throwable;)V
 	public final fun callOnCancellation (Lkotlin/jvm/functions/Function1;Ljava/lang/Throwable;)V
@@ -768,23 +768,6 @@ public final class kotlinx/coroutines/channels/ClosedSendChannelException : java
 	public fun <init> (Ljava/lang/String;)V
 }
 
-public final class kotlinx/coroutines/channels/ConflatedBroadcastChannel : kotlinx/coroutines/channels/BroadcastChannel {
-	public fun <init> ()V
-	public fun <init> (Ljava/lang/Object;)V
-	public synthetic fun cancel (Ljava/lang/Throwable;)Z
-	public fun cancel (Ljava/util/concurrent/CancellationException;)V
-	public fun close (Ljava/lang/Throwable;)Z
-	public fun getOnSend ()Lkotlinx/coroutines/selects/SelectClause2;
-	public final fun getValue ()Ljava/lang/Object;
-	public final fun getValueOrNull ()Ljava/lang/Object;
-	public fun invokeOnClose (Lkotlin/jvm/functions/Function1;)V
-	public fun isClosedForSend ()Z
-	public fun offer (Ljava/lang/Object;)Z
-	public fun openSubscription ()Lkotlinx/coroutines/channels/ReceiveChannel;
-	public fun send (Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
-	public fun trySend-JP2dKIU (Ljava/lang/Object;)Ljava/lang/Object;
-}
-
 public final class kotlinx/coroutines/channels/ProduceKt {
 	public static final fun awaitClose (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static synthetic fun awaitClose$default (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
@@ -855,6 +838,9 @@ public final class kotlinx/coroutines/channels/TickerMode : java/lang/Enum {
 	public static fun values ()[Lkotlinx/coroutines/channels/TickerMode;
 }
 
+public abstract interface class kotlinx/coroutines/channels/Waiter {
+}
+
 public final class kotlinx/coroutines/debug/internal/DebugCoroutineInfo {
 	public fun <init> (Lkotlinx/coroutines/debug/internal/DebugCoroutineInfoImpl;Lkotlin/coroutines/CoroutineContext;)V
 	public final fun getContext ()Lkotlin/coroutines/CoroutineContext;
@@ -1277,7 +1263,7 @@ protected final class kotlinx/coroutines/selects/SelectImplementation$ClauseData
 	public final fun tryRegister (Lkotlinx/coroutines/selects/SelectImplementation;)Z
 }
 
-public abstract interface class kotlinx/coroutines/selects/SelectInstance {
+public abstract interface class kotlinx/coroutines/selects/SelectInstance : kotlinx/coroutines/channels/Waiter {
 	public abstract fun disposeOnCompletion (Lkotlinx/coroutines/DisposableHandle;)V
 	public abstract fun getContext ()Lkotlin/coroutines/CoroutineContext;
 	public abstract fun selectInRegistrationPhase (Ljava/lang/Object;)V

Diff for: kotlinx-coroutines-core/build.gradle

@@ -241,6 +241,7 @@ task jvmStressTest(type: Test, dependsOn: compileTestKotlinJvm) {
     systemProperty 'kotlinx.coroutines.scheduler.keep.alive.sec', '100000' // any unpark problem hangs test
     systemProperty 'kotlinx.coroutines.semaphore.segmentSize', '2'
     systemProperty 'kotlinx.coroutines.semaphore.maxSpinCycles', '10'
+    systemProperty 'kotlinx.coroutines.bufferedChannel.segmentSize', '2'
 }
 
 task jvmLincheckTest(type: Test, dependsOn: compileTestKotlinJvm) {
@@ -259,6 +260,7 @@ static void configureJvmForLincheck(task) {
                     '--add-exports', 'java.base/jdk.internal.util=ALL-UNNAMED'] // in the model checking mode
     task.systemProperty 'kotlinx.coroutines.semaphore.segmentSize', '2'
     task.systemProperty 'kotlinx.coroutines.semaphore.maxSpinCycles', '1' // better for the model checking mode
+    task.systemProperty 'kotlinx.coroutines.bufferedChannel.segmentSize', '2'
 }
 
 // Always check additional test sets

Diff for: kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt

@@ -5,6 +5,7 @@
 package kotlinx.coroutines
 
 import kotlinx.atomicfu.*
+import kotlinx.coroutines.channels.Waiter
 import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
@@ -26,7 +27,7 @@ internal val RESUME_TOKEN = Symbol("RESUME_TOKEN")
 internal open class CancellableContinuationImpl<in T>(
     final override val delegate: Continuation<T>,
     resumeMode: Int
-) : DispatchedTask<T>(resumeMode), CancellableContinuation<T>, CoroutineStackFrame {
+) : DispatchedTask<T>(resumeMode), CancellableContinuation<T>, CoroutineStackFrame, Waiter {
     init {
         assert { resumeMode != MODE_UNINITIALIZED } // invalid mode for CancellableContinuationImpl
     }