Do not try to start sharing strategy in an undispatched manner in native-mt mode as it is not supported

Fixes #3136

Author: qwwdfsad

kotlinx-coroutines-core/common/src/flow/operators/Share.kt

@@ -10,6 +10,7 @@ package kotlinx.coroutines.flow
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.internal.*
+import kotlinx.coroutines.internal.*
 import kotlin.coroutines.*
 import kotlin.jvm.*
 
@@ -204,8 +205,9 @@ private fun <T> CoroutineScope.launchSharing(
      * * Delayed sharing strategies have a chance to immediately observe consecutive subscriptions.
      *   E.g. in the cases like `flow.shareIn(...); flow.take(1)` we want sharing strategy to see the initial subscription
      * * Eager sharing does not start immediately, so the subscribers have actual chance to subscribe _prior_ to sharing.
+     * // TODO: kludge: native-mt doesn't support undispatched
      */
-    val start = if (started == SharingStarted.Eagerly) CoroutineStart.DEFAULT else CoroutineStart.UNDISPATCHED
+    val start = if (started == SharingStarted.Eagerly || isNativeMt) CoroutineStart.DEFAULT else CoroutineStart.UNDISPATCHED
     return launch(context, start = start) { // the single coroutine to rule the sharing
         // Optimize common built-in started strategies
         when {

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

@@ -17,3 +17,5 @@ public expect open class SynchronizedObject() // marker abstract class
  */
 @InternalCoroutinesApi
 public expect inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T
+
+internal expect val isNativeMt: Boolean

kotlinx-coroutines-core/common/test/flow/sharing/ShareInTest.kt

@@ -6,6 +6,7 @@ package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.internal.*
 import kotlin.test.*
 
 class ShareInTest : TestBase() {
@@ -213,6 +214,7 @@ class ShareInTest : TestBase() {
 
     @Test
     fun testShouldStart() = runTest {
+        if (isNativeMt) return@runTest
         val flow = flow {
             expect(2)
             emit(1)
@@ -229,6 +231,7 @@ class ShareInTest : TestBase() {
 
     @Test
     fun testShouldStartScalar() = runTest {
+        if (isNativeMt) return@runTest
         val j = Job()
         val shared = flowOf(239).stateIn(this + j, SharingStarted.Lazily, 42)
         assertEquals(42, shared.first())

kotlinx-coroutines-core/concurrent/test/StressUtil.common.kt

@@ -18,3 +18,5 @@ public actual typealias SynchronizedObject = Any
 @InternalCoroutinesApi
 public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T =
     block()
+
+internal actual val isNativeMt: Boolean = false

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

@@ -18,3 +18,5 @@ public actual typealias SynchronizedObject = Any
 @InternalCoroutinesApi
 public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T =
     kotlin.synchronized(lock, block)
+
+internal actual val isNativeMt: Boolean = false

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

@@ -18,3 +18,6 @@ public actual typealias SynchronizedObject = kotlinx.atomicfu.locks.Synchronized
  */
 @InternalCoroutinesApi
 public actual inline fun <T> synchronized(lock: SynchronizedObject, block: () -> T): T = lock.withLock2(block)
+
+@OptIn(ExperimentalStdlibApi::class)
+internal actual val isNativeMt: Boolean = !kotlin.native.isExperimentalMM()