Incorporate new Native memory model into kotlinx-coroutines mainline

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
@@ -289,7 +289,11 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
             // todo: we should unpark only when this delayed task became first in the queue
             unpark()
         } else {
-            DefaultExecutor.enqueue(task)
+            if (multithreadingSupported) {
+                DefaultExecutor.enqueue(task)
+            } else {
+                error("Cannot execute task because event loop was shut down")
+            }
         }
     }
 
@@ -530,3 +534,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/internal/Concurrent.common.kt

@@ -22,3 +22,5 @@ internal expect class ReentrantLock() {
 internal expect inline fun <T> ReentrantLock.withLock(action: () -> T): T
 
 internal expect fun <E> identitySet(expectedSize: Int): MutableSet<E>
+
+internal expect val multithreadingSupported: Boolean

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,10 @@ public expect val stressTestMultiplier: Int
 @Suppress("NO_ACTUAL_FOR_EXPECT")
 public expect class TestResult
 
+public expect val isNative: Boolean
+// "Speedup" native stress tests
+public val stressTestNativeDivisor = if (isNative) 10 else 1
+
 public expect open class TestBase constructor() {
     /*
      * In common tests we emulate parameterized tests
@@ -43,6 +49,19 @@ public expect open class TestBase constructor() {
     ): TestResult
 }
 
+public fun TestBase.runMtTest(
+    expected: ((Throwable) -> Boolean)? = null,
+    unhandled: List<(Throwable) -> Boolean> = emptyList(),
+    block: suspend CoroutineScope.() -> Unit
+): TestResult {
+    // Never invoked on JS, not extracted to separate 'concurrent' source set
+    // as it doesn't help with the suppress anyway and requires an additional ceremony
+    // and spreads TestBase extensions over the codebase
+    @Suppress("CAST_NEVER_SUCCEEDS")
+    if (!multithreadingSupported) return Unit as TestResult
+    return runTest(expected, unhandled, block)
+}
+
 public suspend inline fun hang(onCancellation: () -> Unit) {
     try {
         suspendCancellableCoroutine<Unit> { }

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/concurrent/src/MultithreadedDispatchers.common.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/jvm/src/internal/LockFreeLinkedList.kt → kotlinx-coroutines-core/concurrent/src/internal/LockFreeLinkedList.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/concurrent/test/AbstractDispatcherConcurrencyTest.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/jvm/test/AtomicCancellationTest.kt → kotlinx-coroutines-core/concurrent/test/AtomicCancellationTest.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/concurrent/test/ConcurrentExceptionsStressTest.kt

@@ -0,0 +1,65 @@
+/*
+ * 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 val workers = Array(if (multithreadingSupported) nWorkers else 0) { index ->
+        newSingleThreadContext("JobExceptionsStressTest-$index")
+    }
+
+    @AfterTest
+    fun tearDown() {
+        workers.forEach {
+            it.close()
+        }
+    }
+
+    @Test
+    fun testStress() = runMtTest {
+        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()
+}
+