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
|
/*
* Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines
import kotlinx.cinterop.*
import platform.posix.*
import kotlin.coroutines.*
/**
* Runs new coroutine and **blocks** current thread _interruptibly_ until its completion.
* This function should not be used from coroutine. It is designed to bridge regular blocking code
* to libraries that are written in suspending style, to be used in `main` functions and in tests.
*
* The default [CoroutineDispatcher] for this builder in an implementation of [EventLoop] that processes continuations
* in this blocked thread until the completion of this coroutine.
* See [CoroutineDispatcher] for the other implementations that are provided by `kotlinx.coroutines`.
*
* When [CoroutineDispatcher] is explicitly specified in the [context], then the new coroutine runs in the context of
* the specified dispatcher while the current thread is blocked. If the specified dispatcher implements [EventLoop]
* interface and this `runBlocking` invocation is performed from inside of the this event loop's thread, then
* this event loop is processed using its [processNextEvent][EventLoop.processNextEvent] method until coroutine completes.
*
* If this blocked thread is interrupted (see [Thread.interrupt]), then the coroutine job is cancelled and
* this `runBlocking` invocation throws [InterruptedException].
*
* See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
*
* @param context context of the coroutine. The default value is an implementation of [EventLoop].
* @param block the coroutine code.
*/
public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext, block: suspend CoroutineScope.() -> T): T {
val contextInterceptor = context[ContinuationInterceptor]
val eventLoop: EventLoop?
var newContext: CoroutineContext = context // todo: kludge for data flow analysis error
if (contextInterceptor == null) {
// create or use private event loop if no dispatcher is specified
eventLoop = ThreadLocalEventLoop.eventLoop
newContext = GlobalScope.newCoroutineContext(context + eventLoop)
} else {
// See if context's interceptor is an event loop that we shall use (to support TestContext)
// or take an existing thread-local event loop if present to avoid blocking it (but don't create one)
eventLoop = (contextInterceptor as? EventLoop)?.takeIf { it.shouldBeProcessedFromContext() }
?: ThreadLocalEventLoop.currentOrNull()
newContext = GlobalScope.newCoroutineContext(context)
}
val coroutine = BlockingCoroutine<T>(newContext, eventLoop)
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
return coroutine.joinBlocking()
}
private class BlockingCoroutine<T>(
parentContext: CoroutineContext,
private val eventLoop: EventLoop?
) : AbstractCoroutine<T>(parentContext, true, true) {
override val isScopedCoroutine: Boolean get() = true
@Suppress("UNCHECKED_CAST")
fun joinBlocking(): T = memScoped {
try {
eventLoop?.incrementUseCount()
val timespec = alloc<timespec>()
while (true) {
val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
// note: process next even may loose unpark flag, so check if completed before parking
if (isCompleted) break
timespec.tv_sec = (parkNanos / 1000000000L).convert() // 1e9 ns -> sec
timespec.tv_nsec = (parkNanos % 1000000000L).convert() // % 1e9
nanosleep(timespec.ptr, null)
}
} finally { // paranoia
eventLoop?.decrementUseCount()
}
// now return result
val state = state.unboxState()
(state as? CompletedExceptionally)?.let { throw it.cause }
state as T
}
}
|
