diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f1be5240b0..fc3bad4281f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## 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..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 @@ -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 @@ -49,64 +50,31 @@ sealed interface SmartMutex { override val lockStateFlow: StateFlow = _lockStateFlow.asStateFlow() private val internalChangesMutex = Mutex() + private val backingMutex = Mutex(locked = locked) 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 = withContext(NonCancellable) { - if (_lockStateFlow.value) { - internalChangesMutex.withLock { - if (_lockStateFlow.value) { - _lockStateFlow.value = false - true - } else { - false - } - } - } else { - false + suspend fun unlock(): Boolean { + val tryUnlock = internalChangesMutex.withLock { + _lockStateFlow.value.also { if (it) _lockStateFlow.value = false } } + if (tryUnlock) backingMutex.unlock() + return tryUnlock } } } @@ -134,4 +102,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..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 @@ -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 @@ -60,36 +61,34 @@ sealed interface SmartSemaphore { override val permitsStateFlow: StateFlow = _freePermitsStateFlow.asStateFlow() private val internalChangesMutex = Mutex(false) + private val acquirePermitsMutex = Mutex(false) 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 */ 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).takeIf { it > 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 } } @@ -97,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) } /** @@ -118,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 } } @@ -137,20 +137,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 +189,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 } +} diff --git a/coroutines/src/commonTest/kotlin/SmartMutexTests.kt b/coroutines/src/commonTest/kotlin/SmartMutexTests.kt index b75a26b9c2d..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() @@ -32,7 +43,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()) } 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() 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/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/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) + } +} 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