merge: Migrate to flow-based simulation for low-level models

This pull request converts the `opendc-simulator-resources` module into a flow
simulator and adapts the existing low-level models (e.g., CPU, network, disk)
to this new flow simulator.

The flow simulator works differently from the uniform resource consumption
model, in that it models flow through a system of connections, as opposed to
resource consumptions. Concretely, this means that while in the uniform
resource consumption model, consumptions with the same usage are propagated to
the resources, in the flow simulator, only changes to the flow in the system
are propagated. 

Overall, this leads to less updates in the system and therefore higher
performance. The benchmarks shows that the new implementation obtains more than
double the performance of the old implementation. We have focused in the new
implementation on reducing the amount of work and memory
allocations/loads/stores per updates.

* Migrate from kotlinx-benchmark to jmh-gradle (for better profiling support)
* Use longer traces for benchmarks (to prevent measuring the benchmark
overhead)
* Use direct field access for perf-sensitive code
* Combine work and deadline to duration
* Add support for pushing flow from context (to eliminate the allocation for
every `SimResourceCommand`)
* Reduce memory allocations in SimResourceInterpreter, by revamping the way
timers are allocated.
* Simplify max-min aggregator implementation (by utilizing the new push
mechanism) 
* Invoke consumer callback on every invalidation (in order to propagate changes
downstream)
* Lazily push changes to resource context (by not updating the flow rate
immediately after a push, but only after an update)
* Remove onUpdate callback 
* Merge distributor and aggregator into switch 
* Separate push and pull flags 
* Remove failure callback from FlowSource 
* Create separate callbacks for remaining events 
* Make convergence callback optional
* Reduce field accesses in FlowConsumerContextImpl
* Optimize hot path in SimTraceWorkload
* Expose CPU time counters directly on hypervisor 
* Optimize telemetry collection

**Breaking API Changes**

* The entire `opendc-simulator-resources` module has been replaced by the
`opendc-simulator-flow` module.
* `SimHypervisor.Listener` has been removed in favour of a new interface that
exposes the performance counters of the hypervisor directly. To listen for
convergence, use `FlowConvergenceListener`.

33 files changed, 530 insertions, 421 deletions

diff --git a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
index 30797089..d654d58a 100644
--- a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
@@ -36,8 +36,9 @@ import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.workload.SimTraceWorkload
 import org.opendc.simulator.core.SimulationCoroutineScope
 import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.flow.FlowEngine
 import org.openjdk.jmh.annotations.*
+import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeUnit
 
 @State(Scope.Thread)
@@ -47,13 +48,13 @@ import java.util.concurrent.TimeUnit
 @OptIn(ExperimentalCoroutinesApi::class)
 class SimMachineBenchmarks {
     private lateinit var scope: SimulationCoroutineScope
-    private lateinit var interpreter: SimResourceInterpreter
+    private lateinit var engine: FlowEngine
     private lateinit var machineModel: MachineModel
 
     @Setup
     fun setUp() {
         scope = SimulationCoroutineScope()
-        interpreter = SimResourceInterpreter(scope.coroutineContext, scope.clock)
+        engine = FlowEngine(scope.coroutineContext, scope.clock)
 
         val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
 
@@ -63,22 +64,15 @@ class SimMachineBenchmarks {
         )
     }
 
-    @State(Scope.Benchmark)
+    @State(Scope.Thread)
     class Workload {
         lateinit var trace: Sequence<SimTraceWorkload.Fragment>
 
         @Setup
         fun setUp() {
-            trace = sequenceOf(
-                SimTraceWorkload.Fragment(0, 1000, 28.0, 1),
-                SimTraceWorkload.Fragment(1000, 1000, 3500.0, 1),
-                SimTraceWorkload.Fragment(2000, 1000, 0.0, 1),
-                SimTraceWorkload.Fragment(3000, 1000, 183.0, 1),
-                SimTraceWorkload.Fragment(4000, 1000, 400.0, 1),
-                SimTraceWorkload.Fragment(5000, 1000, 100.0, 1),
-                SimTraceWorkload.Fragment(6000, 1000, 3000.0, 1),
-                SimTraceWorkload.Fragment(7000, 1000, 4500.0, 1),
-            )
+            val random = ThreadLocalRandom.current()
+            val entries = List(10000) { SimTraceWorkload.Fragment(it * 1000L, 1000, random.nextDouble(0.0, 4500.0), 1) }
+            trace = entries.asSequence()
         }
     }
 
@@ -86,7 +80,7 @@ class SimMachineBenchmarks {
     fun benchmarkBareMetal(state: Workload) {
         return scope.runBlockingSimulation {
             val machine = SimBareMetalMachine(
-                interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+                engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
             return@runBlockingSimulation machine.run(SimTraceWorkload(state.trace))
         }
@@ -96,9 +90,9 @@ class SimMachineBenchmarks {
     fun benchmarkSpaceSharedHypervisor(state: Workload) {
         return scope.runBlockingSimulation {
             val machine = SimBareMetalMachine(
-                interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+                engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            val hypervisor = SimSpaceSharedHypervisor(interpreter)
+            val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
             launch { machine.run(hypervisor) }
 
@@ -117,9 +111,9 @@ class SimMachineBenchmarks {
     fun benchmarkFairShareHypervisorSingle(state: Workload) {
         return scope.runBlockingSimulation {
             val machine = SimBareMetalMachine(
-                interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+                engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            val hypervisor = SimFairShareHypervisor(interpreter)
+            val hypervisor = SimFairShareHypervisor(engine, null, null, null)
 
             launch { machine.run(hypervisor) }
 
@@ -138,9 +132,9 @@ class SimMachineBenchmarks {
     fun benchmarkFairShareHypervisorDouble(state: Workload) {
         return scope.runBlockingSimulation {
             val machine = SimBareMetalMachine(
-                interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+                engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            val hypervisor = SimFairShareHypervisor(interpreter)
+            val hypervisor = SimFairShareHypervisor(engine, null, null, null)
 
             launch { machine.run(hypervisor) }
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
index f9db048d..60a10f20 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
@@ -30,22 +30,22 @@ import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.NetworkAdapter
 import org.opendc.simulator.compute.model.StorageDevice
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.resources.*
+import org.opendc.simulator.flow.*
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
 
 /**
  * Abstract implementation of the [SimMachine] interface.
  *
- * @param interpreter The interpreter to manage the machine's resources.
+ * @param engine The engine to manage the machine's resources.
  * @param parent The parent simulation system.
  * @param model The model of the machine.
  */
 public abstract class SimAbstractMachine(
-    protected val interpreter: SimResourceInterpreter,
-    final override val parent: SimResourceSystem?,
+    protected val engine: FlowEngine,
+    private val parent: FlowConvergenceListener?,
     final override val model: MachineModel
-) : SimMachine, SimResourceSystem {
+) : SimMachine, FlowConvergenceListener {
     /**
      * The resources allocated for this machine.
      */
@@ -54,17 +54,17 @@ public abstract class SimAbstractMachine(
     /**
      * The memory interface of the machine.
      */
-    public val memory: SimMemory = Memory(SimResourceSource(model.memory.sumOf { it.size }.toDouble(), interpreter), model.memory)
+    public val memory: SimMemory = Memory(FlowSink(engine, model.memory.sumOf { it.size }.toDouble()), model.memory)
 
     /**
      * The network interfaces available to the machine.
      */
-    public val net: List<SimNetworkInterface> = model.net.mapIndexed { i, adapter -> NetworkAdapterImpl(adapter, i) }
+    public val net: List<SimNetworkInterface> = model.net.mapIndexed { i, adapter -> NetworkAdapterImpl(engine, adapter, i) }
 
     /**
      * The network interfaces available to the machine.
      */
-    public val storage: List<SimStorageInterface> = model.storage.mapIndexed { i, device -> StorageDeviceImpl(interpreter, device, i) }
+    public val storage: List<SimStorageInterface> = model.storage.mapIndexed { i, device -> StorageDeviceImpl(engine, device, i) }
 
     /**
      * The peripherals of the machine.
@@ -82,7 +82,7 @@ public abstract class SimAbstractMachine(
     private var cont: Continuation<Unit>? = null
 
     /**
-     * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
+     * Converge the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
      */
     override suspend fun run(workload: SimWorkload, meta: Map<String, Any>) {
         check(!isTerminated) { "Machine is terminated" }
@@ -96,14 +96,14 @@ public abstract class SimAbstractMachine(
             // Cancel all cpus on cancellation
             cont.invokeOnCancellation {
                 this.cont = null
-                interpreter.batch {
+                engine.batch {
                     for (cpu in cpus) {
                         cpu.cancel()
                     }
                 }
             }
 
-            interpreter.batch { workload.onStart(ctx) }
+            engine.batch { workload.onStart(ctx) }
         }
     }
 
@@ -116,11 +116,15 @@ public abstract class SimAbstractMachine(
         cancel()
     }
 
+    override fun onConverge(now: Long, delta: Long) {
+        parent?.onConverge(now, delta)
+    }
+
     /**
      * Cancel the workload that is currently running on the machine.
      */
     private fun cancel() {
-        interpreter.batch {
+        engine.batch {
             for (cpu in cpus) {
                 cpu.cancel()
             }
@@ -137,8 +141,8 @@ public abstract class SimAbstractMachine(
      * The execution context in which the workload runs.
      */
     private inner class Context(override val meta: Map<String, Any>) : SimMachineContext {
-        override val interpreter: SimResourceInterpreter
-            get() = this@SimAbstractMachine.interpreter
+        override val engine: FlowEngine
+            get() = this@SimAbstractMachine.engine
 
         override val cpus: List<SimProcessingUnit> = this@SimAbstractMachine.cpus
 
@@ -154,7 +158,7 @@ public abstract class SimAbstractMachine(
     /**
      * The [SimMemory] implementation for a machine.
      */
-    private class Memory(source: SimResourceSource, override val models: List<MemoryUnit>) : SimMemory, SimResourceProvider by source {
+    private class Memory(source: FlowSink, override val models: List<MemoryUnit>) : SimMemory, FlowConsumer by source {
         override fun toString(): String = "SimAbstractMachine.Memory"
     }
 
@@ -162,6 +166,7 @@ public abstract class SimAbstractMachine(
      * The [SimNetworkAdapter] implementation for a machine.
      */
     private class NetworkAdapterImpl(
+        private val engine: FlowEngine,
         model: NetworkAdapter,
         index: Int
     ) : SimNetworkAdapter(), SimNetworkInterface {
@@ -169,18 +174,18 @@ public abstract class SimAbstractMachine(
 
         override val bandwidth: Double = model.bandwidth
 
-        override val provider: SimResourceProvider
+        override val provider: FlowConsumer
             get() = _rx
 
-        override fun createConsumer(): SimResourceConsumer = _tx
+        override fun createConsumer(): FlowSource = _tx
 
-        override val tx: SimResourceProvider
+        override val tx: FlowConsumer
             get() = _tx
-        private val _tx = SimResourceForwarder()
+        private val _tx = FlowForwarder(engine)
 
-        override val rx: SimResourceConsumer
+        override val rx: FlowSource
             get() = _rx
-        private val _rx = SimResourceForwarder()
+        private val _rx = FlowForwarder(engine)
 
         override fun toString(): String = "SimAbstractMachine.NetworkAdapterImpl[name=$name,bandwidth=$bandwidth]"
     }
@@ -189,7 +194,7 @@ public abstract class SimAbstractMachine(
      * The [SimStorageInterface] implementation for a machine.
      */
     private class StorageDeviceImpl(
-        interpreter: SimResourceInterpreter,
+        engine: FlowEngine,
         model: StorageDevice,
         index: Int
     ) : SimStorageInterface {
@@ -197,9 +202,9 @@ public abstract class SimAbstractMachine(
 
         override val capacity: Double = model.capacity
 
-        override val read: SimResourceProvider = SimResourceSource(model.readBandwidth, interpreter)
+        override val read: FlowConsumer = FlowSink(engine, model.readBandwidth)
 
-        override val write: SimResourceProvider = SimResourceSource(model.writeBandwidth, interpreter)
+        override val write: FlowConsumer = FlowSink(engine, model.writeBandwidth)
 
         override fun toString(): String = "SimAbstractMachine.StorageDeviceImpl[name=$name,capacity=$capacity]"
     }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
index 639ca450..9140d31b 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
@@ -26,8 +26,9 @@ import org.opendc.simulator.compute.device.SimPsu
 import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.power.PowerDriver
-import org.opendc.simulator.resources.*
-import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.FlowEngine
+import kotlin.math.max
 
 /**
  * A simulated bare-metal machine that is able to run a single workload.
@@ -35,30 +36,38 @@ import org.opendc.simulator.resources.SimResourceInterpreter
  * A [SimBareMetalMachine] is a stateful object, and you should be careful when operating this object concurrently. For
  * example, the class expects only a single concurrent call to [run].
  *
- * @param interpreter The [SimResourceInterpreter] to drive the simulation.
+ * @param engine The [FlowEngine] to drive the simulation.
  * @param model The machine model to simulate.
  * @param powerDriver The power driver to use.
  * @param psu The power supply of the machine.
  * @param parent The parent simulation system.
  */
 public class SimBareMetalMachine(
-    interpreter: SimResourceInterpreter,
+    engine: FlowEngine,
     model: MachineModel,
     powerDriver: PowerDriver,
     public val psu: SimPsu = SimPsu(500.0, mapOf(1.0 to 1.0)),
-    parent: SimResourceSystem? = null,
-) : SimAbstractMachine(interpreter, parent, model) {
+    parent: FlowConvergenceListener? = null,
+) : SimAbstractMachine(engine, parent, model) {
     /**
-     * The power draw of the machine onto the PSU.
+     * The current power usage of the machine (without PSU loss) in W.
      */
-    public val powerDraw: Double
-        get() = powerDriverLogic.computePower()
+    public val powerUsage: Double
+        get() = _powerUsage
+    private var _powerUsage = 0.0
+
+    /**
+     * The total energy usage of the machine (without PSU loss) in Joules.
+     */
+    public val energyUsage: Double
+        get() = _energyUsage
+    private var _energyUsage = 0.0
 
     /**
      * The processing units of the machine.
      */
     override val cpus: List<SimProcessingUnit> = model.cpus.map { cpu ->
-        Cpu(SimResourceSource(cpu.frequency, interpreter, this@SimBareMetalMachine), cpu)
+        Cpu(FlowSink(engine, cpu.frequency, this@SimBareMetalMachine), cpu)
     }
 
     /**
@@ -66,8 +75,20 @@ public class SimBareMetalMachine(
      */
     private val powerDriverLogic = powerDriver.createLogic(this, cpus)
 
-    override fun onConverge(timestamp: Long) {
+    private var _lastConverge = Long.MAX_VALUE
+
+    override fun onConverge(now: Long, delta: Long) {
+        // Update the PSU stage
         psu.update()
+
+        val lastConverge = _lastConverge
+        _lastConverge = now
+        val duration = max(0, now - lastConverge)
+        if (duration > 0) {
+            // Compute the power and energy usage of the machine
+            _energyUsage += _powerUsage * (duration / 1000.0)
+            _powerUsage = powerDriverLogic.computePower()
+        }
     }
 
     init {
@@ -78,9 +99,9 @@ public class SimBareMetalMachine(
      * A [SimProcessingUnit] of a bare-metal machine.
      */
     private class Cpu(
-        private val source: SimResourceSource,
+        private val source: FlowSink,
         override val model: ProcessingUnit
-    ) : SimProcessingUnit, SimResourceProvider by source {
+    ) : SimProcessingUnit, FlowConsumer by source {
         override var capacity: Double
             get() = source.capacity
             set(value) {
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt
index d8dd8205..ab0b56ae 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachine.kt
@@ -41,7 +41,7 @@ public interface SimMachine : AutoCloseable {
     public val peripherals: List<SimPeripheral>
 
     /**
-     * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
+     * Converge the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
      */
     public suspend fun run(workload: SimWorkload, meta: Map<String, Any> = emptyMap())
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt
index 6996a30d..1317f728 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt
@@ -22,7 +22,7 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * A simulated execution context in which a bootable image runs. This interface represents the
@@ -31,9 +31,9 @@ import org.opendc.simulator.resources.SimResourceInterpreter
  */
 public interface SimMachineContext : AutoCloseable {
     /**
-     * The resource interpreter that simulates the machine.
+     * The [FlowEngine] that simulates the machine.
      */
-    public val interpreter: SimResourceInterpreter
+    public val engine: FlowEngine
 
     /**
      * The metadata associated with the context.
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMemory.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMemory.kt
index 6623df23..b1aef495 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMemory.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMemory.kt
@@ -23,12 +23,12 @@
 package org.opendc.simulator.compute
 
 import org.opendc.simulator.compute.model.MemoryUnit
-import org.opendc.simulator.resources.SimResourceProvider
+import org.opendc.simulator.flow.FlowConsumer
 
 /**
  * An interface to control the memory usage of simulated workloads.
  */
-public interface SimMemory : SimResourceProvider {
+public interface SimMemory : FlowConsumer {
     /**
      * The models representing the static information of the memory units supporting this interface.
      */
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimNetworkInterface.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimNetworkInterface.kt
index 1ac126ae..660b2871 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimNetworkInterface.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimNetworkInterface.kt
@@ -22,8 +22,8 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceProvider
+import org.opendc.simulator.flow.FlowConsumer
+import org.opendc.simulator.flow.FlowSource
 
 /**
  * A firmware interface to a network adapter.
@@ -42,10 +42,10 @@ public interface SimNetworkInterface {
     /**
      * The resource provider for the transmit channel of the network interface.
      */
-    public val tx: SimResourceProvider
+    public val tx: FlowConsumer
 
     /**
      * The resource consumer for the receive channel of the network interface.
      */
-    public val rx: SimResourceConsumer
+    public val rx: FlowSource
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimProcessingUnit.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimProcessingUnit.kt
index 93c9ddfa..c9f36ece 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimProcessingUnit.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimProcessingUnit.kt
@@ -23,12 +23,12 @@
 package org.opendc.simulator.compute
 
 import org.opendc.simulator.compute.model.ProcessingUnit
-import org.opendc.simulator.resources.SimResourceProvider
+import org.opendc.simulator.flow.FlowConsumer
 
 /**
  * A simulated processing unit.
  */
-public interface SimProcessingUnit : SimResourceProvider {
+public interface SimProcessingUnit : FlowConsumer {
     /**
      * The capacity of the processing unit, which can be adjusted by the workload if supported by the machine.
      */
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimStorageInterface.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimStorageInterface.kt
index 21a801f1..3d648671 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimStorageInterface.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimStorageInterface.kt
@@ -22,7 +22,7 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.resources.SimResourceProvider
+import org.opendc.simulator.flow.FlowConsumer
 
 /**
  * A firmware interface to a storage device.
@@ -41,10 +41,10 @@ public interface SimStorageInterface {
     /**
      * The resource provider for the read operations of the storage device.
      */
-    public val read: SimResourceProvider
+    public val read: FlowConsumer
 
     /**
      * The resource consumer for the write operation of the storage device.
      */
-    public val write: SimResourceProvider
+    public val write: FlowConsumer
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/device/SimPsu.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/device/SimPsu.kt
index 0a7dc40f..09defbb5 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/device/SimPsu.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/device/SimPsu.kt
@@ -23,11 +23,9 @@
 package org.opendc.simulator.compute.device
 
 import org.opendc.simulator.compute.power.PowerDriver
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
 import org.opendc.simulator.power.SimPowerInlet
-import org.opendc.simulator.resources.SimResourceCommand
-import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
-import org.opendc.simulator.resources.SimResourceEvent
 import java.util.*
 
 /**
@@ -55,7 +53,7 @@ public class SimPsu(
     /**
      * The consumer context.
      */
-    private var _ctx: SimResourceContext? = null
+    private var _ctx: FlowConnection? = null
 
     /**
      * The driver that is connected to the PSU.
@@ -70,7 +68,7 @@ public class SimPsu(
      * Update the power draw of the PSU.
      */
     public fun update() {
-        _ctx?.interrupt()
+        _ctx?.pull()
     }
 
     /**
@@ -82,23 +80,24 @@ public class SimPsu(
         update()
     }
 
-    override fun createConsumer(): SimResourceConsumer = object : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext): SimResourceCommand {
-            val powerDraw = computePowerDraw(_driver?.computePower() ?: 0.0)
+    override fun createSource(): FlowSource = object : FlowSource {
+        override fun onStart(conn: FlowConnection, now: Long) {
+            _ctx = conn
+            conn.shouldSourceConverge = true
+        }
 
-            return if (powerDraw > 0.0)
-                SimResourceCommand.Consume(Double.POSITIVE_INFINITY, powerDraw, Long.MAX_VALUE)
-            else
-                SimResourceCommand.Idle()
+        override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+            _ctx = null
+        }
+
+        override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+            val powerDraw = computePowerDraw(_driver?.computePower() ?: 0.0)
+            conn.push(powerDraw)
+            return Long.MAX_VALUE
         }
 
-        override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
-            when (event) {
-                SimResourceEvent.Start -> _ctx = ctx
-                SimResourceEvent.Run -> _powerDraw = ctx.speed
-                SimResourceEvent.Exit -> _ctx = null
-                else -> {}
-            }
+        override fun onConverge(conn: FlowConnection, now: Long, delta: Long) {
+            _powerDraw = conn.rate
         }
     }
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt
index 98271fb0..aac8b959 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt
@@ -28,29 +28,31 @@ import org.opendc.simulator.compute.kernel.cpufreq.ScalingPolicy
 import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain
 import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.ProcessingUnit
-import org.opendc.simulator.resources.*
-import org.opendc.simulator.resources.SimResourceSwitch
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.mux.FlowMultiplexer
+import kotlin.math.roundToLong
 
 /**
  * Abstract implementation of the [SimHypervisor] interface.
  *
- * @param interpreter The resource interpreter to use.
+ * @param engine The [FlowEngine] to drive the simulation.
  * @param scalingGovernor The scaling governor to use for scaling the CPU frequency of the underlying hardware.
  */
 public abstract class SimAbstractHypervisor(
-    private val interpreter: SimResourceInterpreter,
-    private val scalingGovernor: ScalingGovernor? = null,
+    protected val engine: FlowEngine,
+    private val listener: FlowConvergenceListener?,
+    private val scalingGovernor: ScalingGovernor?,
     protected val interferenceDomain: VmInterferenceDomain? = null
-) : SimHypervisor {
+) : SimHypervisor, FlowConvergenceListener {
     /**
      * The machine on which the hypervisor runs.
      */
-    private lateinit var context: SimMachineContext
+    protected lateinit var context: SimMachineContext
 
     /**
      * The resource switch to use.
      */
-    private lateinit var switch: SimResourceSwitch
+    protected abstract val mux: FlowMultiplexer
 
     /**
      * The virtual machines running on this hypervisor.
@@ -62,39 +64,73 @@ public abstract class SimAbstractHypervisor(
     /**
      * The resource counters associated with the hypervisor.
      */
-    public override val counters: SimResourceCounters
-        get() = switch.counters
+    public override val counters: SimHypervisorCounters
+        get() = _counters
+    private val _counters = object : SimHypervisorCounters {
+        @JvmField var d = 1.0 // Number of CPUs divided by total CPU capacity
+
+        override var cpuActiveTime: Long = 0L
+        override var cpuIdleTime: Long = 0L
+        override var cpuStealTime: Long = 0L
+        override var cpuLostTime: Long = 0L
+
+        private var _previousDemand = 0.0
+        private var _previousActual = 0.0
+        private var _previousRemaining = 0.0
+        private var _previousInterference = 0.0
+
+        /**
+         * Record the CPU time of the hypervisor.
+         */
+        fun record() {
+            val counters = mux.counters
+            val demand = counters.demand
+            val actual = counters.actual
+            val remaining = counters.remaining
+            val interference = counters.interference
+
+            val demandDelta = demand - _previousDemand
+            val actualDelta = actual - _previousActual
+            val remainingDelta = remaining - _previousRemaining
+            val interferenceDelta = interference - _previousInterference
+
+            _previousDemand = demand
+            _previousActual = actual
+            _previousRemaining = remaining
+            _previousInterference = interference
+
+            cpuActiveTime += (actualDelta * d).roundToLong()
+            cpuIdleTime += (remainingDelta * d).roundToLong()
+            cpuStealTime += ((demandDelta - actualDelta) * d).roundToLong()
+            cpuLostTime += (interferenceDelta * d).roundToLong()
+        }
+    }
 
     /**
-     * The scaling governors attached to the physical CPUs backing this hypervisor.
+     * The CPU capacity of the hypervisor in MHz.
      */
-    private val governors = mutableListOf<ScalingGovernor.Logic>()
+    override val cpuCapacity: Double
+        get() = mux.capacity
 
     /**
-     * Construct the [SimResourceSwitch] implementation that performs the actual scheduling of the CPUs.
+     * The CPU demand of the hypervisor in MHz.
      */
-    public abstract fun createSwitch(ctx: SimMachineContext): SimResourceSwitch
+    override val cpuDemand: Double
+        get() = mux.demand
 
     /**
-     * Check whether the specified machine model fits on this hypervisor.
+     * The CPU usage of the hypervisor in MHz.
      */
-    public abstract fun canFit(model: MachineModel, switch: SimResourceSwitch): Boolean
+    override val cpuUsage: Double
+        get() = mux.rate
 
     /**
-     * Trigger the governors to recompute the scaling limits.
+     * The scaling governors attached to the physical CPUs backing this hypervisor.
      */
-    protected fun triggerGovernors(load: Double) {
-        for (governor in governors) {
-            governor.onLimit(load)
-        }
-    }
+    private val governors = mutableListOf<ScalingGovernor.Logic>()
 
     /* SimHypervisor */
-    override fun canFit(model: MachineModel): Boolean {
-        return canFit(model, switch)
-    }
-
-    override fun createMachine(model: MachineModel, interferenceId: String?): SimMachine {
+    override fun createMachine(model: MachineModel, interferenceId: String?): SimVirtualMachine {
         require(canFit(model)) { "Machine does not fit" }
         val vm = VirtualMachine(model, interferenceId)
         _vms.add(vm)
@@ -104,7 +140,13 @@ public abstract class SimAbstractHypervisor(
     /* SimWorkload */
     override fun onStart(ctx: SimMachineContext) {
         context = ctx
-        switch = createSwitch(ctx)
+
+        _cpuCount = ctx.cpus.size
+        _cpuCapacity = ctx.cpus.sumOf { it.model.frequency }
+        _counters.d = _cpuCount / _cpuCapacity * 1000L
+
+        // Clear the existing outputs of the multiplexer
+        mux.clearOutputs()
 
         for (cpu in ctx.cpus) {
             val governor = scalingGovernor?.createLogic(ScalingPolicyImpl(cpu))
@@ -113,16 +155,31 @@ public abstract class SimAbstractHypervisor(
                 governor.onStart()
             }
 
-            switch.addInput(cpu)
+            cpu.startConsumer(mux.newOutput())
         }
     }
 
+    private var _cpuCount = 0
+    private var _cpuCapacity = 0.0
+
+    /* FlowConvergenceListener */
+    override fun onConverge(now: Long, delta: Long) {
+        _counters.record()
+
+        val load = cpuDemand / cpuCapacity
+        for (governor in governors) {
+            governor.onLimit(load)
+        }
+
+        listener?.onConverge(now, delta)
+    }
+
     /**
      * A virtual machine running on the hypervisor.
      *
      * @param model The machine model of the virtual machine.
      */
-    private inner class VirtualMachine(model: MachineModel, interferenceId: String? = null) : SimAbstractMachine(interpreter, parent = null, model) {
+    private inner class VirtualMachine(model: MachineModel, interferenceId: String? = null) : SimAbstractMachine(engine, parent = null, model), SimVirtualMachine {
         /**
          * The interference key of this virtual machine.
          */
@@ -131,7 +188,42 @@ public abstract class SimAbstractHypervisor(
         /**
          * The vCPUs of the machine.
          */
-        override val cpus = model.cpus.map { VCpu(switch.newOutput(interferenceKey), it) }
+        override val cpus = model.cpus.map { VCpu(mux, mux.newInput(interferenceKey), it) }
+
+        /**
+         * The resource counters associated with the hypervisor.
+         */
+        override val counters: SimHypervisorCounters
+            get() = _counters
+        private val _counters = object : SimHypervisorCounters {
+            private val d = cpus.size / cpus.sumOf { it.model.frequency } * 1000
+
+            override val cpuActiveTime: Long
+                get() = (cpus.sumOf { it.counters.actual } * d).roundToLong()
+            override val cpuIdleTime: Long
+                get() = (cpus.sumOf { it.counters.actual + it.counters.remaining } * d).roundToLong()
+            override val cpuStealTime: Long
+                get() = (cpus.sumOf { it.counters.demand - it.counters.actual } * d).roundToLong()
+            override val cpuLostTime: Long = 0L
+        }
+
+        /**
+         * The CPU capacity of the hypervisor in MHz.
+         */
+        override val cpuCapacity: Double
+            get() = cpus.sumOf(FlowConsumer::capacity)
+
+        /**
+         * The CPU demand of the hypervisor in MHz.
+         */
+        override val cpuDemand: Double
+            get() = cpus.sumOf(FlowConsumer::demand)
+
+        /**
+         * The CPU usage of the hypervisor in MHz.
+         */
+        override val cpuUsage: Double
+            get() = cpus.sumOf(FlowConsumer::rate)
 
         override fun close() {
             super.close()
@@ -145,17 +237,16 @@ public abstract class SimAbstractHypervisor(
                 interferenceDomain?.leave(interferenceKey)
             }
         }
-
-        override fun onConverge(timestamp: Long) {}
     }
 
     /**
      * A [SimProcessingUnit] of a virtual machine.
      */
     private class VCpu(
-        private val source: SimResourceCloseableProvider,
+        private val switch: FlowMultiplexer,
+        private val source: FlowConsumer,
         override val model: ProcessingUnit
-    ) : SimProcessingUnit, SimResourceCloseableProvider by source {
+    ) : SimProcessingUnit, FlowConsumer by source {
         override var capacity: Double
             get() = source.capacity
             set(_) {
@@ -163,6 +254,13 @@ public abstract class SimAbstractHypervisor(
             }
 
         override fun toString(): String = "SimAbstractHypervisor.VCpu[model=$model]"
+
+        /**
+         * Close the CPU
+         */
+        fun close() {
+            switch.removeInput(source)
+        }
     }
 
     /**
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt
index 3b44292d..36f76650 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt
@@ -23,79 +23,34 @@
 package org.opendc.simulator.compute.kernel
 
 import org.opendc.simulator.compute.SimMachine
-import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor
 import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain
 import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.SimResourceSwitch
-import org.opendc.simulator.resources.SimResourceSwitchMaxMin
-import org.opendc.simulator.resources.SimResourceSystem
+import org.opendc.simulator.flow.FlowConvergenceListener
+import org.opendc.simulator.flow.FlowEngine
+import org.opendc.simulator.flow.mux.FlowMultiplexer
+import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
 
 /**
  * A [SimHypervisor] that distributes the computing requirements of multiple [SimWorkload]s on a single [SimMachine]
  * concurrently using weighted fair sharing.
  *
- * @param interpreter The interpreter to manage the machine's resources.
- * @param parent The parent simulation system.
+ * @param engine The [FlowEngine] to manage the machine's resources.
+ * @param listener The listener for the convergence of the system.
  * @param scalingGovernor The CPU frequency scaling governor to use for the hypervisor.
  * @param interferenceDomain The resource interference domain to which the hypervisor belongs.
- * @param listener The hypervisor listener to use.
  */
 public class SimFairShareHypervisor(
-    private val interpreter: SimResourceInterpreter,
-    private val parent: SimResourceSystem? = null,
-    scalingGovernor: ScalingGovernor? = null,
-    interferenceDomain: VmInterferenceDomain? = null,
-    private val listener: SimHypervisor.Listener? = null
-) : SimAbstractHypervisor(interpreter, scalingGovernor, interferenceDomain) {
-
-    override fun canFit(model: MachineModel, switch: SimResourceSwitch): Boolean = true
-
-    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
-        return SwitchSystem(ctx).switch
-    }
-
-    private inner class SwitchSystem(private val ctx: SimMachineContext) : SimResourceSystem {
-        val switch = SimResourceSwitchMaxMin(interpreter, this, interferenceDomain)
-
-        override val parent: SimResourceSystem? = this@SimFairShareHypervisor.parent
-
-        private var lastCpuUsage = 0.0
-        private var lastCpuDemand = 0.0
-        private var lastDemand = 0.0
-        private var lastActual = 0.0
-        private var lastOvercommit = 0.0
-        private var lastInterference = 0.0
-        private var lastReport = Long.MIN_VALUE
-
-        override fun onConverge(timestamp: Long) {
-            val listener = listener ?: return
-            val counters = switch.counters
-
-            if (timestamp > lastReport) {
-                listener.onSliceFinish(
-                    this@SimFairShareHypervisor,
-                    counters.demand - lastDemand,
-                    counters.actual - lastActual,
-                    counters.overcommit - lastOvercommit,
-                    counters.interference - lastInterference,
-                    lastCpuUsage,
-                    lastCpuDemand
-                )
-            }
-            lastReport = timestamp
-
-            lastCpuDemand = switch.inputs.sumOf { it.demand }
-            lastCpuUsage = switch.inputs.sumOf { it.speed }
-            lastDemand = counters.demand
-            lastActual = counters.actual
-            lastOvercommit = counters.overcommit
-            lastInterference = counters.interference
-
-            val load = lastCpuDemand / ctx.cpus.sumOf { it.model.frequency }
-            triggerGovernors(load)
-        }
-    }
+    engine: FlowEngine,
+    listener: FlowConvergenceListener?,
+    scalingGovernor: ScalingGovernor?,
+    interferenceDomain: VmInterferenceDomain?,
+) : SimAbstractHypervisor(engine, listener, scalingGovernor, interferenceDomain) {
+    /**
+     * The multiplexer that distributes the computing capacity.
+     */
+    override val mux: FlowMultiplexer = MaxMinFlowMultiplexer(engine, this, interferenceDomain)
+
+    override fun canFit(model: MachineModel): Boolean = true
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt
index 8d0592ec..3136f4c8 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt
@@ -24,8 +24,8 @@ package org.opendc.simulator.compute.kernel
 
 import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor
 import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.SimResourceSystem
+import org.opendc.simulator.flow.FlowConvergenceListener
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * A [SimHypervisorProvider] for the [SimFairShareHypervisor] implementation.
@@ -34,16 +34,9 @@ public class SimFairShareHypervisorProvider : SimHypervisorProvider {
     override val id: String = "fair-share"
 
     override fun create(
-        interpreter: SimResourceInterpreter,
-        parent: SimResourceSystem?,
+        engine: FlowEngine,
+        listener: FlowConvergenceListener?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
-        listener: SimHypervisor.Listener?
-    ): SimHypervisor = SimFairShareHypervisor(
-        interpreter,
-        parent,
-        scalingGovernor = scalingGovernor,
-        interferenceDomain = interferenceDomain,
-        listener = listener
-    )
+    ): SimHypervisor = SimFairShareHypervisor(engine, listener, scalingGovernor, interferenceDomain)
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt
index 3b49d515..57d4cf20 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt
@@ -25,7 +25,6 @@ package org.opendc.simulator.compute.kernel
 import org.opendc.simulator.compute.SimMachine
 import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.resources.SimResourceCounters
 
 /**
  * A SimHypervisor facilitates the execution of multiple concurrent [SimWorkload]s, while acting as a single workload
@@ -40,7 +39,22 @@ public interface SimHypervisor : SimWorkload {
     /**
      * The resource counters associated with the hypervisor.
      */
-    public val counters: SimResourceCounters
+    public val counters: SimHypervisorCounters
+
+    /**
+     * The CPU usage of the hypervisor in MHz.
+     */
+    public val cpuUsage: Double
+
+    /**
+     * The CPU usage of the hypervisor in MHz.
+     */
+    public val cpuDemand: Double
+
+    /**
+     * The CPU capacity of the hypervisor in MHz.
+     */
+    public val cpuCapacity: Double
 
     /**
      * Determine whether the specified machine characterized by [model] can fit on this hypervisor at this moment.
@@ -53,23 +67,5 @@ public interface SimHypervisor : SimWorkload {
      * @param model The machine to create.
      * @param interferenceId An identifier for the interference model.
      */
-    public fun createMachine(model: MachineModel, interferenceId: String? = null): SimMachine
-
-    /**
-     * Event listener for hypervisor events.
-     */
-    public interface Listener {
-        /**
-         * This method is invoked when a slice is finished.
-         */
-        public fun onSliceFinish(
-            hypervisor: SimHypervisor,
-            totalWork: Double,
-            grantedWork: Double,
-            overcommittedWork: Double,
-            interferedWork: Double,
-            cpuUsage: Double,
-            cpuDemand: Double
-        )
-    }
+    public fun createMachine(model: MachineModel, interferenceId: String? = null): SimVirtualMachine
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorCounters.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorCounters.kt
new file mode 100644
index 00000000..030d9c5f
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorCounters.kt
@@ -0,0 +1,48 @@
+/*
+ * 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.kernel
+
+/**
+ * Performance counters of a [SimHypervisor].
+ */
+public interface SimHypervisorCounters {
+    /**
+     * The amount of time (in milliseconds) the CPUs of the hypervisor were actively running.
+     */
+    public val cpuActiveTime: Long
+
+    /**
+     * The amount of time (in milliseconds) the CPUs of the hypervisor were idle.
+     */
+    public val cpuIdleTime: Long
+
+    /**
+     * The amount of CPU time (in milliseconds) that virtual machines were ready to run, but were not able to.
+     */
+    public val cpuStealTime: Long
+
+    /**
+     * The amount of CPU time (in milliseconds) that was lost due to interference between virtual machines.
+     */
+    public val cpuLostTime: Long
+}
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt
index b307a34d..483217af 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt
@@ -24,8 +24,8 @@ package org.opendc.simulator.compute.kernel
 
 import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor
 import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.SimResourceSystem
+import org.opendc.simulator.flow.FlowConvergenceListener
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * A service provider interface for constructing a [SimHypervisor].
@@ -40,13 +40,12 @@ public interface SimHypervisorProvider {
     public val id: String
 
     /**
-     * Create a [SimHypervisor] instance with the specified [listener].
+     * Create a new [SimHypervisor] instance.
      */
     public fun create(
-        interpreter: SimResourceInterpreter,
-        parent: SimResourceSystem? = null,
+        engine: FlowEngine,
+        listener: FlowConvergenceListener? = null,
         scalingGovernor: ScalingGovernor? = null,
         interferenceDomain: VmInterferenceDomain? = null,
-        listener: SimHypervisor.Listener? = null
     ): SimHypervisor
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt
index ac1c0250..82f8df38 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt
@@ -22,21 +22,24 @@
 
 package org.opendc.simulator.compute.kernel
 
-import org.opendc.simulator.compute.SimMachineContext
+import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor
 import org.opendc.simulator.compute.model.MachineModel
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.SimResourceSwitch
-import org.opendc.simulator.resources.SimResourceSwitchExclusive
+import org.opendc.simulator.flow.FlowConvergenceListener
+import org.opendc.simulator.flow.FlowEngine
+import org.opendc.simulator.flow.mux.FlowMultiplexer
+import org.opendc.simulator.flow.mux.ForwardingFlowMultiplexer
 
 /**
  * A [SimHypervisor] that allocates its sub-resources exclusively for the virtual machine that it hosts.
  */
-public class SimSpaceSharedHypervisor(interpreter: SimResourceInterpreter) : SimAbstractHypervisor(interpreter) {
-    override fun canFit(model: MachineModel, switch: SimResourceSwitch): Boolean {
-        return switch.inputs.size - switch.outputs.size >= model.cpus.size
-    }
+public class SimSpaceSharedHypervisor(
+    engine: FlowEngine,
+    listener: FlowConvergenceListener?,
+    scalingGovernor: ScalingGovernor?,
+) : SimAbstractHypervisor(engine, listener, scalingGovernor) {
+    override val mux: FlowMultiplexer = ForwardingFlowMultiplexer(engine)
 
-    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
-        return SimResourceSwitchExclusive()
+    override fun canFit(model: MachineModel): Boolean {
+        return mux.outputs.size - mux.inputs.size >= model.cpus.size
     }
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt
index 3906cb9a..dd6fb0b1 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt
@@ -24,8 +24,8 @@ package org.opendc.simulator.compute.kernel
 
 import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor
 import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.SimResourceSystem
+import org.opendc.simulator.flow.FlowConvergenceListener
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * A [SimHypervisorProvider] for the [SimSpaceSharedHypervisor] implementation.
@@ -34,10 +34,9 @@ public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider {
     override val id: String = "space-shared"
 
     override fun create(
-        interpreter: SimResourceInterpreter,
-        parent: SimResourceSystem?,
+        engine: FlowEngine,
+        listener: FlowConvergenceListener?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
-        listener: SimHypervisor.Listener?
-    ): SimHypervisor = SimSpaceSharedHypervisor(interpreter)
+    ): SimHypervisor = SimSpaceSharedHypervisor(engine, listener, scalingGovernor)
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimVirtualMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimVirtualMachine.kt
new file mode 100644
index 00000000..36219ef2
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimVirtualMachine.kt
@@ -0,0 +1,50 @@
+/*
+ * 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.kernel
+
+import org.opendc.simulator.compute.SimMachine
+
+/**
+ * A virtual [SimMachine] running on top of another [SimMachine].
+ */
+public interface SimVirtualMachine : SimMachine {
+    /**
+     * The resource counters associated with the virtual machine.
+     */
+    public val counters: SimHypervisorCounters
+
+    /**
+     * The CPU usage of the VM in MHz.
+     */
+    public val cpuUsage: Double
+
+    /**
+     * The CPU usage of the VM in MHz.
+     */
+    public val cpuDemand: Double
+
+    /**
+     * The CPU capacity of the VM in MHz.
+     */
+    public val cpuCapacity: Double
+}
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt
index 1801fcd0..b737d61a 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt
@@ -22,8 +22,8 @@
 
 package org.opendc.simulator.compute.kernel.interference
 
-import org.opendc.simulator.resources.interference.InterferenceDomain
-import org.opendc.simulator.resources.interference.InterferenceKey
+import org.opendc.simulator.flow.interference.InterferenceDomain
+import org.opendc.simulator.flow.interference.InterferenceKey
 
 /**
  * The interference domain of a hypervisor.
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt
index c2e00c8e..b3d72507 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt
@@ -22,7 +22,7 @@
 
 package org.opendc.simulator.compute.kernel.interference
 
-import org.opendc.simulator.resources.interference.InterferenceKey
+import org.opendc.simulator.flow.interference.InterferenceKey
 import java.util.*
 
 /**
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/PStatePowerDriver.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/PStatePowerDriver.kt
index 6577fbfc..f71446f8 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/PStatePowerDriver.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/PStatePowerDriver.kt
@@ -46,7 +46,7 @@ public class PStatePowerDriver(states: Map<Double, PowerModel>) : PowerDriver {
 
             for (cpu in cpus) {
                 targetFreq = max(cpu.capacity, targetFreq)
-                totalSpeed += cpu.speed
+                totalSpeed += cpu.rate
             }
 
             val maxFreq = states.lastKey()
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SimplePowerDriver.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SimplePowerDriver.kt
index bf7aeff1..34e91c35 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SimplePowerDriver.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/power/SimplePowerDriver.kt
@@ -37,7 +37,7 @@ public class SimplePowerDriver(private val model: PowerModel) : PowerDriver {
 
             for (cpu in cpus) {
                 targetFreq += cpu.capacity
-                totalSpeed += cpu.speed
+                totalSpeed += cpu.rate
             }
 
             return model.computePower(totalSpeed / targetFreq)
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
index a01fa20c..99f4a1e1 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
@@ -23,7 +23,7 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.flow.source.FixedFlowSource
 
 /**
  * A [SimWorkload] that models applications as a static number of floating point operations ([flops]) executed on
@@ -44,7 +44,7 @@ public class SimFlopsWorkload(
     override fun onStart(ctx: SimMachineContext) {
         val lifecycle = SimWorkloadLifecycle(ctx)
         for (cpu in ctx.cpus) {
-            cpu.startConsumer(lifecycle.waitFor(SimWorkConsumer(flops.toDouble() / ctx.cpus.size, utilization)))
+            cpu.startConsumer(lifecycle.waitFor(FixedFlowSource(flops.toDouble() / ctx.cpus.size, utilization)))
         }
     }
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
index 4ee56689..2ef3bc43 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
@@ -23,7 +23,7 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.resources.consumer.SimWorkConsumer
+import org.opendc.simulator.flow.source.FixedFlowSource
 
 /**
  * A [SimWorkload] that models application execution as a single duration.
@@ -44,7 +44,7 @@ public class SimRuntimeWorkload(
         val lifecycle = SimWorkloadLifecycle(ctx)
         for (cpu in ctx.cpus) {
             val limit = cpu.capacity * utilization
-            cpu.startConsumer(lifecycle.waitFor(SimWorkConsumer((limit / 1000) * duration, utilization)))
+            cpu.startConsumer(lifecycle.waitFor(FixedFlowSource((limit / 1000) * duration, utilization)))
         }
     }
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
index 5a4c4f44..49ae5933 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
@@ -24,9 +24,8 @@ package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.compute.model.ProcessingUnit
-import org.opendc.simulator.resources.SimResourceCommand
-import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
 import kotlin.math.min
 
 /**
@@ -54,14 +53,15 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
      * Obtain the fragment with a timestamp equal or greater than [now].
      */
     private fun pullFragment(now: Long): Fragment? {
+        // Return the most recent fragment if its starting time + duration is later than `now`
         var fragment = fragment
-        if (fragment != null && !fragment.isExpired(now)) {
+        if (fragment != null && fragment.timestamp + offset + fragment.duration > now) {
             return fragment
         }
 
         while (iterator.hasNext()) {
             fragment = iterator.next()
-            if (!fragment.isExpired(now)) {
+            if (fragment.timestamp + offset + fragment.duration > now) {
                 this.fragment = fragment
                 return fragment
             }
@@ -71,38 +71,38 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
         return null
     }
 
-    /**
-     * Determine if the specified [Fragment] is expired, i.e., it has already passed.
-     */
-    private fun Fragment.isExpired(now: Long): Boolean {
-        val timestamp = this.timestamp + offset
-        return now >= timestamp + duration
-    }
+    private inner class Consumer(cpu: ProcessingUnit) : FlowSource {
+        private val offset = this@SimTraceWorkload.offset
+        private val id = cpu.id
+        private val coreCount = cpu.node.coreCount
+
+        override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+            val fragment = pullFragment(now)
+
+            if (fragment == null) {
+                conn.close()
+                return Long.MAX_VALUE
+            }
 
-    private inner class Consumer(val cpu: ProcessingUnit) : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext): SimResourceCommand {
-            val now = ctx.clock.millis()
-            val fragment = pullFragment(now) ?: return SimResourceCommand.Exit
             val timestamp = fragment.timestamp + offset
 
             // Fragment is in the future
             if (timestamp > now) {
-                return SimResourceCommand.Idle(timestamp)
+                conn.push(0.0)
+                return timestamp - now
             }
 
-            val cores = min(cpu.node.coreCount, fragment.cores)
+            val cores = min(coreCount, fragment.cores)
             val usage = if (fragment.cores > 0)
                 fragment.usage / cores
             else
                 0.0
             val deadline = timestamp + fragment.duration
             val duration = deadline - now
-            val work = duration * usage / 1000
 
-            return if (cpu.id < cores && work > 0.0)
-                SimResourceCommand.Consume(work, usage, deadline)
-            else
-                SimResourceCommand.Idle(deadline)
+            conn.push(if (id < cores && usage > 0.0) usage else 0.0)
+
+            return duration
         }
     }
 
@@ -114,5 +114,10 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
      * @param usage The CPU usage during the fragment.
      * @param cores The amount of cores utilized during the fragment.
      */
-    public data class Fragment(val timestamp: Long, val duration: Long, val usage: Double, val cores: Int)
+    public data class Fragment(
+        @JvmField val timestamp: Long,
+        @JvmField val duration: Long,
+        @JvmField val usage: Double,
+        @JvmField val cores: Int
+    )
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt
index 5dd18271..cc4f1f6a 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt
@@ -23,9 +23,8 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
-import org.opendc.simulator.resources.SimResourceEvent
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
 
 /**
  * A helper class to manage the lifecycle of a [SimWorkload]
@@ -34,40 +33,29 @@ public class SimWorkloadLifecycle(private val ctx: SimMachineContext) {
     /**
      * The resource consumers which represent the lifecycle of the workload.
      */
-    private val waiting = mutableSetOf<SimResourceConsumer>()
+    private val waiting = mutableSetOf<FlowSource>()
 
     /**
      * Wait for the specified [consumer] to complete before ending the lifecycle of the workload.
      */
-    public fun waitFor(consumer: SimResourceConsumer): SimResourceConsumer {
+    public fun waitFor(consumer: FlowSource): FlowSource {
         waiting.add(consumer)
-        return object : SimResourceConsumer by consumer {
-            override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
+        return object : FlowSource by consumer {
+            override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
                 try {
-                    consumer.onEvent(ctx, event)
-                } finally {
-                    if (event == SimResourceEvent.Exit) {
-                        complete(consumer)
-                    }
-                }
-            }
-
-            override fun onFailure(ctx: SimResourceContext, cause: Throwable) {
-                try {
-                    consumer.onFailure(ctx, cause)
+                    consumer.onStop(conn, now, delta)
                 } finally {
                     complete(consumer)
                 }
             }
-
             override fun toString(): String = "SimWorkloadLifecycle.Consumer[delegate=$consumer]"
         }
     }
 
     /**
-     * Complete the specified [SimResourceConsumer].
+     * Complete the specified [FlowSource].
      */
-    private fun complete(consumer: SimResourceConsumer) {
+    private fun complete(consumer: FlowSource) {
         if (waiting.remove(consumer) && waiting.isEmpty()) {
             ctx.close()
         }
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
index 81268879..0bb24ed8 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
@@ -34,10 +34,10 @@ import org.opendc.simulator.compute.workload.SimFlopsWorkload
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.compute.workload.SimWorkloadLifecycle
 import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.flow.FlowEngine
+import org.opendc.simulator.flow.source.FixedFlowSource
 import org.opendc.simulator.network.SimNetworkSink
 import org.opendc.simulator.power.SimPowerSource
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
  * Test suite for the [SimBareMetalMachine] class.
@@ -60,7 +60,7 @@ class SimMachineTest {
     @Test
     fun testFlopsWorkload() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -83,7 +83,7 @@ class SimMachineTest {
             memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -100,13 +100,13 @@ class SimMachineTest {
 
     @Test
     fun testPower() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            interpreter,
+            engine,
             machineModel,
             SimplePowerDriver(LinearPowerModel(100.0, 50.0))
         )
-        val source = SimPowerSource(interpreter, capacity = 1000.0)
+        val source = SimPowerSource(engine, capacity = 1000.0)
         source.connect(machine.psu)
 
         try {
@@ -125,7 +125,7 @@ class SimMachineTest {
     @Test
     fun testCapacityClamp() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -151,7 +151,7 @@ class SimMachineTest {
     @Test
     fun testMemory() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -171,7 +171,7 @@ class SimMachineTest {
     @Test
     fun testMemoryUsage() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -180,7 +180,7 @@ class SimMachineTest {
             machine.run(object : SimWorkload {
                 override fun onStart(ctx: SimMachineContext) {
                     val lifecycle = SimWorkloadLifecycle(ctx)
-                    ctx.memory.startConsumer(lifecycle.waitFor(SimWorkConsumer(ctx.memory.capacity, utilization = 0.8)))
+                    ctx.memory.startConsumer(lifecycle.waitFor(FixedFlowSource(ctx.memory.capacity, utilization = 0.8)))
                 }
             })
 
@@ -192,22 +192,22 @@ class SimMachineTest {
 
     @Test
     fun testNetUsage() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            interpreter,
+            engine,
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
         val adapter = (machine.peripherals[0] as SimNetworkAdapter)
-        adapter.connect(SimNetworkSink(interpreter, adapter.bandwidth))
+        adapter.connect(SimNetworkSink(engine, adapter.bandwidth))
 
         try {
             machine.run(object : SimWorkload {
                 override fun onStart(ctx: SimMachineContext) {
                     val lifecycle = SimWorkloadLifecycle(ctx)
                     val iface = ctx.net[0]
-                    iface.tx.startConsumer(lifecycle.waitFor(SimWorkConsumer(iface.bandwidth, utilization = 0.8)))
+                    iface.tx.startConsumer(lifecycle.waitFor(FixedFlowSource(iface.bandwidth, utilization = 0.8)))
                 }
             })
 
@@ -219,9 +219,9 @@ class SimMachineTest {
 
     @Test
     fun testDiskReadUsage() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            interpreter,
+            engine,
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -231,7 +231,7 @@ class SimMachineTest {
                 override fun onStart(ctx: SimMachineContext) {
                     val lifecycle = SimWorkloadLifecycle(ctx)
                     val disk = ctx.storage[0]
-                    disk.read.startConsumer(lifecycle.waitFor(SimWorkConsumer(disk.read.capacity, utilization = 0.8)))
+                    disk.read.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.read.capacity, utilization = 0.8)))
                 }
             })
 
@@ -243,9 +243,9 @@ class SimMachineTest {
 
     @Test
     fun testDiskWriteUsage() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            interpreter,
+            engine,
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -255,7 +255,7 @@ class SimMachineTest {
                 override fun onStart(ctx: SimMachineContext) {
                     val lifecycle = SimWorkloadLifecycle(ctx)
                     val disk = ctx.storage[0]
-                    disk.write.startConsumer(lifecycle.waitFor(SimWorkConsumer(disk.write.capacity, utilization = 0.8)))
+                    disk.write.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.write.capacity, utilization = 0.8)))
                 }
             })
 
@@ -268,7 +268,7 @@ class SimMachineTest {
     @Test
     fun testCancellation() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -290,7 +290,7 @@ class SimMachineTest {
     @Test
     fun testConcurrentRuns() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -313,7 +313,7 @@ class SimMachineTest {
     @Test
     fun testClose() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt
index 6c9ec7bd..e5b509f0 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/device/SimPsuTest.kt
@@ -29,8 +29,8 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.compute.power.PowerDriver
 import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.power.SimPowerSource
-import org.opendc.simulator.resources.SimResourceInterpreter
 
 /**
  * Test suite for [SimPsu]
@@ -55,8 +55,8 @@ internal class SimPsuTest {
         val ratedOutputPower = 240.0
         val energyEfficiency = mapOf(0.0 to 1.0)
 
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val source = SimPowerSource(interpreter, capacity = ratedOutputPower)
+        val engine = FlowEngine(coroutineContext, clock)
+        val source = SimPowerSource(engine, capacity = ratedOutputPower)
 
         val cpuLogic = mockk<PowerDriver.Logic>()
         every { cpuLogic.computePower() } returns 0.0
@@ -78,8 +78,8 @@ internal class SimPsuTest {
             1.0 to 0.94,
         )
 
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val source = SimPowerSource(interpreter, capacity = ratedOutputPower)
+        val engine = FlowEngine(coroutineContext, clock)
+        val source = SimPowerSource(engine, capacity = ratedOutputPower)
 
         val cpuLogic = mockk<PowerDriver.Logic>()
         every { cpuLogic.computePower() } returnsMany listOf(50.0, 100.0, 150.0, 200.0)
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
index 1f010338..9db2e6ec 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
@@ -40,13 +40,13 @@ import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.compute.workload.SimTraceWorkload
 import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * Test suite for the [SimHypervisor] class.
  */
 @OptIn(ExperimentalCoroutinesApi::class)
-internal class SimHypervisorTest {
+internal class SimFairShareHypervisorTest {
     private lateinit var model: MachineModel
 
     @BeforeEach
@@ -63,26 +63,6 @@ internal class SimHypervisorTest {
      */
     @Test
     fun testOvercommittedSingle() = runBlockingSimulation {
-        val listener = object : SimHypervisor.Listener {
-            var totalRequestedWork = 0.0
-            var totalGrantedWork = 0.0
-            var totalOvercommittedWork = 0.0
-
-            override fun onSliceFinish(
-                hypervisor: SimHypervisor,
-                totalWork: Double,
-                grantedWork: Double,
-                overcommittedWork: Double,
-                interferedWork: Double,
-                cpuUsage: Double,
-                cpuDemand: Double
-            ) {
-                totalRequestedWork += totalWork
-                totalGrantedWork += grantedWork
-                totalOvercommittedWork += overcommittedWork
-            }
-        }
-
         val duration = 5 * 60L
         val workloadA =
             SimTraceWorkload(
@@ -94,24 +74,26 @@ internal class SimHypervisorTest {
                 ),
             )
 
-        val platform = SimResourceInterpreter(coroutineContext, clock)
+        val platform = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0)))
-        val hypervisor = SimFairShareHypervisor(platform, scalingGovernor = PerformanceScalingGovernor(), listener = listener)
+        val hypervisor = SimFairShareHypervisor(platform, null, PerformanceScalingGovernor(), null)
 
         launch {
             machine.run(hypervisor)
             println("Hypervisor finished")
         }
         yield()
+
         val vm = hypervisor.createMachine(model)
         vm.run(workloadA)
+
         yield()
         machine.close()
 
         assertAll(
-            { assertEquals(1113300.0, listener.totalRequestedWork, "Requested Burst does not match") },
-            { assertEquals(1023300.0, listener.totalGrantedWork, "Granted Burst does not match") },
-            { assertEquals(90000.0, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(319781, hypervisor.counters.cpuActiveTime, "Active time does not match") },
+            { assertEquals(880219, hypervisor.counters.cpuIdleTime, "Idle time does not match") },
+            { assertEquals(28125, hypervisor.counters.cpuStealTime, "Steal time does not match") },
             { assertEquals(1200000, clock.millis()) { "Current time is correct" } }
         )
     }
@@ -121,26 +103,6 @@ internal class SimHypervisorTest {
      */
     @Test
     fun testOvercommittedDual() = runBlockingSimulation {
-        val listener = object : SimHypervisor.Listener {
-            var totalRequestedWork = 0.0
-            var totalGrantedWork = 0.0
-            var totalOvercommittedWork = 0.0
-
-            override fun onSliceFinish(
-                hypervisor: SimHypervisor,
-                totalWork: Double,
-                grantedWork: Double,
-                overcommittedWork: Double,
-                interferedWork: Double,
-                cpuUsage: Double,
-                cpuDemand: Double
-            ) {
-                totalRequestedWork += totalWork
-                totalGrantedWork += grantedWork
-                totalOvercommittedWork += overcommittedWork
-            }
-        }
-
         val duration = 5 * 60L
         val workloadA =
             SimTraceWorkload(
@@ -161,11 +123,11 @@ internal class SimHypervisorTest {
                 )
             )
 
-        val platform = SimResourceInterpreter(coroutineContext, clock)
+        val platform = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimFairShareHypervisor(platform, listener = listener)
+        val hypervisor = SimFairShareHypervisor(platform, null, null, null)
 
         launch {
             machine.run(hypervisor)
@@ -187,9 +149,9 @@ internal class SimHypervisorTest {
         yield()
 
         assertAll(
-            { assertEquals(2073600.0, listener.totalRequestedWork, "Requested Burst does not match") },
-            { assertEquals(1053600.0, listener.totalGrantedWork, "Granted Burst does not match") },
-            { assertEquals(1020000.0, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
+            { assertEquals(329250, hypervisor.counters.cpuActiveTime, "Active time does not match") },
+            { assertEquals(870750, hypervisor.counters.cpuIdleTime, "Idle time does not match") },
+            { assertEquals(318750, hypervisor.counters.cpuStealTime, "Steal time does not match") },
             { assertEquals(1200000, clock.millis()) }
         )
     }
@@ -202,11 +164,9 @@ internal class SimHypervisorTest {
             memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
 
-        val platform = SimResourceInterpreter(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimFairShareHypervisor(platform)
+        val platform = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimFairShareHypervisor(platform, null, null, null)
 
         assertDoesNotThrow {
             launch {
@@ -232,11 +192,11 @@ internal class SimHypervisorTest {
         )
         val interferenceModel = VmInterferenceModel(groups)
 
-        val platform = SimResourceInterpreter(coroutineContext, clock)
+        val platform = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimFairShareHypervisor(platform, interferenceDomain = interferenceModel.newDomain())
+        val hypervisor = SimFairShareHypervisor(platform, null, null, interferenceModel.newDomain())
 
         val duration = 5 * 60L
         val workloadA =
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt
index 3d3feb2a..b05ffd22 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt
@@ -40,7 +40,7 @@ import org.opendc.simulator.compute.workload.SimFlopsWorkload
 import org.opendc.simulator.compute.workload.SimRuntimeWorkload
 import org.opendc.simulator.compute.workload.SimTraceWorkload
 import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * A test suite for the [SimSpaceSharedHypervisor].
@@ -74,11 +74,9 @@ internal class SimSpaceSharedHypervisorTest {
                 ),
             )
 
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock), machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val engine = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         val vm = hypervisor.createMachine(machineModel)
@@ -98,11 +96,9 @@ internal class SimSpaceSharedHypervisorTest {
     fun testRuntimeWorkload() = runBlockingSimulation {
         val duration = 5 * 60L * 1000
         val workload = SimRuntimeWorkload(duration)
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val engine = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -121,11 +117,11 @@ internal class SimSpaceSharedHypervisorTest {
     fun testFlopsWorkload() = runBlockingSimulation {
         val duration = 5 * 60L * 1000
         val workload = SimFlopsWorkload((duration * 3.2).toLong(), 1.0)
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+            engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -142,11 +138,11 @@ internal class SimSpaceSharedHypervisorTest {
     @Test
     fun testTwoWorkloads() = runBlockingSimulation {
         val duration = 5 * 60L * 1000
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+            engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -155,6 +151,8 @@ internal class SimSpaceSharedHypervisorTest {
         vm.run(SimRuntimeWorkload(duration))
         vm.close()
 
+        yield()
+
         val vm2 = hypervisor.createMachine(machineModel)
         vm2.run(SimRuntimeWorkload(duration))
         vm2.close()
@@ -168,11 +166,9 @@ internal class SimSpaceSharedHypervisorTest {
      */
     @Test
     fun testConcurrentWorkloadFails() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val engine = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -192,11 +188,11 @@ internal class SimSpaceSharedHypervisorTest {
      */
     @Test
     fun testConcurrentWorkloadSucceeds() = runBlockingSimulation {
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val interpreter = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val hypervisor = SimSpaceSharedHypervisor(interpreter, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/PStatePowerDriverTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/PStatePowerDriverTest.kt
index c39859bf..f557c8d3 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/PStatePowerDriverTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/power/PStatePowerDriverTest.kt
@@ -55,7 +55,7 @@ internal class PStatePowerDriverTest {
         val cpu = mockk<SimProcessingUnit>(relaxUnitFun = true)
 
         every { cpu.capacity } returns 3200.0
-        every { cpu.speed } returns 1200.0
+        every { cpu.rate } returns 1200.0
 
         val driver = PStatePowerDriver(
             sortedMapOf(
@@ -77,10 +77,10 @@ internal class PStatePowerDriverTest {
         val cpus = listOf(cpu, cpu)
 
         every { cpus[0].capacity } returns 1000.0
-        every { cpus[0].speed } returns 1200.0
+        every { cpus[0].rate } returns 1200.0
 
         every { cpus[1].capacity } returns 3500.0
-        every { cpus[1].speed } returns 1200.0
+        every { cpus[1].rate } returns 1200.0
 
         val driver = PStatePowerDriver(
             sortedMapOf(
@@ -112,11 +112,11 @@ internal class PStatePowerDriverTest {
 
         val logic = driver.createLogic(machine, listOf(cpu))
 
-        every { cpu.speed } returns 1400.0
+        every { cpu.rate } returns 1400.0
         every { cpu.capacity } returns 1400.0
         assertEquals(150.0, logic.computePower())
 
-        every { cpu.speed } returns 1400.0
+        every { cpu.rate } returns 1400.0
         every { cpu.capacity } returns 4000.0
         assertEquals(235.0, logic.computePower())
     }
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
index 78019c2e..cdbffe4b 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
@@ -31,7 +31,7 @@ import org.opendc.simulator.compute.model.*
 import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
 import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.simulator.resources.SimResourceInterpreter
+import org.opendc.simulator.flow.FlowEngine
 
 /**
  * Test suite for the [SimTraceWorkloadTest] class.
@@ -52,7 +52,7 @@ class SimTraceWorkloadTest {
     @Test
     fun testSmoke() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -79,7 +79,7 @@ class SimTraceWorkloadTest {
     @Test
     fun testOffset() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -106,7 +106,7 @@ class SimTraceWorkloadTest {
     @Test
     fun testSkipFragment() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -134,7 +134,7 @@ class SimTraceWorkloadTest {
     @Test
     fun testZeroCores() = runBlockingSimulation {
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock),
+            FlowEngine(coroutineContext, clock),
             machineModel,
             SimplePowerDriver(ConstantPowerModel(0.0))
         )