From 00ac59e8e9d6a41c2eac55aa25420dce8fa9c6e0 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 9 Nov 2022 21:24:08 +0000 Subject: refactor(sim/core): Re-implement SimulationScheduler as Dispatcher This change updates the `SimulationScheduler` class to implement the `Dispatcher` interface from the OpenDC Common module, so that OpenDC modules only need to depend on the common module for dispatching future task (possibly in simulation). --- .../opendc-simulator-core/build.gradle.kts | 1 + .../org/opendc/simulator/SimulationDispatcher.java | 243 ++++++++++++++++++++ .../org/opendc/simulator/SimulationScheduler.java | 247 --------------------- .../opendc/simulator/kotlin/SimulationBuilders.kt | 74 ++++-- .../simulator/kotlin/SimulationController.kt | 34 ++- .../kotlin/SimulationCoroutineDispatcher.kt | 98 -------- .../simulator/kotlin/SimulationCoroutineScope.kt | 36 ++- .../opendc/simulator/SimulationDispatcherTest.kt | 107 +++++++++ .../opendc/simulator/SimulationSchedulerTest.kt | 106 --------- .../simulator/kotlin/SimulationBuildersTest.kt | 98 ++++++++ 10 files changed, 545 insertions(+), 499 deletions(-) create mode 100644 opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationDispatcher.java delete mode 100644 opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java delete 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/test/kotlin/org/opendc/simulator/SimulationDispatcherTest.kt delete mode 100644 opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt create mode 100644 opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/kotlin/SimulationBuildersTest.kt (limited to 'opendc-simulator/opendc-simulator-core') diff --git a/opendc-simulator/opendc-simulator-core/build.gradle.kts b/opendc-simulator/opendc-simulator-core/build.gradle.kts index 0de96a8e..0ae95d42 100644 --- a/opendc-simulator/opendc-simulator-core/build.gradle.kts +++ b/opendc-simulator/opendc-simulator-core/build.gradle.kts @@ -28,5 +28,6 @@ plugins { } dependencies { + api(projects.opendc.opendcCommon) api(libs.kotlinx.coroutines) } diff --git a/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationDispatcher.java b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationDispatcher.java new file mode 100644 index 00000000..8c74aacf --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationDispatcher.java @@ -0,0 +1,243 @@ +/* + * Copyright (c) 2022 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; + +import java.time.Instant; +import java.time.InstantSource; +import org.opendc.common.Dispatcher; +import org.opendc.common.DispatcherHandle; + +/** + * A {@link Dispatcher} used by simulations to manage execution of (future) tasks, providing a controllable (virtual) + * clock to skip over delays. + * + *

+ * The dispatcher can be queried to advance the time (via {@link #advanceBy}), run all the scheduled tasks advancing the + * virtual time as needed (via {@link #advanceUntilIdle}), or run the tasks that are scheduled to run as soon as + * possible but have not yet been dispatched (via {@link #runCurrent}). These methods execute the pending tasks using + * a single thread. + * + *

+ * This class is not thread-safe and must not be used concurrently by multiple threads. + */ +public final class SimulationDispatcher implements Dispatcher { + /** + * The {@link TaskQueue} containing the pending tasks. + */ + private final TaskQueue queue = new TaskQueue(); + + /** + * The current time of the scheduler in milliseconds since epoch. + */ + private long currentTime; + + /** + * A counter to establish total order on the events that happen at the same virtual time. + */ + private int count = 0; + + /** + * The {@link InstantSource} instance linked to this scheduler. + */ + private final SimulationClock timeSource = new SimulationClock(this); + + /** + * Construct a {@link SimulationDispatcher} instance with the specified initial time. + * + * @param initialTimeMs The initial virtual time of the scheduler in milliseconds since epoch. + */ + public SimulationDispatcher(long initialTimeMs) { + this.currentTime = initialTimeMs; + } + + /** + * Construct a {@link SimulationDispatcher} instance with the initial time set to UNIX Epoch 0. + */ + public SimulationDispatcher() { + this(0); + } + + /** + * Return the current virtual timestamp of the dispatcher (in milliseconds since epoch). + * + * @return A long value representing the virtual timestamp of the dispatcher in milliseconds since epoch. + */ + public long getCurrentTime() { + return currentTime; + } + + /** + * Return the virtual time source associated with this dispatcher. + * + * @return A {@link InstantSource} tracking the virtual time of the dispatcher. + */ + @Override + public InstantSource getTimeSource() { + return timeSource; + } + + @Override + public void schedule(long delayMs, Runnable command) { + internalSchedule(delayMs, command); + } + + @Override + public DispatcherHandle scheduleCancellable(long delayMs, Runnable command) { + long target = currentTime + delayMs; + if (target < 0) { + target = Long.MAX_VALUE; + } + + long deadline = target; + int id = internalSchedule(delayMs, command); + return () -> internalCancel(deadline, id); + } + + /** + * Run the enqueued tasks in the specified order, advancing the virtual time as needed until there are no more + * tasks in the queue of this scheduler. + */ + public void advanceUntilIdle() { + final TaskQueue queue = this.queue; + + while (true) { + long deadline = queue.peekDeadline(); + Runnable task = queue.poll(); + + if (task == null) { + break; + } + + currentTime = deadline; + task.run(); + } + } + + /** + * Move the virtual clock of this dispatcher forward by the specified amount, running the scheduled tasks in the + * meantime. + * + * @param delayMs The amount of time to move the virtual clock forward (in milliseconds). + * @throws IllegalStateException if passed a negative delay. + */ + public void advanceBy(long delayMs) { + if (delayMs < 0) { + throw new IllegalArgumentException("Can not advance time by a negative delay: " + delayMs + " ms"); + } + + long target = currentTime + delayMs; + if (target < 0) { + target = Long.MAX_VALUE; + } + + final TaskQueue queue = this.queue; + long deadline; + + while ((deadline = queue.peekDeadline()) < target) { + Runnable task = queue.poll(); // Cannot be null since while condition is always false on an empty queue + + task.run(); + currentTime = deadline; + } + + currentTime = target; + } + + /** + * Execute the tasks that are scheduled to execute at this moment of virtual time. + */ + public void runCurrent() { + final TaskQueue queue = this.queue; + long currentTime = this.currentTime; + + while (queue.peekDeadline() == currentTime) { + Runnable task = queue.poll(); + + if (task == null) { + break; + } + + task.run(); + } + } + + /** + * Schedule a task that executes after the specified delayMs. + * + * @param delayMs The time from now until the execution of the task (in milliseconds). + * @param task The task to execute after the delay. + * @return The identifier of the task that can be used together with the timestamp of the task to cancel it. + */ + private int internalSchedule(long delayMs, Runnable task) { + if (delayMs < 0) { + throw new IllegalArgumentException( + "Attempted scheduling an event earlier in time (delay " + delayMs + " ms)"); + } + + long target = currentTime + delayMs; + if (target < 0) { + target = Long.MAX_VALUE; + } + + int id = count++; + queue.add(target, id, task); + return id; + } + + /** + * Cancel a pending task. + * + * @param deadline The deadline of the task. + * @param id The identifier of the task (returned by {@link #internalSchedule(long, Runnable)}). + * @return A boolean indicating whether a task was actually cancelled. + */ + private boolean internalCancel(long deadline, int id) { + return queue.remove(deadline, id); + } + + /** + * A {@link InstantSource} implementation for a {@link SimulationDispatcher}. + */ + private static class SimulationClock implements InstantSource { + private final SimulationDispatcher dispatcher; + + SimulationClock(SimulationDispatcher dispatcher) { + this.dispatcher = dispatcher; + } + + @Override + public Instant instant() { + return Instant.ofEpochMilli(dispatcher.currentTime); + } + + @Override + public long millis() { + return dispatcher.currentTime; + } + + @Override + public String toString() { + return "SimulationDispatcher.InstantSource[time=" + millis() + "ms]"; + } + } +} diff --git a/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java b/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java deleted file mode 100644 index 305bdf5e..00000000 --- a/opendc-simulator/opendc-simulator-core/src/main/java/org/opendc/simulator/SimulationScheduler.java +++ /dev/null @@ -1,247 +0,0 @@ -/* - * Copyright (c) 2022 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; - -import java.time.Clock; -import java.time.Instant; -import java.time.ZoneId; -import java.util.concurrent.Executor; - -/** - * A scheduler is used by simulations to manage execution of (future) tasks, providing a controllable (virtual) clock to - * skip over delays. - * - *

- * The scheduler can be queried to advance the time (via {@link #advanceBy}), run all the scheduled tasks advancing the - * virtual time as needed (via {@link #advanceUntilIdle}), or run the tasks that are scheduled to run as soon as - * possible but have not yet been dispatched (via {@link #runCurrent}). These methods execute the pending tasks using - * a single thread. - * - *

- * This class is not thread-safe and must not be used concurrently by multiple threads. - */ -public final class SimulationScheduler implements Executor { - /** - * The {@link TaskQueue} containing the pending tasks. - */ - private final TaskQueue queue = new TaskQueue(); - - /** - * The current time of the scheduler in milliseconds since epoch. - */ - private long currentTime; - - /** - * A counter to establish total order on the events that happen at the same virtual time. - */ - private int count = 0; - - /** - * The {@link Clock} instance linked to this scheduler. - */ - private final SimulationClock clock = new SimulationClock(this, ZoneId.systemDefault()); - - /** - * Construct a {@link SimulationScheduler} instance with the specified initial time. - * - * @param initialTimeMs The initial virtual time of the scheduler in milliseconds since epoch. - */ - public SimulationScheduler(long initialTimeMs) { - this.currentTime = initialTimeMs; - } - - /** - * Construct a {@link SimulationScheduler} instance with the initial time set to UNIX Epoch 0. - */ - public SimulationScheduler() { - this(0); - } - - /** - * Return the virtual clock associated with this dispatcher. - * - * @return A {@link Clock} tracking the virtual time of the dispatcher. - */ - public Clock getClock() { - return clock; - } - - /** - * Return the current virtual timestamp of the dispatcher (in milliseconds since epoch). - * - * @return A long value representing the virtual timestamp of the dispatcher in milliseconds since epoch. - */ - public long getCurrentTime() { - return currentTime; - } - - /** - * Schedule a task that executes after the specified delayMs. - * - * @param delayMs The time from now until the execution of the task (in milliseconds). - * @param task The task to execute after the delay. - * @return The identifier of the task that can be used together with the timestamp of the task to cancel it. - */ - public int schedule(long delayMs, Runnable task) { - if (delayMs < 0) { - throw new IllegalArgumentException( - "Attempted scheduling an event earlier in time (delay " + delayMs + " ms)"); - } - - long target = currentTime + delayMs; - if (target < 0) { - target = Long.MAX_VALUE; - } - - int id = count++; - queue.add(target, id, task); - return id; - } - - /** - * Cancel a pending task. - * - * @param deadline The deadline of the task. - * @param id The identifier of the task (returned by {@link #schedule(long, Runnable)}). - * @return A boolean indicating whether a task was actually cancelled. - */ - public boolean cancel(long deadline, int id) { - return queue.remove(deadline, id); - } - - /** - * Run the enqueued tasks in the specified order, advancing the virtual time as needed until there are no more - * tasks in the queue of this scheduler. - */ - public void advanceUntilIdle() { - final TaskQueue queue = this.queue; - - while (true) { - long deadline = queue.peekDeadline(); - Runnable task = queue.poll(); - - if (task == null) { - break; - } - - currentTime = deadline; - task.run(); - } - } - - /** - * Move the virtual clock of this dispatcher forward by the specified amount, running the scheduled tasks in the - * meantime. - * - * @param delayMs The amount of time to move the virtual clock forward (in milliseconds). - * @throws IllegalStateException if passed a negative delay. - */ - public void advanceBy(long delayMs) { - if (delayMs < 0) { - throw new IllegalArgumentException("Can not advance time by a negative delay: " + delayMs + " ms"); - } - - long target = currentTime + delayMs; - if (target < 0) { - target = Long.MAX_VALUE; - } - - final TaskQueue queue = this.queue; - long deadline; - - while ((deadline = queue.peekDeadline()) < target) { - Runnable task = queue.poll(); // Cannot be null since while condition is always false on an empty queue - - task.run(); - currentTime = deadline; - } - - currentTime = target; - } - - /** - * Execute the tasks that are scheduled to execute at this moment of virtual time. - */ - public void runCurrent() { - final TaskQueue queue = this.queue; - long currentTime = this.currentTime; - - while (queue.peekDeadline() == currentTime) { - Runnable task = queue.poll(); - - if (task == null) { - break; - } - - task.run(); - } - } - - /** - * Schedule the specified command to run at this moment of virtual time. - * - * @param command The command to execute. - */ - @Override - public void execute(Runnable command) { - schedule(0, command); - } - - /** - * A {@link Clock} implementation for a {@link SimulationScheduler}. - */ - private static class SimulationClock extends Clock { - private final SimulationScheduler scheduler; - private final ZoneId zone; - - SimulationClock(SimulationScheduler scheduler, ZoneId zone) { - this.scheduler = scheduler; - this.zone = zone; - } - - @Override - public ZoneId getZone() { - return zone; - } - - @Override - public Clock withZone(ZoneId zoneId) { - return new SimulationClock(scheduler, zone); - } - - @Override - public Instant instant() { - return Instant.ofEpochMilli(scheduler.currentTime); - } - - @Override - public long millis() { - return scheduler.currentTime; - } - - @Override - public String toString() { - return "SimulationClock[time=" + millis() + "ms]"; - } - } -} 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 index 882a0fc5..6e568137 100644 --- 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 @@ -22,11 +22,18 @@ package org.opendc.simulator.kotlin +import kotlinx.coroutines.CoroutineDispatcher import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable.children import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.async -import org.opendc.simulator.SimulationScheduler +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import org.opendc.common.DispatcherProvider +import org.opendc.common.asCoroutineDispatcher +import org.opendc.simulator.SimulationDispatcher +import java.time.InstantSource import kotlin.coroutines.ContinuationInterceptor import kotlin.coroutines.CoroutineContext import kotlin.coroutines.EmptyCoroutineContext @@ -54,16 +61,16 @@ import kotlin.coroutines.EmptyCoroutineContext * The simulation is run in a single thread, unless other [CoroutineDispatcher] are used for child coroutines. * Because of this, child coroutines are not executed in parallel to [body]. * In order for the spawned-off asynchronous code to actually be executed, one must either [yield] or suspend the - * body some other way, or use commands that control scheduling (see [SimulationScheduler]). + * body some other way, or use commands that control scheduling (see [SimulationDispatcher]). */ @OptIn(ExperimentalCoroutinesApi::class) public fun runSimulation( context: CoroutineContext = EmptyCoroutineContext, - scheduler: SimulationScheduler = SimulationScheduler(), + scheduler: SimulationDispatcher = SimulationDispatcher(), body: suspend SimulationCoroutineScope.() -> Unit ) { - val (safeContext, dispatcher) = context.checkArguments(scheduler) - val startingJobs = safeContext.activeJobs() + val (safeContext, job, dispatcher) = context.checkArguments(scheduler) + val startingJobs = job.activeJobs() val scope = SimulationCoroutineScope(safeContext) val deferred = scope.async { body(scope) @@ -72,7 +79,7 @@ public fun runSimulation( deferred.getCompletionExceptionOrNull()?.let { throw it } - val endingJobs = safeContext.activeJobs() + val endingJobs = job.activeJobs() if ((endingJobs - startingJobs).isNotEmpty()) { throw IllegalStateException("Test finished with active jobs: $endingJobs") } @@ -82,24 +89,51 @@ public fun runSimulation( * Convenience method for calling [runSimulation] on an existing [SimulationCoroutineScope]. */ public fun SimulationCoroutineScope.runSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit = - runSimulation(coroutineContext, scheduler, block) + runSimulation(coroutineContext, dispatcher, block) + +private fun CoroutineContext.checkArguments(scheduler: SimulationDispatcher): Triple { + val job = get(Job) ?: SupervisorJob() + val dispatcher = get(ContinuationInterceptor) ?: scheduler.asCoroutineDispatcher() + val simulationDispatcher = dispatcher.asSimulationDispatcher() + return Triple(this + dispatcher + job, job, simulationDispatcher.asController()) +} + +private fun Job.activeJobs(): Set { + return children.filter { it.isActive }.toSet() +} /** - * Convenience method for calling [runSimulation] on an existing [SimulationCoroutineDispatcher]. + * Convert a [ContinuationInterceptor] into a [SimulationDispatcher] if possible. */ -public fun SimulationCoroutineDispatcher.runSimulation(block: suspend SimulationCoroutineScope.() -> Unit): Unit = - runSimulation(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) - } +internal fun ContinuationInterceptor.asSimulationDispatcher(): SimulationDispatcher { + val provider = this as? DispatcherProvider ?: throw IllegalArgumentException( + "DispatcherProvider such as SimulatorCoroutineDispatcher as the ContinuationInterceptor(Dispatcher) is required" + ) - val job = get(Job) ?: SupervisorJob() - return Pair(this + dispatcher + job, dispatcher as SimulationController) + return provider.dispatcher as? SimulationDispatcher ?: throw IllegalArgumentException("Active dispatcher is not a SimulationDispatcher") } -private fun CoroutineContext.activeJobs(): Set { - return checkNotNull(this[Job]).children.filter { it.isActive }.toSet() +/** + * Helper method to convert a [SimulationDispatcher] into a [SimulationController]. + */ +internal fun SimulationDispatcher.asController(): SimulationController { + return object : SimulationController { + override val dispatcher: SimulationDispatcher + get() = this@asController + + override val timeSource: InstantSource + get() = this@asController.timeSource + + override fun advanceUntilIdle() { + dispatcher.advanceUntilIdle() + } + + override fun advanceBy(delayMs: Long) { + dispatcher.advanceBy(delayMs) + } + + override fun runCurrent() { + dispatcher.runCurrent() + } + } } 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 index f96b2326..f7470ad9 100644 --- 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 @@ -23,30 +23,48 @@ package org.opendc.simulator.kotlin import kotlinx.coroutines.CoroutineDispatcher -import org.opendc.simulator.SimulationScheduler -import java.time.Clock +import org.opendc.simulator.SimulationDispatcher +import java.time.InstantSource /** - * Control the virtual clock of a [CoroutineDispatcher]. + * Interface to 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 + public val timeSource: InstantSource /** - * The [SimulationScheduler] driving the simulation. + * The current virtual timestamp of the dispatcher (in milliseconds since epoch). */ - public val scheduler: SimulationScheduler + public val currentTime: Long + get() = timeSource.millis() + + /** + * Return the [SimulationDispatcher] driving the simulation. + */ + public val dispatcher: SimulationDispatcher /** * 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. + */ + public fun advanceUntilIdle() + + /** + * Move the virtual clock of this dispatcher forward by the specified amount, running the scheduled tasks in the + * meantime. * - * @return the amount of delay-time that this Dispatcher's clock has been forwarded in milliseconds. + * @param delayMs The amount of time to move the virtual clock forward (in milliseconds). + * @throws IllegalStateException if passed a negative delay. + */ + public fun advanceBy(delayMs: Long) + + /** + * Execute the tasks that are scheduled to execute at this moment of virtual time. */ - public fun advanceUntilIdle(): Long + public fun runCurrent() } 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 deleted file mode 100644 index cacbbbf7..00000000 --- a/opendc-simulator/opendc-simulator-core/src/main/kotlin/org/opendc/simulator/kotlin/SimulationCoroutineDispatcher.kt +++ /dev/null @@ -1,98 +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.kotlin - -import kotlinx.coroutines.CancellableContinuation -import kotlinx.coroutines.CoroutineDispatcher -import kotlinx.coroutines.Delay -import kotlinx.coroutines.DisposableHandle -import kotlinx.coroutines.ExperimentalCoroutinesApi -import kotlinx.coroutines.InternalCoroutinesApi -import org.opendc.simulator.SimulationScheduler -import java.lang.Runnable -import java.time.Clock -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 index 6be8e67a..ca49fc53 100644 --- 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 @@ -24,7 +24,8 @@ package org.opendc.simulator.kotlin import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope -import org.opendc.simulator.SimulationScheduler +import org.opendc.common.asCoroutineDispatcher +import org.opendc.simulator.SimulationDispatcher import kotlin.coroutines.ContinuationInterceptor import kotlin.coroutines.CoroutineContext import kotlin.coroutines.EmptyCoroutineContext @@ -34,33 +35,28 @@ import kotlin.coroutines.EmptyCoroutineContext */ 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. + * scope adds a dispatcher automatically. */ -@Suppress("FunctionName") public fun SimulationCoroutineScope( context: CoroutineContext = EmptyCoroutineContext, - scheduler: SimulationScheduler = SimulationScheduler() + scheduler: SimulationDispatcher = SimulationDispatcher() ): SimulationCoroutineScope { var safeContext = context - if (context[ContinuationInterceptor] == null) safeContext += SimulationCoroutineDispatcher(scheduler) - return SimulationCoroutineScopeImpl(safeContext) -} + val simulationDispatcher: SimulationDispatcher + val interceptor = context[ContinuationInterceptor] -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)" - ) + if (interceptor != null) { + simulationDispatcher = interceptor.asSimulationDispatcher() + } else { + simulationDispatcher = scheduler + safeContext += scheduler.asCoroutineDispatcher() } + + return object : SimulationCoroutineScope, SimulationController by simulationDispatcher.asController() { + override val coroutineContext: CoroutineContext = safeContext + } +} diff --git a/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationDispatcherTest.kt b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationDispatcherTest.kt new file mode 100644 index 00000000..600102be --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationDispatcherTest.kt @@ -0,0 +1,107 @@ +/* + * Copyright (c) 2022 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 + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import java.time.Instant + +/** + * Test suite for the [SimulationDispatcher] class. + */ +class SimulationDispatcherTest { + /** + * Test the basic functionality of [SimulationDispatcher.runCurrent]. + */ + @Test + fun testRunCurrent() { + val scheduler = SimulationDispatcher() + var count = 0 + + scheduler.schedule(1) { count += 1 } + scheduler.schedule(2) { count += 1 } + + scheduler.advanceBy(1) + assertEquals(0, count) + scheduler.runCurrent() + assertEquals(1, count) + scheduler.advanceBy(1) + assertEquals(1, count) + scheduler.runCurrent() + assertEquals(2, count) + assertEquals(2, scheduler.currentTime) + + scheduler.advanceBy(Long.MAX_VALUE) + scheduler.runCurrent() + assertEquals(Long.MAX_VALUE, scheduler.currentTime) + } + + /** + * Test the clock of the [SimulationDispatcher]. + */ + @Test + fun testClock() { + val scheduler = SimulationDispatcher() + var count = 0 + + scheduler.schedule(1) { count += 1 } + scheduler.schedule(2) { count += 1 } + + scheduler.advanceBy(2) + assertEquals(2, scheduler.currentTime) + assertEquals(2, scheduler.timeSource.millis()) + assertEquals(Instant.ofEpochMilli(2), scheduler.timeSource.instant()) + } + + /** + * Test large delays. + */ + @Test + fun testAdvanceByLargeDelays() { + val scheduler = SimulationDispatcher() + var count = 0 + + scheduler.schedule(1) { count += 1 } + + scheduler.advanceBy(10) + + scheduler.schedule(Long.MAX_VALUE) { count += 1 } + scheduler.scheduleCancellable(Long.MAX_VALUE) { count += 1 } + scheduler.schedule(100_000_000) { count += 1 } + + scheduler.advanceUntilIdle() + assertEquals(4, count) + } + + /** + * Test negative delays. + */ + @Test + fun testNegativeDelays() { + val scheduler = SimulationDispatcher() + + assertThrows { scheduler.schedule(-100) { } } + assertThrows { scheduler.advanceBy(-100) } + } +} diff --git a/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt deleted file mode 100644 index eca3b582..00000000 --- a/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/SimulationSchedulerTest.kt +++ /dev/null @@ -1,106 +0,0 @@ -/* - * Copyright (c) 2022 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 - -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Test -import org.junit.jupiter.api.assertThrows -import java.time.Instant - -/** - * Test suite for the [SimulationScheduler] class. - */ -class SimulationSchedulerTest { - /** - * Test the basic functionality of [SimulationScheduler.runCurrent]. - */ - @Test - fun testRunCurrent() { - val scheduler = SimulationScheduler() - var count = 0 - - scheduler.schedule(1) { count += 1 } - scheduler.schedule(2) { count += 1 } - - scheduler.advanceBy(1) - assertEquals(0, count) - scheduler.runCurrent() - assertEquals(1, count) - scheduler.advanceBy(1) - assertEquals(1, count) - scheduler.runCurrent() - assertEquals(2, count) - assertEquals(2, scheduler.currentTime) - - scheduler.advanceBy(Long.MAX_VALUE) - scheduler.runCurrent() - assertEquals(Long.MAX_VALUE, scheduler.currentTime) - } - - /** - * Test the clock of the [SimulationScheduler]. - */ - @Test - fun testClock() { - val scheduler = SimulationScheduler() - var count = 0 - - scheduler.schedule(1) { count += 1 } - scheduler.schedule(2) { count += 1 } - - scheduler.advanceBy(2) - assertEquals(2, scheduler.currentTime) - assertEquals(2, scheduler.clock.millis()) - assertEquals(Instant.ofEpochMilli(2), scheduler.clock.instant()) - } - - /** - * Test large delays. - */ - @Test - fun testAdvanceByLargeDelays() { - val scheduler = SimulationScheduler() - var count = 0 - - scheduler.schedule(1) { count += 1 } - - scheduler.advanceBy(10) - - scheduler.schedule(Long.MAX_VALUE) { count += 1 } - scheduler.schedule(100_000_000) { count += 1 } - - scheduler.advanceUntilIdle() - assertEquals(3, count) - } - - /** - * Test negative delays. - */ - @Test - fun testNegativeDelays() { - val scheduler = SimulationScheduler() - - assertThrows { scheduler.schedule(-100) { } } - assertThrows { scheduler.advanceBy(-100) } - } -} diff --git a/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/kotlin/SimulationBuildersTest.kt b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/kotlin/SimulationBuildersTest.kt new file mode 100644 index 00000000..26419a50 --- /dev/null +++ b/opendc-simulator/opendc-simulator-core/src/test/kotlin/org/opendc/simulator/kotlin/SimulationBuildersTest.kt @@ -0,0 +1,98 @@ +/* + * Copyright (c) 2022 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.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.suspendCancellableCoroutine +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows + +/** + * Test suite for the Kotlin simulation builders. + */ +class SimulationBuildersTest { + @Test + fun testDelay() = runSimulation { + assertEquals(0, currentTime) + delay(100) + assertEquals(100, currentTime) + } + + @Test + fun testController() = runSimulation { + var completed = false + + launch { + delay(20) + completed = true + } + + advanceBy(10) + assertFalse(completed) + advanceBy(11) + assertTrue(completed) + + completed = false + launch { completed = true } + runCurrent() + assertTrue(completed) + } + + @Test + fun testFailOnActiveJobs() { + assertThrows { + runSimulation { + launch { suspendCancellableCoroutine {} } + } + } + } + + @Test + fun testPropagateException() { + assertThrows { + runSimulation { + throw IllegalStateException("Test") + } + } + } + + @Test + fun testInvalidDispatcher() { + assertThrows { + runSimulation(Dispatchers.Default) { } + } + } + + @Test + fun testExistingJob() { + runSimulation(Job()) { + delay(10) + } + } +} -- cgit v1.2.3