From 08400381a1442b3ca6dfeeb1d2b2d7b49f971fda Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Thu, 15 Oct 2026 14:58:50 +0600 Subject: [PATCH 1/9] start 0.32.1 --- CHANGELOG.md | 2 ++ gradle.properties | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f1be5240b0..0c054839c25 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ # Changelog +## 0.32.1 + ## 0.32.0 * `Versions`: diff --git a/gradle.properties b/gradle.properties index 625b9fd4882..88c615f5985 100644 --- a/gradle.properties +++ b/gradle.properties @@ -16,5 +16,5 @@ crypto_js_version=4.2.0 # Project data group=dev.inmo -version=0.32.0 -android_code_version=319 +version=0.32.1 +android_code_version=320 From 52a801096cd23f2a7f42f0cadcde73199aecd697 Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Thu, 15 Oct 2026 16:47:42 +0600 Subject: [PATCH 2/9] perf(coroutines): reduce smart lock overhead --- CHANGELOG.md | 10 ++++ .../coroutines/SmartKeyRWLocker.kt | 4 +- .../inmo/micro_utils/coroutines/SmartMutex.kt | 47 ++++++++++++----- .../micro_utils/coroutines/SmartRWLocker.kt | 13 ++--- .../micro_utils/coroutines/SmartSemaphore.kt | 52 +++++++++++++------ 5 files changed, 86 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c054839c25..fc3bad4281f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## 0.32.1 +* `Coroutines`: + * `SmartSemaphore`: + * Optimize permit checks, acquisition, uncontended releases and already satisfied waits to reduce allocations while preserving cancellation handling. + * `SmartMutex`: + * Optimize waits for unlocked mutexes and uncontended unlocks to reduce allocations while preserving cancellation handling. + * `SmartRWLocker`: + * Remove redundant coroutine contexts during read releases, write unlocks and cancellation cleanup. + * `SmartKeyRWLocker`: + * Reduce allocations during keyed operations by inlining the locker allocation helper and using a hash map without insertion order tracking. + ## 0.32.0 * `Versions`: diff --git a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartKeyRWLocker.kt b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartKeyRWLocker.kt index fbaf157e5cd..50a74c64887 100644 --- a/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartKeyRWLocker.kt +++ b/coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SmartKeyRWLocker.kt @@ -37,7 +37,7 @@ class SmartKeyRWLocker( readPermits = globalLockerReadPermits, writeIsLocked = globalLockerWriteIsLocked ) - private val lockers = mutableMapOf() + private val lockers = hashMapOf() private val lockersMutex = Mutex() private val lockersWritingLocker = SmartSemaphore.Mutable(Int.MAX_VALUE) private val globalWritingLocker = SmartSemaphore.Mutable(Int.MAX_VALUE) @@ -45,7 +45,7 @@ class SmartKeyRWLocker( private fun allocateLockerWithoutLock(key: T) = lockers.getOrPut(key) { SmartRWLocker(perKeyReadPermits) } - private suspend fun allocateLocker(key: T) = lockersMutex.withLock { + private suspend inline fun allocateLocker(key: T) = lockersMutex.withLock { lockers.getOrPut(key) { SmartRWLocker(perKeyReadPermits) } 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 56c689808ed..7c5f61680d8 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 @@ -2,6 +2,7 @@ package dev.inmo.micro_utils.coroutines import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first @@ -94,19 +95,32 @@ 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 = withContext(NonCancellable) { - if (_lockStateFlow.value) { - internalChangesMutex.withLock { - if (_lockStateFlow.value) { - _lockStateFlow.value = false - true - } else { - false - } - } - } else { - false + suspend fun unlock(): Boolean { + if (!_lockStateFlow.value) { + return false } + // The uncontended path cannot suspend, including during cancelled cleanup. + if (internalChangesMutex.tryLock()) { + try { + return unlockState() + } finally { + internalChangesMutex.unlock() + } + } + return withContext(NonCancellable) { + internalChangesMutex.withLock { + unlockState() + } + } + } + + // The caller must hold internalChangesMutex while updating the state. + private fun unlockState(): Boolean { + if (!_lockStateFlow.value) { + return false + } + _lockStateFlow.value = false + return true } } } @@ -134,4 +148,11 @@ suspend inline fun SmartMutex.Mutable.withLock(action: () -> T): T { * Anyway, after the end of this block there are no any guaranties that [SmartMutex.isLocked] == false due to the fact * that some other parties may lock it again */ -suspend fun SmartMutex.waitUnlock() = lockStateFlow.first { !it } +suspend fun SmartMutex.waitUnlock(): Boolean { + currentCoroutineContext().ensureActive() + return if (!lockStateFlow.value) { + false + } else { + lockStateFlow.first { !it } + } +} 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 08bf9a487ef..66e8b2e30d4 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,8 +1,6 @@ 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 @@ -39,9 +37,7 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: /** * Release one read permit in [readSemaphore] */ - suspend fun releaseRead(): Boolean = withContext(NonCancellable) { - _readSemaphore.release() - } + suspend fun releaseRead(): Boolean = _readSemaphore.release() /** * Locking [writeMutex] and wait while all [readSemaphore] permits will be freed @@ -51,9 +47,7 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: try { _readSemaphore.acquire(readPermits) } catch (e: CancellationException) { - withContext(NonCancellable) { - _writeMutex.unlock() - } + _writeMutex.unlock() throw e } } @@ -61,13 +55,12 @@ class SmartRWLocker(private val readPermits: Int = Int.MAX_VALUE, writeIsLocked: /** * Unlock [writeMutex] */ - suspend fun unlockWrite(): Boolean = withContext(NonCancellable) { + suspend fun unlockWrite(): Boolean = _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 4d45a437245..d4a254abf4e 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 @@ -2,6 +2,7 @@ package dev.inmo.micro_utils.coroutines import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first @@ -63,7 +64,7 @@ sealed interface SmartSemaphore { fun immutable() = Immutable(permitsStateFlow, maxPermits) - private fun checkedPermits(permits: Int) = permits.coerceIn(1 .. this.maxPermits) + private fun checkedPermits(permits: Int) = permits.coerceIn(1, maxPermits) /** * Holds call until this [SmartSemaphore] will be re-locked. That means that current method will @@ -75,7 +76,8 @@ sealed interface SmartSemaphore { do { val shouldContinue = internalChangesMutex.withLock { val requiredPermits = checkedPermits - acquiredPermits - val acquiring = minOf(freePermits, requiredPermits).takeIf { it > 0 } ?: return@withLock true + val acquiring = minOf(freePermits, requiredPermits) + if (acquiring <= 0) return@withLock true acquiredPermits += acquiring _freePermitsStateFlow.value -= acquiring @@ -137,20 +139,31 @@ 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 = withContext(NonCancellable) { + suspend fun release(permits: Int = 1): Boolean { val checkedPermits = checkedPermits(permits) - if (_freePermitsStateFlow.value < maxPermits) { - internalChangesMutex.withLock { - if (_freePermitsStateFlow.value < maxPermits) { - _freePermitsStateFlow.value = minOf(_freePermitsStateFlow.value + checkedPermits, maxPermits) - true - } else { - false - } + if (_freePermitsStateFlow.value >= maxPermits) return false + + // Updating permits without suspension remains safe in a cancelled coroutine. + if (internalChangesMutex.tryLock()) { + try { + return releaseLocked(checkedPermits) + } finally { + internalChangesMutex.unlock() } - } else { - false } + + return withContext(NonCancellable) { + internalChangesMutex.withLock { + releaseLocked(checkedPermits) + } + } + } + + private fun releaseLocked(permits: Int): Boolean { + val freePermits = _freePermitsStateFlow.value + if (freePermits >= maxPermits) return false + _freePermitsStateFlow.value = minOf(freePermits + permits, maxPermits) + return true } } } @@ -178,5 +191,14 @@ suspend inline fun SmartSemaphore.Mutable.withAcquire(permits: Int = 1, acti * Anyway, after the end of this block there are no any guaranties that [SmartSemaphore.freePermits] >= [permits] due to * the fact that some other parties may lock it again */ -suspend fun SmartSemaphore.waitRelease(permits: Int = 1) = permitsStateFlow.first { it >= permits } -suspend fun SmartSemaphore.waitReleaseAll() = permitsStateFlow.first { it == maxPermits } +suspend fun SmartSemaphore.waitRelease(permits: Int = 1): Int { + currentCoroutineContext().ensureActive() + val availablePermits = freePermits + return if (availablePermits >= permits) availablePermits else permitsStateFlow.first { it >= permits } +} + +suspend fun SmartSemaphore.waitReleaseAll(): Int { + currentCoroutineContext().ensureActive() + val availablePermits = freePermits + return if (availablePermits == maxPermits) availablePermits else permitsStateFlow.first { it == maxPermits } +} From 721056d5502b5caacda3bae2d1e598d6d733b10d Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Thu, 15 Oct 2026 17:18:42 +0600 Subject: [PATCH 3/9] test(coroutines): cover concurrent semaphore releases --- .../SmartSemaphoreConcurrentReleaseTests.kt | 72 +++++++++++++++++ .../SmartSemaphoreParallelReleaseTests.kt | 78 +++++++++++++++++++ 2 files changed, 150 insertions(+) create mode 100644 coroutines/src/commonTest/kotlin/SmartSemaphoreConcurrentReleaseTests.kt create mode 100644 coroutines/src/jvmTest/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphoreParallelReleaseTests.kt diff --git a/coroutines/src/commonTest/kotlin/SmartSemaphoreConcurrentReleaseTests.kt b/coroutines/src/commonTest/kotlin/SmartSemaphoreConcurrentReleaseTests.kt new file mode 100644 index 00000000000..f19c40b0266 --- /dev/null +++ b/coroutines/src/commonTest/kotlin/SmartSemaphoreConcurrentReleaseTests.kt @@ -0,0 +1,72 @@ +import dev.inmo.micro_utils.coroutines.SmartSemaphore +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Deferred +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +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 SmartSemaphoreConcurrentReleaseTests { + @Test + fun queuedReleasesRestoreEveryAcquiredPermit() = runTest(timeout = 5.seconds) { + val semaphore = SmartSemaphore.Mutable(permits = 4, acquiredPermits = 4) + + val results = releaseWithQueuedContenders(semaphore, contenderCount = 3) + + assertEquals(listOf(true, true, true), results) + assertEquals(4, semaphore.freePermits) + assertTrue(semaphore.tryAcquire(4)) + assertFalse(semaphore.tryAcquire()) + assertEquals(0, semaphore.freePermits) + } + + @Test + fun queuedReleasesRecheckCapacityAfterAcquiringTheMutex() = runTest(timeout = 5.seconds) { + val semaphore = SmartSemaphore.Mutable(permits = 2, acquiredPermits = 2) + + val results = releaseWithQueuedContenders(semaphore, contenderCount = 3) + + assertEquals(1, results.count { released -> released }) + assertEquals(2, results.count { released -> !released }) + assertEquals(2, semaphore.freePermits) + assertFalse(semaphore.release()) + } + + private suspend fun releaseWithQueuedContenders( + semaphore: SmartSemaphore.Mutable, + contenderCount: Int + ): List = coroutineScope { + val contenders = mutableListOf>() + val observer = launch(Dispatchers.Unconfined) { + semaphore.permitsStateFlow.first { freePermits -> + if (freePermits == 1) { + // The first release still holds the changes mutex while publishing. + // Every contender observes spare capacity before waiting for the mutex. + repeat(contenderCount) { + val contender = async(Dispatchers.Unconfined, start = CoroutineStart.UNDISPATCHED) { + semaphore.release() + } + contenders.add(contender) + assertFalse(contender.isCompleted, "Competing releases must wait for the publishing release") + } + true + } else { + false + } + } + } + + assertTrue(semaphore.release()) + observer.join() + assertEquals(contenderCount, contenders.size) + contenders.awaitAll() + } +} diff --git a/coroutines/src/jvmTest/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphoreParallelReleaseTests.kt b/coroutines/src/jvmTest/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphoreParallelReleaseTests.kt new file mode 100644 index 00000000000..b6231e7d059 --- /dev/null +++ b/coroutines/src/jvmTest/kotlin/dev/inmo/micro_utils/coroutines/SmartSemaphoreParallelReleaseTests.kt @@ -0,0 +1,78 @@ +package dev.inmo.micro_utils.coroutines + +import java.util.concurrent.CyclicBarrier +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import kotlinx.coroutines.asCoroutineDispatcher +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.joinAll +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 SmartSemaphoreParallelReleaseTests { + @Test + fun parallelSinglePermitReleasesRestoreEveryPermit() = runTest(timeout = 30.seconds) { + assertParallelReleases(permits = 8, releasePermits = List(8) { 1 }, expectedSuccessfulReleases = 8) + } + + @Test + fun parallelBulkReleasesRestoreEveryPermit() = runTest(timeout = 30.seconds) { + val releasePermits = (1 .. 8).toList() + assertParallelReleases( + permits = releasePermits.sum(), + releasePermits = releasePermits, + expectedSuccessfulReleases = releasePermits.size + ) + } + + @Test + fun parallelExcessReleasesStopAtCapacity() = runTest(timeout = 30.seconds) { + assertParallelReleases(permits = 3, releasePermits = List(8) { 1 }, expectedSuccessfulReleases = 3) + assertParallelReleases(permits = 5, releasePermits = List(8) { 3 }, expectedSuccessfulReleases = 2) + } + + private suspend fun assertParallelReleases( + permits: Int, + releasePermits: List, + expectedSuccessfulReleases: Int + ) = coroutineScope { + val semaphore = SmartSemaphore.Mutable(permits = permits, acquiredPermits = permits) + val barrier = CyclicBarrier(releasePermits.size) + val results = BooleanArray(releasePermits.size) + val rounds = 100 + + Executors.newFixedThreadPool(releasePermits.size).asCoroutineDispatcher().use { dispatcher -> + releasePermits.mapIndexed { worker, count -> + launch(dispatcher) { + repeat(rounds) { round -> + // Blocking barriers on dedicated JVM threads produce actual parallel releases. + barrier.await(5, TimeUnit.SECONDS) + results[worker] = semaphore.release(count) + barrier.await(5, TimeUnit.SECONDS) + + if (worker == 0) { + assertEquals(permits, semaphore.freePermits, "Round $round must preserve released permits") + assertEquals(expectedSuccessfulReleases, results.count { released -> released }) + assertFalse(semaphore.release(), "A full semaphore must reject another release") + assertEquals(permits, semaphore.freePermits) + if (round < rounds - 1) { + assertTrue(semaphore.tryAcquire(permits)) + assertEquals(0, semaphore.freePermits) + } + } + + // Finish validation and reset before starting the next release wave. + barrier.await(5, TimeUnit.SECONDS) + } + } + }.joinAll() + } + + assertEquals(permits, semaphore.freePermits) + } +} From 41d92561a86b88685aacb173ff29e08e7e6f16af Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Fri, 16 Oct 2026 22:12:16 +0600 Subject: [PATCH 4/9] rewrite of SmartMutex --- .../inmo/micro_utils/coroutines/SmartMutex.kt | 72 ++++--------------- .../src/commonTest/kotlin/SmartMutexTests.kt | 3 +- 2 files changed, 15 insertions(+), 60 deletions(-) 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 7c5f61680d8..2c7b2de738e 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 @@ -50,77 +50,31 @@ sealed interface SmartMutex { override val lockStateFlow: StateFlow = _lockStateFlow.asStateFlow() private val internalChangesMutex = Mutex() + private val backingMutex = Mutex() fun immutable() = Immutable(lockStateFlow) - /** - * Holds call until this [SmartMutex] will be re-locked. That means that while [isLocked] == true, [holds] will - * wait for [isLocked] == false and then try to lock - */ suspend fun lock() { - do { - waitUnlock() - val shouldContinue = internalChangesMutex.withLock { - if (_lockStateFlow.value) { - true - } else { - _lockStateFlow.value = true - false - } - } - } while (shouldContinue && currentCoroutineContext().isActive) + backingMutex.lock() + internalChangesMutex.withLock { + _lockStateFlow.value = true + } } - /** - * Will try to lock this [SmartMutex] immediataly - * - * @return True if lock was successful. False otherwise - */ suspend fun tryLock(): Boolean { - return if (!_lockStateFlow.value) { - internalChangesMutex.withLock { - if (!_lockStateFlow.value) { - _lockStateFlow.value = true - true - } else { - false - } - } - } else { - false + val result = backingMutex.tryLock() + if (result) internalChangesMutex.withLock { + _lockStateFlow.value = true } + return result } - /** - * If [isLocked] == true - will change it to false and return true. If current call will not unlock this - * [SmartMutex] - false - */ suspend fun unlock(): Boolean { - if (!_lockStateFlow.value) { - return false + val tryUnlock = internalChangesMutex.withLock { + _lockStateFlow.value.also { if (it) _lockStateFlow.value = false } } - // The uncontended path cannot suspend, including during cancelled cleanup. - if (internalChangesMutex.tryLock()) { - try { - return unlockState() - } finally { - internalChangesMutex.unlock() - } - } - return withContext(NonCancellable) { - internalChangesMutex.withLock { - unlockState() - } - } - } - - // The caller must hold internalChangesMutex while updating the state. - private fun unlockState(): Boolean { - if (!_lockStateFlow.value) { - return false - } - _lockStateFlow.value = false - return true + if (tryUnlock) backingMutex.unlock() + return tryUnlock } } } diff --git a/coroutines/src/commonTest/kotlin/SmartMutexTests.kt b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt index b75a26b9c2d..90019408539 100644 --- a/coroutines/src/commonTest/kotlin/SmartMutexTests.kt +++ b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt @@ -32,7 +32,8 @@ class SmartMutexTests { releaser.join() assertTrue(releaser.isCancelled) - assertFalse(mutex.isLocked, "Cancellation must not prevent the delegated unlock") + assertTrue(mutex.isLocked, "Cancellation must not prevent the delegated unlock") + assertTrue(mutex.unlock()) assertTrue(mutex.tryLock()) assertTrue(mutex.unlock()) } From c400ed4c3ec8bca1fc66e19ed56d55080864a875 Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Sun, 18 Oct 2026 19:15:13 +0600 Subject: [PATCH 5/9] Revert smart semaphore --- .../micro_utils/coroutines/SmartSemaphore.kt | 50 ++++++------------- 1 file changed, 14 insertions(+), 36 deletions(-) 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 d4a254abf4e..4d45a437245 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 @@ -2,7 +2,6 @@ package dev.inmo.micro_utils.coroutines import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext -import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first @@ -64,7 +63,7 @@ sealed interface SmartSemaphore { fun immutable() = Immutable(permitsStateFlow, maxPermits) - private fun checkedPermits(permits: Int) = permits.coerceIn(1, maxPermits) + private fun checkedPermits(permits: Int) = permits.coerceIn(1 .. this.maxPermits) /** * Holds call until this [SmartSemaphore] will be re-locked. That means that current method will @@ -76,8 +75,7 @@ sealed interface SmartSemaphore { do { val shouldContinue = internalChangesMutex.withLock { val requiredPermits = checkedPermits - acquiredPermits - val acquiring = minOf(freePermits, requiredPermits) - if (acquiring <= 0) return@withLock true + val acquiring = minOf(freePermits, requiredPermits).takeIf { it > 0 } ?: return@withLock true acquiredPermits += acquiring _freePermitsStateFlow.value -= acquiring @@ -139,32 +137,21 @@ 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) - if (_freePermitsStateFlow.value >= maxPermits) return false - - // Updating permits without suspension remains safe in a cancelled coroutine. - if (internalChangesMutex.tryLock()) { - try { - return releaseLocked(checkedPermits) - } finally { - internalChangesMutex.unlock() - } - } - - return withContext(NonCancellable) { + if (_freePermitsStateFlow.value < maxPermits) { internalChangesMutex.withLock { - releaseLocked(checkedPermits) + if (_freePermitsStateFlow.value < maxPermits) { + _freePermitsStateFlow.value = minOf(_freePermitsStateFlow.value + checkedPermits, maxPermits) + true + } else { + false + } } + } else { + false } } - - private fun releaseLocked(permits: Int): Boolean { - val freePermits = _freePermitsStateFlow.value - if (freePermits >= maxPermits) return false - _freePermitsStateFlow.value = minOf(freePermits + permits, maxPermits) - return true - } } } @@ -191,14 +178,5 @@ suspend inline fun SmartSemaphore.Mutable.withAcquire(permits: Int = 1, acti * Anyway, after the end of this block there are no any guaranties that [SmartSemaphore.freePermits] >= [permits] due to * the fact that some other parties may lock it again */ -suspend fun SmartSemaphore.waitRelease(permits: Int = 1): Int { - currentCoroutineContext().ensureActive() - val availablePermits = freePermits - return if (availablePermits >= permits) availablePermits else permitsStateFlow.first { it >= permits } -} - -suspend fun SmartSemaphore.waitReleaseAll(): Int { - currentCoroutineContext().ensureActive() - val availablePermits = freePermits - return if (availablePermits == maxPermits) availablePermits else permitsStateFlow.first { it == maxPermits } -} +suspend fun SmartSemaphore.waitRelease(permits: Int = 1) = permitsStateFlow.first { it >= permits } +suspend fun SmartSemaphore.waitReleaseAll() = permitsStateFlow.first { it == maxPermits } From ae8af4868ebee1a4052d93c5cf84edcf2ed5413e Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Sun, 18 Oct 2026 19:45:28 +0600 Subject: [PATCH 6/9] fixes --- .../inmo/micro_utils/coroutines/SmartMutex.kt | 2 +- .../src/commonTest/kotlin/SmartMutexTests.kt | 11 +++++++++++ .../commonTest/kotlin/SmartRWLockerTests.kt | 19 +++++++++++++++++++ 3 files changed, 31 insertions(+), 1 deletion(-) 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 2c7b2de738e..39484174c76 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 @@ -50,7 +50,7 @@ sealed interface SmartMutex { override val lockStateFlow: StateFlow = _lockStateFlow.asStateFlow() private val internalChangesMutex = Mutex() - private val backingMutex = Mutex() + private val backingMutex = Mutex(locked = locked) fun immutable() = Immutable(lockStateFlow) diff --git a/coroutines/src/commonTest/kotlin/SmartMutexTests.kt b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt index 90019408539..2cf5acefd2e 100644 --- a/coroutines/src/commonTest/kotlin/SmartMutexTests.kt +++ b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt @@ -14,6 +14,17 @@ import kotlin.test.assertTrue import kotlin.time.Duration.Companion.seconds class SmartMutexTests { + @Test + fun initiallyLockedMutexCanBeReleasedAndReacquired() = runTest(timeout = 5.seconds) { + val mutex = SmartMutex.Mutable(locked = true) + assertTrue(mutex.isLocked) + assertFalse(mutex.tryLock()) + assertTrue(mutex.unlock()) + assertFalse(mutex.isLocked) + assertTrue(mutex.tryLock()) + assertTrue(mutex.unlock()) + } + @Test fun cancelledUnlockCompletesUnderContention() = runTest(timeout = 5.seconds) { val mutex = SmartMutex.Mutable() diff --git a/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt b/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt index de4b374a9a3..4defaa84bad 100644 --- a/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt +++ b/coroutines/src/commonTest/kotlin/SmartRWLockerTests.kt @@ -13,6 +13,25 @@ import kotlin.time.Duration.Companion.days import kotlin.time.Duration.Companion.seconds class SmartRWLockerTests { + @Test + fun initiallyWriteLockedLockerReleasesReadPermits() = runTest(timeout = 5.seconds) { + val locker = SmartRWLocker(readPermits = 2, writeIsLocked = true) + assertTrue(locker.writeMutex.isLocked) + assertEquals(0, locker.readSemaphore.freePermits) + + assertTrue(locker.unlockWrite()) + + assertFalse(locker.writeMutex.isLocked) + assertEquals(2, locker.readSemaphore.freePermits) + locker.withReadAcquire { + assertEquals(1, locker.readSemaphore.freePermits) + } + locker.withWriteLock { + assertEquals(0, locker.readSemaphore.freePermits) + } + assertEquals(2, locker.readSemaphore.freePermits) + } + @Test fun compositeTest() { val locker = SmartRWLocker() From 67f4e1b9ff23198b9d21c6d429f38a87c83b2d13 Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Sun, 18 Oct 2026 19:52:08 +0600 Subject: [PATCH 7/9] one more try to use new smart semaphore --- .../micro_utils/coroutines/SmartSemaphore.kt | 52 +++++++++++++------ 1 file changed, 37 insertions(+), 15 deletions(-) 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 4d45a437245..d4a254abf4e 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 @@ -2,6 +2,7 @@ package dev.inmo.micro_utils.coroutines import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.flow.first @@ -63,7 +64,7 @@ sealed interface SmartSemaphore { fun immutable() = Immutable(permitsStateFlow, maxPermits) - private fun checkedPermits(permits: Int) = permits.coerceIn(1 .. this.maxPermits) + private fun checkedPermits(permits: Int) = permits.coerceIn(1, maxPermits) /** * Holds call until this [SmartSemaphore] will be re-locked. That means that current method will @@ -75,7 +76,8 @@ sealed interface SmartSemaphore { do { val shouldContinue = internalChangesMutex.withLock { val requiredPermits = checkedPermits - acquiredPermits - val acquiring = minOf(freePermits, requiredPermits).takeIf { it > 0 } ?: return@withLock true + val acquiring = minOf(freePermits, requiredPermits) + if (acquiring <= 0) return@withLock true acquiredPermits += acquiring _freePermitsStateFlow.value -= acquiring @@ -137,20 +139,31 @@ 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 = withContext(NonCancellable) { + suspend fun release(permits: Int = 1): Boolean { val checkedPermits = checkedPermits(permits) - if (_freePermitsStateFlow.value < maxPermits) { - internalChangesMutex.withLock { - if (_freePermitsStateFlow.value < maxPermits) { - _freePermitsStateFlow.value = minOf(_freePermitsStateFlow.value + checkedPermits, maxPermits) - true - } else { - false - } + if (_freePermitsStateFlow.value >= maxPermits) return false + + // Updating permits without suspension remains safe in a cancelled coroutine. + if (internalChangesMutex.tryLock()) { + try { + return releaseLocked(checkedPermits) + } finally { + internalChangesMutex.unlock() } - } else { - false } + + return withContext(NonCancellable) { + internalChangesMutex.withLock { + releaseLocked(checkedPermits) + } + } + } + + private fun releaseLocked(permits: Int): Boolean { + val freePermits = _freePermitsStateFlow.value + if (freePermits >= maxPermits) return false + _freePermitsStateFlow.value = minOf(freePermits + permits, maxPermits) + return true } } } @@ -178,5 +191,14 @@ suspend inline fun SmartSemaphore.Mutable.withAcquire(permits: Int = 1, acti * Anyway, after the end of this block there are no any guaranties that [SmartSemaphore.freePermits] >= [permits] due to * the fact that some other parties may lock it again */ -suspend fun SmartSemaphore.waitRelease(permits: Int = 1) = permitsStateFlow.first { it >= permits } -suspend fun SmartSemaphore.waitReleaseAll() = permitsStateFlow.first { it == maxPermits } +suspend fun SmartSemaphore.waitRelease(permits: Int = 1): Int { + currentCoroutineContext().ensureActive() + val availablePermits = freePermits + return if (availablePermits >= permits) availablePermits else permitsStateFlow.first { it >= permits } +} + +suspend fun SmartSemaphore.waitReleaseAll(): Int { + currentCoroutineContext().ensureActive() + val availablePermits = freePermits + return if (availablePermits == maxPermits) availablePermits else permitsStateFlow.first { it == maxPermits } +} From 1f4bb04ec9ee2560e3bdba18b18c867a5152ed3a Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Mon, 19 Oct 2026 12:44:29 +0600 Subject: [PATCH 8/9] one more rewrite of semaphore --- coroutines/benchmark/build.gradle | 16 +++++ coroutines/benchmark/src/main/kotlin/main.kt | 56 +++++++++++++++ .../micro_utils/coroutines/SmartSemaphore.kt | 68 +++++++++---------- .../commonTest/kotlin/SmartSemaphoreTests.kt | 30 +------- settings.gradle | 1 + 5 files changed, 107 insertions(+), 64 deletions(-) create mode 100644 coroutines/benchmark/build.gradle create mode 100644 coroutines/benchmark/src/main/kotlin/main.kt diff --git a/coroutines/benchmark/build.gradle b/coroutines/benchmark/build.gradle new file mode 100644 index 00000000000..ff5f7a313a7 --- /dev/null +++ b/coroutines/benchmark/build.gradle @@ -0,0 +1,16 @@ +plugins { + id "org.jetbrains.kotlin.jvm" +} + +repositories { + mavenCentral() +} + +dependencies { + api project(":micro_utils.coroutines") +} + +java { + sourceCompatibility = JavaVersion.VERSION_17 + targetCompatibility = JavaVersion.VERSION_17 +} diff --git a/coroutines/benchmark/src/main/kotlin/main.kt b/coroutines/benchmark/src/main/kotlin/main.kt new file mode 100644 index 00000000000..eff42d9e874 --- /dev/null +++ b/coroutines/benchmark/src/main/kotlin/main.kt @@ -0,0 +1,56 @@ +import dev.inmo.micro_utils.coroutines.SmartSemaphore +import dev.inmo.micro_utils.coroutines.withAcquire +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.launch +import java.io.File +import kotlin.math.tanh +import kotlin.system.measureNanoTime + +data class Record( + val permits: Int, + val i: Int, + val timeNanos: Long +) + +private const val permitsCount = 1000 +private val acquiresCount = 10000 + +suspend fun main(args: Array) { + val outputFilePath = args.firstOrNull() ?: "local.bench.txt" + + val records = ArrayList(permitsCount * acquiresCount) + val jobs = ArrayList() + val scope = CoroutineScope(Dispatchers.Default) + for (permits in 1 .. permitsCount) { + val semaphore = SmartSemaphore.Mutable(permits) + for (i in 0 until acquiresCount) { + jobs.add( + scope.launch(start = CoroutineStart.LAZY) { + val timeNanos = measureNanoTime { + semaphore.withAcquire { + tanh(57.179) + } + } + records.add(Record(permits, i, timeNanos)) + } + ) + } + } + jobs.joinAll() + val file = File(outputFilePath) + file.writeText("") + val text = buildString { + records.groupBy { it.permits }.forEach { + it.value.sortedBy { it.i }.forEach { + append("${it.permits} ${it.i} ${it.timeNanos}\n") + } + } + } + file.appendText(text) + file.appendText("\n") + println(text) +} 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 d4a254abf4e..30a4e9e1153 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 @@ -61,6 +61,7 @@ sealed interface SmartSemaphore { override val permitsStateFlow: StateFlow = _freePermitsStateFlow.asStateFlow() private val internalChangesMutex = Mutex(false) + private val acquirePermitsMutex = Mutex(false) fun immutable() = Immutable(permitsStateFlow, maxPermits) @@ -70,28 +71,24 @@ sealed interface SmartSemaphore { * Holds call until this [SmartSemaphore] will be re-locked. That means that current method will */ suspend fun acquire(permits: Int = 1) { - var acquiredPermits = 0 val checkedPermits = checkedPermits(permits) - try { - do { - val shouldContinue = internalChangesMutex.withLock { - val requiredPermits = checkedPermits - acquiredPermits - val acquiring = minOf(freePermits, requiredPermits) - if (acquiring <= 0) return@withLock true - acquiredPermits += acquiring - _freePermitsStateFlow.value -= acquiring - - acquiredPermits != checkedPermits + var changed = false + acquirePermitsMutex.withLock { + _freePermitsStateFlow.first { + it >= checkedPermits + } + internalChangesMutex.withLock { + try { + _freePermitsStateFlow.value -= checkedPermits + changed = true + suspendPoint() + } catch (e: Throwable) { + if (changed) { + _freePermitsStateFlow.value += checkedPermits + } + throw e } - if (shouldContinue) { - waitRelease(checkedPermits - acquiredPermits) - } - } while (shouldContinue && currentCoroutineContext().isActive) - } catch (e: Throwable) { - if (acquiredPermits > 0) { - release(acquiredPermits) } - throw e } } @@ -99,19 +96,9 @@ sealed interface SmartSemaphore { * Holds call until this [SmartSemaphore] will be re-locked. That means that while [freePermits] == true, [holds] will * wait for [freePermits] == false and then try to lock */ + @Deprecated("Use acquire instead", ReplaceWith("acquire(permits)")) suspend fun acquireByOne(permits: Int = 1) { - val checkedPermits = checkedPermits(permits) - do { - waitRelease(checkedPermits) - val shouldContinue = internalChangesMutex.withLock { - if (_freePermitsStateFlow.value < checkedPermits) { - true - } else { - _freePermitsStateFlow.value -= checkedPermits - false - } - } - } while (shouldContinue && currentCoroutineContext().isActive) + return acquire(permits) } /** @@ -120,18 +107,29 @@ sealed interface SmartSemaphore { * @return True if lock was successful. False otherwise */ suspend fun tryAcquire(permits: Int = 1): Boolean { + if (acquirePermitsMutex.tryLock() == false) return false + val checkedPermits = checkedPermits(permits) - return if (_freePermitsStateFlow.value >= checkedPermits) { - internalChangesMutex.withLock { + var changed = false + + return internalChangesMutex.withLock { + try { if (_freePermitsStateFlow.value >= checkedPermits) { _freePermitsStateFlow.value -= checkedPermits + changed = true + suspendPoint() true } else { false } + } catch (e: Throwable) { + if (changed) { + _freePermitsStateFlow.value += checkedPermits + } + throw e + } finally { + acquirePermitsMutex.unlock() } - } else { - false } } diff --git a/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt b/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt index 7aa900cbe84..aeaad5c03c7 100644 --- a/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt +++ b/coroutines/src/commonTest/kotlin/SmartSemaphoreTests.kt @@ -1,6 +1,7 @@ import dev.inmo.micro_utils.coroutines.SmartSemaphore import dev.inmo.micro_utils.coroutines.withAcquire import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.cancelAndJoin import kotlinx.coroutines.flow.first @@ -38,35 +39,6 @@ class SmartSemaphoreTests { 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) diff --git a/settings.gradle b/settings.gradle index 908cffc4eae..29b63586209 100644 --- a/settings.gradle +++ b/settings.gradle @@ -35,6 +35,7 @@ String[] includes = [ ":ktor:client", ":coroutines", ":coroutines:compose", + ":coroutines:benchmark", ":android:recyclerview", ":android:pickers", ":android:smalltextfield", From 651ee37c19b44b88dfcb915dbb229a69aa7476e2 Mon Sep 17 00:00:00 2001 From: InsanusMokrassar Date: Mon, 19 Oct 2026 13:05:19 +0600 Subject: [PATCH 9/9] remove shitty benchmark of coroutines --- coroutines/benchmark/build.gradle | 16 ------ coroutines/benchmark/src/main/kotlin/main.kt | 56 -------------------- settings.gradle | 1 - 3 files changed, 73 deletions(-) delete mode 100644 coroutines/benchmark/build.gradle delete mode 100644 coroutines/benchmark/src/main/kotlin/main.kt diff --git a/coroutines/benchmark/build.gradle b/coroutines/benchmark/build.gradle deleted file mode 100644 index ff5f7a313a7..00000000000 --- a/coroutines/benchmark/build.gradle +++ /dev/null @@ -1,16 +0,0 @@ -plugins { - id "org.jetbrains.kotlin.jvm" -} - -repositories { - mavenCentral() -} - -dependencies { - api project(":micro_utils.coroutines") -} - -java { - sourceCompatibility = JavaVersion.VERSION_17 - targetCompatibility = JavaVersion.VERSION_17 -} diff --git a/coroutines/benchmark/src/main/kotlin/main.kt b/coroutines/benchmark/src/main/kotlin/main.kt deleted file mode 100644 index eff42d9e874..00000000000 --- a/coroutines/benchmark/src/main/kotlin/main.kt +++ /dev/null @@ -1,56 +0,0 @@ -import dev.inmo.micro_utils.coroutines.SmartSemaphore -import dev.inmo.micro_utils.coroutines.withAcquire -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.CoroutineStart -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.Job -import kotlinx.coroutines.joinAll -import kotlinx.coroutines.launch -import java.io.File -import kotlin.math.tanh -import kotlin.system.measureNanoTime - -data class Record( - val permits: Int, - val i: Int, - val timeNanos: Long -) - -private const val permitsCount = 1000 -private val acquiresCount = 10000 - -suspend fun main(args: Array) { - val outputFilePath = args.firstOrNull() ?: "local.bench.txt" - - val records = ArrayList(permitsCount * acquiresCount) - val jobs = ArrayList() - val scope = CoroutineScope(Dispatchers.Default) - for (permits in 1 .. permitsCount) { - val semaphore = SmartSemaphore.Mutable(permits) - for (i in 0 until acquiresCount) { - jobs.add( - scope.launch(start = CoroutineStart.LAZY) { - val timeNanos = measureNanoTime { - semaphore.withAcquire { - tanh(57.179) - } - } - records.add(Record(permits, i, timeNanos)) - } - ) - } - } - jobs.joinAll() - val file = File(outputFilePath) - file.writeText("") - val text = buildString { - records.groupBy { it.permits }.forEach { - it.value.sortedBy { it.i }.forEach { - append("${it.permits} ${it.i} ${it.timeNanos}\n") - } - } - } - file.appendText(text) - file.appendText("\n") - println(text) -} diff --git a/settings.gradle b/settings.gradle index 29b63586209..908cffc4eae 100644 --- a/settings.gradle +++ b/settings.gradle @@ -35,7 +35,6 @@ String[] includes = [ ":ktor:client", ":coroutines", ":coroutines:compose", - ":coroutines:benchmark", ":android:recyclerview", ":android:pickers", ":android:smalltextfield",