Skip to content

Fix MutexCancellationStressTest flakiness #3724

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Apr 26, 2023
Merged

Fix MutexCancellationStressTest flakiness #3724

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ba4976a
Fix MutexCancellationStressTest flakiness
qwwdfsad Apr 25, 2023
715a04a
~restore diff/after-squash history
qwwdfsad Apr 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 60 additions & 49 deletions kotlinx-coroutines-core/jvm/test/MutexCancellationStressTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,69 +8,80 @@ import kotlinx.coroutines.internal.*
import kotlinx.coroutines.selects.*
import kotlinx.coroutines.sync.*
import org.junit.*
import org.junit.Test
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicInteger
import kotlin.test.*

class MutexCancellationStressTest : TestBase() {
@Test
fun testStressCancellationDoesNotBreakMutex() = runTest {
val mutex = Mutex()
val mutexJobNumber = 3
val mutexOwners = Array(mutexJobNumber) { "$it" }
val dispatcher = Executors.newFixedThreadPool(mutexJobNumber + 2).asCoroutineDispatcher()
var counter = 0
val counterLocal = Array(mutexJobNumber) { LocalAtomicInt(0) }
val completed = LocalAtomicInt(0)
val mutexJobLauncher: (jobNumber: Int) -> Job = { jobId ->
val coroutineName = "MutexJob-$jobId"
launch(dispatcher + CoroutineName(coroutineName)) {
while (completed.value == 0) {
mutex.holdsLock(mutexOwners[(jobId + 1) % mutexJobNumber])
if (mutex.tryLock(mutexOwners[jobId])) {
counterLocal[jobId].incrementAndGet()
counter++
mutex.unlock(mutexOwners[jobId])
}
mutex.withLock(mutexOwners[jobId]) {
counterLocal[jobId].incrementAndGet()
counter++
}
select<Unit> {
mutex.onLock(mutexOwners[jobId]) {
fun testStressCancellationDoesNotBreakMutex() {
runTest {
val mutex = Mutex()
val mutexJobNumber = 3
val mutexOwners = Array(mutexJobNumber) { "$it" }
val dispatcher = Executors.newFixedThreadPool(mutexJobNumber + 2).asCoroutineDispatcher()
var counter = 0
val counterLocal = Array(mutexJobNumber) { AtomicInteger(0) }
val completed = AtomicBoolean(false)
val mutexJobLauncher: (jobNumber: Int) -> Job = { jobId ->
val coroutineName = "MutexJob-$jobId"
// ATOMIC to always have a chance to proceed
launch(dispatcher + CoroutineName(coroutineName), CoroutineStart.ATOMIC) {
while (!completed.get()) {
// Stress out holdsLock
mutex.holdsLock(mutexOwners[(jobId + 1) % mutexJobNumber])
// Stress out lock-like primitives
if (mutex.tryLock(mutexOwners[jobId])) {
counterLocal[jobId].incrementAndGet()
counter++
mutex.unlock(mutexOwners[jobId])
}
mutex.withLock(mutexOwners[jobId]) {
counterLocal[jobId].incrementAndGet()
counter++
}
select<Unit> {
mutex.onLock(mutexOwners[jobId]) {
counterLocal[jobId].incrementAndGet()
counter++
mutex.unlock(mutexOwners[jobId])
}
}
}
}
}
}
val mutexJobs = (0 until mutexJobNumber).map { mutexJobLauncher(it) }.toMutableList()
val checkProgressJob = launch(dispatcher + CoroutineName("checkProgressJob")) {
var lastCounterLocalSnapshot = (0 until mutexJobNumber).map { 0 }
while (completed.value == 0) {
delay(1000)
val c = counterLocal.map { it.value }
for (i in 0 until mutexJobNumber) {
assert(c[i] > lastCounterLocalSnapshot[i]) { "No progress in MutexJob-$i" }
val mutexJobs = (0 until mutexJobNumber).map { mutexJobLauncher(it) }.toMutableList()
val checkProgressJob = launch(dispatcher + CoroutineName("checkProgressJob")) {
var lastCounterLocalSnapshot = (0 until mutexJobNumber).map { 0 }
while (!completed.get()) {
delay(500)
// If we've caught the completion after delay, then there is a chance no progress were made whatsoever, bail out
if (completed.get()) return@launch
val c = counterLocal.map { it.value }
for (i in 0 until mutexJobNumber) {
assert(c[i] > lastCounterLocalSnapshot[i]) { "No progress in MutexJob-$i, last observed state: ${c[i]}" }
}
lastCounterLocalSnapshot = c
}
lastCounterLocalSnapshot = c
}
}
val cancellationJob = launch(dispatcher + CoroutineName("cancellationJob")) {
var cancellingJobId = 0
while (completed.value == 0) {
val jobToCancel = mutexJobs.removeFirst()
jobToCancel.cancelAndJoin()
mutexJobs += mutexJobLauncher(cancellingJobId)
cancellingJobId = (cancellingJobId + 1) % mutexJobNumber
val cancellationJob = launch(dispatcher + CoroutineName("cancellationJob")) {
var cancellingJobId = 0
while (!completed.get()) {
val jobToCancel = mutexJobs.removeFirst()
jobToCancel.cancelAndJoin()
mutexJobs += mutexJobLauncher(cancellingJobId)
cancellingJobId = (cancellingJobId + 1) % mutexJobNumber
}
}
delay(2000L * stressTestMultiplier)
completed.set(true)
cancellationJob.join()
mutexJobs.forEach { it.join() }
checkProgressJob.join()
assertEquals(counter, counterLocal.sumOf { it.value })
dispatcher.close()
}
delay(2000L * stressTestMultiplier)
completed.value = 1
cancellationJob.join()
mutexJobs.forEach { it.join() }
checkProgressJob.join()
check(counter == counterLocal.sumOf { it.value })
dispatcher.close()
}
}