1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
/*
* Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines
import org.junit.*
import org.junit.Test
import java.util.concurrent.*
class AwaitStressTest : TestBase() {
private val iterations = 50_000 * stressTestMultiplier
private val pool = newFixedThreadPoolContext(4, "AwaitStressTest")
@After
fun tearDown() {
pool.close()
}
@Test
fun testMultipleExceptions() = runTest {
val ctx = pool + NonCancellable
repeat(iterations) {
val barrier = CyclicBarrier(4)
val d1 = async(ctx) {
barrier.await()
throw TestException()
}
val d2 = async(ctx) {
barrier.await()
throw TestException()
}
val d3 = async(ctx) {
barrier.await()
1L
}
try {
barrier.await()
awaitAll(d1, d2, d3)
expectUnreached()
} catch (e: TestException) {
// Expected behaviour
}
barrier.reset()
}
}
@Test
fun testAwaitAll() = runTest {
val barrier = CyclicBarrier(3)
repeat(iterations) {
val d1 = async(pool) {
barrier.await()
1L
}
val d2 = async(pool) {
barrier.await()
2L
}
barrier.await()
awaitAll(d1, d2)
require(d1.isCompleted && d2.isCompleted)
barrier.reset()
}
}
@Test
fun testConcurrentCancellation() = runTest {
var cancelledOnce = false
repeat(iterations) {
val barrier = CyclicBarrier(3)
val d1 = async(pool) {
barrier.await()
delay(10_000)
yield()
}
val d2 = async(pool) {
barrier.await()
d1.cancel()
}
barrier.await()
try {
awaitAll(d1, d2)
} catch (e: CancellationException) {
cancelledOnce = true
}
}
require(cancelledOnce) { "Cancellation exception wasn't properly caught" }
}
@Test
fun testMutatingCollection() = runTest {
val barrier = CyclicBarrier(4)
repeat(iterations) {
// thread-safe collection that we are going to modify
val deferreds = CopyOnWriteArrayList<Deferred<Long>>()
deferreds += async(pool) {
barrier.await()
1L
}
deferreds += async(pool) {
barrier.await()
2L
}
deferreds += async(pool) {
barrier.await()
deferreds.removeAt(2)
3L
}
val allJobs = ArrayList(deferreds)
barrier.await()
val results = deferreds.awaitAll() // shouldn't hang
check(results == listOf(1L, 2L, 3L) || results == listOf(1L, 2L))
allJobs.awaitAll()
barrier.reset()
}
}
}
|
