More thread-safety in Native counterpart and one more test from native-mt

Author: qwwdfsad

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 kotlin.test.*
+
+class ConcurrentExceptionsStressTest : TestBase() {
+    private val nWorkers = 4
+    private val nRepeat = 1000 * stressTestMultiplier
+
+    private val workers = Array(nWorkers) { index ->
+        newSingleThreadContext("JobExceptionsStressTest-$index")
+    }
+
+    @AfterTest
+    fun tearDown() {
+        workers.forEach {
+            it.close()
+        }
+    }
+
+    @Test
+    fun testStress() = runTest {
+        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 {
+        e?.printStackTrace()
+        throw IllegalStateException("Unexpected $msg exception", e)
+    }
+
+    private class StressException(val index: Int) : SuppressSupportingThrowable()
+}
+

kotlinx-coroutines-core/concurrent/test/ConcurrentTestUtilities.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.exceptions
+
+internal expect open class SuppressSupportingThrowable() : Throwable
+expect val Throwable.suppressed: Array<Throwable>
+expect fun Throwable.printStackTrace()
+
+expect fun randomWait()

kotlinx-coroutines-core/jvm/test/ConcurrentTestUtilities.kt

@@ -0,0 +1,29 @@
+/*
+ * 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 kotlin.random.*
+
+actual fun randomWait() {
+    val n = Random.nextInt(1000)
+    if (n < 500) return // no wait 50% of time
+    repeat(n) {
+        BlackHole.sink *= 3
+    }
+    if (n > 900) Thread.yield()
+}
+
+private object BlackHole {
+    @Volatile
+    var sink = 1
+}
+
+
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+internal actual typealias SuppressSupportingThrowable = Throwable
+
+@Suppress("EXTENSION_SHADOWED_BY_MEMBER")
+actual fun Throwable.printStackTrace() = printStackTrace()
+

kotlinx-coroutines-core/jvm/test/exceptions/Exceptions.kt

@@ -15,7 +15,7 @@ import kotlin.test.*
  * but run only under JDK 1.8
  */
 @Suppress("ConflictingExtensionProperty")
-val Throwable.suppressed: Array<Throwable> get() {
+actual val Throwable.suppressed: Array<Throwable> get() {
     val method = this::class.java.getMethod("getSuppressed") ?: error("This test can only be run using JDK 1.7")
     @Suppress("UNCHECKED_CAST")
     return method.invoke(this) as Array<Throwable>

kotlinx-coroutines-core/native/src/Exceptions.kt

@@ -4,6 +4,9 @@
 
 package kotlinx.coroutines
 
+import kotlinx.coroutines.internal.*
+import kotlinx.coroutines.internal.SuppressSupportingThrowableImpl
+
 /**
  * Thrown by cancellable suspending functions if the [Job] of the coroutine is cancelled while it is suspending.
  * It indicates _normal_ cancellation of a coroutine.
@@ -31,7 +34,9 @@ internal actual class JobCancellationException public actual constructor(
 }
 
 @Suppress("NOTHING_TO_INLINE")
-internal actual inline fun Throwable.addSuppressedThrowable(other: Throwable) { /* empty */ }
+internal actual inline fun Throwable.addSuppressedThrowable(other: Throwable) {
+    if (this is SuppressSupportingThrowableImpl) addSuppressed(other)
+}
 
 // For use in tests
 internal actual val RECOVER_STACK_TRACES: Boolean = false

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

@@ -4,6 +4,7 @@
 
 package kotlinx.coroutines.internal
 
+import kotlinx.atomicfu.*
 import kotlinx.atomicfu.locks.withLock as withLock2
 
 @Suppress("ACTUAL_WITHOUT_EXPECT")
@@ -14,3 +15,17 @@ internal actual inline fun <T> ReentrantLock.withLock(action: () -> T): T = this
 internal actual fun <E> subscriberList(): MutableList<E> = CopyOnWriteList<E>()
 
 internal actual fun <E> identitySet(expectedSize: Int): MutableSet<E> = HashSet()
+
+// "Suppress-supporting throwable" is currently used for tests only
+internal open class SuppressSupportingThrowableImpl : Throwable() {
+    private val _suppressed = atomic<Array<Throwable>>(emptyArray())
+
+    val suppressed: Array<Throwable>
+        get() = _suppressed.value
+
+    fun addSuppressed(other: Throwable) {
+        _suppressed.update { current ->
+            current + other
+        }
+    }
+}

kotlinx-coroutines-core/native/src/internal/Synchronized.kt

@@ -5,16 +5,15 @@
 package kotlinx.coroutines.internal
 
 import kotlinx.coroutines.*
-
+import kotlinx.atomicfu.locks.withLock as withLock2
 /**
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
-public actual typealias SynchronizedObject = Any
+public actual typealias SynchronizedObject = kotlinx.atomicfu.locks.SynchronizedObject
 
 /**
  * @suppress **This an internal API and should not be used from general code.**
  */
 @InternalCoroutinesApi
-public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T =
-    block()
+public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T = lock.withLock2(block)

kotlinx-coroutines-core/native/test/ConcurrentTestUtilities.kt

@@ -0,0 +1,34 @@
+/*
+ * 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.atomicfu.*
+import kotlinx.coroutines.internal.*
+import platform.posix.*
+import kotlin.native.concurrent.*
+import kotlin.random.*
+
+actual fun randomWait() {
+    val n = Random.nextInt(1000)
+    if (n < 500) return // no wait 50% of time
+    repeat(n) {
+        BlackHole.sink *= 3
+    }
+    if (n > 900) sched_yield()
+}
+
+@ThreadLocal
+private object BlackHole {
+    var sink = 1
+}
+
+internal actual typealias SuppressSupportingThrowable = SuppressSupportingThrowableImpl
+
+actual val Throwable.suppressed: Array<Throwable>
+    get() = (this as? SuppressSupportingThrowableImpl)?.suppressed ?: emptyArray()
+
+@Suppress("EXTENSION_SHADOWED_BY_MEMBER")
+actual fun Throwable.printStackTrace() = printStackTrace()
+