Roman Elizarov | f16fd27 | 2017-02-07 11:26:00 +0300 | [diff] [blame] | 1 | /* |
Roman Elizarov | 1f74a2d | 2018-06-29 19:19:45 +0300 | [diff] [blame^] | 2 | * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. |
Roman Elizarov | f16fd27 | 2017-02-07 11:26:00 +0300 | [diff] [blame] | 3 | */ |
| 4 | |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 5 | package kotlinx.coroutines.experimental.channels |
| 6 | |
| 7 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 8 | import kotlinx.coroutines.experimental.selects.* |
| 9 | import org.junit.* |
| 10 | import org.junit.Assert.* |
| 11 | import org.junit.runner.* |
| 12 | import org.junit.runners.* |
| 13 | import java.util.concurrent.atomic.* |
| 14 | import kotlin.coroutines.experimental.* |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 15 | |
| 16 | @RunWith(Parameterized::class) |
| 17 | class ChannelSendReceiveStressTest( |
| 18 | val kind: TestChannelKind, |
| 19 | val nSenders: Int, |
| 20 | val nReceivers: Int |
Roman Elizarov | ebe18b4 | 2017-02-28 17:50:55 +0300 | [diff] [blame] | 21 | ) : TestBase() { |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 22 | companion object { |
| 23 | @Parameterized.Parameters(name = "{0}, nSenders={1}, nReceivers={2}") |
| 24 | @JvmStatic |
| 25 | fun params(): Collection<Array<Any>> = |
| 26 | listOf(1, 2, 10).flatMap { nSenders -> |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 27 | listOf(1, 10).flatMap { nReceivers -> |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 28 | TestChannelKind.values().map { arrayOf<Any>(it, nSenders, nReceivers) } |
| 29 | } |
| 30 | } |
| 31 | } |
| 32 | |
Roman Elizarov | ebe18b4 | 2017-02-28 17:50:55 +0300 | [diff] [blame] | 33 | val timeLimit = 30_000L * stressTestMultiplier // 30 sec |
Roman Elizarov | 7f1380a | 2017-10-21 17:27:27 +0300 | [diff] [blame] | 34 | val nEvents = 200_000 * stressTestMultiplier |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 35 | |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 36 | val maxBuffer = 10_000 // artificial limit for LinkedListChannel |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 37 | |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 38 | val channel = kind.create() |
| 39 | val sendersCompleted = AtomicInteger() |
| 40 | val receiversCompleted = AtomicInteger() |
| 41 | val dupes = AtomicInteger() |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 42 | val sentTotal = AtomicInteger() |
Roman Elizarov | cbd8e40 | 2017-02-28 23:55:41 +0300 | [diff] [blame] | 43 | val received = AtomicIntegerArray(nEvents) |
| 44 | val receivedTotal = AtomicInteger() |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 45 | val receivedBy = IntArray(nReceivers) |
| 46 | |
| 47 | @Test |
| 48 | fun testSendReceiveStress() = runBlocking { |
Roman Elizarov | d3d335b | 2017-10-21 17:43:53 +0300 | [diff] [blame] | 49 | println("--- ChannelSendReceiveStressTest $kind with nSenders=$nSenders, nReceivers=$nReceivers") |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 50 | val receivers = List(nReceivers) { receiverIndex -> |
| 51 | // different event receivers use different code |
| 52 | launch(CommonPool + CoroutineName("receiver$receiverIndex")) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 53 | when (receiverIndex % 5) { |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 54 | 0 -> doReceive(receiverIndex) |
| 55 | 1 -> doReceiveOrNull(receiverIndex) |
| 56 | 2 -> doIterator(receiverIndex) |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 57 | 3 -> doReceiveSelect(receiverIndex) |
| 58 | 4 -> doReceiveSelectOrNull(receiverIndex) |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 59 | } |
| 60 | receiversCompleted.incrementAndGet() |
| 61 | } |
| 62 | } |
| 63 | val senders = List(nSenders) { senderIndex -> |
| 64 | launch(CommonPool + CoroutineName("sender$senderIndex")) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 65 | when (senderIndex % 2) { |
| 66 | 0 -> doSend(senderIndex) |
| 67 | 1 -> doSendSelect(senderIndex) |
| 68 | } |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 69 | sendersCompleted.incrementAndGet() |
| 70 | } |
| 71 | } |
Roman Elizarov | 3aed4ee | 2017-03-06 12:21:05 +0300 | [diff] [blame] | 72 | // print progress |
Roman Elizarov | 43e3af7 | 2017-07-21 16:01:31 +0300 | [diff] [blame] | 73 | val progressJob = launch(coroutineContext) { |
Roman Elizarov | 3aed4ee | 2017-03-06 12:21:05 +0300 | [diff] [blame] | 74 | var seconds = 0 |
| 75 | while (true) { |
| 76 | delay(1000) |
| 77 | println("${++seconds}: Sent ${sentTotal.get()}, received ${receivedTotal.get()}") |
| 78 | } |
| 79 | } |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 80 | try { |
| 81 | withTimeout(timeLimit) { |
| 82 | senders.forEach { it.join() } |
| 83 | channel.close() |
| 84 | receivers.forEach { it.join() } |
| 85 | } |
| 86 | } catch (e: CancellationException) { |
| 87 | println("!!! Test timed out $e") |
| 88 | } |
Roman Elizarov | 3aed4ee | 2017-03-06 12:21:05 +0300 | [diff] [blame] | 89 | progressJob.cancel() |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 90 | println("Tested $kind with nSenders=$nSenders, nReceivers=$nReceivers") |
| 91 | println("Completed successfully ${sendersCompleted.get()} sender coroutines") |
| 92 | println("Completed successfully ${receiversCompleted.get()} receiver coroutines") |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 93 | println(" Sent ${sentTotal.get()} events") |
Roman Elizarov | cbd8e40 | 2017-02-28 23:55:41 +0300 | [diff] [blame] | 94 | println(" Received ${receivedTotal.get()} events") |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 95 | println(" Received dupes ${dupes.get()}") |
| 96 | repeat(nReceivers) { receiveIndex -> |
| 97 | println(" Received by #$receiveIndex ${receivedBy[receiveIndex]}") |
| 98 | } |
| 99 | assertEquals(nSenders, sendersCompleted.get()) |
| 100 | assertEquals(nReceivers, receiversCompleted.get()) |
| 101 | assertEquals(0, dupes.get()) |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 102 | assertEquals(nEvents, sentTotal.get()) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 103 | if (!kind.isConflated) assertEquals(nEvents, receivedTotal.get()) |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 104 | repeat(nReceivers) { receiveIndex -> |
Roman Elizarov | 12f961d | 2017-02-06 15:44:03 +0300 | [diff] [blame] | 105 | assertTrue("Each receiver should have received something", receivedBy[receiveIndex] > 0) |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 106 | } |
| 107 | } |
| 108 | |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 109 | private suspend fun doSent() { |
| 110 | sentTotal.incrementAndGet() |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 111 | if (!kind.isConflated) { |
Roman Elizarov | 3aed4ee | 2017-03-06 12:21:05 +0300 | [diff] [blame] | 112 | while (sentTotal.get() > receivedTotal.get() + maxBuffer) |
| 113 | yield() // throttle fast senders to prevent OOM with LinkedListChannel |
| 114 | } |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 115 | } |
| 116 | |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 117 | private suspend fun doSend(senderIndex: Int) { |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 118 | for (i in senderIndex until nEvents step nSenders) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 119 | channel.send(i) |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 120 | doSent() |
| 121 | } |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 122 | } |
| 123 | |
| 124 | private suspend fun doSendSelect(senderIndex: Int) { |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 125 | for (i in senderIndex until nEvents step nSenders) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 126 | select<Unit> { channel.onSend(i) { Unit } } |
Roman Elizarov | bc296bb | 2017-03-01 00:37:37 +0300 | [diff] [blame] | 127 | doSent() |
| 128 | } |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 129 | } |
| 130 | |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 131 | private fun doReceived(receiverIndex: Int, event: Int) { |
Roman Elizarov | cbd8e40 | 2017-02-28 23:55:41 +0300 | [diff] [blame] | 132 | if (!received.compareAndSet(event, 0, 1)) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 133 | println("Duplicate event $event at $receiverIndex") |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 134 | dupes.incrementAndGet() |
| 135 | } |
Roman Elizarov | cbd8e40 | 2017-02-28 23:55:41 +0300 | [diff] [blame] | 136 | receivedTotal.incrementAndGet() |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 137 | receivedBy[receiverIndex]++ |
| 138 | } |
| 139 | |
| 140 | private suspend fun doReceive(receiverIndex: Int) { |
| 141 | while (true) { |
| 142 | try { doReceived(receiverIndex, channel.receive()) } |
| 143 | catch (ex: ClosedReceiveChannelException) { break } |
| 144 | } |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 145 | } |
| 146 | |
| 147 | private suspend fun doReceiveOrNull(receiverIndex: Int) { |
| 148 | while (true) { |
| 149 | doReceived(receiverIndex, channel.receiveOrNull() ?: break) |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | private suspend fun doIterator(receiverIndex: Int) { |
| 154 | for (event in channel) { |
| 155 | doReceived(receiverIndex, event) |
| 156 | } |
| 157 | } |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 158 | |
| 159 | private suspend fun doReceiveSelect(receiverIndex: Int) { |
| 160 | while (true) { |
| 161 | try { |
| 162 | val event = select<Int> { channel.onReceive { it } } |
| 163 | doReceived(receiverIndex, event) |
| 164 | } catch (ex: ClosedReceiveChannelException) { break } |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | private suspend fun doReceiveSelectOrNull(receiverIndex: Int) { |
| 169 | while (true) { |
| 170 | val event = select<Int?> { channel.onReceiveOrNull { it } } ?: break |
| 171 | doReceived(receiverIndex, event) |
| 172 | } |
| 173 | } |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 174 | } |