~

Author: qwwdfsad

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

@@ -14,6 +14,7 @@ internal expect class LocalAtomicInt(value: Int) {
     fun get(): Int
     fun set(value: Int)
     fun decrementAndGet(): Int
+    fun incrementAndGet(): Int
 }
 
 internal inline var LocalAtomicInt.value

kotlinx-coroutines-core/common/src/sync/Mutex.kt

@@ -172,13 +172,50 @@ internal open class MutexImpl(locked: Boolean) : SemaphoreImpl(1, if (locked) 1
 
     override fun tryLock(owner: Any?): Boolean =
         if (tryAcquire()) {
-            assert { this.owner.value === NO_OWNER }
+            val previousOwner = this.owner.value
+            // First setup the invariant, then fail on assert of things went south
             this.owner.value = owner
+            assert { previousOwner === NO_OWNER }
             true
         } else {
+            /*
+             * We have to ensure invariant that any lock attempt on
+             * mutex already locked with the same owner fails with ISE.
+             * Here we can have the following logical race:
+             * - T1: lock(O1) holds
+             * - T2: tryLock(O2) fails, we fall into this branch
+             * - T1: unlock(O1), lock(O3)
+             * - T2: reads O3 in else branch
+             */
+            check(owner === null || owner !== this.owner.value) {
+                "Invariant violation: already locked by $owner"
+            }
             false
         }
 
+//    override fun tryLock(owner: Any?): Boolean {
+//        /*
+//         * Here we are maintaining invariant "tryLock(owner) on locked with this very owner mutex throws ISE".
+//         * In order to achieve that and avoid check-and-act, we do stamped comparison, restarting the operation if
+//         * stamped check failed.
+//         */
+//        while (true) {
+//            val ownerBeforeAcquisition = this.owner.value
+//            return if (tryAcquire()) {
+//                assert { this.owner.value === NO_OWNER }
+//                this.owner.value = owner
+//                true
+//            } else {
+//                // TODO handle nulls as well and do not retry
+//                val ownerAfterAcquisition = this.owner.value
+//                // Owner changed -> we are racing with lock/unlock and cannot deterministically check the invariant, retry
+//                if (ownerBeforeAcquisition !== ownerAfterAcquisition) continue
+//                check(owner !== ownerAfterAcquisition) { "Invariant violation: already locked by $owner" }
+//                false
+//            }
+//        }
+//    }
+
     override fun unlock(owner: Any?) {
         while (true) {
             // Is this mutex locked?

kotlinx-coroutines-core/common/test/sync/MutexTest.kt

@@ -138,4 +138,13 @@ class MutexTest : TestBase() {
         assertTrue(mutex.holdsLock(owner2))
         finish(4)
     }
+
+    @Test
+    fun testIllegalStateInvariant() = runTest {
+        val mutex = Mutex()
+        val owner = Any()
+        assertTrue(mutex.tryLock(owner))
+        assertFailsWith<IllegalStateException> { mutex.tryLock(owner) }
+        assertFailsWith<IllegalStateException> { mutex.lock(owner) }
+    }
 }

kotlinx-coroutines-core/concurrent/test/sync/MutexStressTest.kt

@@ -6,6 +6,7 @@ package kotlinx.coroutines.sync
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.exceptions.*
+import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.selects.*
 import kotlin.test.*
 
@@ -32,7 +33,7 @@ class MutexStressTest : TestBase() {
 
     @Test
     fun testMultiThreadedContext() = runTest {
-       newFixedThreadPoolContext(8, "testMultiThreadedContext").use {
+        newFixedThreadPoolContext(8, "testMultiThreadedContext").use {
             testBody(it)
         }
     }
@@ -134,4 +135,39 @@ class MutexStressTest : TestBase() {
             assertFalse { mutex.isLocked }
         }
     }
+
+    @Test
+    fun testTryLockInvariant() = runTest {
+        val mutex = Mutex()
+        val owner = Any()
+        val iterations = n * 10 * stressTestMultiplierSqrt
+        val failures = LocalAtomicInt(0)
+        val job = launch(Dispatchers.Default) {
+            repeat(iterations) {
+                try {
+                    mutex.lock(owner)
+                    yield()
+                    mutex.unlock()
+                } catch (e: IllegalStateException) {
+                    failures.incrementAndGet()
+                }
+            }
+        }
+
+        val job2 = launch(Dispatchers.Default) {
+            while (job.isActive) {
+                try {
+                    val locked = mutex.tryLock(owner)
+                    if (locked) {
+                        mutex.unlock(owner)
+                    }
+                } catch (e: IllegalStateException) {
+                    failures.incrementAndGet()
+                }
+            }
+        }
+
+        joinAll(job, job2)
+        assertTrue(failures.value > 0, "At least one ISE was expected during this stress-test")
+    }
 }

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

@@ -12,4 +12,5 @@ internal actual class LocalAtomicInt actual constructor(private var value: Int)
     actual fun get(): Int = value
 
     actual fun decrementAndGet(): Int = --value
+    actual fun incrementAndGet(): Int = ++value
 }

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

@@ -17,4 +17,6 @@ internal actual class LocalAtomicInt actual constructor(value: Int) {
     actual fun get(): Int = iRef.value
 
     actual fun decrementAndGet(): Int = iRef.decrementAndGet()
+
+    actual fun incrementAndGet(): Int = iRef.incrementAndGet()
 }