kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/internal/LockFreeLinkedListAtomicStressLFTest.kt

/*
 * Copyright 2016-2017 JetBrains s.r.o.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package kotlinx.coroutines.experimental.internal

import kotlinx.atomicfu.LockFreedomTestEnvironment
import kotlinx.coroutines.experimental.TestBase
import org.junit.Assert.*
import org.junit.Test
import java.util.*
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference

/**
 * This stress test has 4 threads adding randomly to the list and them immediately undoing
 * this addition by remove, and 4 threads trying to remove nodes from two lists simultaneously (atomically).
 */
class LockFreeLinkedListAtomicStressLFTest : TestBase() {
    private val env = LockFreedomTestEnvironment("LockFreeLinkedListAtomicStressLFTest")

    data class IntNode(val i: Int) : LockFreeLinkedListNode()

    private val TEST_DURATION_SEC = 5 * stressTestMultiplier

    val nLists = 4
    val nAdderThreads = 4
    val nRemoverThreads = 4

    val lists = Array(nLists) { LockFreeLinkedListHead() }

    val undone = AtomicLong()
    val missed = AtomicLong()
    val removed = AtomicLong()
    val error = AtomicReference<Throwable>()

    @Test
    fun testStress() {
        repeat(nAdderThreads) { threadId ->
            val rnd = Random()
            env.testThread(name = "adder-$threadId") {
                when (rnd.nextInt(4)) {
                    0 -> {
                        val list = lists[rnd.nextInt(nLists)]
                        val node = IntNode(threadId)
                        list.addLast(node)
                        burnTime(rnd)
                        tryRemove(node)
                    }
                    1 -> {
                        // just to test conditional add
                        val list = lists[rnd.nextInt(nLists)]
                        val node = IntNode(threadId)
                        assertTrue(list.addLastIf(node, { true }))
                        burnTime(rnd)
                        tryRemove(node)
                    }
                    2 -> {
                        // just to test failed conditional add and burn some time
                        val list = lists[rnd.nextInt(nLists)]
                        val node = IntNode(threadId)
                        assertFalse(list.addLastIf(node, { false }))
                        burnTime(rnd)
                    }
                    3 -> {
                        // add two atomically
                        val idx1 = rnd.nextInt(nLists - 1)
                        val idx2 = idx1 + 1 + rnd.nextInt(nLists - idx1 - 1)
                        check(idx1 < idx2) // that is our global order
                        val list1 = lists[idx1]
                        val list2 = lists[idx2]
                        val node1 = IntNode(threadId)
                        val node2 = IntNode(-threadId - 1)
                        val add1 = list1.describeAddLast(node1)
                        val add2 = list2.describeAddLast(node2)
                        val op = object : AtomicOp<Any?>() {
                            override fun prepare(affected: Any?): Any? =
                                add1.prepare(this) ?:
                                add2.prepare(this)
                            override fun complete(affected: Any?, failure: Any?) {
                                add1.complete(this, failure)
                                add2.complete(this, failure)
                            }
                        }
                        assertTrue(op.perform(null) == null)
                        burnTime(rnd)
                        tryRemove(node1)
                        tryRemove(node2)
                    }
                    else -> error("Cannot happen")
                }
            }
        }
        repeat(nRemoverThreads) { threadId ->
            val rnd = Random()
            env.testThread(name = "remover-$threadId") {
                val idx1 = rnd.nextInt(nLists - 1)
                val idx2 = idx1 + 1 + rnd.nextInt(nLists - idx1 - 1)
                check(idx1 < idx2) // that is our global order
                val list1 = lists[idx1]
                val list2 = lists[idx2]
                val remove1 = list1.describeRemoveFirst()
                val remove2 = list2.describeRemoveFirst()
                val op = object : AtomicOp<Any?>() {
                    override fun prepare(affected: Any?): Any? =
                        remove1.prepare(this) ?:
                        remove2.prepare(this)
                    override fun complete(affected: Any?, failure: Any?) {
                        remove1.complete(this, failure)
                        remove2.complete(this, failure)
                    }
                }
                val success = op.perform(null) == null
                if (success) removed.addAndGet(2)
            }
        }
        env.performTest(TEST_DURATION_SEC) {
            val _undone = undone.get()
            val _missed = missed.get()
            val _removed = removed.get()
            println("  Adders undone $_undone node additions")
            println("  Adders missed $_missed nodes")
            println("Remover removed $_removed nodes")
        }
        error.get()?.let { throw it }
        assertEquals(missed.get(), removed.get())
        assertTrue(undone.get() > 0)
        assertTrue(missed.get() > 0)
        lists.forEach { it.validate() }
    }

    private val sink = IntArray(1024)

    private fun burnTime(rnd: Random) {
        if (rnd.nextInt(100) < 95) return // be quick, no wait 95% of time
        do {
            val x = rnd.nextInt(100)
            val i = rnd.nextInt(sink.size)
            repeat(x) { sink[i] += it }
        } while (x >= 90)
    }

    private fun tryRemove(node: IntNode) {
        if (node.remove())
            undone.incrementAndGet()
        else
            missed.incrementAndGet()
    }
}