Roman Elizarov | f16fd27 | 2017-02-07 11:26:00 +0300 | [diff] [blame] | 1 | /* |
| 2 | * Copyright 2016-2017 JetBrains s.r.o. |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 17 | package kotlinx.coroutines.experimental |
| 18 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 19 | import kotlinx.atomicfu.* |
| 20 | import kotlinx.coroutines.experimental.internal.* |
Vsevolod Tolstopyatov | cd00643 | 2018-04-26 16:03:40 +0300 | [diff] [blame] | 21 | import kotlinx.coroutines.experimental.timeunit.* |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 22 | import java.util.concurrent.locks.* |
| 23 | import kotlin.coroutines.experimental.* |
Vsevolod Tolstopyatov | cd00643 | 2018-04-26 16:03:40 +0300 | [diff] [blame] | 24 | import kotlin.jvm.* |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 25 | |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 26 | /** |
| 27 | * Implemented by [CoroutineDispatcher] implementations that have event loop inside and can |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 28 | * be asked to process next event from their event queue. |
| 29 | * |
| 30 | * It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 31 | * continue processing events when invoked from the event dispatch thread. |
| 32 | */ |
Roman Elizarov | bddb1d7 | 2017-12-25 17:16:08 +0300 | [diff] [blame] | 33 | public interface EventLoop { |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 34 | /** |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 35 | * Processes next event in this event loop. |
| 36 | * |
| 37 | * The result of this function is to be interpreted like this: |
| 38 | * * `<= 0` -- there are potentially more events for immediate processing; |
| 39 | * * `> 0` -- a number of nanoseconds to wait for next scheduled event; |
| 40 | * * [Long.MAX_VALUE] -- no more events, or was invoked from the wrong thread. |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 41 | */ |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 42 | public fun processNextEvent(): Long |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 43 | |
Roman Elizarov | e2ceda8 | 2017-07-21 18:19:47 +0300 | [diff] [blame] | 44 | /** @suppress **Deprecated **/ |
| 45 | @Deprecated(message = "Companion object to be removed, no replacement") |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 46 | public companion object Factory { |
Roman Elizarov | e2ceda8 | 2017-07-21 18:19:47 +0300 | [diff] [blame] | 47 | /** @suppress **Deprecated **/ |
| 48 | @Deprecated("Replaced with top-level function", level = DeprecationLevel.HIDDEN) |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 49 | public operator fun invoke(thread: Thread = Thread.currentThread(), parentJob: Job? = null): CoroutineDispatcher = |
| 50 | EventLoopImpl(thread).apply { |
| 51 | if (parentJob != null) initParentJob(parentJob) |
| 52 | } |
| 53 | } |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 54 | } |
| 55 | |
Roman Elizarov | e2ceda8 | 2017-07-21 18:19:47 +0300 | [diff] [blame] | 56 | /** |
| 57 | * Creates a new event loop that is bound the specified [thread] (current thread by default) and |
| 58 | * stops accepting new events when [parentJob] completes. Every continuation that is scheduled |
| 59 | * onto this event loop unparks the specified thread via [LockSupport.unpark]. |
| 60 | * |
| 61 | * The main event-processing loop using the resulting `eventLoop` object should look like this: |
| 62 | * ``` |
| 63 | * while (needsToBeRunning) { |
| 64 | * if (Thread.interrupted()) break // or handle somehow |
| 65 | * LockSupport.parkNanos(eventLoop.processNextEvent()) // event loop will unpark |
| 66 | * } |
| 67 | * ``` |
| 68 | */ |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 69 | @Suppress("FunctionName") |
Roman Elizarov | e2ceda8 | 2017-07-21 18:19:47 +0300 | [diff] [blame] | 70 | public fun EventLoop(thread: Thread = Thread.currentThread(), parentJob: Job? = null): CoroutineDispatcher = |
| 71 | EventLoopImpl(thread).apply { |
| 72 | if (parentJob != null) initParentJob(parentJob) |
| 73 | } |
| 74 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 75 | private const val DELAYED = 0 |
| 76 | private const val REMOVED = 1 |
| 77 | private const val RESCHEDULED = 2 |
| 78 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 79 | @Suppress("PrivatePropertyName") |
| 80 | private val CLOSED_EMPTY = Symbol("CLOSED_EMPTY") |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 81 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 82 | private typealias Queue<T> = LockFreeMPSCQueueCore<T> |
| 83 | |
| 84 | internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop { |
| 85 | // null | CLOSED_EMPTY | task | Queue<Runnable> |
| 86 | private val _queue = atomic<Any?>(null) |
| 87 | |
| 88 | // Allocated only only once |
| 89 | private val _delayed = atomic<ThreadSafeHeap<DelayedTask>?>(null) |
| 90 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 91 | protected abstract val isCompleted: Boolean |
| 92 | protected abstract fun unpark() |
| 93 | protected abstract fun isCorrectThread(): Boolean |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 94 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 95 | protected val isEmpty: Boolean |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 96 | get() = isQueueEmpty && isDelayedEmpty |
| 97 | |
| 98 | private val isQueueEmpty: Boolean get() { |
| 99 | val queue = _queue.value |
| 100 | return when (queue) { |
| 101 | null -> true |
| 102 | is Queue<*> -> queue.isEmpty |
| 103 | else -> queue === CLOSED_EMPTY |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | private val isDelayedEmpty: Boolean get() { |
| 108 | val delayed = _delayed.value |
| 109 | return delayed == null || delayed.isEmpty |
| 110 | } |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 111 | |
| 112 | private val nextTime: Long |
| 113 | get() { |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 114 | if (!isQueueEmpty) return 0 |
| 115 | val delayed = _delayed.value ?: return Long.MAX_VALUE |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 116 | val nextDelayedTask = delayed.peek() ?: return Long.MAX_VALUE |
| 117 | return (nextDelayedTask.nanoTime - timeSource.nanoTime()).coerceAtLeast(0) |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 118 | } |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 119 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 120 | override fun dispatch(context: CoroutineContext, block: Runnable) = |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 121 | execute(block) |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 122 | |
| 123 | override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) = |
| 124 | schedule(DelayedResumeTask(time, unit, continuation)) |
| 125 | |
| 126 | override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle = |
| 127 | DelayedRunnableTask(time, unit, block).also { schedule(it) } |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 128 | |
| 129 | override fun processNextEvent(): Long { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 130 | if (!isCorrectThread()) return Long.MAX_VALUE |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 131 | // queue all delayed tasks that are due to be executed |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 132 | val delayed = _delayed.value |
| 133 | if (delayed != null && !delayed.isEmpty) { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 134 | val now = timeSource.nanoTime() |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 135 | while (true) { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 136 | // make sure that moving from delayed to queue removes from delayed only after it is added to queue |
| 137 | // to make sure that 'isEmpty' and `nextTime` that check both of them |
| 138 | // do not transiently report that both delayed and queue are empty during move |
| 139 | delayed.removeFirstIf { |
| 140 | if (it.timeToExecute(now)) { |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 141 | enqueueImpl(it) |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 142 | } else |
| 143 | false |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 144 | } ?: break // quit loop when nothing more to remove or enqueueImpl returns false on "isComplete" |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 145 | } |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 146 | } |
| 147 | // then process one event from queue |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 148 | dequeue()?.run() |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 149 | return nextTime |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 150 | } |
| 151 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 152 | @Suppress("MemberVisibilityCanBePrivate") // todo: remove suppress when KT-22030 is fixed |
| 153 | internal fun execute(task: Runnable) { |
| 154 | if (enqueueImpl(task)) { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 155 | // todo: we should unpark only when this delayed task became first in the queue |
| 156 | unpark() |
| 157 | } else |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 158 | DefaultExecutor.execute(task) |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 159 | } |
| 160 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 161 | @Suppress("UNCHECKED_CAST") |
| 162 | private fun enqueueImpl(task: Runnable): Boolean { |
| 163 | _queue.loop { queue -> |
| 164 | if (isCompleted) return false // fail fast if already completed, may still add, but queues will close |
| 165 | when (queue) { |
| 166 | null -> if (_queue.compareAndSet(null, task)) return true |
| 167 | is Queue<*> -> { |
| 168 | when ((queue as Queue<Runnable>).addLast(task)) { |
| 169 | Queue.ADD_SUCCESS -> return true |
| 170 | Queue.ADD_CLOSED -> return false |
| 171 | Queue.ADD_FROZEN -> _queue.compareAndSet(queue, queue.next()) |
| 172 | } |
| 173 | } |
| 174 | else -> when { |
| 175 | queue === CLOSED_EMPTY -> return false |
| 176 | else -> { |
| 177 | // update to full-blown queue to add one more |
| 178 | val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY) |
| 179 | newQueue.addLast(queue as Runnable) |
| 180 | newQueue.addLast(task) |
| 181 | if (_queue.compareAndSet(queue, newQueue)) return true |
| 182 | } |
| 183 | } |
| 184 | } |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 185 | } |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 186 | } |
| 187 | |
| 188 | @Suppress("UNCHECKED_CAST") |
| 189 | private fun dequeue(): Runnable? { |
| 190 | _queue.loop { queue -> |
| 191 | when (queue) { |
| 192 | null -> return null |
| 193 | is Queue<*> -> { |
| 194 | val result = (queue as Queue<Runnable>).removeFirstOrNull() |
| 195 | if (result !== Queue.REMOVE_FROZEN) return result as Runnable? |
| 196 | _queue.compareAndSet(queue, queue.next()) |
| 197 | } |
| 198 | else -> when { |
| 199 | queue === CLOSED_EMPTY -> return null |
| 200 | else -> if (_queue.compareAndSet(queue, null)) return queue as Runnable |
| 201 | } |
| 202 | } |
| 203 | } |
| 204 | } |
| 205 | |
| 206 | protected fun closeQueue() { |
| 207 | assert(isCompleted) |
| 208 | _queue.loop { queue -> |
| 209 | when (queue) { |
| 210 | null -> if (_queue.compareAndSet(null, CLOSED_EMPTY)) return |
| 211 | is Queue<*> -> { |
| 212 | queue.close() |
| 213 | return |
| 214 | } |
| 215 | else -> when { |
| 216 | queue === CLOSED_EMPTY -> return |
| 217 | else -> { |
| 218 | // update to full-blown queue to close |
| 219 | val newQueue = Queue<Runnable>(Queue.INITIAL_CAPACITY) |
| 220 | newQueue.addLast(queue as Runnable) |
| 221 | if (_queue.compareAndSet(queue, newQueue)) return |
| 222 | } |
| 223 | } |
| 224 | } |
| 225 | } |
| 226 | |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 227 | } |
| 228 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 229 | internal fun schedule(delayedTask: DelayedTask) { |
| 230 | if (scheduleImpl(delayedTask)) { |
| 231 | // todo: we should unpark only when this delayed task became first in the queue |
| 232 | unpark() |
| 233 | } else |
| 234 | DefaultExecutor.schedule(delayedTask) |
| 235 | } |
| 236 | |
| 237 | private fun scheduleImpl(delayedTask: DelayedTask): Boolean { |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 238 | if (isCompleted) return false |
| 239 | val delayed = _delayed.value ?: run { |
| 240 | _delayed.compareAndSet(null, ThreadSafeHeap()) |
| 241 | _delayed.value!! |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 242 | } |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 243 | return delayed.addLastIf(delayedTask) { !isCompleted } |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 244 | } |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 245 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 246 | internal fun removeDelayedImpl(delayedTask: DelayedTask) { |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 247 | _delayed.value?.remove(delayedTask) |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 248 | } |
Roman Elizarov | d528e3e | 2017-01-23 15:40:05 +0300 | [diff] [blame] | 249 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 250 | // It performs "hard" shutdown for test cleanup purposes |
| 251 | protected fun resetAll() { |
| 252 | _queue.value = null |
| 253 | _delayed.value = null |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 254 | } |
| 255 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 256 | // This is a "soft" (normal) shutdown |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 257 | protected fun rescheduleAllDelayed() { |
| 258 | while (true) { |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 259 | val delayedTask = _delayed.value?.removeFirstOrNull() ?: break |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 260 | delayedTask.rescheduleOnShutdown() |
| 261 | } |
| 262 | } |
| 263 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 264 | internal abstract inner class DelayedTask( |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 265 | time: Long, timeUnit: TimeUnit |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 266 | ) : Runnable, Comparable<DelayedTask>, DisposableHandle, ThreadSafeHeapNode { |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 267 | override var index: Int = -1 |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 268 | var state = DELAYED // Guarded by by lock on this task for reschedule/dispose purposes |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 269 | @JvmField val nanoTime: Long = timeSource.nanoTime() + timeUnit.toNanos(time) |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 270 | |
| 271 | override fun compareTo(other: DelayedTask): Int { |
| 272 | val dTime = nanoTime - other.nanoTime |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 273 | return when { |
| 274 | dTime > 0 -> 1 |
| 275 | dTime < 0 -> -1 |
| 276 | else -> 0 |
| 277 | } |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 278 | } |
| 279 | |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 280 | fun timeToExecute(now: Long): Boolean = now - nanoTime >= 0L |
| 281 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 282 | fun rescheduleOnShutdown() = synchronized(this) { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 283 | if (state != DELAYED) return |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 284 | if (_delayed.value!!.remove(this)) { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 285 | state = RESCHEDULED |
| 286 | DefaultExecutor.schedule(this) |
| 287 | } else |
| 288 | state = REMOVED |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 289 | } |
| 290 | |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 291 | final override fun dispose() = synchronized(this) { |
Roman Elizarov | 9483301 | 2018-01-12 18:28:24 +0300 | [diff] [blame] | 292 | when (state) { |
| 293 | DELAYED -> _delayed.value?.remove(this) |
| 294 | RESCHEDULED -> DefaultExecutor.removeDelayedImpl(this) |
| 295 | else -> return |
| 296 | } |
| 297 | state = REMOVED |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 298 | } |
Roman Elizarov | 769d482 | 2017-07-10 13:38:54 +0300 | [diff] [blame] | 299 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 300 | override fun toString(): String = "Delayed[nanos=$nanoTime]" |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 301 | } |
| 302 | |
| 303 | private inner class DelayedResumeTask( |
| 304 | time: Long, timeUnit: TimeUnit, |
| 305 | private val cont: CancellableContinuation<Unit> |
| 306 | ) : DelayedTask(time, timeUnit) { |
Vsevolod Tolstopyatov | cd00643 | 2018-04-26 16:03:40 +0300 | [diff] [blame] | 307 | |
| 308 | init { |
| 309 | // Note that this operation isn't lock-free, but very short |
Roman Elizarov | 6d9f40f | 2018-04-28 14:44:02 +0300 | [diff] [blame] | 310 | cont.disposeOnCancellation(this) |
Vsevolod Tolstopyatov | cd00643 | 2018-04-26 16:03:40 +0300 | [diff] [blame] | 311 | } |
| 312 | |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 313 | override fun run() { |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 314 | with(cont) { resumeUndispatched(Unit) } |
| 315 | } |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 316 | } |
| 317 | |
| 318 | private inner class DelayedRunnableTask( |
| 319 | time: Long, timeUnit: TimeUnit, |
| 320 | private val block: Runnable |
| 321 | ) : DelayedTask(time, timeUnit) { |
Roman Elizarov | a047a11 | 2017-07-10 18:50:07 +0300 | [diff] [blame] | 322 | override fun run() { block.run() } |
Roman Elizarov | 769d482 | 2017-07-10 13:38:54 +0300 | [diff] [blame] | 323 | override fun toString(): String = super.toString() + block.toString() |
Roman Elizarov | 4518e05 | 2017-03-02 22:48:27 +0300 | [diff] [blame] | 324 | } |
Roman Elizarov | 53a0a40 | 2017-01-19 20:21:57 +0300 | [diff] [blame] | 325 | } |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 326 | |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 327 | internal abstract class ThreadEventLoop( |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 328 | private val thread: Thread |
| 329 | ) : EventLoopBase() { |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 330 | override fun isCorrectThread(): Boolean = Thread.currentThread() === thread |
| 331 | |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 332 | override fun unpark() { |
| 333 | if (Thread.currentThread() !== thread) |
| 334 | timeSource.unpark(thread) |
| 335 | } |
| 336 | |
| 337 | fun shutdown() { |
Roman Elizarov | 5d94a26 | 2017-12-28 00:23:39 +0300 | [diff] [blame] | 338 | closeQueue() |
Roman Elizarov | 35d2c34 | 2017-07-20 14:54:39 +0300 | [diff] [blame] | 339 | assert(isCorrectThread()) |
| 340 | // complete processing of all queued tasks |
| 341 | while (processNextEvent() <= 0) { /* spin */ } |
| 342 | // reschedule the rest of delayed tasks |
| 343 | rescheduleAllDelayed() |
| 344 | } |
| 345 | } |
| 346 | |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 347 | private class EventLoopImpl(thread: Thread) : ThreadEventLoop(thread) { |
| 348 | private var parentJob: Job? = null |
| 349 | |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 350 | override val isCompleted: Boolean get() = parentJob?.isCompleted == true |
| 351 | |
| 352 | fun initParentJob(parentJob: Job) { |
| 353 | require(this.parentJob == null) |
| 354 | this.parentJob = parentJob |
| 355 | } |
| 356 | } |
| 357 | |
| 358 | internal class BlockingEventLoop(thread: Thread) : ThreadEventLoop(thread) { |
Roman Elizarov | 8b38fa2 | 2017-09-27 17:44:31 +0300 | [diff] [blame] | 359 | @Volatile |
| 360 | public override var isCompleted: Boolean = false |
| 361 | } |