From b1c4d1f94e35445bdba5a56b614d0ec28a332624 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Thu, 11 Jan 2018 16:27:05 +0100 Subject: refactor(#18): Redesign core simulation API This change contains the redesign of the core simulation API and provides a cleaner interface for developing simulation models for the users. --- .../com/atlarge/opendc/omega/MessageContainer.kt | 48 ++++ .../kotlin/com/atlarge/opendc/omega/OmegaKernel.kt | 315 +++++++++++++++++++++ .../com/atlarge/opendc/omega/OmegaKernelFactory.kt | 47 +++ .../main/kotlin/com/atlarge/opendc/omega/Resume.kt | 39 +++ .../kotlin/com/atlarge/opendc/omega/Timeout.kt | 35 +++ .../kotlin/com/atlarge/opendc/omega/SmokeTest.kt | 116 ++++++++ 6 files changed, 600 insertions(+) create mode 100644 opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/MessageContainer.kt create mode 100644 opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernel.kt create mode 100644 opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernelFactory.kt create mode 100644 opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Resume.kt create mode 100644 opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Timeout.kt create mode 100644 opendc-kernel-omega/src/test/kotlin/com/atlarge/opendc/omega/SmokeTest.kt (limited to 'opendc-kernel-omega/src') diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/MessageContainer.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/MessageContainer.kt new file mode 100644 index 00000000..af13d1fd --- /dev/null +++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/MessageContainer.kt @@ -0,0 +1,48 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 com.atlarge.opendc.omega + +import com.atlarge.opendc.simulator.Instant +import com.atlarge.opendc.simulator.Entity +import com.atlarge.opendc.simulator.Envelope + +/** + * A wrapper around a message that has been scheduled for processing. + * + * @property message The message to wrap. + * @property time The point in time to deliver the message. + * @property sender The sender of the message. + * @property destination The destination of the message. + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +internal data class MessageContainer(override val message: Any, + val time: Instant, + override val sender: Entity<*, *>?, + override val destination: Entity<*, *>) : Envelope { + /** + * A flag to indicate the message has been canceled. + */ + internal var canceled: Boolean = false +} diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernel.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernel.kt new file mode 100644 index 00000000..fb5ce24b --- /dev/null +++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernel.kt @@ -0,0 +1,315 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 com.atlarge.opendc.omega + +import com.atlarge.opendc.simulator.* +import com.atlarge.opendc.simulator.kernel.Kernel +import com.atlarge.opendc.simulator.Bootstrap +import mu.KotlinLogging +import java.util.* +import kotlin.coroutines.experimental.* + +/** + * The Omega simulation kernel is the reference simulation kernel implementation for the OpenDC Simulator core. + * + * This simulator implementation is a single-threaded implementation, running simulation kernels synchronously and + * provides a single priority queue for all events (messages, ticks, etc) that occur in the entities. + * + * @property model The model that is simulated. + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +internal class OmegaKernel(bootstrap: Bootstrap) : Kernel, Bootstrap.Context { + /** + * The logger instance to use for the simulator. + */ + private val logger = KotlinLogging.logger {} + + /** + * The registry of the simulation kernels used in the experiment. + */ + private val registry: MutableMap, OmegaContext<*>> = HashMap() + + /** + * The message queue. + */ + private val queue: Queue = PriorityQueue(Comparator.comparingLong(MessageContainer::time)) + + /** + * The simulation time. + */ + override var time: Instant = 0 + + /** + * The model of simulation. + */ + override val model: M = bootstrap.bootstrap(this) + + override val , S> E.state: S + get() = context?.state ?: initialState + + /** + * The context associated with an [Entity]. + */ + @Suppress("UNCHECKED_CAST") + private val , S, M> E.context: OmegaContext? + get() = registry[this] as? OmegaContext + + override fun register(entity: Entity<*, M>): Boolean { + if (!registry.containsKey(entity) && entity !is Process) { + return false + } + + @Suppress("UNCHECKED_CAST") + val process = entity as Process + val context = OmegaContext(entity).also { registry.put(entity, it) } + + // Bootstrap the process coroutine + val block: suspend () -> Unit = { process.run { context.run() } } + block.startCoroutine(context) + + return true + } + + override fun deregister(entity: Entity<*, M>): Boolean { + val context = entity.context ?: return false + context.resume(Unit) + return true + } + + override fun schedule(message: Any, destination: Entity<*, *>, sender: Entity<*, *>?, delay: Duration) = + schedule(prepare(message, destination, sender, delay)) + + override fun step() { + while (true) { + val envelope = queue.peek() ?: return + val delivery = envelope.time + + if (delivery > time) { + // Tick has yet to occur + // Jump in time to next event + time = delivery + break + } else if (delivery < time) { + // Tick has already occurred + logger.warn { "message processed out of order" } + } + + queue.poll() + + // If the sender has canceled the message, we move on to the next message + if (envelope.canceled) { + continue + } + + val context = envelope.destination.context ?: continue + + if (envelope.message !is Interrupt) { + context.continuation.resume(envelope) + } else { + context.continuation.resumeWithException(envelope.message) + } + + context.last = time + } + } + + + override fun run() { + while (queue.isNotEmpty()) { + step() + } + } + + override fun run(until: Instant) { + require(until > 0) { "The given instant must be a non-zero positive number" } + + if (time >= until) { + return + } + + while (time < until && queue.isNotEmpty()) { + step() + } + + // Fix clock if step() jumped too far in time to give the impression to the user that simulation stopped at + // exactly the tick it gave. This has not effect on the actual simulation results as the next call to run() will + // just jump forward again. + if (time > until) { + time = until + } + } + + private fun schedule(envelope: MessageContainer) { + queue.add(envelope) + } + + private fun prepare(message: Any, destination: Entity<*, *>, sender: Entity<*, *>? = null, + delay: Duration): MessageContainer { + require(delay >= 0) { "The amount of time to delay the message must be a positive number" } + return MessageContainer(message, time + delay, sender, destination) + } + + /** + * This internal class provides the default implementation for the [Context] interface for this simulator. + */ + private inner class OmegaContext(val process: Process) : Context, Continuation { + /** + * The continuation to resume the execution of the process. + */ + lateinit var continuation: Continuation> + + /** + * The last point in time the process has done some work. + */ + var last: Instant = -1 + + /** + * The model in which the process exists. + */ + override val model: M + get() = this@OmegaKernel.model + + /** + * The state of the entity. + */ + override var state: S = process.initialState + + /** + * The current point in simulation time. + */ + override val time: Instant + get() = this@OmegaKernel.time + + /** + * The duration between the current point in simulation time and the last point in simulation time where the + * [Context] has executed some work. + */ + override val delta: Duration + get() = maxOf(time - last, 0) + + /** + * The [CoroutineContext] for a [Context]. + */ + override val context: CoroutineContext = EmptyCoroutineContext + + /** + * The observable state of an [Entity] within the simulation is provided by the context of the simulation. + */ + override val , S> T.state: S + get() = context?.state ?: initialState + + /** + * Retrieve and remove and single message from the mailbox of the [Entity] and suspend the [Context] until the + * message has been received. + * + * @return The envelope containing the message. + */ + suspend fun receiveEnvelope(): Envelope<*> = suspendCoroutine { continuation = it } + + suspend override fun receive(transform: suspend Envelope<*>.(Any) -> T): T { + val envelope = receiveEnvelope() + return transform(envelope, envelope.message) + } + + + suspend override fun receive(timeout: Duration, transform: suspend Envelope<*>.(Any) -> T): T? { + val send = prepare(Timeout, process, process, timeout).also { schedule(it) } + + try { + val received = receiveEnvelope() + + if (received.message !is Timeout) { + send.canceled = true + return transform(received, received.message) + } + + return null + } finally { + send.canceled = true + } + } + + suspend override fun Entity<*, *>.send(msg: Any, delay: Duration) = send(msg, process, delay) + + suspend override fun Entity<*, *>.send(msg: Any, sender: Entity<*, *>, delay: Duration) = + schedule(prepare(msg, sender, delay = delay)) + + suspend override fun Entity<*, *>.interrupt() = send(Interrupt) + + suspend override fun hold(duration: Duration) { + require(duration >= 0) { "The amount of time to hold must be a positive number" } + val envelope = prepare(Resume, process, process, duration).also { schedule(it) } + + try { + while (true) { + if (receive() is Resume) + return + } + } finally { + envelope.canceled = true + } + } + + suspend override fun hold(duration: Duration, queue: Queue) { + require(duration >= 0) { "The amount of time to hold must be a positive number" } + val envelope = prepare(Resume, process, process, duration).also { schedule(it) } + + try { + while (true) { + val msg = receive() + if (msg is Resume) + return + queue.add(msg) + } + } finally { + envelope.canceled = true + } + } + + + // Completion continuation implementation + /** + * Resume the execution of this continuation with the given value. + * + * @param value The value to resume with. + */ + override fun resume(value: Unit) { + // Deregister process from registry in order to have the GC collect this context + registry.remove(process) + } + + /** + * Resume the execution of this continuation with an exception. + * + * @param exception The exception to resume with. + */ + override fun resumeWithException(exception: Throwable) { + // Deregister process from registry in order to have the GC collect this context:w + registry.remove(process) + + logger.error(exception) { "An exception occurred during the execution of a process" } + } + } +} diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernelFactory.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernelFactory.kt new file mode 100644 index 00000000..dcad4dce --- /dev/null +++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/OmegaKernelFactory.kt @@ -0,0 +1,47 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 com.atlarge.opendc.omega + +import com.atlarge.opendc.simulator.kernel.KernelFactory +import com.atlarge.opendc.simulator.kernel.Kernel +import com.atlarge.opendc.simulator.Bootstrap + +/** + * The Omega simulation kernel is the reference simulation kernel implementation for the OpenDC Simulator core. + * + * This simulator implementation is a single-threaded implementation, running simulation kernels synchronously and + * provides a single priority queue for all events (messages, ticks, etc) that occur in the entities. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +object OmegaKernelFactory : KernelFactory { + /** + * Create a simulation over the given model facilitated by this simulation kernel. + * + * @param bootstrap The bootstrap procedure to bootstrap the simulation with. + * @return A [Kernel] instance to control the simulation. + */ + override fun create(bootstrap: Bootstrap): Kernel = OmegaKernel(bootstrap) +} diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Resume.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Resume.kt new file mode 100644 index 00000000..d4bd8536 --- /dev/null +++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Resume.kt @@ -0,0 +1,39 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 com.atlarge.opendc.omega + +import com.atlarge.opendc.simulator.Context +import com.atlarge.opendc.simulator.Process + +/** + * An internal message used by the Omega simulation kernel to indicate to a suspended [Process], that it should wake up + * and resume execution. + * + * This message is not guaranteed to work on other simulation kernels and [Context.interrupt] should be preferred to + * wake up a process from another entity. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +object Resume diff --git a/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Timeout.kt b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Timeout.kt new file mode 100644 index 00000000..c205f6b5 --- /dev/null +++ b/opendc-kernel-omega/src/main/kotlin/com/atlarge/opendc/omega/Timeout.kt @@ -0,0 +1,35 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 com.atlarge.opendc.omega + +import com.atlarge.opendc.simulator.Process + +/** + * An internal message used by the Omega simulation kernel to indicate to a suspended [Process], that a timeout has been + * reached and that it should wake up and resume execution. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +object Timeout diff --git a/opendc-kernel-omega/src/test/kotlin/com/atlarge/opendc/omega/SmokeTest.kt b/opendc-kernel-omega/src/test/kotlin/com/atlarge/opendc/omega/SmokeTest.kt new file mode 100644 index 00000000..b358d618 --- /dev/null +++ b/opendc-kernel-omega/src/test/kotlin/com/atlarge/opendc/omega/SmokeTest.kt @@ -0,0 +1,116 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 com.atlarge.opendc.omega + +import com.atlarge.opendc.simulator.Context +import com.atlarge.opendc.simulator.Process +import com.atlarge.opendc.simulator.Bootstrap +import org.junit.jupiter.api.Test + +/** + * This test suite checks for smoke when running a large amount of simulations. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +internal class SmokeTest { + class EchoProcess : Process { + override val initialState = Unit + suspend override fun Context.run() { + while (true) { + receive { + sender?.send(message) + } + } + } + } + + /** + * Run a large amount of simulations and test if any exceptions occur. + */ + @Test + fun smoke() { + val n = 1000 + val messages = 100 + val bootstrap: Bootstrap = Bootstrap.create { ctx -> + repeat(n) { + EchoProcess().also { + ctx.register(it) + + for (i in 1 until messages) { + ctx.schedule(i, it, delay = i.toLong()) + } + } + } + } + val kernel = OmegaKernelFactory.create(bootstrap) + kernel.run() + } + + class NullProcess : Process { + override val initialState = Unit + suspend override fun Context.run() {} + } + + /** + * Test if the kernel allows sending messages to [Context] instances that have already stopped. + */ + @Test + fun `sending message to process that has gracefully stopped`() { + val process = NullProcess() + val bootstrap: Bootstrap = Bootstrap.create { ctx -> + process.also { + ctx.register(it) + ctx.schedule(0, it) + } + } + + val kernel = OmegaKernelFactory.create(bootstrap) + kernel.run() + } + + class CrashProcess : Process { + override val initialState = Unit + suspend override fun Context.run() { + TODO("This process should crash") + } + } + + /** + * Test if the kernel allows sending messages to [Context] instances that have crashed. + */ + @Test + fun `sending message to process that has crashed`() { + val process = CrashProcess() + val bootstrap: Bootstrap = Bootstrap.create { ctx -> + process.also { + ctx.register(it) + ctx.schedule(0, it) + } + } + + val kernel = OmegaKernelFactory.create(bootstrap) + kernel.run() + } +} -- cgit v1.2.3