core/kotlinx-coroutines-core/test/JoinStressTest.kt

package kotlinx.coroutines.experimental

import org.junit.*
import org.junit.Test
import java.io.*
import java.util.concurrent.*
import kotlin.test.*

class JoinStressTest : TestBase() {

    private val iterations = 50_000 * stressTestMultiplier
    private val pool = newFixedThreadPoolContext(3, "JoinStressTest")

    @After
    fun tearDown() {
        pool.close()
    }

    class TestException : Exception() {
        override fun fillInStackTrace(): Throwable = this
    }

    @Test
    fun testExceptionalJoinWithCancellation() = runBlocking {
        val results = IntArray(2)

        repeat(iterations) {
            val barrier = CyclicBarrier(3)
            val exceptionalJob = async(pool) {
                barrier.await()
                throw TestException()
            }


            val awaiterJob = async(pool) {
                barrier.await()
                try {
                    exceptionalJob.await()
                } catch (e: TestException) {
                    0
                } catch (e: CancellationException) {
                    1
                }
            }

            barrier.await()
            exceptionalJob.cancel()
            ++results[awaiterJob.await()]
            require(!exceptionalJob.cancel())
        }

        // Check that concurrent cancellation of job which throws TestException without suspends doesn't suppress TestException
        assertEquals(iterations, results[0], results.toList().toString())
        assertEquals(0, results[1], results.toList().toString())
    }

    @Test
    fun testExceptionalJoinWithMultipleCancellations() = runBlocking {
        val results = IntArray(2)
        var successfulCancellations = 0

        repeat(iterations) {
            val barrier = CyclicBarrier(4)
            val exceptionalJob = async(pool) {
                barrier.await()
                throw TestException()
            }

            val awaiterJob = async(pool) {
                barrier.await()
                try {
                    exceptionalJob.await()
                } catch (e: TestException) {
                    0
                } catch (e: CancellationException) {
                    1
                }
            }

            val canceller = async(pool) {
                barrier.await()
                exceptionalJob.cancel(IOException())
            }

            barrier.await()
            val result = exceptionalJob.cancel()
            ++results[awaiterJob.await()]
            val cancellerResult = canceller.await()

            // None or one cancel can succeed
            require(!(result && cancellerResult))
            require(!exceptionalJob.cancel())

            if (result || cancellerResult) {
                ++successfulCancellations
            }
        }

        assertEquals(iterations, results[0], results.toList().toString())
        assertEquals(0, results[1], results.toList().toString())
        require(successfulCancellations > 0) { "Cancellation never succeeds, something wrong with stress test infra" }
    }
}