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

kotlinx-coroutines-core/jvm/src/CompletionHandler.kt renamed to kotlinx-coroutines-core/concurrent/src/CompletionHandler.kt

@@ -0,0 +1,11 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+@ExperimentalCoroutinesApi
+public expect fun newSingleThreadContext(name: String): CloseableCoroutineDispatcher
+
+@ExperimentalCoroutinesApi
+public expect fun newFixedThreadPoolContext(nThreads: Int, name: String): CloseableCoroutineDispatcher

kotlinx-coroutines-core/concurrent/src/MultithreadedDispatchers.common.kt

@@ -7,6 +7,8 @@ package kotlinx.coroutines.internal
 
 import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
+import kotlin.jvm.*
+import kotlin.native.concurrent.*
 
 private typealias Node = LockFreeLinkedListNode
 
@@ -20,9 +22,11 @@ internal const val SUCCESS: Int = 1
 internal const val FAILURE: Int = 2
 
 @PublishedApi
+@SharedImmutable
 internal val CONDITION_FALSE: Any = Symbol("CONDITION_FALSE")
 
 @PublishedApi
+@SharedImmutable
 internal val LIST_EMPTY: Any = Symbol("LIST_EMPTY")
 
 /** @suppress **This is unstable API and it is subject to change.** */
@@ -616,7 +620,7 @@ public actual open class LockFreeLinkedListNode {
         assert { next === this._next.value }
     }
 
-    override fun toString(): String = "${this::class.java.simpleName}@${Integer.toHexString(System.identityHashCode(this))}"
+    override fun toString(): String = "${this::classSimpleName}@${this.hexAddress}"
 }
 
 private class Removed(@JvmField val ref: Node) {

kotlinx-coroutines-core/jvm/src/internal/LockFreeLinkedList.kt renamed to kotlinx-coroutines-core/concurrent/src/internal/LockFreeLinkedList.kt

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+package kotlinx.coroutines
+
+import kotlinx.coroutines.channels.*
+import kotlin.test.*
+
+
+abstract class AbstractDispatcherConcurrencyTest : TestBase() {
+
+   public abstract val dispatcher: CoroutineDispatcher
+
+    @Test
+    fun testLaunchAndJoin() = runMtTest {
+        expect(1)
+        var capturedMutableState = 0
+        val job = GlobalScope.launch(dispatcher) {
+            ++capturedMutableState
+            expect(2)
+        }
+        runBlocking { job.join() }
+        assertEquals(1, capturedMutableState)
+        finish(3)
+    }
+
+    @Test
+    fun testDispatcherHasOwnThreads() = runMtTest {
+        val channel = Channel<Int>()
+        GlobalScope.launch(dispatcher) {
+            channel.send(42)
+        }
+
+        var result = ChannelResult.failure<Int>()
+        while (!result.isSuccess) {
+            result = channel.tryReceive()
+            // Block the thread, wait
+        }
+        // Delivery was successful, let's check it
+        assertEquals(42, result.getOrThrow())
+    }
+
+    @Test
+    fun testDelayInDispatcher() = runMtTest {
+        expect(1)
+        val job = GlobalScope.launch(dispatcher) {
+            expect(2)
+            delay(100)
+            expect(3)
+        }
+        runBlocking { job.join() }
+        finish(4)
+    }
+}

kotlinx-coroutines-core/concurrent/test/AbstractDispatcherConcurrencyTest.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-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines
@@ -142,4 +142,4 @@ class AtomicCancellationTest : TestBase() {
         yield() // to jobToJoin & canceller
         expect(6)
     }
-}
+}

kotlinx-coroutines-core/jvm/test/AtomicCancellationTest.kt renamed to kotlinx-coroutines-core/concurrent/test/AtomicCancellationTest.kt

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.coroutines.exceptions.*
+import kotlinx.coroutines.internal.*
+import kotlin.test.*
+
+class ConcurrentExceptionsStressTest : TestBase() {
+    private val nWorkers = 4
+    private val nRepeat = 1000 * stressTestMultiplier
+
+    private var workers: Array<CloseableCoroutineDispatcher> = emptyArray()
+
+    @AfterTest
+    fun tearDown() {
+        workers.forEach {
+            it.close()
+        }
+    }
+
+    @Test
+    fun testStress() = runMtTest {
+        workers = Array(nWorkers) { index ->
+            newSingleThreadContext("JobExceptionsStressTest-$index")
+        }
+
+        repeat(nRepeat) {
+            testOnce()
+        }
+    }
+
+    @Suppress("SuspendFunctionOnCoroutineScope") // workaround native inline fun stacktraces
+    private suspend fun CoroutineScope.testOnce() {
+        val deferred = async(NonCancellable) {
+            repeat(nWorkers) { index ->
+                // Always launch a coroutine even if parent job was already cancelled (atomic start)
+                launch(workers[index], start = CoroutineStart.ATOMIC) {
+                    randomWait()
+                    throw StressException(index)
+                }
+            }
+        }
+        deferred.join()
+        assertTrue(deferred.isCancelled)
+        val completionException = deferred.getCompletionExceptionOrNull()
+        val cause = completionException as? StressException
+            ?: unexpectedException("completion", completionException)
+        val suppressed = cause.suppressed
+        val indices = listOf(cause.index) + suppressed.mapIndexed { index, e ->
+            (e as? StressException)?.index ?: unexpectedException("suppressed $index", e)
+        }
+        repeat(nWorkers) { index ->
+            assertTrue(index in indices, "Exception $index is missing: $indices")
+        }
+        assertEquals(nWorkers, indices.size, "Duplicated exceptions in list: $indices")
+    }
+
+    private fun unexpectedException(msg: String, e: Throwable?): Nothing {
+        throw IllegalStateException("Unexpected $msg exception", e)
+    }
+
+    private class StressException(val index: Int) : SuppressSupportingThrowable()
+}
+

kotlinx-coroutines-core/concurrent/test/ConcurrentExceptionsStressTest.kt

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.exceptions
+
+import kotlinx.coroutines.*
+
+internal expect open class SuppressSupportingThrowable() : Throwable
+expect val Throwable.suppressed: Array<Throwable>
+expect fun Throwable.printStackTrace()
+
+expect fun randomWait()
+
+expect fun currentThreadName(): String
+
+inline fun CloseableCoroutineDispatcher.use(block: (CloseableCoroutineDispatcher) -> Unit) {
+    try {
+        block(this)
+    } finally {
+        close()
+    }
+}