Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +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 | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 3 | */ |
| 4 | |
| 5 | package kotlinx.coroutines.experimental.channels |
| 6 | |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 7 | import kotlinx.coroutines.experimental.* |
Roman Elizarov | 656bcd9 | 2017-06-07 09:24:19 +0300 | [diff] [blame] | 8 | import org.junit.After |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 9 | import org.junit.Test |
| 10 | import java.util.concurrent.atomic.AtomicInteger |
| 11 | import java.util.concurrent.atomic.AtomicReference |
Roman Elizarov | 9fe5f46 | 2018-02-21 19:05:52 +0300 | [diff] [blame] | 12 | import kotlin.coroutines.experimental.* |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 13 | |
| 14 | class ConflatedChannelCloseStressTest : TestBase() { |
| 15 | |
| 16 | val nSenders = 2 |
| 17 | val testSeconds = 3 * stressTestMultiplier |
| 18 | |
| 19 | val curChannel = AtomicReference<ConflatedChannel<Int>>(ConflatedChannel()) |
| 20 | val sent = AtomicInteger() |
| 21 | val closed = AtomicInteger() |
| 22 | val received = AtomicInteger() |
| 23 | |
Roman Elizarov | 656bcd9 | 2017-06-07 09:24:19 +0300 | [diff] [blame] | 24 | val pool = newFixedThreadPoolContext(nSenders + 2, "TestStressClose") |
| 25 | |
| 26 | @After |
Roman Elizarov | d9ae2bc | 2017-10-20 17:36:56 +0800 | [diff] [blame] | 27 | fun tearDown() { |
| 28 | pool.close() |
| 29 | } |
Roman Elizarov | 656bcd9 | 2017-06-07 09:24:19 +0300 | [diff] [blame] | 30 | |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 31 | @Test |
| 32 | fun testStressClose() = runBlocking<Unit> { |
Roman Elizarov | d3d335b | 2017-10-21 17:43:53 +0300 | [diff] [blame] | 33 | println("--- ConflatedChannelCloseStressTest with nSenders=$nSenders") |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 34 | val senderJobs = List(nSenders) { Job() } |
| 35 | val senders = List(nSenders) { senderId -> |
Roman Elizarov | 656bcd9 | 2017-06-07 09:24:19 +0300 | [diff] [blame] | 36 | launch(pool) { |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 37 | var x = senderId |
| 38 | try { |
| 39 | while (isActive) { |
| 40 | try { |
| 41 | curChannel.get().offer(x) |
| 42 | x += nSenders |
| 43 | sent.incrementAndGet() |
| 44 | } catch (e: ClosedSendChannelException) { |
| 45 | // ignore |
| 46 | } |
| 47 | } |
| 48 | } finally { |
| 49 | senderJobs[senderId].cancel() |
| 50 | } |
| 51 | } |
| 52 | } |
| 53 | val closerJob = Job() |
Roman Elizarov | 656bcd9 | 2017-06-07 09:24:19 +0300 | [diff] [blame] | 54 | val closer = launch(pool) { |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 55 | try { |
| 56 | while (isActive) { |
| 57 | flipChannel() |
| 58 | closed.incrementAndGet() |
| 59 | yield() |
| 60 | } |
| 61 | } finally { |
| 62 | closerJob.cancel() |
| 63 | } |
| 64 | } |
Roman Elizarov | 656bcd9 | 2017-06-07 09:24:19 +0300 | [diff] [blame] | 65 | val receiver = async(pool) { |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 66 | while (isActive) { |
| 67 | curChannel.get().receiveOrNull() |
| 68 | received.incrementAndGet() |
| 69 | } |
| 70 | } |
| 71 | // print stats while running |
| 72 | repeat(testSeconds) { |
| 73 | delay(1000) |
| 74 | printStats() |
| 75 | } |
| 76 | println("Stopping") |
| 77 | senders.forEach { it.cancel() } |
| 78 | closer.cancel() |
| 79 | // wait them to complete |
| 80 | println("waiting for senders...") |
| 81 | senderJobs.forEach { it.join() } |
| 82 | println("waiting for closer...") |
| 83 | closerJob.join() |
| 84 | // close cur channel |
| 85 | println("Closing channel and signalling receiver...") |
| 86 | flipChannel() |
| 87 | curChannel.get().close(StopException()) |
| 88 | /// wait for receiver do complete |
| 89 | println("Waiting for receiver...") |
| 90 | try { |
| 91 | receiver.await() |
Roman Elizarov | 448528e | 2017-07-21 18:11:20 +0300 | [diff] [blame] | 92 | error("Receiver should not complete normally") |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 93 | } catch (e: StopException) { |
| 94 | // ok |
| 95 | } |
| 96 | // print stats |
| 97 | println("--- done") |
| 98 | printStats() |
| 99 | } |
| 100 | |
| 101 | private fun flipChannel() { |
| 102 | val oldChannel = curChannel.get() |
| 103 | val newChannel = ConflatedChannel<Int>() |
| 104 | curChannel.set(newChannel) |
Roman Elizarov | 448528e | 2017-07-21 18:11:20 +0300 | [diff] [blame] | 105 | check(oldChannel.close()) |
Roman Elizarov | e6e8ce8 | 2017-06-05 17:04:39 +0300 | [diff] [blame] | 106 | } |
| 107 | |
| 108 | private fun printStats() { |
| 109 | println("sent ${sent.get()}, closed ${closed.get()}, received ${received.get()}") |
| 110 | } |
| 111 | |
| 112 | class StopException : Exception() |
| 113 | } |