Incorporate new Native memory model into kotlinx-coroutines mainline (#2833)

* Support of new K/N memory model
* Dispatchers.Default backed by a pool of workers on Linux and by global_queue on iOS-like
* Implementation of Dispatchers.Main that uses the main queue on iOS and default dispatcher on other platforms (#2858)
* Introduced newSingleThreadDispatcher and newFixedThreadPoolDispatcher
* Use proper reentrant locking and CoW arrays on new memory model, make TestBase _almost_ race-free
* More thread-safety in Native counterpart and one more test from native-mt
* Source-set sharing for tests shared between JVM and K/N
* Wrap Obj-C interop into autorelease pool to avoid memory leaks

Fixes #2914

Co-authored-by: dkhalanskyjb <[email protected]>

Author: qwwdfsad

buildSrc/src/main/kotlin/SourceSets.kt

@@ -16,4 +16,4 @@ fun KotlinSourceSet.configureMultiplatform() {
         optInAnnotations.forEach { optIn(it) }
         progressiveMode = true
     }
-}
+}

kotlinx-coroutines-core/build.gradle

@@ -74,19 +74,50 @@ kotlin {
         SourceSetsKt.configureMultiplatform(it)
     }
 
-    configure(targets) {
-        // Configure additional binaries and test runs -- one for each OS
-        if (["macos", "linux", "mingw"].any { name.startsWith(it) }) {
-            binaries {
-                // Test for memory leaks using a special entry point that does not exit but returns from main
-                binaries.getTest("DEBUG").freeCompilerArgs += ["-e", "kotlinx.coroutines.mainNoExit"]
-                // Configure a separate test where code runs in background
-                test("background", [org.jetbrains.kotlin.gradle.plugin.mpp.NativeBuildType.DEBUG]) {
-                    freeCompilerArgs += ["-e", "kotlinx.coroutines.mainBackground"]
-                }
+    /*
+     * Configure four test runs:
+     * 1) Old memory model, Main thread
+     * 2) New memory model, Main thread
+     * 3) Old memory model, BG thread
+     * 4) New memory model, BG thread (required for Dispatchers.Main tests on Darwin)
+     *
+     * All new MM targets are build with optimize = true to have stress tests properly run.
+     */
+    targets.withType(org.jetbrains.kotlin.gradle.plugin.mpp.KotlinNativeTargetWithTests.class).configureEach {
+        binaries {
+            // Test for memory leaks using a special entry point that does not exit but returns from main
+            binaries.getTest("DEBUG").freeCompilerArgs += ["-e", "kotlinx.coroutines.mainNoExit"]
+        }
+
+        binaries.test("newMM", [DEBUG]) {
+            def thisTest = it
+            freeCompilerArgs += ["-e", "kotlinx.coroutines.mainNoExit"]
+            optimized = true
+            binaryOptions["memoryModel"] = "experimental"
+            testRuns.create("newMM") {
+                setExecutionSourceFrom(thisTest)
+                // A hack to get different suffixes in the aggregated report.
+                executionTask.configure { targetName = "$targetName new MM" }
+            }
+        }
+
+        binaries.test("worker",  [DEBUG]) {
+            def thisTest = it
+            freeCompilerArgs += ["-e", "kotlinx.coroutines.mainBackground"]
+            testRuns.create("worker") {
+                setExecutionSourceFrom(thisTest)
+                executionTask.configure { targetName = "$targetName worker" }
             }
-            testRuns {
-                background { setExecutionSourceFrom(binaries.backgroundDebugTest) }
+        }
+
+        binaries.test("workerWithNewMM", [DEBUG]) {
+            def thisTest = it
+            optimized = true
+            freeCompilerArgs += ["-e", "kotlinx.coroutines.mainBackground"]
+            binaryOptions["memoryModel"] = "experimental"
+            testRuns.create("workerWithNewMM") {
+                setExecutionSourceFrom(thisTest)
+                executionTask.configure { targetName = "$targetName worker with new MM" }
             }
         }
     }
@@ -97,6 +128,7 @@ kotlin {
     }
 }
 
+
 configurations {
     configureKotlinJvmPlatform(kotlinCompilerPluginClasspath)
 }

kotlinx-coroutines-core/common/src/EventLoop.common.kt

@@ -276,7 +276,7 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
         // then process one event from queue
         val task = dequeue()
         if (task != null) {
-            task.run()
+            platformAutoreleasePool { task.run() }
             return 0
         }
         return nextTime
@@ -530,3 +530,14 @@ internal expect fun nanoTime(): Long
 internal expect object DefaultExecutor {
     public fun enqueue(task: Runnable)
 }
+
+/**
+ * Used by Darwin targets to wrap a [Runnable.run] call in an Objective-C Autorelease Pool. It is a no-op on JVM, JS and
+ * non-Darwin native targets.
+ *
+ * Coroutines on Darwin targets can call into the Objective-C world, where a callee may push a to-be-returned object to
+ * the Autorelease Pool, so as to avoid a premature ARC release before it reaches the caller. This means the pool must
+ * be eventually drained to avoid leaks. Since Kotlin Coroutines does not use [NSRunLoop], which provides automatic
+ * pool management, it must manage the pool creation and pool drainage manually.
+ */
+internal expect inline fun platformAutoreleasePool(crossinline block: () -> Unit)

kotlinx-coroutines-core/common/src/channels/ArrayBroadcastChannel.kt

@@ -33,6 +33,11 @@ internal class ArrayBroadcastChannel<E>(
         require(capacity >= 1) { "ArrayBroadcastChannel capacity must be at least 1, but $capacity was specified" }
     }
 
+    /**
+     * NB: prior to changing any logic of ArrayBroadcastChannel internals, please ensure that
+     * you do not break internal invariants of the SubscriberList implementation on K/N and KJS
+     */
+
     /*
      *  Writes to buffer are guarded by bufferLock, but reads from buffer are concurrent with writes
      *    - Write element to buffer then write "tail" (volatile)
@@ -60,6 +65,7 @@ internal class ArrayBroadcastChannel<E>(
         get() = _size.value
         set(value) { _size.value = value }
 
+    @Suppress("DEPRECATION")
     private val subscribers = subscriberList<Subscriber<E>>()
 
     override val isBufferAlwaysFull: Boolean get() = false

kotlinx-coroutines-core/common/src/internal/Concurrent.common.kt

@@ -8,10 +8,12 @@ package kotlinx.coroutines.internal
  * Special kind of list intended to be used as collection of subscribers in `ArrayBroadcastChannel`
  * On JVM it's CopyOnWriteList and on JS it's MutableList.
  *
- * Note that this alias is intentionally not named as CopyOnWriteList to avoid accidental misusage outside of ArrayBroadcastChannel
+ * Note that this alias is intentionally not named as CopyOnWriteList to avoid accidental misusage outside of the ArrayBroadcastChannel
  */
 internal typealias SubscribersList<E> = MutableList<E>
 
+@Deprecated(message = "Implementation of this primitive is tailored to specific ArrayBroadcastChannel usages on K/N " +
+    "and K/JS platforms and it is unsafe to use it anywhere else")
 internal expect fun <E> subscriberList(): SubscribersList<E>
 
 internal expect class ReentrantLock() {

kotlinx-coroutines-core/common/test/EmptyContext.kt

@@ -7,10 +7,6 @@ package kotlinx.coroutines
 import kotlinx.coroutines.intrinsics.*
 import kotlin.coroutines.*
 
-suspend fun <T> withEmptyContext(block: suspend () -> T): T {
-    val baseline = Result.failure<T>(IllegalStateException("Block was suspended"))
-    var result: Result<T> = baseline
-    block.startCoroutineUnintercepted(Continuation(EmptyCoroutineContext) { result = it })
-    while (result == baseline) yield()
-    return result.getOrThrow()
+suspend fun <T> withEmptyContext(block: suspend () -> T): T = suspendCoroutine { cont ->
+    block.startCoroutineUnintercepted(Continuation(EmptyCoroutineContext) { cont.resumeWith(it) })
 }

kotlinx-coroutines-core/common/test/TestBase.common.kt

@@ -7,11 +7,13 @@
 package kotlinx.coroutines
 
 import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.test.*
 
 public expect val isStressTest: Boolean
 public expect val stressTestMultiplier: Int
+public expect val stressTestMultiplierSqrt: Int
 
 /**
  * The result of a multiplatform asynchronous test.
@@ -20,6 +22,8 @@ public expect val stressTestMultiplier: Int
 @Suppress("NO_ACTUAL_FOR_EXPECT")
 public expect class TestResult
 
+public expect val isNative: Boolean
+
 public expect open class TestBase constructor() {
     /*
      * In common tests we emulate parameterized tests

kotlinx-coroutines-core/common/test/flow/FlowInvariantsTest.kt

@@ -235,7 +235,7 @@ class FlowInvariantsTest : TestBase() {
             }
             expectUnreached()
         } catch (e: IllegalStateException) {
-            assertTrue(e.message!!.contains("Flow invariant is violated"))
+            assertTrue(e.message!!.contains("Flow invariant is violated"), "But had: ${e.message}")
             finish(2)
         }
     }

kotlinx-coroutines-core/concurrent/src/Builders.concurrent.kt

@@ -0,0 +1,14 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlin.coroutines.*
+
+/**
+ * Runs a new coroutine and **blocks** the current thread until its completion.
+ * This function should not be used from a coroutine. It is designed to bridge regular blocking code
+ * to libraries that are written in suspending style, to be used in `main` functions and in tests.
+ */
+public expect fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T