BroadcastChannelMultiReceiveStressTest streamlined,

consistent stdout headers in all stress-tests

Author: elizarov

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelMultiReceiveStressTest.kt

@@ -40,7 +40,7 @@ class BroadcastChannelMultiReceiveStressTest(
     }
 
     private val nReceivers = if (isStressTest) 10 else 5
-    private val nSeconds = 5 * stressTestMultiplier
+    private val nSeconds = 3 * stressTestMultiplier
 
     private val broadcast = kind.create<Long>()
     private val pool = newFixedThreadPoolContext(nReceivers + 1, "BroadcastChannelMultiReceiveStressTest")
@@ -57,6 +57,7 @@ class BroadcastChannelMultiReceiveStressTest(
 
     @Test
     fun testStress() = runBlocking {
+        println("--- BroadcastChannelMultiReceiveStressTest $kind with nReceivers=$nReceivers")
         val ctx = pool + coroutineContext[Job]!!
         val sender =
             launch(context = ctx + CoroutineName("Sender")) {
@@ -67,31 +68,35 @@ class BroadcastChannelMultiReceiveStressTest(
                 }
             }
         val receivers = mutableListOf<Job>()
-        repeat(nSeconds) { sec ->
-            // launch new receiver up to max
-            if (receivers.size < nReceivers) {
-                val receiverIndex = receivers.size
-                val name = "Receiver$receiverIndex"
-                println("$sec: Launching $name")
-                receivers += launch(ctx + CoroutineName(name)) {
-                    broadcast.openSubscription().use { sub ->
-                        when (receiverIndex % 5) {
-                            0 -> doReceive(sub, receiverIndex)
-                            1 -> doReceiveOrNull(sub, receiverIndex)
-                            2 -> doIterator(sub, receiverIndex)
-                            3 -> doReceiveSelect(sub, receiverIndex)
-                            4 -> doReceiveSelectOrNull(sub, receiverIndex)
-                        }
+        fun printProgress() {
+            println("Sent ${sentTotal.get()}, received ${receivedTotal.get()}, receivers=${receivers.size}")
+        }
+        // ramp up receivers
+        repeat(nReceivers) {
+            delay(100) // wait 0.1 sec
+            val receiverIndex = receivers.size
+            val name = "Receiver$receiverIndex"
+            println("Launching $name")
+            receivers += launch(ctx + CoroutineName(name)) {
+                broadcast.openSubscription().use { sub ->
+                    when (receiverIndex % 5) {
+                        0 -> doReceive(sub, receiverIndex)
+                        1 -> doReceiveOrNull(sub, receiverIndex)
+                        2 -> doIterator(sub, receiverIndex)
+                        3 -> doReceiveSelect(sub, receiverIndex)
+                        4 -> doReceiveSelectOrNull(sub, receiverIndex)
                     }
                 }
             }
-            // wait a sec
-            delay(100)
-            // print progress
-            println("${sec + 1}: Sent ${sentTotal.get()}, received ${receivedTotal.get()}, receivers=${receivers.size}")
+            printProgress()
+        }
+        // wait
+        repeat(nSeconds) { sec ->
+            delay(1000)
+            printProgress()
         }
         sender.cancelAndJoin()
-        println("Tested with nReceivers=$nReceivers")
+        println("Tested $kind with nReceivers=$nReceivers")
         val total = sentTotal.get()
         println("      Sent $total events, waiting for receivers")
         stopOnReceive.set(total)

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/BroadcastChannelSubStressTest.kt

@@ -47,6 +47,7 @@ class BroadcastChannelSubStressTest(
 
     @Test
     fun testStress() = runBlocking {
+        println("--- BroadcastChannelSubStressTest $kind")
         val ctx = coroutineContext + CommonPool
         val sender =
             launch(context = ctx + CoroutineName("Sender")) {

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ChannelAtomicCancelStressTest.kt

@@ -72,6 +72,7 @@ class ChannelAtomicCancelStressTest(val kind: TestChannelKind) : TestBase() {
 
     @Test
     fun testAtomicCancelStress() = runBlocking<Unit> {
+        println("--- ChannelAtomicCancelStressTest $kind")
         val deadline = System.currentTimeMillis() + TEST_DURATION
         launchSender()
         launchReceiver()

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ChannelSendReceiveStressTest.kt

@@ -59,8 +59,7 @@ class ChannelSendReceiveStressTest(
 
     @Test
     fun testSendReceiveStress() = runBlocking {
-        println("-------------------------------------")
-        println("Testing $kind with nSenders=$nSenders, nReceivers=$nReceivers")
+        println("--- ChannelSendReceiveStressTest $kind with nSenders=$nSenders, nReceivers=$nReceivers")
         val receivers = List(nReceivers) { receiverIndex ->
             // different event receivers use different code
             launch(CommonPool + CoroutineName("receiver$receiverIndex")) {

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ConflatedBroadcastChannelNotifyStressTest.kt

@@ -37,6 +37,7 @@ class ConflatedBroadcastChannelNotifyStressTest : TestBase() {
 
     @Test
     fun testStressNotify()= runBlocking<Unit> {
+        println("--- ConflatedBroadcastChannelNotifyStressTest")
         val senders = List(nSenders) { senderId ->
             launch(CommonPool + CoroutineName("Sender$senderId")) {
                 repeat(nEvents) { i ->

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/channels/ConflatedChannelCloseStressTest.kt

@@ -41,6 +41,7 @@ class ConflatedChannelCloseStressTest : TestBase() {
 
     @Test
     fun testStressClose() = runBlocking<Unit> {
+        println("--- ConflatedChannelCloseStressTest with nSenders=$nSenders")
         val senderJobs = List(nSenders) { Job() }
         val senders = List(nSenders) { senderId ->
             launch(pool) {

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

@@ -43,6 +43,7 @@ class LockFreeLinkedListLongStressTest : TestBase() {
 
     @Test
     fun testStress() {
+        println("--- LockFreeLinkedListLongStressTest")
         for (j in 0 until nAddThreads)
             threads += thread(start = false, name = "adder-$j") {
                 for (i in j until nAdded step nAddThreads) {

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

@@ -46,6 +46,7 @@ class LockFreeLinkedListShortStressTest : TestBase() {
 
     @Test
     fun testStress() {
+        println("--- LockFreeLinkedListShortStressTest")
         val deadline = System.currentTimeMillis() + TEST_DURATION
         repeat(nAdderThreads) { threadId ->
             threads += thread(start = false, name = "adder-$threadId") {

core/kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectPhilosophersStressTest.kt

@@ -49,6 +49,7 @@ class SelectPhilosophersStressTest : TestBase() {
 
     @Test
     fun testPhilosophers() = runBlocking<Unit> {
+        println("--- SelectPhilosophersStressTest")
         val timeLimit = System.currentTimeMillis() + TEST_DURATION
         val philosophers = List<Deferred<Int>>(n) { id ->
             async(CommonPool) {