test(coroutines): cover concurrent semaphore releases

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<Boolean> = coroutineScope {
+        val contenders = mutableListOf<Deferred<Boolean>>()
+        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()
+    }
+}

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