simulator: Make hypervisors generic for the resource type

This change moves the hypervisor implementations to the
opendc-simulator-resources module and makes them generic to the resource
type that is being used (e.g., CPU, disk or networking).

5 files changed, 792 insertions, 78 deletions

diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
new file mode 100644
index 00000000..e7642dc1
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceContextTest.kt
@@ -0,0 +1,156 @@
+/*
+ * 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.resources
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.runBlockingTest
+import org.junit.jupiter.api.*
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+
+/**
+ * A test suite for the [SimAbstractResourceContext] class.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class SimResourceContextTest {
+    data class SimCpu(val speed: Double) : SimResource {
+        override val capacity: Double
+            get() = speed
+    }
+
+    @Test
+    fun testFlushWithoutCommand() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+
+        val resource = SimCpu(4200.0)
+
+        val consumer = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(10.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        val context = object : SimAbstractResourceContext<SimCpu>(resource, clock, consumer) {
+            override fun onIdle(deadline: Long) {
+            }
+
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {
+            }
+
+            override fun onFinish() {
+            }
+
+            override fun onFailure(cause: Throwable) {
+            }
+        }
+
+        context.flush()
+    }
+
+    @Test
+    fun testIntermediateFlush() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val resource = SimCpu(4200.0)
+
+        val consumer = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(10.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        var counter = 0
+        val context = object : SimAbstractResourceContext<SimCpu>(resource, clock, consumer) {
+            override fun onIdle(deadline: Long) {
+            }
+
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {
+                counter++
+            }
+
+            override fun onFinish() {
+            }
+
+            override fun onFailure(cause: Throwable) {
+            }
+        }
+
+        context.start()
+        delay(1) // Delay 1 ms to prevent hitting the fast path
+        context.flush(isIntermediate = true)
+        assertEquals(2, counter)
+    }
+
+    @Test
+    fun testIntermediateFlushIdle() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val resource = SimCpu(4200.0)
+
+        val consumer = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Idle(10)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        var counter = 0
+        var isFinished = false
+        val context = object : SimAbstractResourceContext<SimCpu>(resource, clock, consumer) {
+            override fun onIdle(deadline: Long) {
+                counter++
+            }
+
+            override fun onConsume(work: Double, limit: Double, deadline: Long) {
+            }
+
+            override fun onFinish() {
+                isFinished = true
+            }
+
+            override fun onFailure(cause: Throwable) {
+            }
+        }
+
+        context.start()
+        delay(5)
+        context.flush(isIntermediate = true)
+        delay(5)
+        context.flush(isIntermediate = true)
+
+        assertAll(
+            { assertEquals(1, counter) },
+            { assertTrue(isFinished) }
+        )
+    }
+}
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
new file mode 100644
index 00000000..ced1bd98
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceForwarderTest.kt
@@ -0,0 +1,92 @@
+/*
+ * 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.resources
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runBlockingTest
+import org.junit.jupiter.api.Test
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+import org.opendc.utils.TimerScheduler
+
+/**
+ * A test suite for the [SimResourceForwarder] class.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class SimResourceForwarderTest {
+
+    data class SimCpu(val speed: Double) : SimResource {
+        override val capacity: Double
+            get() = speed
+    }
+
+    @Test
+    fun testExitImmediately() = runBlockingTest {
+        val forwarder = SimResourceForwarder(SimCpu(1000.0))
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(SimCpu(2000.0), clock, scheduler)
+
+        launch {
+            source.consume(forwarder)
+            source.close()
+        }
+
+        forwarder.consume(object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        })
+        forwarder.close()
+        scheduler.close()
+    }
+
+    @Test
+    fun testExit() = runBlockingTest {
+        val forwarder = SimResourceForwarder(SimCpu(1000.0))
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(SimCpu(2000.0), clock, scheduler)
+
+        launch {
+            source.consume(forwarder)
+            source.close()
+        }
+
+        forwarder.consume(object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(1.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        })
+
+        forwarder.close()
+    }
+}
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
index 8b380efb..4f7825fc 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSourceTest.kt
@@ -24,38 +24,27 @@ package org.opendc.simulator.resources
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.toList
-import kotlinx.coroutines.test.TestCoroutineScope
 import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
-import java.time.Clock
 
 /**
- * A test suite for the [SimResourceScheduler] class.
+ * A test suite for the [SimResourceSource] class.
  */
 @OptIn(ExperimentalCoroutinesApi::class)
 class SimResourceSourceTest {
-
-    private lateinit var scope: TestCoroutineScope
-    private lateinit var clock: Clock
-
     data class SimCpu(val speed: Double) : SimResource {
         override val capacity: Double
             get() = speed
     }
 
-    @BeforeEach
-    fun setUp() {
-        scope = TestCoroutineScope()
-        clock = DelayControllerClockAdapter(scope)
-    }
-
     @Test
-    fun testSpeed() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testSpeed() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -67,21 +56,25 @@ class SimResourceSourceTest {
             }
         }
 
-        scope.runBlockingTest {
+        try {
             val res = mutableListOf<Double>()
             val job = launch { provider.speed.toList(res) }
 
             provider.consume(consumer)
 
             job.cancel()
-            assertEquals(listOf(0.0, resource.speed, 0.0), res) { "Speed is reported correctly" }
+            assertEquals(listOf(0.0, provider.resource.speed, 0.0), res) { "Speed is reported correctly" }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
     }
 
     @Test
-    fun testSpeedLimit() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testSpeedLimit() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -93,21 +86,29 @@ class SimResourceSourceTest {
             }
         }
 
-        scope.runBlockingTest {
+        try {
             val res = mutableListOf<Double>()
             val job = launch { provider.speed.toList(res) }
 
             provider.consume(consumer)
 
             job.cancel()
-            assertEquals(listOf(0.0, resource.speed, 0.0), res) { "Speed is reported correctly" }
+            assertEquals(listOf(0.0, provider.resource.speed, 0.0), res) { "Speed is reported correctly" }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
     }
 
+    /**
+     * Test to see whether no infinite recursion occurs when interrupting during [SimResourceConsumer.onStart] or
+     * [SimResourceConsumer.onNext].
+     */
     @Test
-    fun testInterrupt() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testIntermediateInterrupt() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -120,17 +121,52 @@ class SimResourceSourceTest {
             }
         }
 
-        assertDoesNotThrow {
-            scope.runBlockingTest {
-                provider.consume(consumer)
+        try {
+            provider.consume(consumer)
+        } finally {
+            scheduler.close()
+            provider.close()
+        }
+    }
+
+    @Test
+    fun testInterrupt() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+        lateinit var resCtx: SimResourceContext<SimCpu>
+
+        val consumer = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                resCtx = ctx
+                return SimResourceCommand.Consume(4.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                assertEquals(0.0, remainingWork)
+                return SimResourceCommand.Exit
             }
         }
+
+        try {
+            launch {
+                yield()
+                resCtx.interrupt()
+            }
+            provider.consume(consumer)
+
+            assertEquals(0, currentTime)
+        } finally {
+            scheduler.close()
+            provider.close()
+        }
     }
 
     @Test
-    fun testFailure() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testFailure() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -142,17 +178,21 @@ class SimResourceSourceTest {
             }
         }
 
-        assertThrows<IllegalStateException> {
-            scope.runBlockingTest {
+        try {
+            assertThrows<IllegalStateException> {
                 provider.consume(consumer)
             }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
     }
 
     @Test
-    fun testExceptionPropagationOnNext() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testExceptionPropagationOnNext() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -164,15 +204,21 @@ class SimResourceSourceTest {
             }
         }
 
-        assertThrows<IllegalStateException> {
-            scope.runBlockingTest { provider.consume(consumer) }
+        try {
+            assertThrows<IllegalStateException> {
+                provider.consume(consumer)
+            }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
     }
 
     @Test
-    fun testConcurrentConsumption() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testConcurrentConsumption() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -184,18 +230,24 @@ class SimResourceSourceTest {
             }
         }
 
-        assertThrows<IllegalStateException> {
-            scope.runBlockingTest {
-                launch { provider.consume(consumer) }
-                launch { provider.consume(consumer) }
+        try {
+            assertThrows<IllegalStateException> {
+                coroutineScope {
+                    launch { provider.consume(consumer) }
+                    provider.consume(consumer)
+                }
             }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
     }
 
     @Test
-    fun testClosedConsumption() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testClosedConsumption() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -207,18 +259,22 @@ class SimResourceSourceTest {
             }
         }
 
-        assertThrows<IllegalStateException> {
-            scope.runBlockingTest {
+        try {
+            assertThrows<IllegalStateException> {
                 provider.close()
                 provider.consume(consumer)
             }
+        } finally {
+            scheduler.close()
+            provider.close()
         }
     }
 
     @Test
-    fun testCloseDuringConsumption() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testCloseDuringConsumption() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -230,19 +286,23 @@ class SimResourceSourceTest {
             }
         }
 
-        scope.runBlockingTest {
+        try {
             launch { provider.consume(consumer) }
             delay(500)
             provider.close()
-        }
 
-        assertEquals(500, scope.currentTime)
+            assertEquals(500, currentTime)
+        } finally {
+            scheduler.close()
+            provider.close()
+        }
     }
 
     @Test
-    fun testIdle() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
+    fun testIdle() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
 
         val consumer = object : SimResourceConsumer<SimCpu> {
             override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
@@ -254,31 +314,40 @@ class SimResourceSourceTest {
             }
         }
 
-        scope.runBlockingTest {
+        try {
             provider.consume(consumer)
-        }
 
-        assertEquals(500, scope.currentTime)
+            assertEquals(500, currentTime)
+        } finally {
+            scheduler.close()
+            provider.close()
+        }
     }
 
     @Test
     fun testInfiniteSleep() {
-        val resource = SimCpu(4200.0)
-        val provider = SimResourceSource(resource, clock, TimerScheduler(scope, clock))
-
-        val consumer = object : SimResourceConsumer<SimCpu> {
-            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
-                return SimResourceCommand.Idle()
-            }
-
-            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
-                return SimResourceCommand.Exit
-            }
-        }
-
         assertThrows<IllegalStateException> {
-            scope.runBlockingTest {
-                provider.consume(consumer)
+            runBlockingTest {
+                val clock = DelayControllerClockAdapter(this)
+                val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+                val provider = SimResourceSource(SimCpu(4200.0), clock, scheduler)
+
+                val consumer = object : SimResourceConsumer<SimCpu> {
+                    override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                        return SimResourceCommand.Idle()
+                    }
+
+                    override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                        return SimResourceCommand.Exit
+                    }
+                }
+
+                try {
+                    provider.consume(consumer)
+                } finally {
+                    scheduler.close()
+                    provider.close()
+                }
             }
         }
     }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
new file mode 100644
index 00000000..ca6558bf
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusiveTest.kt
@@ -0,0 +1,190 @@
+/*
+ * 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.resources
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.yield
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertAll
+import org.junit.jupiter.api.assertThrows
+import org.opendc.simulator.resources.consumer.SimTraceConsumer
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+import org.opendc.utils.TimerScheduler
+import java.lang.IllegalStateException
+
+/**
+ * Test suite for the [SimResourceSwitchExclusive] class.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class SimResourceSwitchExclusiveTest {
+    class SimCpu(val speed: Double) : SimResource {
+        override val capacity: Double
+            get() = speed
+    }
+
+    /**
+     * Test a trace workload.
+     */
+    @Test
+    fun testTrace() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val speed = mutableListOf<Double>()
+
+        val duration = 5 * 60L
+        val workload =
+            SimTraceConsumer(
+                sequenceOf(
+                    SimTraceConsumer.Fragment(duration * 1000, 28.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 3500.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 0.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 183.0)
+                ),
+            )
+
+        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
+        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+
+        switch.addInput(source)
+
+        val provider = switch.addOutput(SimCpu(3200.0))
+        val job = launch { source.speed.toList(speed) }
+
+        try {
+            provider.consume(workload)
+            yield()
+        } finally {
+            job.cancel()
+            provider.close()
+        }
+
+        assertAll(
+            { assertEquals(listOf(0.0, 28.0, 3200.0, 0.0, 183.0, 0.0), speed) { "Correct speed" } },
+            { assertEquals(5 * 60L * 4000, currentTime) { "Took enough time" } }
+        )
+    }
+
+    /**
+     * Test runtime workload on hypervisor.
+     */
+    @Test
+    fun testRuntimeWorkload() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val duration = 5 * 60L * 1000
+        val workload = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(duration / 1000.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
+        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+
+        switch.addInput(source)
+
+        val provider = switch.addOutput(SimCpu(3200.0))
+
+        try {
+            provider.consume(workload)
+            yield()
+        } finally {
+            provider.close()
+        }
+        assertEquals(duration, currentTime) { "Took enough time" }
+    }
+
+    /**
+     * Test two workloads running sequentially.
+     */
+    @Test
+    fun testTwoWorkloads() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val duration = 5 * 60L * 1000
+        val workload = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(duration / 1000.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
+        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+
+        switch.addInput(source)
+
+        val provider = switch.addOutput(SimCpu(3200.0))
+
+        try {
+            provider.consume(workload)
+            yield()
+            provider.consume(workload)
+        } finally {
+            provider.close()
+        }
+        assertEquals(duration * 2, currentTime) { "Took enough time" }
+    }
+
+    /**
+     * Test concurrent workloads on the machine.
+     */
+    @Test
+    fun testConcurrentWorkloadFails() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val duration = 5 * 60L * 1000
+        val workload = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(duration.toDouble(), 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        val switch = SimResourceSwitchExclusive<SimCpu>(coroutineContext)
+        val source = SimResourceSource(SimCpu(3200.0), clock, scheduler)
+
+        switch.addInput(source)
+
+        switch.addOutput(SimCpu(3200.0))
+        assertThrows<IllegalStateException> { switch.addOutput(SimCpu(3200.0)) }
+    }
+}
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
new file mode 100644
index 00000000..698c1700
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/test/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMinTest.kt
@@ -0,0 +1,207 @@
+/*
+ * 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.resources
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runBlockingTest
+import kotlinx.coroutines.yield
+import org.junit.jupiter.api.*
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.opendc.simulator.resources.consumer.SimTraceConsumer
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+import org.opendc.utils.TimerScheduler
+
+/**
+ * Test suite for the [SimResourceSwitch] implementations
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class SimResourceSwitchMaxMinTest {
+    class SimCpu(val speed: Double) : SimResource {
+        override val capacity: Double
+            get() = speed
+    }
+
+    @Test
+    fun testSmoke() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val switch = SimResourceSwitchMaxMin<SimCpu>(clock, coroutineContext)
+
+        val sources = List(2) { SimResourceSource(SimCpu(2000.0), clock, scheduler) }
+        sources.forEach { switch.addInput(it) }
+
+        val provider = switch.addOutput(SimCpu(1000.0))
+
+        val consumer = object : SimResourceConsumer<SimCpu> {
+            override fun onStart(ctx: SimResourceContext<SimCpu>): SimResourceCommand {
+                return SimResourceCommand.Consume(1.0, 1.0)
+            }
+
+            override fun onNext(ctx: SimResourceContext<SimCpu>, remainingWork: Double): SimResourceCommand {
+                return SimResourceCommand.Exit
+            }
+        }
+
+        try {
+            provider.consume(consumer)
+            yield()
+        } finally {
+            switch.close()
+            scheduler.close()
+        }
+    }
+
+    /**
+     * Test overcommitting of resources via the hypervisor with a single VM.
+     */
+    @Test
+    fun testOvercommittedSingle() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val listener = object : SimResourceSwitchMaxMin.Listener<SimCpu> {
+            var totalRequestedWork = 0L
+            var totalGrantedWork = 0L
+            var totalOvercommittedWork = 0L
+
+            override fun onSliceFinish(
+                switch: SimResourceSwitchMaxMin<SimCpu>,
+                requestedWork: Long,
+                grantedWork: Long,
+                overcommittedWork: Long,
+                interferedWork: Long,
+                cpuUsage: Double,
+                cpuDemand: Double
+            ) {
+                totalRequestedWork += requestedWork
+                totalGrantedWork += grantedWork
+                totalOvercommittedWork += overcommittedWork
+            }
+        }
+
+        val duration = 5 * 60L
+        val workload =
+            SimTraceConsumer(
+                sequenceOf(
+                    SimTraceConsumer.Fragment(duration * 1000, 28.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 3500.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 0.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 183.0)
+                ),
+            )
+
+        val switch = SimResourceSwitchMaxMin(clock, coroutineContext, listener)
+        val provider = switch.addOutput(SimCpu(3200.0))
+
+        try {
+            switch.addInput(SimResourceSource(SimCpu(3200.0), clock, scheduler))
+            provider.consume(workload)
+            yield()
+        } finally {
+            switch.close()
+            scheduler.close()
+        }
+
+        assertAll(
+            { assertEquals(1113300, listener.totalRequestedWork, "Requested Burst does not match") },
+            { assertEquals(1023300, listener.totalGrantedWork, "Granted Burst does not match") },
+            { assertEquals(90000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(1200000, currentTime) }
+        )
+    }
+
+    /**
+     * Test overcommitting of resources via the hypervisor with two VMs.
+     */
+    @Test
+    fun testOvercommittedDual() = runBlockingTest {
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+
+        val listener = object : SimResourceSwitchMaxMin.Listener<SimCpu> {
+            var totalRequestedWork = 0L
+            var totalGrantedWork = 0L
+            var totalOvercommittedWork = 0L
+
+            override fun onSliceFinish(
+                switch: SimResourceSwitchMaxMin<SimCpu>,
+                requestedWork: Long,
+                grantedWork: Long,
+                overcommittedWork: Long,
+                interferedWork: Long,
+                cpuUsage: Double,
+                cpuDemand: Double
+            ) {
+                totalRequestedWork += requestedWork
+                totalGrantedWork += grantedWork
+                totalOvercommittedWork += overcommittedWork
+            }
+        }
+
+        val duration = 5 * 60L
+        val workloadA =
+            SimTraceConsumer(
+                sequenceOf(
+                    SimTraceConsumer.Fragment(duration * 1000, 28.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 3500.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 0.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 183.0)
+                ),
+            )
+        val workloadB =
+            SimTraceConsumer(
+                sequenceOf(
+                    SimTraceConsumer.Fragment(duration * 1000, 28.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 3100.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 0.0),
+                    SimTraceConsumer.Fragment(duration * 1000, 73.0)
+                )
+            )
+
+        val switch = SimResourceSwitchMaxMin(clock, coroutineContext, listener)
+        val providerA = switch.addOutput(SimCpu(3200.0))
+        val providerB = switch.addOutput(SimCpu(3200.0))
+
+        try {
+            switch.addInput(SimResourceSource(SimCpu(3200.0), clock, scheduler))
+
+            coroutineScope {
+                launch { providerA.consume(workloadA) }
+                providerB.consume(workloadB)
+            }
+
+            yield()
+        } finally {
+            switch.close()
+            scheduler.close()
+        }
+        assertAll(
+            { assertEquals(2082000, listener.totalRequestedWork, "Requested Burst does not match") },
+            { assertEquals(1062000, listener.totalGrantedWork, "Granted Burst does not match") },
+            { assertEquals(1020000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(1200000, currentTime) }
+        )
+    }
+}