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
|
/*
* Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
*/
package kotlinx.coroutines.scheduling
import kotlinx.coroutines.*
import kotlinx.coroutines.internal.*
import java.util.concurrent.*
// TODO most of these fields will be moved to 'object ExperimentalDispatcher'
internal const val DEFAULT_SCHEDULER_NAME = "DefaultDispatcher"
// 100us as default
@JvmField
internal val WORK_STEALING_TIME_RESOLUTION_NS = systemProp(
"kotlinx.coroutines.scheduler.resolution.ns", 100000L
)
@JvmField
internal val BLOCKING_DEFAULT_PARALLELISM = systemProp(
"kotlinx.coroutines.scheduler.blocking.parallelism", 16
)
// NOTE: we coerce default to at least two threads to give us chances that multi-threading problems
// get reproduced even on a single-core machine, but support explicit setting of 1 thread scheduler if needed.
@JvmField
internal val CORE_POOL_SIZE = systemProp(
"kotlinx.coroutines.scheduler.core.pool.size",
AVAILABLE_PROCESSORS.coerceAtLeast(2), // !!! at least two here
minValue = CoroutineScheduler.MIN_SUPPORTED_POOL_SIZE
)
@JvmField
internal val MAX_POOL_SIZE = systemProp(
"kotlinx.coroutines.scheduler.max.pool.size",
(AVAILABLE_PROCESSORS * 128).coerceIn(
CORE_POOL_SIZE,
CoroutineScheduler.MAX_SUPPORTED_POOL_SIZE
),
maxValue = CoroutineScheduler.MAX_SUPPORTED_POOL_SIZE
)
@JvmField
internal val IDLE_WORKER_KEEP_ALIVE_NS = TimeUnit.SECONDS.toNanos(
systemProp("kotlinx.coroutines.scheduler.keep.alive.sec", 60L)
)
@JvmField
internal var schedulerTimeSource: TimeSource = NanoTimeSource
/**
* Marker indicating that task is CPU-bound and will not block
*/
internal const val TASK_NON_BLOCKING = 0
/**
* Marker indicating that task may potentially block, thus giving scheduler a hint that additional thread may be required
*/
internal const val TASK_PROBABLY_BLOCKING = 1
internal interface TaskContext {
val taskMode: Int // TASK_XXX
fun afterTask()
}
internal object NonBlockingContext : TaskContext {
override val taskMode: Int = TASK_NON_BLOCKING
override fun afterTask() {
// Nothing for non-blocking context
}
}
internal abstract class Task(
@JvmField var submissionTime: Long,
@JvmField var taskContext: TaskContext
) : Runnable {
constructor() : this(0, NonBlockingContext)
inline val mode: Int get() = taskContext.taskMode // TASK_XXX
}
internal inline val Task.isBlocking get() = taskContext.taskMode == TASK_PROBABLY_BLOCKING
// Non-reusable Task implementation to wrap Runnable instances that do not otherwise implement task
internal class TaskImpl(
@JvmField val block: Runnable,
submissionTime: Long,
taskContext: TaskContext
) : Task(submissionTime, taskContext) {
override fun run() {
try {
block.run()
} finally {
taskContext.afterTask()
}
}
override fun toString(): String =
"Task[${block.classSimpleName}@${block.hexAddress}, $submissionTime, $taskContext]"
}
// Open for tests
internal class GlobalQueue : LockFreeTaskQueue<Task>(singleConsumer = false)
internal abstract class TimeSource {
abstract fun nanoTime(): Long
}
internal object NanoTimeSource : TimeSource() {
override fun nanoTime() = System.nanoTime()
}
|
