diff --git a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartMutex.kt b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartMutex.kt index 5ae306a725d..56c689808ed 100644 --- a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartMutex.kt +++ b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartMutex.kt @@ -1,5 +1,6 @@ package dev.inmo.micro_utils.coroutines +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -7,6 +8,7 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.isActive import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext import kotlin.contracts.ExperimentalContracts import kotlin.contracts.InvocationKind import kotlin.contracts.contract @@ -92,8 +94,8 @@ sealed interface SmartMutex { * If [isLocked] == true - will change it to false and return true. If current call will not unlock this * [SmartMutex] - false */ - suspend fun unlock(): Boolean { - return if (_lockStateFlow.value) { + suspend fun unlock(): Boolean = withContext(NonCancellable) { + if (_lockStateFlow.value) { internalChangesMutex.withLock { if (_lockStateFlow.value) { _lockStateFlow.value = false diff --git a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartRWLocker.kt b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartRWLocker.kt index aea220812cd..f1f74e7b59f 100644 --- a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartRWLocker.kt +++ b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartRWLocker.kt @@ -1,6 +1,8 @@ package dev.inmo.micro_utils.coroutines import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.withContext import kotlin.contracts.ExperimentalContracts import kotlin.contracts.InvocationKind import kotlin.contracts.contract @@ -21,6 +23,7 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: val readSemaphore: SmartSemaphore.Immutable = _readSemaphore.immutable() val writeMutex: SmartMutex.Immutable = _writeMutex.immutable() + /** * Do lock in [readSemaphore] inside of [writeMutex] locking */ @@ -32,8 +35,8 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: /** * Release one read permit in [readSemaphore] */ - suspend fun releaseRead(): Boolean { - return _readSemaphore.release() + suspend fun releaseRead(): Boolean = withContext(NonCancellable) { + _readSemaphore.release() } /** @@ -44,7 +47,9 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: try { _readSemaphore.acquire(readPermits) } catch (e: CancellationException) { - _writeMutex.unlock() + withContext(NonCancellable) { + _writeMutex.unlock() + } throw e } } @@ -52,9 +57,9 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: /** * Unlock [writeMutex] */ - suspend fun unlockWrite(): Boolean { - return _writeMutex.unlock().also { - if (it) { + suspend fun unlockWrite(): Boolean = withContext(NonCancellable) { + _writeMutex.unlock().also { unlocked -> + if (unlocked) { _readSemaphore.release(readPermits) } } diff --git a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphore.kt b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphore.kt index eb06cafd9c7..43f0550e4fc 100644 --- a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphore.kt +++ b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphore.kt @@ -1,5 +1,6 @@ package dev.inmo.micro_utils.coroutines +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow @@ -8,6 +9,7 @@ import kotlinx.coroutines.isActive import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext import kotlin.contracts.ExperimentalContracts import kotlin.contracts.InvocationKind import kotlin.contracts.contract @@ -76,7 +78,9 @@ sealed interface SmartSemaphore { } } while (shouldContinue && currentCoroutineContext().isActive) } catch (e: Throwable) { - release(acquiredPermits) + if (acquiredPermits > 0) { + release(acquiredPermits) + } throw e } } @@ -107,9 +111,9 @@ sealed interface SmartSemaphore { */ suspend fun tryAcquire(permits: Int = 1): Boolean { val checkedPermits = checkedPermits(permits) - return if (_freePermitsStateFlow.value < checkedPermits) { + return if (_freePermitsStateFlow.value >= checkedPermits) { internalChangesMutex.withLock { - if (_freePermitsStateFlow.value < checkedPermits) { + if (_freePermitsStateFlow.value >= checkedPermits) { _freePermitsStateFlow.value -= checkedPermits true } else { @@ -125,12 +129,12 @@ sealed interface SmartSemaphore { * If [freePermits] == true - will change it to false and return true. If current call will not unlock this * [SmartSemaphore] - false */ - suspend fun release(permits: Int = 1): Boolean { + suspend fun release(permits: Int = 1): Boolean = withContext(NonCancellable) { val checkedPermits = checkedPermits(permits) - return if (_freePermitsStateFlow.value < this.maxPermits) { + if (_freePermitsStateFlow.value < maxPermits) { internalChangesMutex.withLock { - if (_freePermitsStateFlow.value < this.maxPermits) { - _freePermitsStateFlow.value = minOf(_freePermitsStateFlow.value + checkedPermits, this.maxPermits) + if (_freePermitsStateFlow.value < maxPermits) { + _freePermitsStateFlow.value = minOf(_freePermitsStateFlow.value + checkedPermits, maxPermits) true } else { false diff --git a/coroutines/src/commonTest/kotlin/SmartMutexTests.kt b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt new file mode 100644 index 00000000000..b75a26b9c2d --- /dev/null +++ b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt @@ -0,0 +1,73 @@ +import dev.inmo.micro_utils.coroutines.SmartMutex +import dev.inmo.micro_utils.coroutines.withLock +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.cancel +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertFalse +import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.seconds + +class SmartMutexTests { + @Test + fun cancelledUnlockCompletesUnderContention() = runTest(timeout = 5.seconds) { + val mutex = SmartMutex.Mutable() + + // Delegate this acquisition's release to another coroutine. The + // unconfined collector runs while lock() still holds its internal mutex. + val releaser = launch(Dispatchers.Unconfined) { + mutex.lockStateFlow.first { it } + currentCoroutineContext().cancel() + mutex.unlock() + } + + // Keep acquisition on the normal test dispatcher so the collector + // attempts cancelled cleanup before the internal mutex is released. + mutex.lock() + releaser.join() + + assertTrue(releaser.isCancelled) + assertFalse(mutex.isLocked, "Cancellation must not prevent the delegated unlock") + assertTrue(mutex.tryLock()) + assertTrue(mutex.unlock()) + } + + @Test + fun cancellingWithLockBodyReleasesMutex() = runTest(timeout = 5.seconds) { + val mutex = SmartMutex.Mutable() + val holder = launch(Dispatchers.Unconfined) { + mutex.withLock { + awaitCancellation() + } + } + assertTrue(mutex.isLocked) + + holder.cancelAndJoin() + + assertFalse(mutex.isLocked) + } + + @Test + fun cancelledWaiterDoesNotEnterOrReleaseHeldMutex() = runTest(timeout = 5.seconds) { + val mutex = SmartMutex.Mutable() + var entered = false + mutex.withLock { + val waiter = launch(Dispatchers.Unconfined) { + mutex.withLock { + entered = true + } + } + + waiter.cancelAndJoin() + + assertFalse(entered) + assertTrue(mutex.isLocked) + } + assertFalse(mutex.isLocked) + } +} diff --git a/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt b/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt index 210ce222b15..de4b374a9a3 100644 --- a/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt +++ b/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt @@ -9,6 +9,7 @@ import kotlin.test.assertEquals import kotlin.test.assertFails import kotlin.test.assertFalse import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.days import kotlin.time.Duration.Companion.seconds class SmartRWLockerTests { @@ -109,6 +110,59 @@ class SmartRWLockerTests { } } + @Test + fun failureOnReadFreeingRead() = runTest { + val locker = SmartRWLocker() + val job = launch { + locker.withReadAcquire { + while (isActive) { + delay(1.days) + } + } + } + + locker.readSemaphore.permitsStateFlow.first { + it == locker.readSemaphore.maxPermits - 1 + } + job.cancelAndJoin() + locker.readSemaphore.permitsStateFlow.first { + it == locker.readSemaphore.maxPermits + } + } + + @Test + fun cancelledReaderReleasesPermitUnderContention() = runTest(timeout = 5.seconds) { + val locker = SmartRWLocker(readPermits = 2) + val reader = launch(Dispatchers.Unconfined) { + locker.withReadAcquire { + awaitCancellation() + } + } + assertEquals(1, locker.readSemaphore.freePermits) + + // Observe the second acquisition synchronously while it still holds the + // semaphore's internal mutex. Cancelling the unconfined reader makes its + // cleanup contend for that mutex before the acquisition can release it. + val cancellation = launch(Dispatchers.Unconfined) { + locker.readSemaphore.permitsStateFlow.first { it == 0 } + reader.cancel() + } + + // Keep this acquisition on the normal test dispatcher: making it + // unconfined would change the ordering that forces cleanup contention. + locker.withReadAcquire { + cancellation.join() + reader.join() + assertTrue(reader.isCancelled) + assertEquals( + 1, + locker.readSemaphore.freePermits, + "The cancelled reader must release its permit while the other reader still holds one" + ) + } + assertEquals(2, locker.readSemaphore.freePermits) + } + @Test fun simpleWithReadAcquireTest() { val locker = SmartRWLocker() diff --git a/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt b/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt new file mode 100644 index 00000000000..7aa900cbe84 --- /dev/null +++ b/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt @@ -0,0 +1,104 @@ +import dev.inmo.micro_utils.coroutines.SmartSemaphore +import dev.inmo.micro_utils.coroutines.withAcquire +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue +import kotlin.time.Duration.Companion.seconds + +class SmartSemaphoreTests { + @Test + fun cancelledHolderReleasesPermitUnderContention() = runTest(timeout = 5.seconds) { + val semaphore = SmartSemaphore.Mutable(permits = 2) + val holder = launch(Dispatchers.Unconfined) { + semaphore.withAcquire { + awaitCancellation() + } + } + assertEquals(1, semaphore.freePermits) + + // The synchronous observer cancels the holder while the second + // acquisition still owns the semaphore's internal changes mutex. + val cancellation = launch(Dispatchers.Unconfined) { + semaphore.permitsStateFlow.first { it == 0 } + holder.cancel() + } + semaphore.withAcquire { + cancellation.join() + holder.join() + assertTrue(holder.isCancelled) + assertEquals(1, semaphore.freePermits, "The cancelled holder must return its permit") + } + assertEquals(2, semaphore.freePermits) + } + + @Test + fun cancelledAcquireReturnsPartialPermitsUnderContention() = runTest(timeout = 5.seconds) { + // One permit belongs to another holder; the waiter can acquire two + // permits immediately, but must wait for the third. + val semaphore = SmartSemaphore.Mutable(permits = 3, acquiredPermits = 1) + lateinit var waiter: kotlinx.coroutines.Job + val cancellation = launch(Dispatchers.Unconfined) { + semaphore.permitsStateFlow.first { it == 1 } + waiter.cancel() + } + waiter = launch(Dispatchers.Unconfined) { + semaphore.acquire(3) + } + assertEquals(0, semaphore.freePermits) + assertFalse(waiter.isCompleted) + + // Publishing this release resumes the observer while the internal + // mutex is held. The cancelled acquire must wait to roll back safely. + semaphore.release() + cancellation.join() + waiter.join() + assertTrue(waiter.isCancelled) + assertEquals(3, semaphore.freePermits, "Cancellation must return both partially acquired permits") + semaphore.withAcquire(3) { + assertEquals(0, semaphore.freePermits) + } + assertEquals(3, semaphore.freePermits) + } + + @Test + fun cancelledAcquireWithoutPermitsDoesNotReleaseAnotherHoldersPermit() = runTest(timeout = 5.seconds) { + val semaphore = SmartSemaphore.Mutable(permits = 1, acquiredPermits = 1) + val waiter = launch(Dispatchers.Unconfined) { + semaphore.acquire() + } + assertFalse(waiter.isCompleted) + + waiter.cancelAndJoin() + assertEquals(0, semaphore.freePermits, "A cancelled waiter that acquired nothing must release nothing") + assertTrue(semaphore.release()) + assertEquals(1, semaphore.freePermits) + } + + @Test + fun tryAcquireUsesAvailablePermits() = runTest { + val semaphore = SmartSemaphore.Mutable(permits = 3) + assertTrue(semaphore.tryAcquire(2)) + assertEquals(1, semaphore.freePermits) + assertTrue(semaphore.tryAcquire()) + assertEquals(0, semaphore.freePermits) + assertTrue(semaphore.release(3)) + assertEquals(3, semaphore.freePermits) + } + + @Test + fun tryAcquireWithInsufficientPermitsLeavesStateUnchanged() = runTest { + val semaphore = SmartSemaphore.Mutable(permits = 3, acquiredPermits = 2) + assertFalse(semaphore.tryAcquire(2)) + assertEquals(1, semaphore.freePermits) + semaphore.acquire() + assertFalse(semaphore.tryAcquire()) + assertEquals(0, semaphore.freePermits) + } +}