Roman Elizarov | f16fd27 | 2017-02-07 11:26:00 +0300 | [diff] [blame] | 1 | /* |
| 2 | * Copyright 2016-2017 JetBrains s.r.o. |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 17 | package kotlinx.coroutines.experimental.channels |
| 18 | |
| 19 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 20 | import kotlinx.coroutines.experimental.selects.select |
| 21 | import org.junit.Assert.assertEquals |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 22 | import org.junit.Test |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 23 | import org.junit.runner.RunWith |
| 24 | import org.junit.runners.Parameterized |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 25 | import java.util.* |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 26 | import java.util.concurrent.atomic.AtomicReference |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 27 | |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 28 | /** |
| 29 | * Tests cancel atomicity for channel send & receive operations, including their select versions. |
| 30 | */ |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 31 | @RunWith(Parameterized::class) |
Roman Elizarov | ebe18b4 | 2017-02-28 17:50:55 +0300 | [diff] [blame] | 32 | class ChannelAtomicCancelStressTest(val kind: TestChannelKind) : TestBase() { |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 33 | companion object { |
| 34 | @Parameterized.Parameters(name = "{0}") |
| 35 | @JvmStatic |
| 36 | fun params(): Collection<Array<Any>> = TestChannelKind.values().map { arrayOf<Any>(it) } |
| 37 | } |
| 38 | |
Roman Elizarov | ebe18b4 | 2017-02-28 17:50:55 +0300 | [diff] [blame] | 39 | val TEST_DURATION = 3000L * stressTestMultiplier |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 40 | |
Roman Elizarov | 7b2d8b0 | 2017-02-02 20:09:14 +0300 | [diff] [blame] | 41 | val channel = kind.create() |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 42 | val senderDone = ArrayChannel<Boolean>(1) |
| 43 | val receiverDone = ArrayChannel<Boolean>(1) |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 44 | |
| 45 | var lastSent = 0 |
| 46 | var lastReceived = 0 |
| 47 | |
| 48 | var stoppedSender = 0 |
| 49 | var stoppedReceiver = 0 |
| 50 | |
| 51 | var missedCnt = 0 |
| 52 | var dupCnt = 0 |
| 53 | |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 54 | val failed = AtomicReference<Throwable>() |
| 55 | |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 56 | lateinit var sender: Job |
| 57 | lateinit var receiver: Job |
| 58 | |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 59 | fun fail(e: Throwable) = failed.compareAndSet(null, e) |
| 60 | |
| 61 | inline fun cancellable(done: ArrayChannel<Boolean>, block: () -> Unit) { |
| 62 | try { |
| 63 | block() |
| 64 | } catch (e: Throwable) { |
| 65 | if (e !is CancellationException) fail(e) |
| 66 | throw e |
| 67 | } finally { |
| 68 | if (!done.offer(true)) |
| 69 | fail(IllegalStateException("failed to offer to done channel")) |
| 70 | } |
| 71 | } |
| 72 | |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 73 | @Test |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 74 | fun testAtomicCancelStress() = runBlocking<Unit> { |
Roman Elizarov | d3d335b | 2017-10-21 17:43:53 +0300 | [diff] [blame] | 75 | println("--- ChannelAtomicCancelStressTest $kind") |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 76 | val deadline = System.currentTimeMillis() + TEST_DURATION |
| 77 | launchSender() |
| 78 | launchReceiver() |
| 79 | val rnd = Random() |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 80 | while (System.currentTimeMillis() < deadline && failed.get() == null) { |
Roman Elizarov | 6c63aea | 2017-02-02 12:20:48 +0300 | [diff] [blame] | 81 | when (rnd.nextInt(3)) { |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 82 | 0 -> { // cancel & restart sender |
| 83 | stopSender() |
| 84 | launchSender() |
| 85 | } |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 86 | 1 -> { // cancel & restart receiver |
| 87 | stopReceiver() |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 88 | launchReceiver() |
| 89 | } |
| 90 | 2 -> yield() // just yield (burn a little time) |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 91 | } |
| 92 | } |
| 93 | stopSender() |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 94 | stopReceiver() |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 95 | println(" Sent $lastSent ints to channel") |
| 96 | println(" Received $lastReceived ints from channel") |
| 97 | println(" Stopped sender $stoppedSender times") |
| 98 | println("Stopped receiver $stoppedReceiver times") |
| 99 | println(" Missed $missedCnt ints") |
| 100 | println(" Duplicated $dupCnt ints") |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 101 | failed.get()?.let { throw it } |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 102 | assertEquals(0, dupCnt) |
Roman Elizarov | e3aa8ff | 2017-04-27 19:16:40 +0300 | [diff] [blame] | 103 | if (!kind.isConflated) { |
Roman Elizarov | 3aed4ee | 2017-03-06 12:21:05 +0300 | [diff] [blame] | 104 | assertEquals(0, missedCnt) |
| 105 | assertEquals(lastSent, lastReceived) |
| 106 | } |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 107 | } |
| 108 | |
| 109 | fun launchSender() { |
Roman Elizarov | a74eb5f | 2017-05-11 20:15:18 +0300 | [diff] [blame] | 110 | sender = launch(CommonPool, start = CoroutineStart.ATOMIC) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 111 | val rnd = Random() |
| 112 | cancellable(senderDone) { |
Roman Elizarov | 2ad0e94 | 2017-02-28 19:14:08 +0300 | [diff] [blame] | 113 | var counter = 0 |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 114 | while (true) { |
| 115 | val trySend = lastSent + 1 |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 116 | when (rnd.nextInt(2)) { |
| 117 | 0 -> channel.send(trySend) |
| 118 | 1 -> select { channel.onSend(trySend) {} } |
| 119 | else -> error("cannot happen") |
| 120 | } |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 121 | lastSent = trySend // update on success |
Roman Elizarov | 2ad0e94 | 2017-02-28 19:14:08 +0300 | [diff] [blame] | 122 | if (counter++ % 1000 == 0) yield() // yield periodically to check cancellation on LinkedListChannel |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 123 | } |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 124 | } |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | suspend fun stopSender() { |
| 129 | stoppedSender++ |
| 130 | sender.cancel() |
| 131 | senderDone.receive() |
| 132 | } |
| 133 | |
| 134 | fun launchReceiver() { |
Roman Elizarov | a74eb5f | 2017-05-11 20:15:18 +0300 | [diff] [blame] | 135 | receiver = launch(CommonPool, start = CoroutineStart.ATOMIC) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 136 | val rnd = Random() |
| 137 | cancellable(receiverDone) { |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 138 | while (true) { |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 139 | val received = when (rnd.nextInt(2)) { |
| 140 | 0 -> channel.receive() |
| 141 | 1 -> select { channel.onReceive { it } } |
| 142 | else -> error("cannot happen") |
| 143 | } |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 144 | val expected = lastReceived + 1 |
| 145 | if (received > expected) |
| 146 | missedCnt++ |
| 147 | if (received < expected) |
| 148 | dupCnt++ |
| 149 | lastReceived = received |
| 150 | } |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 151 | } |
| 152 | } |
| 153 | } |
| 154 | |
Roman Elizarov | 1216e91 | 2017-02-22 09:57:06 +0300 | [diff] [blame] | 155 | suspend fun stopReceiver() { |
Roman Elizarov | 187eace | 2017-01-31 09:39:58 +0300 | [diff] [blame] | 156 | stoppedReceiver++ |
| 157 | receiver.cancel() |
| 158 | receiverDone.receive() |
| 159 | } |
| 160 | } |