blob: 835e6b427c93ea32568cc7bed5cf56992a0d8bf8 (
plain)
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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
/*
* Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines
import kotlinx.atomicfu.*
import kotlinx.coroutines.channels.*
import org.junit.Test
import java.util.concurrent.locks.*
import kotlin.concurrent.*
import kotlin.test.*
/**
* Tests event loops integration.
* See [https://github.com/Kotlin/kotlinx.coroutines/issues/860].
*/
class EventLoopsTest : TestBase() {
@Test
fun testNestedRunBlocking() {
runBlocking { // outer event loop
// Produce string "OK"
val ch = produce { send("OK") }
// try receive this string in a blocking way:
assertEquals("OK", runBlocking { ch.receive() }) // it should not hang here
}
}
@Test
fun testUnconfinedInRunBlocking() {
var completed = false
runBlocking {
launch(Dispatchers.Unconfined) {
completed = true
}
// should not go into runBlocking loop, but complete here
assertTrue(completed)
}
}
@Test
fun testNestedUnconfined() {
expect(1)
GlobalScope.launch(Dispatchers.Unconfined) {
expect(2)
GlobalScope.launch(Dispatchers.Unconfined) {
// this gets scheduled into outer unconfined loop
expect(4)
}
expect(3) // ^^ executed before the above unconfined
}
finish(5)
}
@Test
fun testEventLoopInDefaultExecutor() = runTest {
expect(1)
withContext(Dispatchers.Unconfined) {
delay(1)
assertTrue(Thread.currentThread().name.startsWith(DefaultExecutor.THREAD_NAME))
expect(2)
// now runBlocking inside default executor thread --> should use outer event loop
DefaultExecutor.enqueue(Runnable {
expect(4) // will execute when runBlocking runs loop
})
expect(3)
runBlocking {
expect(5)
}
}
finish(6)
}
/**
* Simple test for [processNextEventInCurrentThread] API use-case.
*/
@Test
fun testProcessNextEventInCurrentThreadSimple() = runTest {
expect(1)
val event = EventSync()
// this coroutine fires event
launch {
expect(3)
event.fireEvent()
}
// main coroutine waits for event (same thread!)
expect(2)
event.blockingAwait()
finish(4)
}
@Test
fun testSecondThreadRunBlocking() = runTest {
val testThread = Thread.currentThread()
val testContext = coroutineContext
val event = EventSync() // will signal completion
var thread = thread {
runBlocking { // outer event loop
// Produce string "OK"
val ch = produce { send("OK") }
// try receive this string in a blocking way using test context (another thread)
assertEquals("OK", runBlocking(testContext) {
assertEquals(testThread, Thread.currentThread())
ch.receive() // it should not hang here
})
}
event.fireEvent() // done thread
}
event.blockingAwait() // wait for thread to complete
thread.join() // it is safe to join thread now
}
/**
* Test for [processNextEventInCurrentThread] API use-case with delay.
*/
@Test
fun testProcessNextEventInCurrentThreadDelay() = runTest {
expect(1)
val event = EventSync()
// this coroutine fires event
launch {
expect(3)
delay(100)
event.fireEvent()
}
// main coroutine waits for event (same thread!)
expect(2)
event.blockingAwait()
finish(4)
}
class EventSync {
private val waitingThread = atomic<Thread?>(null)
private val fired = atomic(false)
fun fireEvent() {
fired.value = true
waitingThread.value?.let { LockSupport.unpark(it) }
}
fun blockingAwait() {
check(waitingThread.getAndSet(Thread.currentThread()) == null)
while (!fired.getAndSet(false)) {
val time = processNextEventInCurrentThread()
LockSupport.parkNanos(time)
}
waitingThread.value = null
}
}
}
|
