[K/N] Migrate runtime/atomics tests to new testing infra ^KT-61259
This commit is contained in:
Alexander Shabalin
committed by
Space Team
Space Team
parent
ba374bb45c
commit
2f22b0a6b0
@@ -0,0 +1,238 @@
|
||||
/*
|
||||
* Copyright 2010-2023 JetBrains s.r.o. and Kotlin Programming Language contributors.
|
||||
* Use of this source code is governed by the Apache 2.0 license that can be found in the license/LICENSE.txt file.
|
||||
*/
|
||||
|
||||
package test.concurrent
|
||||
|
||||
import kotlin.concurrent.*
|
||||
import kotlin.native.concurrent.Future
|
||||
import kotlin.native.concurrent.TransferMode
|
||||
import kotlin.native.concurrent.Worker
|
||||
import kotlin.test.*
|
||||
|
||||
object ThreadPool {
|
||||
private var _workers: Array<Worker>? = null
|
||||
|
||||
public fun init(threadCount: Int) {
|
||||
_workers = Array(threadCount) { Worker.start() }
|
||||
}
|
||||
|
||||
public fun deinit() {
|
||||
_workers?.forEach {
|
||||
it.requestTermination().result
|
||||
}
|
||||
_workers = null
|
||||
}
|
||||
|
||||
public fun <T> execute(f: (Int) -> T): List<Future<T>> = _workers?.mapIndexed { index, worker ->
|
||||
worker.execute(TransferMode.SAFE, { Pair(f, index) }) { (f, index) ->
|
||||
f(index)
|
||||
}
|
||||
} ?: error("Call ThreadPool.init() first")
|
||||
}
|
||||
|
||||
class AtomicIntStressTest {
|
||||
@BeforeTest fun init() = ThreadPool.init(20)
|
||||
@AfterTest fun deinit() = ThreadPool.deinit()
|
||||
|
||||
@Test fun addAndGet() {
|
||||
val atomic = AtomicInt(10)
|
||||
val futures = ThreadPool.execute {
|
||||
atomic.addAndGet(1000)
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
assertEquals(10 + 1000 * futures.size, atomic.value)
|
||||
}
|
||||
|
||||
@Test fun incrementAndGet() {
|
||||
val initial = 15
|
||||
val atomic = AtomicInt(initial)
|
||||
val futures = ThreadPool.execute {
|
||||
atomic.incrementAndGet()
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
assertEquals(initial + futures.size, atomic.value)
|
||||
}
|
||||
|
||||
@Test fun mutex() {
|
||||
val place = AtomicInt(1)
|
||||
val counter = AtomicInt(0)
|
||||
val futures = ThreadPool.execute { index ->
|
||||
// Here we simulate mutex using [place] location to store tag of the current worker.
|
||||
// When it is negative - worker executes exclusively.
|
||||
val tag = index + 1
|
||||
while (place.compareAndExchange(tag, -tag) != tag) {}
|
||||
assertEquals(index + 1, counter.incrementAndGet())
|
||||
// Now, let the next worker run.
|
||||
val previousPlace = place.compareAndExchange(-tag, tag + 1)
|
||||
assertEquals(-tag, previousPlace)
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
assertEquals(futures.size, counter.value)
|
||||
}
|
||||
}
|
||||
|
||||
class AtomicLongStressTest {
|
||||
@BeforeTest fun init() = ThreadPool.init(20)
|
||||
@AfterTest fun deinit() = ThreadPool.deinit()
|
||||
|
||||
@Test fun addAndGet() {
|
||||
val atomic = AtomicLong(10L)
|
||||
val futures = ThreadPool.execute {
|
||||
atomic.addAndGet(9999999999)
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
assertEquals(10L + 9999999999 * futures.size, atomic.value)
|
||||
}
|
||||
}
|
||||
|
||||
private class LockFreeStack<T> {
|
||||
private val top = AtomicReference<Node<T>?>(null)
|
||||
|
||||
private class Node<T>(val value: T, val next: Node<T>?)
|
||||
|
||||
fun isEmpty(): Boolean = top.value == null
|
||||
|
||||
fun push(value: T) {
|
||||
while(true) {
|
||||
val cur = top.value
|
||||
val upd = Node(value, cur)
|
||||
if (top.compareAndSet(cur, upd)) return
|
||||
}
|
||||
}
|
||||
|
||||
fun pop(): T? {
|
||||
while(true) {
|
||||
val cur = top.value
|
||||
if (cur == null) return null
|
||||
if (top.compareAndSet(cur, cur.next)) return cur.value
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class AtomicStressTest {
|
||||
@BeforeTest fun init() = ThreadPool.init(20)
|
||||
@AfterTest fun deinit() = ThreadPool.deinit()
|
||||
|
||||
@Test fun mutex() {
|
||||
val common = AtomicReference<Data?>(null)
|
||||
val futures = ThreadPool.execute { index ->
|
||||
// Try to publish our own data, until successful, in a tight loop.
|
||||
while (!common.compareAndSet(null, Data(index))) {}
|
||||
}
|
||||
val seen = mutableListOf<Int>()
|
||||
futures.forEach {
|
||||
while(true) {
|
||||
val current = common.value ?: continue
|
||||
// Each worker publishes exactly once
|
||||
assertFalse(seen.contains(current.value))
|
||||
seen.add(current.value)
|
||||
// Let others publish.
|
||||
common.compareAndExchange(current, null)
|
||||
break
|
||||
}
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
assertContentEquals(futures.indices, seen.sorted())
|
||||
}
|
||||
|
||||
@Test fun stackConcurrentPush() {
|
||||
val stack = LockFreeStack<Int>()
|
||||
val writers = ThreadPool.execute {
|
||||
stack.push(it)
|
||||
}
|
||||
writers.forEach { it.result }
|
||||
|
||||
val seen = mutableSetOf<Int>()
|
||||
while(!stack.isEmpty()) {
|
||||
val value = stack.pop()
|
||||
assertNotNull(value)
|
||||
seen.add(value)
|
||||
}
|
||||
assertEquals(writers.size, seen.size)
|
||||
}
|
||||
}
|
||||
|
||||
class AtomicIntArrayStressTest {
|
||||
@BeforeTest fun init() = ThreadPool.init(20)
|
||||
@AfterTest fun deinit() = ThreadPool.deinit()
|
||||
|
||||
@Test fun incrementAndGet() {
|
||||
val intArr = AtomicIntArray(10)
|
||||
val futures = ThreadPool.execute {
|
||||
for (i in 0 until 500) {
|
||||
val index = (0 until intArr.length).random()
|
||||
intArr.incrementAndGet(index)
|
||||
}
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
var sum = 0
|
||||
for (i in 0 until intArr.length) {
|
||||
sum += intArr[i]
|
||||
}
|
||||
assertEquals(futures.size * 500, sum)
|
||||
}
|
||||
}
|
||||
|
||||
class AtomicLongArrayStressTest {
|
||||
@BeforeTest fun init() = ThreadPool.init(20)
|
||||
@AfterTest fun deinit() = ThreadPool.deinit()
|
||||
|
||||
@Test fun incrementAndGet() {
|
||||
val longArr = AtomicLongArray(10)
|
||||
val futures = ThreadPool.execute {
|
||||
for (i in 0 until 500) {
|
||||
val index = (0 until longArr.length).random()
|
||||
longArr.incrementAndGet(index)
|
||||
}
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
var sum = 0L
|
||||
for (i in 0 until longArr.length) {
|
||||
sum += longArr[i]
|
||||
}
|
||||
assertEquals(futures.size.toLong() * 500, sum)
|
||||
}
|
||||
}
|
||||
|
||||
class AtomicArrayStressTest {
|
||||
@BeforeTest fun init() = ThreadPool.init(20)
|
||||
@AfterTest fun deinit() = ThreadPool.deinit()
|
||||
|
||||
@Test fun compareAndSet() {
|
||||
val refArr = AtomicArray(10) { Data(0) }
|
||||
val futures = ThreadPool.execute {
|
||||
for (i in 0 until 500) {
|
||||
val index = (0 until refArr.length).random()
|
||||
while(true) {
|
||||
val cur = refArr[index]
|
||||
val newValue = Data(cur.value + 1)
|
||||
if (refArr.compareAndSet(index, cur, newValue)) break
|
||||
}
|
||||
}
|
||||
}
|
||||
futures.forEach {
|
||||
it.result
|
||||
}
|
||||
var sum = 0
|
||||
for (i in 0 until refArr.length) {
|
||||
sum += refArr[i].value
|
||||
}
|
||||
assertEquals(futures.size * 500, sum)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user