mirror of
https://github.com/InsanusMokrassar/MicroUtils.git
synced 2026-11-02 10:03:01 +00:00
one more rewrite of semaphore
This commit is contained in:
1 parent
67f4e1b9ff
commit
1f4bb04ec9
5 files changed
+107
-64
No files matched your search
+33
-35
@@ -61,6 +61,7 @@ sealed interface SmartSemaphore {
|
||||
override val permitsStateFlow: StateFlow<Int> = _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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user