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 - } - if (shouldContinue) { - waitRelease(checkedPermits - acquiredPermits) - } - } while (shouldContinue && currentCoroutineContext().isActive) - } catch (e: Throwable) { - if (acquiredPermits > 0) { - release(acquiredPermits) + 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 + } } - 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",