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) + } +}