simulator: Add support for transforming resource consumers

This change adds support for transforming the resource commands emitted
by the resource consumers. The SimResourceForwarder is modified to also
support transforming the resource commands.

6 files changed, 87 insertions, 11 deletions

diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
index 81d09f12..8046dd53 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
@@ -142,11 +142,11 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
          */
         override fun close() {
             if (!isTerminated) {
+                isTerminated = true
+
                 cpus.forEach { (_, provider) -> provider.close() }
                 _vms.remove(this)
             }
-
-            isTerminated = true
         }
     }
 
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/jmh/kotlin/org/opendc/simulator/resources/SimResourceBenchmarks.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/jmh/kotlin/org/opendc/simulator/resources/SimResourceBenchmarks.kt
index f2eea97c..937b6966 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/jmh/kotlin/org/opendc/simulator/resources/SimResourceBenchmarks.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/jmh/kotlin/org/opendc/simulator/resources/SimResourceBenchmarks.kt
@@ -71,7 +71,7 @@ class SimResourceBenchmarks {
     fun benchmarkForwardOverhead(state: Workload) {
         return scope.runBlockingTest {
             val provider = SimResourceSource(4200.0, clock, scheduler)
-            val forwarder = SimResourceForwarder()
+            val forwarder = SimResourceTransformer()
             provider.startConsumer(forwarder)
             return@runBlockingTest forwarder.consume(state.consumers[0])
         }
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlow.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlow.kt
new file mode 100644
index 00000000..bbf6ad44
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlow.kt
@@ -0,0 +1,29 @@
+/*
+ * 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
+
+/**
+ * A [SimResourceFlow] acts as both a resource consumer and resource provider at the same time, simplifying bridging
+ * between different components.
+ */
+public interface SimResourceFlow : SimResourceConsumer, SimResourceProvider
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt
index a10f84b6..45e4c220 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt
@@ -38,7 +38,7 @@ public class SimResourceSwitchExclusive : SimResourceSwitch {
     override val outputs: Set<SimResourceProvider>
         get() = _outputs
 
-    private val availableResources = ArrayDeque<SimResourceForwarder>()
+    private val availableResources = ArrayDeque<SimResourceTransformer>()
 
     private val _inputs = mutableSetOf<SimResourceProvider>()
     override val inputs: Set<SimResourceProvider>
@@ -83,7 +83,7 @@ public class SimResourceSwitchExclusive : SimResourceSwitch {
 
     private inner class Provider(
         private val capacity: Double,
-        private val forwarder: SimResourceForwarder
+        private val forwarder: SimResourceTransformer
     ) : SimResourceProvider by forwarder {
         override fun close() {
             // We explicitly do not close the forwarder here in order to re-use it across output resources.
diff --git a/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceForwarder.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt
index 1a05accd..73f18c7c 100644
--- a/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceForwarder.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt
@@ -23,9 +23,15 @@
 package org.opendc.simulator.resources
 
 /**
- * A helper class to construct a [SimResourceProvider] which forwards the requests to a [SimResourceConsumer].
+ * A [SimResourceFlow] that transforms the resource commands emitted by the resource commands to the resource provider.
+ *
+ * @param isCoupled A flag to indicate that the transformer will exit when the resource consumer exits.
+ * @param transform The function to transform the received resource command.
  */
-public class SimResourceForwarder : SimResourceProvider, SimResourceConsumer {
+public class SimResourceTransformer(
+    private val isCoupled: Boolean = false,
+    private val transform: (SimResourceContext, SimResourceCommand) -> SimResourceCommand
+) : SimResourceFlow {
     /**
      * The [SimResourceContext] in which the forwarder runs.
      */
@@ -98,7 +104,7 @@ public class SimResourceForwarder : SimResourceProvider, SimResourceConsumer {
         return if (state == SimResourceState.Stopped) {
             SimResourceCommand.Exit
         } else if (delegate != null) {
-            val command = delegate.onNext(ctx)
+            val command = transform(ctx, delegate.onNext(ctx))
             if (command == SimResourceCommand.Exit) {
                 // Warning: resumption of the continuation might change the entire state of the forwarder. Make sure we
                 // reset beforehand the existing state and check whether it has been updated afterwards
@@ -106,7 +112,7 @@ public class SimResourceForwarder : SimResourceProvider, SimResourceConsumer {
 
                 delegate.onFinish(ctx)
 
-                if (state == SimResourceState.Stopped)
+                if (isCoupled || state == SimResourceState.Stopped)
                     SimResourceCommand.Exit
                 else
                     onNext(ctx)
@@ -154,3 +160,12 @@ public class SimResourceForwarder : SimResourceProvider, SimResourceConsumer {
         }
     }
 }
+
+/**
+ * Constructs a [SimResourceTransformer] that forwards the received resource command with an identity transform.
+ *
+ * @param isCoupled A flag to indicate that the transformer will exit when the resource consumer exits.
+ */
+public fun SimResourceForwarder(isCoupled: Boolean = false): SimResourceTransformer {
+    return SimResourceTransformer(isCoupled) { _, command -> command }
+}
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/SimResourceTransformerTest.kt
index 143dbca9..38598f6b 100644
--- 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/SimResourceTransformerTest.kt
@@ -36,10 +36,10 @@ import org.opendc.simulator.utils.DelayControllerClockAdapter
 import org.opendc.utils.TimerScheduler
 
 /**
- * A test suite for the [SimResourceForwarder] class.
+ * A test suite for the [SimResourceTransformer] class.
  */
 @OptIn(ExperimentalCoroutinesApi::class)
-internal class SimResourceForwarderTest {
+internal class SimResourceTransformerTest {
     @Test
     fun testExitImmediately() = runBlockingTest {
         val forwarder = SimResourceForwarder()
@@ -165,6 +165,23 @@ internal class SimResourceForwarderTest {
     }
 
     @Test
+    fun testExitPropagation() = runBlockingTest {
+        val forwarder = SimResourceForwarder(isCoupled = true)
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(2000.0, clock, scheduler)
+
+        val consumer = mockk<SimResourceConsumer>(relaxUnitFun = true)
+        every { consumer.onNext(any()) } returns SimResourceCommand.Exit
+
+        source.startConsumer(forwarder)
+        forwarder.consume(consumer)
+        yield()
+
+        assertEquals(SimResourceState.Pending, source.state)
+    }
+
+    @Test
     fun testAdjustCapacity() = runBlockingTest {
         val forwarder = SimResourceForwarder()
         val clock = DelayControllerClockAdapter(this)
@@ -183,4 +200,19 @@ internal class SimResourceForwarderTest {
         assertEquals(3000, currentTime)
         verify(exactly = 1) { consumer.onCapacityChanged(any(), true) }
     }
+
+    @Test
+    fun testTransformExit() = runBlockingTest {
+        val forwarder = SimResourceTransformer { _, _ -> SimResourceCommand.Exit }
+        val clock = DelayControllerClockAdapter(this)
+        val scheduler = TimerScheduler<Any>(coroutineContext, clock)
+        val source = SimResourceSource(1.0, clock, scheduler)
+
+        val consumer = spyk(SimWorkConsumer(2.0, 1.0))
+        source.startConsumer(forwarder)
+        forwarder.consume(consumer)
+
+        assertEquals(0, currentTime)
+        verify(exactly = 1) { consumer.onNext(any()) }
+    }
 }