Add advanced energy model to OpenDC (v1)

This pull request is a preparation for the adding an advanced energy model to OpenDC (#90):

* Add benchmarks for the `opendc-simulator-compute` module, which enables us to quantify the effect on future changes w.r.t. to the energy model on the performance.
* Add `SimResourceTransformer` which can transform resource consumptions. This is useful when transforming from CPU frequency to energy usage for instance.
* Move power models to `opendc-simulator-compute` to be able to be used more generically.

**Breaking API Changes**

* Power models have moved to `opendc-simulator-compute` and now implement the `MachinePowerModel` interface.

31 files changed, 332 insertions, 158 deletions

diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt
index 694676bc..3c4b4410 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt
@@ -29,13 +29,12 @@ import org.opendc.compute.api.Flavor
 import org.opendc.compute.api.Server
 import org.opendc.compute.api.ServerState
 import org.opendc.compute.service.driver.*
-import org.opendc.compute.simulator.power.api.CpuPowerModel
-import org.opendc.compute.simulator.power.api.Powerable
-import org.opendc.compute.simulator.power.models.ConstantPowerModel
 import org.opendc.simulator.compute.*
 import org.opendc.simulator.compute.interference.IMAGE_PERF_INTERFERENCE_MODEL
 import org.opendc.simulator.compute.interference.PerformanceInterferenceModel
 import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.power.ConstantPowerModel
+import org.opendc.simulator.compute.power.MachinePowerModel
 import org.opendc.simulator.failures.FailureDomain
 import org.opendc.utils.flow.EventFlow
 import java.time.Clock
@@ -54,9 +53,9 @@ public class SimHost(
     context: CoroutineContext,
     clock: Clock,
     hypervisor: SimHypervisorProvider,
-    cpuPowerModel: CpuPowerModel = ConstantPowerModel(0.0),
-    private val mapper: SimWorkloadMapper = SimMetaWorkloadMapper()
-) : Host, FailureDomain, Powerable, AutoCloseable {
+    powerModel: MachinePowerModel = ConstantPowerModel(0.0),
+    private val mapper: SimWorkloadMapper = SimMetaWorkloadMapper(),
+) : Host, FailureDomain, AutoCloseable {
     /**
      * The [CoroutineScope] of the host bounded by the lifecycle of the host.
      */
@@ -84,7 +83,7 @@ public class SimHost(
     /**
      * The machine to run on.
      */
-    public val machine: SimBareMetalMachine = SimBareMetalMachine(context, clock, model)
+    public val machine: SimBareMetalMachine = SimBareMetalMachine(context, clock, model, powerModel)
 
     /**
      * The hypervisor to run multiple workloads.
@@ -133,8 +132,6 @@ public class SimHost(
 
     override val model: HostModel = HostModel(model.cpus.size, model.memory.map { it.size }.sum())
 
-    override val powerDraw: Flow<Double> = cpuPowerModel.getPowerDraw(machine)
-
     init {
         // Launch hypervisor onto machine
         scope.launch {
@@ -223,7 +220,7 @@ public class SimHost(
     }
 
     private fun onGuestStart(vm: Guest) {
-        guests.forEach { _, guest ->
+        guests.forEach { (_, guest) ->
             if (guest.state == ServerState.RUNNING) {
                 vm.performanceInterferenceModel?.onStart(vm.server.image.name)
             }
@@ -233,7 +230,7 @@ public class SimHost(
     }
 
     private fun onGuestStop(vm: Guest) {
-        guests.forEach { _, guest ->
+        guests.forEach { (_, guest) ->
             if (guest.state == ServerState.RUNNING) {
                 vm.performanceInterferenceModel?.onStop(vm.server.image.name)
             }
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/api/CpuPowerModel.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/api/CpuPowerModel.kt
deleted file mode 100644
index 893f7ab1..00000000
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/api/CpuPowerModel.kt
+++ /dev/null
@@ -1,28 +0,0 @@
-package org.opendc.compute.simulator.power.api
-
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.map
-import org.opendc.simulator.compute.SimMachine
-
-public interface CpuPowerModel {
-    /**
-     * Computes CPU power consumption for each host.
-     *
-     * @param cpuUtil The CPU utilization percentage.
-     * @return A [Double] value of CPU power consumption.
-     * @throws IllegalArgumentException Will throw an error if [cpuUtil] is out of range.
-     */
-    @Throws(IllegalArgumentException::class)
-    public fun computeCpuPower(cpuUtil: Double): Double
-
-    /**
-     * Emits the values of power consumption for servers.
-     *
-     * @param machine The [SimMachine] that the model is measuring.
-     * @return A [Flow] of values representing the server power draw.
-     */
-    public fun getPowerDraw(machine: SimMachine): Flow<Double> =
-        machine.usage.map {
-            computeCpuPower(it)
-        }
-}
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/ZeroIdlePowerDecorator.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/ZeroIdlePowerDecorator.kt
deleted file mode 100644
index b0c3fa4c..00000000
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/ZeroIdlePowerDecorator.kt
+++ /dev/null
@@ -1,14 +0,0 @@
-package org.opendc.compute.simulator.power.models
-
-import org.opendc.compute.simulator.power.api.CpuPowerModel
-
-/**
- * A decorator for ignoring the idle power when computing energy consumption of components.
- *
- * @param delegate The [CpuPowerModel] to delegate to.
- */
-public class ZeroIdlePowerDecorator(private val delegate: CpuPowerModel) : CpuPowerModel {
-    override fun computeCpuPower(cpuUtil: Double): Double {
-        return if (cpuUtil == 0.0) 0.0 else delegate.computeCpuPower(cpuUtil)
-    }
-}
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/package-info.java b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/package-info.java
deleted file mode 100644
index 9354f1f9..00000000
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/package-info.java
+++ /dev/null
@@ -1,12 +0,0 @@
-/**
- * Contains a set of energy models, which are developed based on that of the following cloud simulators.
- * N.B. Different configurations and new features may have been introduced.
- *
- * CloudSim:
- * @see <a href="http://dx.doi.org/10.1002/cpe.1867">Anton Beloglazov, and Rajkumar Buyya, "Optimal Online Deterministic
- * Algorithms and Adaptive Heuristics for Energy and Performance Efficient Dynamic Consolidation of Virtual Machines in
- * Cloud Data Centers", Concurrency and Computation: Practice and Experience (CCPE), Volume 24,
- * Issue 13, Pages: 1397-1420, John Wiley & Sons, Ltd, New York, USA, 2012</a>
- */
-package org.opendc.compute.simulator.power.models;
-// rest of the file is empty
diff --git a/simulator/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt b/simulator/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt
index f327b55d..44436019 100644
--- a/simulator/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt
+++ b/simulator/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/ExperimentHelpers.kt
@@ -205,7 +205,7 @@ public fun attachMonitor(
             }
             .launchIn(coroutineScope)
 
-        (host as SimHost).powerDraw
+        (host as SimHost).machine.powerDraw
             .onEach { monitor.reportPowerConsumption(host, it) }
             .launchIn(coroutineScope)
     }
diff --git a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/MachineDef.kt b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/MachineDef.kt
index b5b3b84b..c7f7c4dc 100644
--- a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/MachineDef.kt
+++ b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/MachineDef.kt
@@ -22,8 +22,8 @@
 
 package org.opendc.format.environment
 
-import org.opendc.compute.simulator.power.api.CpuPowerModel
 import org.opendc.simulator.compute.SimMachineModel
+import org.opendc.simulator.compute.power.MachinePowerModel
 import java.util.*
 
 public data class MachineDef(
@@ -31,5 +31,5 @@ public data class MachineDef(
     val name: String,
     val meta: Map<String, Any>,
     val model: SimMachineModel,
-    val powerModel: CpuPowerModel
+    val powerModel: MachinePowerModel
 )
diff --git a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc18/Sc18EnvironmentReader.kt b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc18/Sc18EnvironmentReader.kt
index 3da8d0b3..1f080c2d 100644
--- a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc18/Sc18EnvironmentReader.kt
+++ b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc18/Sc18EnvironmentReader.kt
@@ -25,13 +25,13 @@ package org.opendc.format.environment.sc18
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
-import org.opendc.compute.simulator.power.models.ConstantPowerModel
 import org.opendc.format.environment.EnvironmentReader
 import org.opendc.format.environment.MachineDef
 import org.opendc.simulator.compute.SimMachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.power.ConstantPowerModel
 import java.io.InputStream
 import java.util.*
 
diff --git a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20ClusterEnvironmentReader.kt b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20ClusterEnvironmentReader.kt
index 9a06a40f..cf90da68 100644
--- a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20ClusterEnvironmentReader.kt
+++ b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20ClusterEnvironmentReader.kt
@@ -22,13 +22,13 @@
 
 package org.opendc.format.environment.sc20
 
-import org.opendc.compute.simulator.power.models.LinearPowerModel
 import org.opendc.format.environment.EnvironmentReader
 import org.opendc.format.environment.MachineDef
 import org.opendc.simulator.compute.SimMachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.power.LinearPowerModel
 import java.io.File
 import java.io.FileInputStream
 import java.io.InputStream
diff --git a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20EnvironmentReader.kt b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20EnvironmentReader.kt
index effd0286..c6a19430 100644
--- a/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20EnvironmentReader.kt
+++ b/simulator/opendc-format/src/main/kotlin/org/opendc/format/environment/sc20/Sc20EnvironmentReader.kt
@@ -25,13 +25,13 @@ package org.opendc.format.environment.sc20
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import com.fasterxml.jackson.module.kotlin.readValue
-import org.opendc.compute.simulator.power.models.LinearPowerModel
 import org.opendc.format.environment.EnvironmentReader
 import org.opendc.format.environment.MachineDef
 import org.opendc.simulator.compute.SimMachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.power.LinearPowerModel
 import java.io.InputStream
 import java.util.*
 
diff --git a/simulator/opendc-runner-web/src/main/kotlin/org/opendc/runner/web/TopologyParser.kt b/simulator/opendc-runner-web/src/main/kotlin/org/opendc/runner/web/TopologyParser.kt
index e7e99a3d..1683cdb8 100644
--- a/simulator/opendc-runner-web/src/main/kotlin/org/opendc/runner/web/TopologyParser.kt
+++ b/simulator/opendc-runner-web/src/main/kotlin/org/opendc/runner/web/TopologyParser.kt
@@ -30,13 +30,13 @@ import com.mongodb.client.model.Filters
 import com.mongodb.client.model.Projections
 import org.bson.Document
 import org.bson.types.ObjectId
-import org.opendc.compute.simulator.power.models.LinearPowerModel
 import org.opendc.format.environment.EnvironmentReader
 import org.opendc.format.environment.MachineDef
 import org.opendc.simulator.compute.SimMachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.power.LinearPowerModel
 import java.util.*
 
 /**
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/build.gradle.kts b/simulator/opendc-simulator/opendc-simulator-compute/build.gradle.kts
index 0005928f..149c0ed2 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/build.gradle.kts
+++ b/simulator/opendc-simulator/opendc-simulator-compute/build.gradle.kts
@@ -20,12 +20,13 @@
  * SOFTWARE.
  */
 
-description = "Library for simulation of cloud computing components"
+description = "Library for simulating computing workloads"
 
 plugins {
     `kotlin-library-conventions`
     `testing-conventions`
     `jacoco-conventions`
+    `benchmark-conventions`
 }
 
 dependencies {
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/BenchmarkHelpers.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/BenchmarkHelpers.kt
new file mode 100644
index 00000000..43bbfd0b
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/BenchmarkHelpers.kt
@@ -0,0 +1,43 @@
+/*
+ * 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.compute
+
+import org.opendc.simulator.compute.workload.SimTraceWorkload
+
+/**
+ * Helper function to create simple consumer workload.
+ */
+fun createSimpleConsumer(): SimTraceWorkload {
+    return SimTraceWorkload(
+        sequenceOf(
+            SimTraceWorkload.Fragment(1000, 28.0, 1),
+            SimTraceWorkload.Fragment(1000, 3500.0, 1),
+            SimTraceWorkload.Fragment(1000, 0.0, 1),
+            SimTraceWorkload.Fragment(1000, 183.0, 1),
+            SimTraceWorkload.Fragment(1000, 400.0, 1),
+            SimTraceWorkload.Fragment(1000, 100.0, 1),
+            SimTraceWorkload.Fragment(1000, 3000.0, 1),
+            SimTraceWorkload.Fragment(1000, 4500.0, 1),
+        ),
+    )
+}
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
new file mode 100644
index 00000000..eb22c855
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
@@ -0,0 +1,145 @@
+/*
+ * 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.compute
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestCoroutineScope
+import kotlinx.coroutines.test.runBlockingTest
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingNode
+import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.workload.SimWorkload
+import org.opendc.simulator.utils.DelayControllerClockAdapter
+import org.opendc.utils.TimerScheduler
+import org.openjdk.jmh.annotations.*
+import java.time.Clock
+import java.util.concurrent.TimeUnit
+
+@State(Scope.Thread)
+@Fork(1)
+@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 3, timeUnit = TimeUnit.SECONDS)
+@OptIn(ExperimentalCoroutinesApi::class)
+class SimMachineBenchmarks {
+    private lateinit var scope: TestCoroutineScope
+    private lateinit var clock: Clock
+    private lateinit var scheduler: TimerScheduler<Any>
+    private lateinit var machineModel: SimMachineModel
+
+    @Setup
+    fun setUp() {
+        scope = TestCoroutineScope()
+        clock = DelayControllerClockAdapter(scope)
+        scheduler = TimerScheduler(scope.coroutineContext, clock)
+
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
+
+        machineModel = SimMachineModel(
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 1000.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+        )
+    }
+
+    @State(Scope.Thread)
+    class Workload {
+        lateinit var workloads: Array<SimWorkload>
+
+        @Setup
+        fun setUp() {
+            workloads = Array(2) { createSimpleConsumer() }
+        }
+    }
+
+    @Benchmark
+    fun benchmarkBareMetal(state: Workload) {
+        return scope.runBlockingTest {
+            val machine = SimBareMetalMachine(scope.coroutineContext, clock, machineModel)
+            return@runBlockingTest machine.run(state.workloads[0])
+        }
+    }
+
+    @Benchmark
+    fun benchmarkSpaceSharedHypervisor(state: Workload) {
+        return scope.runBlockingTest {
+            val machine = SimBareMetalMachine(coroutineContext, clock, machineModel)
+            val hypervisor = SimSpaceSharedHypervisor()
+
+            launch { machine.run(hypervisor) }
+
+            val vm = hypervisor.createMachine(machineModel)
+
+            try {
+                return@runBlockingTest vm.run(state.workloads[0])
+            } finally {
+                vm.close()
+                machine.close()
+            }
+        }
+    }
+
+    @Benchmark
+    fun benchmarkFairShareHypervisorSingle(state: Workload) {
+        return scope.runBlockingTest {
+            val machine = SimBareMetalMachine(coroutineContext, clock, machineModel)
+            val hypervisor = SimFairShareHypervisor()
+
+            launch { machine.run(hypervisor) }
+
+            val vm = hypervisor.createMachine(machineModel)
+
+            try {
+                return@runBlockingTest vm.run(state.workloads[0])
+            } finally {
+                vm.close()
+                machine.close()
+            }
+        }
+    }
+
+    @Benchmark
+    fun benchmarkFairShareHypervisorDouble(state: Workload) {
+        return scope.runBlockingTest {
+            val machine = SimBareMetalMachine(coroutineContext, clock, machineModel)
+            val hypervisor = SimFairShareHypervisor()
+
+            launch { machine.run(hypervisor) }
+
+            coroutineScope {
+                repeat(2) { i ->
+                    val vm = hypervisor.createMachine(machineModel)
+
+                    launch {
+                        try {
+                            vm.run(state.workloads[i])
+                        } finally {
+                            machine.close()
+                        }
+                    }
+                }
+            }
+            machine.close()
+        }
+    }
+}
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-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
index 19479719..f86c4198 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
@@ -23,7 +23,10 @@
 package org.opendc.simulator.compute
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.power.ConstantPowerModel
+import org.opendc.simulator.compute.power.MachinePowerModel
 import org.opendc.simulator.resources.*
 import org.opendc.utils.TimerScheduler
 import java.time.Clock
@@ -43,14 +46,20 @@ import kotlin.coroutines.*
 public class SimBareMetalMachine(
     context: CoroutineContext,
     private val clock: Clock,
-    override val model: SimMachineModel
+    override val model: SimMachineModel,
+    private val powerModel: MachinePowerModel = ConstantPowerModel(0.0)
 ) : SimAbstractMachine(clock) {
     /**
      * The [Job] associated with this machine.
      */
-    private val job = Job()
+    private val scope = CoroutineScope(context + Job())
 
-    override val context: CoroutineContext = context + job
+    /**
+     * The power draw of the machine.
+     */
+    public val powerDraw: StateFlow<Double> = usage.map { powerModel.computeCpuPower(it) }.stateIn(scope, SharingStarted.Eagerly, 0.0)
+
+    override val context: CoroutineContext = scope.coroutineContext
 
     /**
      * The [TimerScheduler] to use for scheduling the interrupts.
@@ -63,6 +72,6 @@ public class SimBareMetalMachine(
     override fun close() {
         super.close()
         scheduler.close()
-        job.cancel()
+        scope.cancel()
     }
 }
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/ConstantPowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/ConstantPowerModel.kt
index 5e80827b..5d7ae8ad 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/ConstantPowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/ConstantPowerModel.kt
@@ -1,10 +1,8 @@
-package org.opendc.compute.simulator.power.models
-
-import org.opendc.compute.simulator.power.api.CpuPowerModel
+package org.opendc.simulator.compute.power
 
 /**
  * A power model which produces a constant value [constant].
  */
-public class ConstantPowerModel(private val constant: Double) : CpuPowerModel {
+public class ConstantPowerModel(private val constant: Double) : MachinePowerModel {
     public override fun computeCpuPower(cpuUtil: Double): Double = constant
 }
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/CubicPowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/CubicPowerModel.kt
index 9008a987..8e47f571 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/CubicPowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/CubicPowerModel.kt
@@ -1,6 +1,5 @@
-package org.opendc.compute.simulator.power.models
+package org.opendc.simulator.compute.power
 
-import org.opendc.compute.simulator.power.api.CpuPowerModel
 import kotlin.math.pow
 
 /**
@@ -15,7 +14,7 @@ import kotlin.math.pow
 public class CubicPowerModel(
     private var maxPower: Double,
     staticPowerPercent: Double
-) : CpuPowerModel {
+) : MachinePowerModel {
     private var staticPower: Double = staticPowerPercent * maxPower
     private var constPower: Double = (maxPower - staticPower) / 100.0.pow(3)
 
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/InterpolationPowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/InterpolationPowerModel.kt
index b01957e4..af7b6e6d 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/InterpolationPowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/InterpolationPowerModel.kt
@@ -1,6 +1,5 @@
-package org.opendc.compute.simulator.power.models
+package org.opendc.simulator.compute.power
 
-import org.opendc.compute.simulator.power.api.CpuPowerModel
 import kotlin.math.ceil
 import kotlin.math.floor
 
@@ -13,7 +12,7 @@ import kotlin.math.floor
  *                      It is the amount of energy consumed even when the host is idle.
  * @property constPower The constant power consumption for each fraction of resource used.
  */
-public abstract class InterpolationPowerModel : CpuPowerModel {
+public abstract class InterpolationPowerModel : MachinePowerModel {
 
     public override fun computeCpuPower(cpuUtil: Double): Double {
         require(cpuUtil in 0.0..1.0) { "CPU utilization must be in [0, 1]" }
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/LinearPowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/LinearPowerModel.kt
index 913095ad..14443aff 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/LinearPowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/LinearPowerModel.kt
@@ -1,6 +1,4 @@
-package org.opendc.compute.simulator.power.models
-
-import org.opendc.compute.simulator.power.api.CpuPowerModel
+package org.opendc.simulator.compute.power
 
 /**
  * The linear power model partially adapted from CloudSim.
@@ -14,7 +12,7 @@ import org.opendc.compute.simulator.power.api.CpuPowerModel
 public class LinearPowerModel(
     private var maxPower: Double,
     staticPowerPercent: Double
-) : CpuPowerModel {
+) : MachinePowerModel {
     private var staticPower: Double = staticPowerPercent * maxPower
     private var constPower: Double = (maxPower - staticPower) / 100
 
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/MachinePowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/MachinePowerModel.kt
new file mode 100644
index 00000000..9bf03b87
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/MachinePowerModel.kt
@@ -0,0 +1,18 @@
+package org.opendc.simulator.compute.power
+
+import org.opendc.simulator.compute.SimMachine
+
+/**
+ * A model for estimating the power usage of a [SimMachine].
+ */
+public interface MachinePowerModel {
+    /**
+     * Computes CPU power consumption for each host.
+     *
+     * @param cpuUtil The CPU utilization percentage.
+     * @return A [Double] value of CPU power consumption.
+     * @throws IllegalArgumentException Will throw an error if [cpuUtil] is out of range.
+     */
+    @Throws(IllegalArgumentException::class)
+    public fun computeCpuPower(cpuUtil: Double): Double
+}
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/PStatePowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/PStatePowerModel.kt
index aea089da..722f478d 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/PStatePowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/PStatePowerModel.kt
@@ -1,4 +1,4 @@
-package org.opendc.compute.simulator.power.models
+package org.opendc.simulator.compute.power
 
 import org.opendc.simulator.compute.SimBareMetalMachine
 import java.time.Clock
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/SqrtPowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SqrtPowerModel.kt
index 85d94ffc..bf177aff 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/SqrtPowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SqrtPowerModel.kt
@@ -1,6 +1,5 @@
-package org.opendc.compute.simulator.power.models
+package org.opendc.simulator.compute.power
 
-import org.opendc.compute.simulator.power.api.CpuPowerModel
 import kotlin.math.sqrt
 
 /**
@@ -15,7 +14,7 @@ import kotlin.math.sqrt
 public class SqrtPowerModel(
     private var maxPower: Double,
     staticPowerPercent: Double
-) : CpuPowerModel {
+) : MachinePowerModel {
     private var staticPower: Double = staticPowerPercent * maxPower
     private var constPower: Double = (maxPower - staticPower) / sqrt(100.0)
 
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/SquarePowerModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SquarePowerModel.kt
index 5f44aa3c..cbfad530 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/models/SquarePowerModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SquarePowerModel.kt
@@ -1,6 +1,5 @@
-package org.opendc.compute.simulator.power.models
+package org.opendc.simulator.compute.power
 
-import org.opendc.compute.simulator.power.api.CpuPowerModel
 import kotlin.math.pow
 
 /**
@@ -15,7 +14,7 @@ import kotlin.math.pow
 public class SquarePowerModel(
     private var maxPower: Double,
     staticPowerPercent: Double
-) : CpuPowerModel {
+) : MachinePowerModel {
     private var staticPower: Double = staticPowerPercent * maxPower
     private var constPower: Double = (maxPower - staticPower) / 100.0.pow(2)
 
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/ZeroIdlePowerDecorator.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/ZeroIdlePowerDecorator.kt
new file mode 100644
index 00000000..01deac5b
--- /dev/null
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/ZeroIdlePowerDecorator.kt
@@ -0,0 +1,12 @@
+package org.opendc.simulator.compute.power
+
+/**
+ * A decorator for ignoring the idle power when computing energy consumption of components.
+ *
+ * @param delegate The [MachinePowerModel] to delegate to.
+ */
+public class ZeroIdlePowerDecorator(private val delegate: MachinePowerModel) : MachinePowerModel {
+    override fun computeCpuPower(cpuUtil: Double): Double {
+        return if (cpuUtil == 0.0) 0.0 else delegate.computeCpuPower(cpuUtil)
+    }
+}
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/power/CpuPowerModelTest.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/MachinePowerModelTest.kt
index cd18b120..9fdd0363 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/power/CpuPowerModelTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/MachinePowerModelTest.kt
@@ -1,28 +1,24 @@
-package org.opendc.compute.simulator.power
+package org.opendc.simulator.compute.power
 
 import io.mockk.*
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
-import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.Arguments
 import org.junit.jupiter.params.provider.MethodSource
-import org.opendc.compute.simulator.power.api.CpuPowerModel
-import org.opendc.compute.simulator.power.models.*
-import org.opendc.simulator.compute.SimBareMetalMachine
 import java.util.stream.Stream
 import kotlin.math.pow
 
 @OptIn(ExperimentalCoroutinesApi::class)
-internal class CpuPowerModelTest {
+internal class MachinePowerModelTest {
     private val epsilon = 10.0.pow(-3)
     private val cpuUtil = 0.9
 
     @ParameterizedTest
-    @MethodSource("cpuPowerModelArgs")
+    @MethodSource("MachinePowerModelArgs")
     fun `compute power consumption given CPU loads`(
-        powerModel: CpuPowerModel,
+        powerModel: MachinePowerModel,
         expectedPowerConsumption: Double
     ) {
         val computedPowerConsumption = powerModel.computeCpuPower(cpuUtil)
@@ -30,9 +26,9 @@ internal class CpuPowerModelTest {
     }
 
     @ParameterizedTest
-    @MethodSource("cpuPowerModelArgs")
+    @MethodSource("MachinePowerModelArgs")
     fun `ignore idle power when computing power consumptions`(
-        powerModel: CpuPowerModel,
+        powerModel: MachinePowerModel,
         expectedPowerConsumption: Double
     ) {
         val zeroPowerModel = ZeroIdlePowerDecorator(powerModel)
@@ -40,33 +36,10 @@ internal class CpuPowerModelTest {
         assertEquals(0.0, computedPowerConsumption)
     }
 
-    @ParameterizedTest
-    @MethodSource("cpuPowerModelArgs")
-    fun `emit power draw for hosts by different models`(
-        powerModel: CpuPowerModel,
-        expectedPowerConsumption: Double
-    ) {
-        runBlockingTest {
-            val cpuLoads = flowOf(cpuUtil, cpuUtil, cpuUtil).stateIn(this)
-            val machine = mockkClass(SimBareMetalMachine::class)
-            every { machine.usage } returns cpuLoads
-
-            val serverPowerDraw = powerModel.getPowerDraw(machine)
-
-            assertEquals(
-                serverPowerDraw.first().toDouble(),
-                flowOf(expectedPowerConsumption).first().toDouble(),
-                epsilon
-            )
-
-            verify(exactly = 1) { machine.usage }
-        }
-    }
-
     @Suppress("unused")
     private companion object {
         @JvmStatic
-        fun cpuPowerModelArgs(): Stream<Arguments> = Stream.of(
+        fun MachinePowerModelArgs(): Stream<Arguments> = Stream.of(
             Arguments.of(ConstantPowerModel(0.0), 0.0),
             Arguments.of(LinearPowerModel(350.0, 200 / 350.0), 335.0),
             Arguments.of(SquarePowerModel(350.0, 200 / 350.0), 321.5),
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/power/PStatePowerModelTest.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/PStatePowerModelTest.kt
index e144e541..9116f928 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/power/PStatePowerModelTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/PStatePowerModelTest.kt
@@ -1,9 +1,8 @@
-package org.opendc.compute.simulator.power
+package org.opendc.simulator.compute.power
 
 import io.mockk.*
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
-import org.opendc.compute.simulator.power.models.PStatePowerModel
 import org.opendc.simulator.compute.SimBareMetalMachine
 import java.time.Clock
 
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-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/api/Powerable.kt b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlow.kt
index 780f2a29..bbf6ad44 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/power/api/Powerable.kt
+++ b/simulator/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlow.kt
@@ -1,7 +1,5 @@
 /*
- * MIT License
- *
- * Copyright (c) 2020 atlarge-research
+ * 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
@@ -22,16 +20,10 @@
  * SOFTWARE.
  */
 
-package org.opendc.compute.simulator.power.api
-
-import kotlinx.coroutines.flow.Flow
+package org.opendc.simulator.resources
 
 /**
- * An entity that is uses power from some power source.
+ * A [SimResourceFlow] acts as both a resource consumer and resource provider at the same time, simplifying bridging
+ * between different components.
  */
-public interface Powerable {
-    /**
-     * The power draw at the device's power supply in watts (W).w
-     */
-    public val powerDraw: Flow<Double>
-}
+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()) }
+    }
 }