From c214a7fe0d46ecc23a71f9237b20281c0ca1c929 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 5 Oct 2022 13:22:40 +0200 Subject: refactor(sim/core): Use SimulationScheduler in coroutine dispatcher This change updates the implementation of `SimulationDispatcher` to use a (possibly user-provided) `SimulationScheduler` for managing the execution of the simulation and future tasks. --- .../opendc/simulator/core/SimulationBuilders.kt | 75 --------- .../opendc/simulator/core/SimulationController.kt | 46 ------ .../core/SimulationCoroutineDispatcher.kt | 167 --------------------- .../simulator/core/SimulationCoroutineScope.kt | 62 -------- .../opendc/simulator/kotlin/SimulationBuilders.kt | 80 ++++++++++ .../simulator/kotlin/SimulationController.kt | 52 +++++++ .../kotlin/SimulationCoroutineDispatcher.kt | 94 ++++++++++++ .../simulator/kotlin/SimulationCoroutineScope.kt | 66 ++++++++ 8 files changed, 292 insertions(+), 350 deletions(-) delete mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationBuilders.kt delete mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationController.kt delete mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt delete mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineScope.kt create mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationBuilders.kt create mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationController.kt create mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt create mode 100644 opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineScope.kt (limited to 'opendc-simulator/opendc-simulator-core/src') diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationBuilders.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationBuilders.kt deleted file mode 100644 index 9b284c11..00000000 --- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationBuilders.kt +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.core - -import kotlinx.coroutines.* -import kotlin.coroutines.ContinuationInterceptor -import kotlin.coroutines.CoroutineContext -import kotlin.coroutines.EmptyCoroutineContext - -/** - * Executes a [body] inside an immediate execution dispatcher. - */ -@OptIn(ExperimentalCoroutinesApi::class) -public fun runBlockingSimulation(context: CoroutineContext = EmptyCoroutineContext, body: suspend SimulationCoroutineScope.() -> Unit) { - val (safeContext, dispatcher) = context.checkArguments() - val startingJobs = safeContext.activeJobs() - val scope = SimulationCoroutineScope(safeContext) - val deferred = scope.async { - body(scope) - } - dispatcher.advanceUntilIdle() - deferred.getCompletionExceptionOrNull()?.let { - throw it - } - val endingJobs = safeContext.activeJobs() - if ((endingJobs - startingJobs).isNotEmpty()) { - throw IllegalStateException("Test finished with active jobs: $endingJobs") - } -} - -/** - * Convenience method for calling [runBlockingSimulation] on an existing [SimulationCoroutineScope]. - */ -public fun SimulationCoroutineScope.runBlockingSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit = - runBlockingSimulation(coroutineContext, block) - -/** - * Convenience method for calling [runBlockingSimulation] on an existing [SimulationCoroutineDispatcher]. - */ -public fun SimulationCoroutineDispatcher.runBlockingSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit = - runBlockingSimulation(this, block) - -private fun CoroutineContext.checkArguments(): Pair { - val dispatcher = get(ContinuationInterceptor).run { - this?.let { require(this is SimulationController) { "Dispatcher must implement SimulationController: $this" } } - this ?: SimulationCoroutineDispatcher() - } - - val job = get(Job) ?: SupervisorJob() - return Pair(this + dispatcher + job, dispatcher as SimulationController) -} - -private fun CoroutineContext.activeJobs(): Set { - return checkNotNull(this[Job]).children.filter { it.isActive }.toSet() -} diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationController.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationController.kt deleted file mode 100644 index 2b670b91..00000000 --- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationController.kt +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.core - -import kotlinx.coroutines.CoroutineDispatcher -import java.time.Clock - -/** - * Control the virtual clock of a [CoroutineDispatcher]. - */ -public interface SimulationController { - /** - * The current virtual clock as it is known to this Dispatcher. - */ - public val clock: Clock - - /** - * Immediately execute all pending tasks and advance the virtual clock-time to the last delay. - * - * If new tasks are scheduled due to advancing virtual time, they will be executed before `advanceUntilIdle` - * returns. - * - * @return the amount of delay-time that this Dispatcher's clock has been forwarded in milliseconds. - */ - public fun advanceUntilIdle(): Long -} diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt deleted file mode 100644 index 908e902a..00000000 --- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineDispatcher.kt +++ /dev/null @@ -1,167 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.core - -import kotlinx.coroutines.* -import java.lang.Runnable -import java.time.Clock -import java.time.Instant -import java.time.ZoneId -import java.util.* -import kotlin.coroutines.CoroutineContext - -/** - * A [CoroutineDispatcher] that performs both immediate execution of coroutines on the main thread and uses a virtual - * clock for time management. - */ -@OptIn(InternalCoroutinesApi::class) -public class SimulationCoroutineDispatcher : CoroutineDispatcher(), SimulationController, Delay { - /** - * Queue of ordered tasks to run. - */ - private val queue = PriorityQueue() - - /** - * Global order counter. - */ - private var _counter = 0L - - /** - * The current virtual time of simulation - */ - private var _clock = SimClock() - - /** - * The virtual clock of this dispatcher. - */ - override val clock: Clock = ClockAdapter(_clock) - - override fun dispatch(context: CoroutineContext, block: Runnable) { - block.run() - } - - override fun dispatchYield(context: CoroutineContext, block: Runnable) { - post(block) - } - - @OptIn(ExperimentalCoroutinesApi::class) - override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation) { - postDelayed(CancellableContinuationRunnable(continuation) { resumeUndispatched(Unit) }, timeMillis) - } - - override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle { - val node = postDelayed(block, timeMillis) - return object : DisposableHandle { - override fun dispose() { - queue.remove(node) - } - } - } - - override fun toString(): String { - return "SimulationCoroutineDispatcher[time=${_clock.time}ms, queued=${queue.size}]" - } - - private fun post(block: Runnable) = - queue.add(TimedRunnable(block, _counter++)) - - private fun postDelayed(block: Runnable, delayTime: Long) = - TimedRunnable(block, _counter++, safePlus(_clock.time, delayTime)) - .also { - queue.add(it) - } - - private fun safePlus(currentTime: Long, delayTime: Long): Long { - check(delayTime >= 0) - val result = currentTime + delayTime - if (result < currentTime) return Long.MAX_VALUE // clamp on overflow - return result - } - - override fun advanceUntilIdle(): Long { - val queue = queue - val clock = _clock - val oldTime = clock.time - - while (true) { - val current = queue.poll() ?: break - - // If the scheduled time is 0 (immediate) use current virtual time - if (current.time != 0L) { - clock.time = current.time - } - - current.run() - } - - return clock.time - oldTime - } - - /** - * A helper class that holds the time of the simulation. - */ - private class SimClock(@JvmField var time: Long = 0) - - /** - * A helper class to expose a [Clock] instance for this dispatcher. - */ - private class ClockAdapter(private val clock: SimClock, private val zone: ZoneId = ZoneId.systemDefault()) : Clock() { - override fun getZone(): ZoneId = zone - - override fun withZone(zone: ZoneId): Clock = ClockAdapter(clock, zone) - - override fun instant(): Instant = Instant.ofEpochMilli(millis()) - - override fun millis(): Long = clock.time - - override fun toString(): String = "SimulationCoroutineDispatcher.ClockAdapter[time=${clock.time}]" - } - - /** - * This class exists to allow cleanup code to avoid throwing for cancelled continuations scheduled - * in the future. - */ - private class CancellableContinuationRunnable( - @JvmField val continuation: CancellableContinuation, - private val block: CancellableContinuation.() -> Unit - ) : Runnable { - override fun run() = continuation.block() - } - - /** - * A Runnable for our event loop that represents a task to perform at a time. - */ - private class TimedRunnable( - @JvmField val runnable: Runnable, - private val count: Long = 0, - @JvmField val time: Long = 0 - ) : Comparable, Runnable by runnable { - override fun compareTo(other: TimedRunnable) = if (time == other.time) { - count.compareTo(other.count) - } else { - time.compareTo(other.time) - } - - override fun toString() = "TimedRunnable[time=$time, run=$runnable]" - } -} diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineScope.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineScope.kt deleted file mode 100644 index 1da7f0fa..00000000 --- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/core/SimulationCoroutineScope.kt +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.core - -import kotlinx.coroutines.CoroutineExceptionHandler -import kotlinx.coroutines.CoroutineScope -import kotlin.coroutines.ContinuationInterceptor -import kotlin.coroutines.CoroutineContext -import kotlin.coroutines.EmptyCoroutineContext - -/** - * A scope which provides detailed control over the execution of coroutines for simulations. - */ -public interface SimulationCoroutineScope : CoroutineScope, SimulationController - -private class SimulationCoroutineScopeImpl( - override val coroutineContext: CoroutineContext -) : - SimulationCoroutineScope, - SimulationController by coroutineContext.simulationController - -/** - * A scope which provides detailed control over the execution of coroutines for simulations. - * - * If the provided context does not provide a [ContinuationInterceptor] (Dispatcher) or [CoroutineExceptionHandler], the - * scope adds [SimulationCoroutineDispatcher] automatically. - */ -@Suppress("FunctionName") -public fun SimulationCoroutineScope(context: CoroutineContext = EmptyCoroutineContext): SimulationCoroutineScope { - var safeContext = context - if (context[ContinuationInterceptor] == null) safeContext += SimulationCoroutineDispatcher() - return SimulationCoroutineScopeImpl(safeContext) -} - -private inline val CoroutineContext.simulationController: SimulationController - get() { - val handler = this[ContinuationInterceptor] - return handler as? SimulationController ?: throw IllegalArgumentException( - "SimulationCoroutineScope requires a SimulationController such as SimulatorCoroutineDispatcher as " + - "the ContinuationInterceptor (Dispatcher)" - ) - } diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationBuilders.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationBuilders.kt new file mode 100644 index 00000000..a291b4e2 --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationBuilders.kt @@ -0,0 +1,80 @@ +/* + * Copyright (c) 2021 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.kotlin + +import kotlinx.coroutines.* +import org.opendc.simulator.SimulationScheduler +import kotlin.coroutines.ContinuationInterceptor +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext + +/** + * Executes a [body] inside an immediate execution dispatcher. + */ +@OptIn(ExperimentalCoroutinesApi::class) +public fun runBlockingSimulation( + context: CoroutineContext = EmptyCoroutineContext, + scheduler: SimulationScheduler = SimulationScheduler(), + body: suspend SimulationCoroutineScope.() -> Unit +) { + val (safeContext, dispatcher) = context.checkArguments(scheduler) + val startingJobs = safeContext.activeJobs() + val scope = SimulationCoroutineScope(safeContext) + val deferred = scope.async { + body(scope) + } + dispatcher.advanceUntilIdle() + deferred.getCompletionExceptionOrNull()?.let { + throw it + } + val endingJobs = safeContext.activeJobs() + if ((endingJobs - startingJobs).isNotEmpty()) { + throw IllegalStateException("Test finished with active jobs: $endingJobs") + } +} + +/** + * Convenience method for calling [runBlockingSimulation] on an existing [SimulationCoroutineScope]. + */ +public fun SimulationCoroutineScope.runBlockingSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit = + runBlockingSimulation(coroutineContext, scheduler, block) + +/** + * Convenience method for calling [runBlockingSimulation] on an existing [SimulationCoroutineDispatcher]. + */ +public fun SimulationCoroutineDispatcher.runBlockingSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit = + runBlockingSimulation(this, scheduler, block) + +private fun CoroutineContext.checkArguments(scheduler: SimulationScheduler): Pair { + val dispatcher = get(ContinuationInterceptor).run { + this?.let { require(this is SimulationController) { "Dispatcher must implement SimulationController: $this" } } + this ?: SimulationCoroutineDispatcher(scheduler) + } + + val job = get(Job) ?: SupervisorJob() + return Pair(this + dispatcher + job, dispatcher as SimulationController) +} + +private fun CoroutineContext.activeJobs(): Set { + return checkNotNull(this[Job]).children.filter { it.isActive }.toSet() +} diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationController.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationController.kt new file mode 100644 index 00000000..f96b2326 --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationController.kt @@ -0,0 +1,52 @@ +/* + * Copyright (c) 2021 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.kotlin + +import kotlinx.coroutines.CoroutineDispatcher +import org.opendc.simulator.SimulationScheduler +import java.time.Clock + +/** + * Control the virtual clock of a [CoroutineDispatcher]. + */ +public interface SimulationController { + /** + * The current virtual clock as it is known to this Dispatcher. + */ + public val clock: Clock + + /** + * The [SimulationScheduler] driving the simulation. + */ + public val scheduler: SimulationScheduler + + /** + * Immediately execute all pending tasks and advance the virtual clock-time to the last delay. + * + * If new tasks are scheduled due to advancing virtual time, they will be executed before `advanceUntilIdle` + * returns. + * + * @return the amount of delay-time that this Dispatcher's clock has been forwarded in milliseconds. + */ + public fun advanceUntilIdle(): Long +} diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt new file mode 100644 index 00000000..21ad1a86 --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt @@ -0,0 +1,94 @@ +/* + * Copyright (c) 2021 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.kotlin + +import kotlinx.coroutines.* +import org.opendc.simulator.SimulationScheduler +import java.lang.Runnable +import java.time.Clock +import java.util.* +import kotlin.coroutines.CoroutineContext + +/** + * A [CoroutineDispatcher] that performs both immediate execution of coroutines on the main thread and uses a virtual + * clock for time management. + * + * @param scheduler The [SimulationScheduler] used to manage the execution of future tasks. + */ +@OptIn(InternalCoroutinesApi::class) +public class SimulationCoroutineDispatcher( + override val scheduler: SimulationScheduler = SimulationScheduler() +) : CoroutineDispatcher(), SimulationController, Delay { + /** + * The virtual clock of this dispatcher. + */ + override val clock: Clock = scheduler.clock + + override fun dispatch(context: CoroutineContext, block: Runnable) { + block.run() + } + + override fun dispatchYield(context: CoroutineContext, block: Runnable) { + scheduler.execute(block) + } + + @OptIn(ExperimentalCoroutinesApi::class) + override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation) { + scheduler.schedule(timeMillis, CancellableContinuationRunnable(continuation) { resumeUndispatched(Unit) }) + } + + override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle { + return object : DisposableHandle { + private val deadline = (scheduler.currentTime + timeMillis).let { if (it >= 0) it else Long.MAX_VALUE } + private val id = scheduler.schedule(timeMillis, block) + + override fun dispose() { + scheduler.cancel(deadline, id) + } + } + } + + override fun toString(): String { + return "SimulationCoroutineDispatcher[time=${scheduler.currentTime}ms]" + } + + override fun advanceUntilIdle(): Long { + val scheduler = scheduler + val oldTime = scheduler.currentTime + + scheduler.advanceUntilIdle() + + return scheduler.currentTime - oldTime + } + + /** + * This class exists to allow cleanup code to avoid throwing for cancelled continuations scheduled + * in the future. + */ + private class CancellableContinuationRunnable( + @JvmField val continuation: CancellableContinuation, + private val block: CancellableContinuation.() -> Unit + ) : Runnable { + override fun run() = continuation.block() + } +} diff --git a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineScope.kt b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineScope.kt new file mode 100644 index 00000000..6be8e67a --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineScope.kt @@ -0,0 +1,66 @@ +/* + * Copyright (c) 2021 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.kotlin + +import kotlinx.coroutines.CoroutineExceptionHandler +import kotlinx.coroutines.CoroutineScope +import org.opendc.simulator.SimulationScheduler +import kotlin.coroutines.ContinuationInterceptor +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext + +/** + * A scope which provides detailed control over the execution of coroutines for simulations. + */ +public interface SimulationCoroutineScope : CoroutineScope, SimulationController + +private class SimulationCoroutineScopeImpl( + override val coroutineContext: CoroutineContext +) : + SimulationCoroutineScope, + SimulationController by coroutineContext.simulationController + +/** + * A scope which provides detailed control over the execution of coroutines for simulations. + * + * If the provided context does not provide a [ContinuationInterceptor] (Dispatcher) or [CoroutineExceptionHandler], the + * scope adds [SimulationCoroutineDispatcher] automatically. + */ +@Suppress("FunctionName") +public fun SimulationCoroutineScope( + context: CoroutineContext = EmptyCoroutineContext, + scheduler: SimulationScheduler = SimulationScheduler() +): SimulationCoroutineScope { + var safeContext = context + if (context[ContinuationInterceptor] == null) safeContext += SimulationCoroutineDispatcher(scheduler) + return SimulationCoroutineScopeImpl(safeContext) +} + +private inline val CoroutineContext.simulationController: SimulationController + get() { + val handler = this[ContinuationInterceptor] + return handler as? SimulationController ?: throw IllegalArgumentException( + "SimulationCoroutineScope requires a SimulationController such as SimulatorCoroutineDispatcher as " + + "the ContinuationInterceptor (Dispatcher)" + ) + } -- cgit v1.2.3