From 0ffd93933228e87a205c9839d1bf04cd0e178e8c Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 30 Sep 2021 12:14:42 +0200
Subject: test(simulator): Use longer traces for benchmarks

This change updates the JMH benchmarks to use longer traces in order to
measure the overhead of running the flow simulation as opposed to
setting up the benchmark.
---
 .../org/opendc/simulator/compute/SimMachineBenchmarks.kt | 16 +++++-----------
 1 file changed, 5 insertions(+), 11 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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..88ad7286 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
@@ -38,6 +38,7 @@ import org.opendc.simulator.core.SimulationCoroutineScope
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.resources.SimResourceInterpreter
 import org.openjdk.jmh.annotations.*
+import java.util.concurrent.ThreadLocalRandom
 import java.util.concurrent.TimeUnit
 
 @State(Scope.Thread)
@@ -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()
         }
     }
 
-- 
cgit v1.2.3


From d575bed5418be222e1d3ad39af862e2390596d61 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Sun, 26 Sep 2021 13:11:10 +0200
Subject: refactor(simulator): Combine work and deadline to duration

This change removes the work and deadline properties from the
SimResourceCommand.Consume class and introduces a new property duration.
This property is now used in conjunction with the limit to compute the amount
of work processed by a resource provider.

Previously, we used both work and deadline to compute the duration and
the amount of remaining work at the end of a consumption. However, with
this change, we ensure that a resource consumption always runs at the
same speed once establishing, drastically simplifying the computation
for the amount of work processed during the consumption.
---
 .../kotlin/org/opendc/simulator/compute/device/SimPsu.kt     |  6 +++---
 .../opendc/simulator/compute/workload/SimTraceWorkload.kt    | 12 +++++-------
 .../simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt |  2 ++
 3 files changed, 10 insertions(+), 10 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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..34ac4418 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
@@ -83,13 +83,13 @@ public class SimPsu(
     }
 
     override fun createConsumer(): SimResourceConsumer = object : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): SimResourceCommand {
             val powerDraw = computePowerDraw(_driver?.computePower() ?: 0.0)
 
             return if (powerDraw > 0.0)
-                SimResourceCommand.Consume(Double.POSITIVE_INFINITY, powerDraw, Long.MAX_VALUE)
+                SimResourceCommand.Consume(powerDraw, Long.MAX_VALUE)
             else
-                SimResourceCommand.Idle()
+                SimResourceCommand.Consume(0.0)
         }
 
         override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
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..527619bd 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
@@ -80,14 +80,13 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
     }
 
     private inner class Consumer(val cpu: ProcessingUnit) : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext): SimResourceCommand {
-            val now = ctx.clock.millis()
+        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): SimResourceCommand {
             val fragment = pullFragment(now) ?: return SimResourceCommand.Exit
             val timestamp = fragment.timestamp + offset
 
             // Fragment is in the future
             if (timestamp > now) {
-                return SimResourceCommand.Idle(timestamp)
+                return SimResourceCommand.Consume(0.0, timestamp - now)
             }
 
             val cores = min(cpu.node.coreCount, fragment.cores)
@@ -97,12 +96,11 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
                 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)
+            return if (cpu.id < cores && usage > 0.0)
+                SimResourceCommand.Consume(usage, duration)
             else
-                SimResourceCommand.Idle(deadline)
+                SimResourceCommand.Consume(0.0, duration)
         }
     }
 
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..55d6d7c4 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
@@ -155,6 +155,8 @@ internal class SimSpaceSharedHypervisorTest {
         vm.run(SimRuntimeWorkload(duration))
         vm.close()
 
+        yield()
+
         val vm2 = hypervisor.createMachine(machineModel)
         vm2.run(SimRuntimeWorkload(duration))
         vm2.close()
-- 
cgit v1.2.3


From 02fa44c0b116ff51c4cbe2876d8b2a225ed68553 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Tue, 28 Sep 2021 11:58:19 +0200
Subject: refactor(simulator): Add support for pushing flow from context

This change adds a new method to `SimResourceContext` called `push`
which allows users to change the requested flow rate directly without
having to interrupt the consumer.
---
 .../org/opendc/simulator/compute/device/SimPsu.kt   | 10 +++-------
 .../simulator/compute/workload/SimTraceWorkload.kt  | 21 +++++++++++++--------
 2 files changed, 16 insertions(+), 15 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 34ac4418..6e6e590f 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
@@ -24,7 +24,6 @@ package org.opendc.simulator.compute.device
 
 import org.opendc.simulator.compute.power.PowerDriver
 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
@@ -83,13 +82,10 @@ public class SimPsu(
     }
 
     override fun createConsumer(): SimResourceConsumer = object : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): SimResourceCommand {
+        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
             val powerDraw = computePowerDraw(_driver?.computePower() ?: 0.0)
-
-            return if (powerDraw > 0.0)
-                SimResourceCommand.Consume(powerDraw, Long.MAX_VALUE)
-            else
-                SimResourceCommand.Consume(0.0)
+            ctx.push(powerDraw)
+            return Long.MAX_VALUE
         }
 
         override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
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 527619bd..dd582bb2 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,7 +24,6 @@ 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 kotlin.math.min
@@ -80,13 +79,20 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
     }
 
     private inner class Consumer(val cpu: ProcessingUnit) : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): SimResourceCommand {
-            val fragment = pullFragment(now) ?: return SimResourceCommand.Exit
+        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
+            val fragment = pullFragment(now)
+
+            if (fragment == null) {
+                ctx.close()
+                return Long.MAX_VALUE
+            }
+
             val timestamp = fragment.timestamp + offset
 
             // Fragment is in the future
             if (timestamp > now) {
-                return SimResourceCommand.Consume(0.0, timestamp - now)
+                ctx.push(0.0)
+                return timestamp - now
             }
 
             val cores = min(cpu.node.coreCount, fragment.cores)
@@ -97,10 +103,9 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
             val deadline = timestamp + fragment.duration
             val duration = deadline - now
 
-            return if (cpu.id < cores && usage > 0.0)
-                SimResourceCommand.Consume(usage, duration)
-            else
-                SimResourceCommand.Consume(0.0, duration)
+            ctx.push(if (cpu.id < cores && usage > 0.0) usage else 0.0)
+
+            return duration
         }
     }
 
-- 
cgit v1.2.3


From d2f15fd7fd16922c11b0dcaa8807e8a321859773 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Wed, 29 Sep 2021 22:05:03 +0200
Subject: refactor(simulator): Merge distributor and aggregator into switch

This change removes the distributor and aggregator interfaces in favour
of a single switch interface. Since the switch interface is as powerful
as both the distributor and aggregator, we don't need the latter two.
---
 .../simulator/compute/kernel/SimAbstractHypervisor.kt      | 14 +++++++++++---
 .../opendc/simulator/compute/kernel/SimHypervisorTest.kt   |  2 ++
 2 files changed, 13 insertions(+), 3 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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..cf9e3230 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
@@ -131,7 +131,7 @@ 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(switch, switch.newOutput(interferenceKey), it) }
 
         override fun close() {
             super.close()
@@ -153,9 +153,10 @@ public abstract class SimAbstractHypervisor(
      * A [SimProcessingUnit] of a virtual machine.
      */
     private class VCpu(
-        private val source: SimResourceCloseableProvider,
+        private val switch: SimResourceSwitch,
+        private val source: SimResourceProvider,
         override val model: ProcessingUnit
-    ) : SimProcessingUnit, SimResourceCloseableProvider by source {
+    ) : SimProcessingUnit, SimResourceProvider by source {
         override var capacity: Double
             get() = source.capacity
             set(_) {
@@ -163,6 +164,13 @@ public abstract class SimAbstractHypervisor(
             }
 
         override fun toString(): String = "SimAbstractHypervisor.VCpu[model=$model]"
+
+        /**
+         * Close the CPU
+         */
+        fun close() {
+            switch.removeOutput(source)
+        }
     }
 
     /**
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/SimHypervisorTest.kt
index 1f010338..8cd535ad 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/SimHypervisorTest.kt
@@ -103,8 +103,10 @@ internal class SimHypervisorTest {
             println("Hypervisor finished")
         }
         yield()
+
         val vm = hypervisor.createMachine(model)
         vm.run(workloadA)
+
         yield()
         machine.close()
 
-- 
cgit v1.2.3


From 4cc1d40d421c8736f8b21b360b61d6b065158b7a Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Wed, 29 Sep 2021 23:56:16 +0200
Subject: refactor(simulator): Migrate to flow-based simulation

This change renames the `opendc-simulator-resources` module into the
`opendc-simulator-flow` module to indicate that the core simulation
model of OpenDC is based around modelling and simulating flows.

Previously, the distinction between resource consumer and provider, and
input and output caused some confusion. By switching to a flow-based
model, this distinction is now clear (as in, the water flows from source
to consumer/sink).
---
 .../simulator/compute/SimMachineBenchmarks.kt      | 20 ++++-----
 .../opendc/simulator/compute/SimAbstractMachine.kt | 49 +++++++++++-----------
 .../simulator/compute/SimBareMetalMachine.kt       | 18 ++++----
 .../org/opendc/simulator/compute/SimMachine.kt     |  2 +-
 .../opendc/simulator/compute/SimMachineContext.kt  |  6 +--
 .../org/opendc/simulator/compute/SimMemory.kt      |  4 +-
 .../simulator/compute/SimNetworkInterface.kt       |  8 ++--
 .../opendc/simulator/compute/SimProcessingUnit.kt  |  4 +-
 .../simulator/compute/SimStorageInterface.kt       |  6 +--
 .../org/opendc/simulator/compute/device/SimPsu.kt  | 24 +++++------
 .../compute/kernel/SimAbstractHypervisor.kt        | 38 ++++++++---------
 .../compute/kernel/SimFairShareHypervisor.kt       | 30 ++++++-------
 .../kernel/SimFairShareHypervisorProvider.kt       | 10 ++---
 .../simulator/compute/kernel/SimHypervisor.kt      |  4 +-
 .../compute/kernel/SimHypervisorProvider.kt        |  8 ++--
 .../compute/kernel/SimSpaceSharedHypervisor.kt     | 16 +++----
 .../kernel/SimSpaceSharedHypervisorProvider.kt     | 10 ++---
 .../kernel/interference/VmInterferenceDomain.kt    |  4 +-
 .../kernel/interference/VmInterferenceModel.kt     |  2 +-
 .../simulator/compute/power/PStatePowerDriver.kt   |  2 +-
 .../simulator/compute/power/SimplePowerDriver.kt   |  2 +-
 .../simulator/compute/workload/SimFlopsWorkload.kt |  4 +-
 .../compute/workload/SimRuntimeWorkload.kt         |  4 +-
 .../simulator/compute/workload/SimTraceWorkload.kt | 14 +++----
 .../compute/workload/SimWorkloadLifecycle.kt       | 26 ++++++------
 .../org/opendc/simulator/compute/SimMachineTest.kt | 48 ++++++++++-----------
 .../opendc/simulator/compute/device/SimPsuTest.kt  | 10 ++---
 .../simulator/compute/kernel/SimHypervisorTest.kt  | 10 ++---
 .../compute/kernel/SimSpaceSharedHypervisorTest.kt | 36 ++++++++--------
 .../compute/power/PStatePowerDriverTest.kt         | 10 ++---
 .../compute/workload/SimTraceWorkloadTest.kt       | 10 ++---
 31 files changed, 219 insertions(+), 220 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 88ad7286..c57919c1 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,7 +36,7 @@ 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
@@ -48,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)
 
@@ -80,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))
         }
@@ -90,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)
 
             launch { machine.run(hypervisor) }
 
@@ -111,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)
 
             launch { machine.run(hypervisor) }
 
@@ -132,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)
 
             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..6a62d8a5 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,
+    final override val parent: FlowSystem?,
     final override val model: MachineModel
-) : SimMachine, SimResourceSystem {
+) : SimMachine, FlowSystem {
     /**
      * 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) }
         }
     }
 
@@ -120,7 +120,7 @@ public abstract class SimAbstractMachine(
      * 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 +137,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 +154,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 +162,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 +170,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 +190,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 +198,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..37cf282b 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,8 @@ 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
 
 /**
  * A simulated bare-metal machine that is able to run a single workload.
@@ -35,19 +35,19 @@ 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: FlowSystem? = null,
+) : SimAbstractMachine(engine, parent, model) {
     /**
      * The power draw of the machine onto the PSU.
      */
@@ -58,7 +58,7 @@ public class SimBareMetalMachine(
      * 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)
     }
 
     /**
@@ -78,9 +78,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 6e6e590f..b05d8ad9 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,10 +23,10 @@
 package org.opendc.simulator.compute.device
 
 import org.opendc.simulator.compute.power.PowerDriver
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowEvent
+import org.opendc.simulator.flow.FlowSource
 import org.opendc.simulator.power.SimPowerInlet
-import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
-import org.opendc.simulator.resources.SimResourceEvent
 import java.util.*
 
 /**
@@ -54,7 +54,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.
@@ -69,7 +69,7 @@ public class SimPsu(
      * Update the power draw of the PSU.
      */
     public fun update() {
-        _ctx?.interrupt()
+        _ctx?.pull()
     }
 
     /**
@@ -81,18 +81,18 @@ public class SimPsu(
         update()
     }
 
-    override fun createConsumer(): SimResourceConsumer = object : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
+    override fun createConsumer(): FlowSource = object : FlowSource {
+        override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
             val powerDraw = computePowerDraw(_driver?.computePower() ?: 0.0)
-            ctx.push(powerDraw)
+            conn.push(powerDraw)
             return Long.MAX_VALUE
         }
 
-        override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
+        override fun onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
             when (event) {
-                SimResourceEvent.Start -> _ctx = ctx
-                SimResourceEvent.Run -> _powerDraw = ctx.speed
-                SimResourceEvent.Exit -> _ctx = null
+                FlowEvent.Start -> _ctx = conn
+                FlowEvent.Converge -> _powerDraw = conn.rate
+                FlowEvent.Exit -> _ctx = null
                 else -> {}
             }
         }
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 cf9e3230..b145eefc 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,17 +28,17 @@ 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
 
 /**
  * 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,
+    protected val engine: FlowEngine,
     private val scalingGovernor: ScalingGovernor? = null,
     protected val interferenceDomain: VmInterferenceDomain? = null
 ) : SimHypervisor {
@@ -50,7 +50,7 @@ public abstract class SimAbstractHypervisor(
     /**
      * The resource switch to use.
      */
-    private lateinit var switch: SimResourceSwitch
+    private lateinit var mux: FlowMultiplexer
 
     /**
      * The virtual machines running on this hypervisor.
@@ -62,8 +62,8 @@ public abstract class SimAbstractHypervisor(
     /**
      * The resource counters associated with the hypervisor.
      */
-    public override val counters: SimResourceCounters
-        get() = switch.counters
+    public override val counters: FlowCounters
+        get() = mux.counters
 
     /**
      * The scaling governors attached to the physical CPUs backing this hypervisor.
@@ -71,14 +71,14 @@ public abstract class SimAbstractHypervisor(
     private val governors = mutableListOf<ScalingGovernor.Logic>()
 
     /**
-     * Construct the [SimResourceSwitch] implementation that performs the actual scheduling of the CPUs.
+     * Construct the [FlowMultiplexer] implementation that performs the actual scheduling of the CPUs.
      */
-    public abstract fun createSwitch(ctx: SimMachineContext): SimResourceSwitch
+    public abstract fun createMultiplexer(ctx: SimMachineContext): FlowMultiplexer
 
     /**
      * Check whether the specified machine model fits on this hypervisor.
      */
-    public abstract fun canFit(model: MachineModel, switch: SimResourceSwitch): Boolean
+    public abstract fun canFit(model: MachineModel, switch: FlowMultiplexer): Boolean
 
     /**
      * Trigger the governors to recompute the scaling limits.
@@ -91,7 +91,7 @@ public abstract class SimAbstractHypervisor(
 
     /* SimHypervisor */
     override fun canFit(model: MachineModel): Boolean {
-        return canFit(model, switch)
+        return canFit(model, mux)
     }
 
     override fun createMachine(model: MachineModel, interferenceId: String?): SimMachine {
@@ -104,7 +104,7 @@ public abstract class SimAbstractHypervisor(
     /* SimWorkload */
     override fun onStart(ctx: SimMachineContext) {
         context = ctx
-        switch = createSwitch(ctx)
+        mux = createMultiplexer(ctx)
 
         for (cpu in ctx.cpus) {
             val governor = scalingGovernor?.createLogic(ScalingPolicyImpl(cpu))
@@ -113,7 +113,7 @@ public abstract class SimAbstractHypervisor(
                 governor.onStart()
             }
 
-            switch.addInput(cpu)
+            mux.addOutput(cpu)
         }
     }
 
@@ -122,7 +122,7 @@ public abstract class SimAbstractHypervisor(
      *
      * @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) {
         /**
          * The interference key of this virtual machine.
          */
@@ -131,7 +131,7 @@ public abstract class SimAbstractHypervisor(
         /**
          * The vCPUs of the machine.
          */
-        override val cpus = model.cpus.map { VCpu(switch, switch.newOutput(interferenceKey), it) }
+        override val cpus = model.cpus.map { VCpu(mux, mux.newInput(interferenceKey), it) }
 
         override fun close() {
             super.close()
@@ -153,10 +153,10 @@ public abstract class SimAbstractHypervisor(
      * A [SimProcessingUnit] of a virtual machine.
      */
     private class VCpu(
-        private val switch: SimResourceSwitch,
-        private val source: SimResourceProvider,
+        private val switch: FlowMultiplexer,
+        private val source: FlowConsumer,
         override val model: ProcessingUnit
-    ) : SimProcessingUnit, SimResourceProvider by source {
+    ) : SimProcessingUnit, FlowConsumer by source {
         override var capacity: Double
             get() = source.capacity
             set(_) {
@@ -169,7 +169,7 @@ public abstract class SimAbstractHypervisor(
          * Close the CPU
          */
         fun close() {
-            switch.removeOutput(source)
+            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..36ab7c1c 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
@@ -28,39 +28,39 @@ 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.FlowEngine
+import org.opendc.simulator.flow.FlowSystem
+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 engine The [FlowEngine] to manage the machine's resources.
  * @param parent The parent simulation 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,
+    engine: FlowEngine,
+    private val parent: FlowSystem? = null,
     scalingGovernor: ScalingGovernor? = null,
     interferenceDomain: VmInterferenceDomain? = null,
     private val listener: SimHypervisor.Listener? = null
-) : SimAbstractHypervisor(interpreter, scalingGovernor, interferenceDomain) {
+) : SimAbstractHypervisor(engine, scalingGovernor, interferenceDomain) {
 
-    override fun canFit(model: MachineModel, switch: SimResourceSwitch): Boolean = true
+    override fun canFit(model: MachineModel, switch: FlowMultiplexer): Boolean = true
 
-    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
+    override fun createMultiplexer(ctx: SimMachineContext): FlowMultiplexer {
         return SwitchSystem(ctx).switch
     }
 
-    private inner class SwitchSystem(private val ctx: SimMachineContext) : SimResourceSystem {
-        val switch = SimResourceSwitchMaxMin(interpreter, this, interferenceDomain)
+    private inner class SwitchSystem(private val ctx: SimMachineContext) : FlowSystem {
+        val switch = MaxMinFlowMultiplexer(engine, this, interferenceDomain)
 
-        override val parent: SimResourceSystem? = this@SimFairShareHypervisor.parent
+        override val parent: FlowSystem? = this@SimFairShareHypervisor.parent
 
         private var lastCpuUsage = 0.0
         private var lastCpuDemand = 0.0
@@ -87,8 +87,8 @@ public class SimFairShareHypervisor(
             }
             lastReport = timestamp
 
-            lastCpuDemand = switch.inputs.sumOf { it.demand }
-            lastCpuUsage = switch.inputs.sumOf { it.speed }
+            lastCpuDemand = switch.outputs.sumOf { it.demand }
+            lastCpuUsage = switch.outputs.sumOf { it.rate }
             lastDemand = counters.demand
             lastActual = counters.actual
             lastOvercommit = counters.overcommit
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..bfa099fb 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.FlowEngine
+import org.opendc.simulator.flow.FlowSystem
 
 /**
  * A [SimHypervisorProvider] for the [SimFairShareHypervisor] implementation.
@@ -34,13 +34,13 @@ public class SimFairShareHypervisorProvider : SimHypervisorProvider {
     override val id: String = "fair-share"
 
     override fun create(
-        interpreter: SimResourceInterpreter,
-        parent: SimResourceSystem?,
+        engine: FlowEngine,
+        parent: FlowSystem?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
         listener: SimHypervisor.Listener?
     ): SimHypervisor = SimFairShareHypervisor(
-        interpreter,
+        engine,
         parent,
         scalingGovernor = scalingGovernor,
         interferenceDomain = 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..1b11ca6b 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,7 @@ 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
+import org.opendc.simulator.flow.FlowCounters
 
 /**
  * A SimHypervisor facilitates the execution of multiple concurrent [SimWorkload]s, while acting as a single workload
@@ -40,7 +40,7 @@ public interface SimHypervisor : SimWorkload {
     /**
      * The resource counters associated with the hypervisor.
      */
-    public val counters: SimResourceCounters
+    public val counters: FlowCounters
 
     /**
      * Determine whether the specified machine characterized by [model] can fit on this hypervisor at this moment.
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..97f07097 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.FlowEngine
+import org.opendc.simulator.flow.FlowSystem
 
 /**
  * A service provider interface for constructing a [SimHypervisor].
@@ -43,8 +43,8 @@ public interface SimHypervisorProvider {
      * Create a [SimHypervisor] instance with the specified [listener].
      */
     public fun create(
-        interpreter: SimResourceInterpreter,
-        parent: SimResourceSystem? = null,
+        engine: FlowEngine,
+        parent: FlowSystem? = null,
         scalingGovernor: ScalingGovernor? = null,
         interferenceDomain: VmInterferenceDomain? = null,
         listener: SimHypervisor.Listener? = null
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..883e0d82 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
@@ -24,19 +24,19 @@ package org.opendc.simulator.compute.kernel
 
 import org.opendc.simulator.compute.SimMachineContext
 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.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) : SimAbstractHypervisor(engine) {
+    override fun canFit(model: MachineModel, switch: FlowMultiplexer): Boolean {
+        return switch.outputs.size - switch.inputs.size >= model.cpus.size
     }
 
-    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
-        return SimResourceSwitchExclusive()
+    override fun createMultiplexer(ctx: SimMachineContext): FlowMultiplexer {
+        return ForwardingFlowMultiplexer(engine)
     }
 }
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..7869d72d 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.FlowEngine
+import org.opendc.simulator.flow.FlowSystem
 
 /**
  * A [SimHypervisorProvider] for the [SimSpaceSharedHypervisor] implementation.
@@ -34,10 +34,10 @@ public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider {
     override val id: String = "space-shared"
 
     override fun create(
-        interpreter: SimResourceInterpreter,
-        parent: SimResourceSystem?,
+        engine: FlowEngine,
+        parent: FlowSystem?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
         listener: SimHypervisor.Listener?
-    ): SimHypervisor = SimSpaceSharedHypervisor(interpreter)
+    ): SimHypervisor = SimSpaceSharedHypervisor(engine)
 }
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 dd582bb2..a877dac1 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,8 +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.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
 import kotlin.math.min
 
 /**
@@ -78,12 +78,12 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
         return now >= timestamp + duration
     }
 
-    private inner class Consumer(val cpu: ProcessingUnit) : SimResourceConsumer {
-        override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
+    private inner class Consumer(val cpu: ProcessingUnit) : FlowSource {
+        override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
             val fragment = pullFragment(now)
 
             if (fragment == null) {
-                ctx.close()
+                conn.close()
                 return Long.MAX_VALUE
             }
 
@@ -91,7 +91,7 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
 
             // Fragment is in the future
             if (timestamp > now) {
-                ctx.push(0.0)
+                conn.push(0.0)
                 return timestamp - now
             }
 
@@ -103,7 +103,7 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
             val deadline = timestamp + fragment.duration
             val duration = deadline - now
 
-            ctx.push(if (cpu.id < cores && usage > 0.0) usage else 0.0)
+            conn.push(if (cpu.id < cores && usage > 0.0) usage else 0.0)
 
             return duration
         }
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..dabe60e0 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,9 @@
 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.FlowEvent
+import org.opendc.simulator.flow.FlowSource
 
 /**
  * A helper class to manage the lifecycle of a [SimWorkload]
@@ -34,27 +34,27 @@ 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 onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
                 try {
-                    consumer.onEvent(ctx, event)
+                    consumer.onEvent(conn, now, event)
                 } finally {
-                    if (event == SimResourceEvent.Exit) {
+                    if (event == FlowEvent.Exit) {
                         complete(consumer)
                     }
                 }
             }
 
-            override fun onFailure(ctx: SimResourceContext, cause: Throwable) {
+            override fun onFailure(conn: FlowConnection, cause: Throwable) {
                 try {
-                    consumer.onFailure(ctx, cause)
+                    consumer.onFailure(conn, cause)
                 } finally {
                     complete(consumer)
                 }
@@ -65,9 +65,9 @@ public class SimWorkloadLifecycle(private val ctx: SimMachineContext) {
     }
 
     /**
-     * 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/SimHypervisorTest.kt
index 8cd535ad..058d5d28 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/SimHypervisorTest.kt
@@ -40,7 +40,7 @@ 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.
@@ -94,7 +94,7 @@ 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)
 
@@ -163,7 +163,7 @@ internal class SimHypervisorTest {
                 )
             )
 
-        val platform = SimResourceInterpreter(coroutineContext, clock)
+        val platform = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -204,7 +204,7 @@ internal class SimHypervisorTest {
             memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
 
-        val platform = SimResourceInterpreter(coroutineContext, clock)
+        val platform = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
             platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
         )
@@ -234,7 +234,7 @@ 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))
         )
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 55d6d7c4..95fb6679 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,11 @@ internal class SimSpaceSharedHypervisorTest {
                 ),
             )
 
-        val interpreter = SimResourceInterpreter(coroutineContext, clock)
+        val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(
-            SimResourceInterpreter(coroutineContext, clock), machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
+            FlowEngine(coroutineContext, clock), machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val hypervisor = SimSpaceSharedHypervisor(engine)
 
         launch { machine.run(hypervisor) }
         val vm = hypervisor.createMachine(machineModel)
@@ -98,11 +98,11 @@ internal class SimSpaceSharedHypervisorTest {
     fun testRuntimeWorkload() = runBlockingSimulation {
         val duration = 5 * 60L * 1000
         val workload = SimRuntimeWorkload(duration)
-        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)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -121,11 +121,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)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -142,11 +142,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)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -170,11 +170,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)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -194,7 +192,7 @@ 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))
         )
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))
         )
-- 
cgit v1.2.3


From b0fc93f818e5e735e972a04f5aa49e0ebe1de181 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 30 Sep 2021 14:35:30 +0200
Subject: refactor(simulator): Remove failure callback from FlowSource

This change removes the `onFailure` method from FlowSource. Instead, the
FlowConsumer will receive the reason for failure of the source.
---
 .../compute/workload/SimWorkloadLifecycle.kt       | 22 +++++++++++++---------
 1 file changed, 13 insertions(+), 9 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 dabe60e0..b85be39d 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
@@ -41,22 +41,26 @@ public class SimWorkloadLifecycle(private val ctx: SimMachineContext) {
      */
     public fun waitFor(consumer: FlowSource): FlowSource {
         waiting.add(consumer)
-        return object : FlowSource by consumer {
+        return object : FlowSource {
+            override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+                return try {
+                    consumer.onPull(conn, now, delta)
+                } catch (cause: Throwable) {
+                    complete(consumer)
+                    throw cause
+                }
+            }
+
             override fun onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
                 try {
                     consumer.onEvent(conn, now, event)
-                } finally {
+
                     if (event == FlowEvent.Exit) {
                         complete(consumer)
                     }
-                }
-            }
-
-            override fun onFailure(conn: FlowConnection, cause: Throwable) {
-                try {
-                    consumer.onFailure(conn, cause)
-                } finally {
+                } catch (cause: Throwable) {
                     complete(consumer)
+                    throw cause
                 }
             }
 
-- 
cgit v1.2.3


From a2ce07026bf3ef17326e72f395dfa2dd9d9b17be Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 30 Sep 2021 15:37:35 +0200
Subject: refactor(simulator): Create separate callbacks for remaining events

This change creates separate callbacks for the remaining events:
onStart, onStop and onConverge.
---
 .../org/opendc/simulator/compute/device/SimPsu.kt  | 18 ++++++++--------
 .../compute/workload/SimWorkloadLifecycle.kt       | 24 ++++------------------
 2 files changed, 14 insertions(+), 28 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 b05d8ad9..8400c225 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
@@ -24,7 +24,6 @@ package org.opendc.simulator.compute.device
 
 import org.opendc.simulator.compute.power.PowerDriver
 import org.opendc.simulator.flow.FlowConnection
-import org.opendc.simulator.flow.FlowEvent
 import org.opendc.simulator.flow.FlowSource
 import org.opendc.simulator.power.SimPowerInlet
 import java.util.*
@@ -82,19 +81,22 @@ public class SimPsu(
     }
 
     override fun createConsumer(): FlowSource = object : FlowSource {
+        override fun onStart(conn: FlowConnection, now: Long) {
+            _ctx = conn
+        }
+
+        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(conn: FlowConnection, now: Long, event: FlowEvent) {
-            when (event) {
-                FlowEvent.Start -> _ctx = conn
-                FlowEvent.Converge -> _powerDraw = conn.rate
-                FlowEvent.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/workload/SimWorkloadLifecycle.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadLifecycle.kt
index b85be39d..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
@@ -24,7 +24,6 @@ package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.flow.FlowConnection
-import org.opendc.simulator.flow.FlowEvent
 import org.opendc.simulator.flow.FlowSource
 
 /**
@@ -41,29 +40,14 @@ public class SimWorkloadLifecycle(private val ctx: SimMachineContext) {
      */
     public fun waitFor(consumer: FlowSource): FlowSource {
         waiting.add(consumer)
-        return object : FlowSource {
-            override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
-                return try {
-                    consumer.onPull(conn, now, delta)
-                } catch (cause: Throwable) {
-                    complete(consumer)
-                    throw cause
-                }
-            }
-
-            override fun onEvent(conn: FlowConnection, now: Long, event: FlowEvent) {
+        return object : FlowSource by consumer {
+            override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
                 try {
-                    consumer.onEvent(conn, now, event)
-
-                    if (event == FlowEvent.Exit) {
-                        complete(consumer)
-                    }
-                } catch (cause: Throwable) {
+                    consumer.onStop(conn, now, delta)
+                } finally {
                     complete(consumer)
-                    throw cause
                 }
             }
-
             override fun toString(): String = "SimWorkloadLifecycle.Consumer[delegate=$consumer]"
         }
     }
-- 
cgit v1.2.3


From 94783ff9d8cd81275fefd5804ac99f98e2dee3a4 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 30 Sep 2021 16:00:05 +0200
Subject: fix(simulator): Fix loss computation for UPS and PDU

This change fixes the loss computation for both the UPS and PDU
implementation that was broken due to the new pushing mechanism. We
implement a new class FlowMapper that can be used to map the flow pushed
by a `FlowSource` using a user-specified method.
---
 .../src/main/kotlin/org/opendc/simulator/compute/device/SimPsu.kt       | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 8400c225..62d91c0b 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
@@ -80,7 +80,7 @@ public class SimPsu(
         update()
     }
 
-    override fun createConsumer(): FlowSource = object : FlowSource {
+    override fun createSource(): FlowSource = object : FlowSource {
         override fun onStart(conn: FlowConnection, now: Long) {
             _ctx = conn
         }
-- 
cgit v1.2.3


From 559ac2327b8aa319fb8ab4558d4f4aa3382349f4 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Thu, 30 Sep 2021 16:27:45 +0200
Subject: perf(simulator): Make convergence callback optional

This change adds two new properties for controlling whether the
convergence callbacks of the source and consumer respectively should be
invoked. This saves a lot of unnecessary calls for stages that do not
have any implementation of the `onConvergence` method.
---
 .../org/opendc/simulator/compute/SimAbstractMachine.kt   |  8 ++++++--
 .../org/opendc/simulator/compute/SimBareMetalMachine.kt  |  4 ++--
 .../kotlin/org/opendc/simulator/compute/device/SimPsu.kt |  1 +
 .../simulator/compute/kernel/SimAbstractHypervisor.kt    |  2 --
 .../simulator/compute/kernel/SimFairShareHypervisor.kt   | 16 ++++++++--------
 .../compute/kernel/SimFairShareHypervisorProvider.kt     |  4 ++--
 .../simulator/compute/kernel/SimHypervisorProvider.kt    |  4 ++--
 .../compute/kernel/SimSpaceSharedHypervisorProvider.kt   |  4 ++--
 8 files changed, 23 insertions(+), 20 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 6a62d8a5..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
@@ -43,9 +43,9 @@ import kotlin.coroutines.resume
  */
 public abstract class SimAbstractMachine(
     protected val engine: FlowEngine,
-    final override val parent: FlowSystem?,
+    private val parent: FlowConvergenceListener?,
     final override val model: MachineModel
-) : SimMachine, FlowSystem {
+) : SimMachine, FlowConvergenceListener {
     /**
      * The resources allocated for this machine.
      */
@@ -116,6 +116,10 @@ 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.
      */
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 37cf282b..0bcf5957 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
@@ -46,7 +46,7 @@ public class SimBareMetalMachine(
     model: MachineModel,
     powerDriver: PowerDriver,
     public val psu: SimPsu = SimPsu(500.0, mapOf(1.0 to 1.0)),
-    parent: FlowSystem? = null,
+    parent: FlowConvergenceListener? = null,
 ) : SimAbstractMachine(engine, parent, model) {
     /**
      * The power draw of the machine onto the PSU.
@@ -66,7 +66,7 @@ public class SimBareMetalMachine(
      */
     private val powerDriverLogic = powerDriver.createLogic(this, cpus)
 
-    override fun onConverge(timestamp: Long) {
+    override fun onConverge(now: Long, delta: Long) {
         psu.update()
     }
 
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 62d91c0b..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
@@ -83,6 +83,7 @@ public class SimPsu(
     override fun createSource(): FlowSource = object : FlowSource {
         override fun onStart(conn: FlowConnection, now: Long) {
             _ctx = conn
+            conn.shouldSourceConverge = true
         }
 
         override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
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 b145eefc..bcba8e8e 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
@@ -145,8 +145,6 @@ public abstract class SimAbstractHypervisor(
                 interferenceDomain?.leave(interferenceKey)
             }
         }
-
-        override fun onConverge(timestamp: Long) {}
     }
 
     /**
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 36ab7c1c..b0515c6e 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
@@ -28,8 +28,8 @@ 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.flow.FlowConvergenceListener
 import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSystem
 import org.opendc.simulator.flow.mux.FlowMultiplexer
 import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
 
@@ -45,7 +45,7 @@ import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
  */
 public class SimFairShareHypervisor(
     engine: FlowEngine,
-    private val parent: FlowSystem? = null,
+    private val parent: FlowConvergenceListener? = null,
     scalingGovernor: ScalingGovernor? = null,
     interferenceDomain: VmInterferenceDomain? = null,
     private val listener: SimHypervisor.Listener? = null
@@ -57,11 +57,9 @@ public class SimFairShareHypervisor(
         return SwitchSystem(ctx).switch
     }
 
-    private inner class SwitchSystem(private val ctx: SimMachineContext) : FlowSystem {
+    private inner class SwitchSystem(private val ctx: SimMachineContext) : FlowConvergenceListener {
         val switch = MaxMinFlowMultiplexer(engine, this, interferenceDomain)
 
-        override val parent: FlowSystem? = this@SimFairShareHypervisor.parent
-
         private var lastCpuUsage = 0.0
         private var lastCpuDemand = 0.0
         private var lastDemand = 0.0
@@ -70,11 +68,11 @@ public class SimFairShareHypervisor(
         private var lastInterference = 0.0
         private var lastReport = Long.MIN_VALUE
 
-        override fun onConverge(timestamp: Long) {
+        override fun onConverge(now: Long, delta: Long) {
             val listener = listener ?: return
             val counters = switch.counters
 
-            if (timestamp > lastReport) {
+            if (now > lastReport) {
                 listener.onSliceFinish(
                     this@SimFairShareHypervisor,
                     counters.demand - lastDemand,
@@ -85,7 +83,7 @@ public class SimFairShareHypervisor(
                     lastCpuDemand
                 )
             }
-            lastReport = timestamp
+            lastReport = now
 
             lastCpuDemand = switch.outputs.sumOf { it.demand }
             lastCpuUsage = switch.outputs.sumOf { it.rate }
@@ -96,6 +94,8 @@ public class SimFairShareHypervisor(
 
             val load = lastCpuDemand / ctx.cpus.sumOf { it.model.frequency }
             triggerGovernors(load)
+
+            parent?.onConverge(now, delta)
         }
     }
 }
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 bfa099fb..e0a70926 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.flow.FlowConvergenceListener
 import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSystem
 
 /**
  * A [SimHypervisorProvider] for the [SimFairShareHypervisor] implementation.
@@ -35,7 +35,7 @@ public class SimFairShareHypervisorProvider : SimHypervisorProvider {
 
     override fun create(
         engine: FlowEngine,
-        parent: FlowSystem?,
+        parent: FlowConvergenceListener?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
         listener: SimHypervisor.Listener?
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 97f07097..dad2cc3b 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.flow.FlowConvergenceListener
 import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSystem
 
 /**
  * A service provider interface for constructing a [SimHypervisor].
@@ -44,7 +44,7 @@ public interface SimHypervisorProvider {
      */
     public fun create(
         engine: FlowEngine,
-        parent: FlowSystem? = null,
+        parent: FlowConvergenceListener? = null,
         scalingGovernor: ScalingGovernor? = null,
         interferenceDomain: VmInterferenceDomain? = null,
         listener: SimHypervisor.Listener? = null
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 7869d72d..93921eb9 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.flow.FlowConvergenceListener
 import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSystem
 
 /**
  * A [SimHypervisorProvider] for the [SimSpaceSharedHypervisor] implementation.
@@ -35,7 +35,7 @@ public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider {
 
     override fun create(
         engine: FlowEngine,
-        parent: FlowSystem?,
+        parent: FlowConvergenceListener?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
         listener: SimHypervisor.Listener?
-- 
cgit v1.2.3


From 6e424e9b44687d01e618e7bc38afc427610cd845 Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Fri, 1 Oct 2021 20:08:21 +0200
Subject: perf(simulator): Optimize hot path in SimTraceWorkload

This change optimizes the hot path of the SimTraceWorkload class. This
class is used by multiple experiments to replay workload traces and
consumes a considerable amount of time of the total time of the
experiments.

By inlining the isExpired method, marking the properties on the Fragment
class as field, and caching several properties locally in the class, we
reduce the amount of virtual method calls in the hot path.
---
 .../simulator/compute/workload/SimTraceWorkload.kt | 28 ++++++++++++----------
 1 file changed, 15 insertions(+), 13 deletions(-)

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 a877dac1..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
@@ -53,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
             }
@@ -70,15 +71,11 @@ 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
 
-    private inner class Consumer(val cpu: ProcessingUnit) : FlowSource {
         override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
             val fragment = pullFragment(now)
 
@@ -95,7 +92,7 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
                 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
@@ -103,7 +100,7 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val
             val deadline = timestamp + fragment.duration
             val duration = deadline - now
 
-            conn.push(if (cpu.id < cores && usage > 0.0) usage else 0.0)
+            conn.push(if (id < cores && usage > 0.0) usage else 0.0)
 
             return duration
         }
@@ -117,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
+    )
 }
-- 
cgit v1.2.3


From 081221684fb826ab5a00c1d8cc5a9886b9e2203c Mon Sep 17 00:00:00 2001
From: Fabian Mastenbroek <mail.fabianm@gmail.com>
Date: Fri, 1 Oct 2021 22:04:35 +0200
Subject: feat(simulator): Expose CPU time counters directly on hypervisor

This change adds a new interface to the SimHypervisor interface that
exposes the CPU time counters directly. These are derived from the flow
counters and will be used by SimHost to expose them via telemetry.
---
 .../simulator/compute/SimMachineBenchmarks.kt      |   6 +-
 .../simulator/compute/SimBareMetalMachine.kt       |  27 +-
 .../compute/kernel/SimAbstractHypervisor.kt        | 144 +++++++++--
 .../compute/kernel/SimFairShareHypervisor.kt       |  67 +----
 .../kernel/SimFairShareHypervisorProvider.kt       |  11 +-
 .../simulator/compute/kernel/SimHypervisor.kt      |  38 ++-
 .../compute/kernel/SimHypervisorCounters.kt        |  48 ++++
 .../compute/kernel/SimHypervisorProvider.kt        |   5 +-
 .../compute/kernel/SimSpaceSharedHypervisor.kt     |  17 +-
 .../kernel/SimSpaceSharedHypervisorProvider.kt     |   5 +-
 .../simulator/compute/kernel/SimVirtualMachine.kt  |  50 ++++
 .../compute/kernel/SimFairShareHypervisorTest.kt   | 238 ++++++++++++++++++
 .../simulator/compute/kernel/SimHypervisorTest.kt  | 280 ---------------------
 .../compute/kernel/SimSpaceSharedHypervisorTest.kt |  20 +-
 14 files changed, 533 insertions(+), 423 deletions(-)
 create mode 100644 opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorCounters.kt
 create mode 100644 opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimVirtualMachine.kt
 create mode 100644 opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
 delete mode 100644 opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt

(limited to 'opendc-simulator/opendc-simulator-compute/src')

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 c57919c1..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
@@ -92,7 +92,7 @@ class SimMachineBenchmarks {
             val machine = SimBareMetalMachine(
                 engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            val hypervisor = SimSpaceSharedHypervisor(engine)
+            val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
             launch { machine.run(hypervisor) }
 
@@ -113,7 +113,7 @@ class SimMachineBenchmarks {
             val machine = SimBareMetalMachine(
                 engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            val hypervisor = SimFairShareHypervisor(engine)
+            val hypervisor = SimFairShareHypervisor(engine, null, null, null)
 
             launch { machine.run(hypervisor) }
 
@@ -134,7 +134,7 @@ class SimMachineBenchmarks {
             val machine = SimBareMetalMachine(
                 engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
             )
-            val hypervisor = SimFairShareHypervisor(engine)
+            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/SimBareMetalMachine.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
index 0bcf5957..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
@@ -28,6 +28,7 @@ import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.power.PowerDriver
 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.
@@ -49,10 +50,18 @@ public class SimBareMetalMachine(
     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.
@@ -66,8 +75,20 @@ public class SimBareMetalMachine(
      */
     private val powerDriverLogic = powerDriver.createLogic(this, cpus)
 
+    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 {
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 bcba8e8e..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
@@ -30,6 +30,7 @@ import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.flow.*
 import org.opendc.simulator.flow.mux.FlowMultiplexer
+import kotlin.math.roundToLong
 
 /**
  * Abstract implementation of the [SimHypervisor] interface.
@@ -39,18 +40,19 @@ import org.opendc.simulator.flow.mux.FlowMultiplexer
  */
 public abstract class SimAbstractHypervisor(
     protected val engine: FlowEngine,
-    private val scalingGovernor: ScalingGovernor? = null,
+    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 mux: FlowMultiplexer
+    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: FlowCounters
-        get() = mux.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 [FlowMultiplexer] implementation that performs the actual scheduling of the CPUs.
+     * The CPU demand of the hypervisor in MHz.
      */
-    public abstract fun createMultiplexer(ctx: SimMachineContext): FlowMultiplexer
+    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: FlowMultiplexer): 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, mux)
-    }
-
-    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
-        mux = createMultiplexer(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()
             }
 
-            mux.addOutput(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(engine, 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.
          */
@@ -133,6 +190,41 @@ public abstract class SimAbstractHypervisor(
          */
         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()
 
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 b0515c6e..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,7 +23,6 @@
 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
@@ -38,64 +37,20 @@ import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
  * concurrently using weighted fair sharing.
  *
  * @param engine The [FlowEngine] to manage the machine's resources.
- * @param parent The parent simulation system.
+ * @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(
     engine: FlowEngine,
-    private val parent: FlowConvergenceListener? = null,
-    scalingGovernor: ScalingGovernor? = null,
-    interferenceDomain: VmInterferenceDomain? = null,
-    private val listener: SimHypervisor.Listener? = null
-) : SimAbstractHypervisor(engine, scalingGovernor, interferenceDomain) {
-
-    override fun canFit(model: MachineModel, switch: FlowMultiplexer): Boolean = true
-
-    override fun createMultiplexer(ctx: SimMachineContext): FlowMultiplexer {
-        return SwitchSystem(ctx).switch
-    }
-
-    private inner class SwitchSystem(private val ctx: SimMachineContext) : FlowConvergenceListener {
-        val switch = MaxMinFlowMultiplexer(engine, this, interferenceDomain)
-
-        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(now: Long, delta: Long) {
-            val listener = listener ?: return
-            val counters = switch.counters
-
-            if (now > lastReport) {
-                listener.onSliceFinish(
-                    this@SimFairShareHypervisor,
-                    counters.demand - lastDemand,
-                    counters.actual - lastActual,
-                    counters.overcommit - lastOvercommit,
-                    counters.interference - lastInterference,
-                    lastCpuUsage,
-                    lastCpuDemand
-                )
-            }
-            lastReport = now
-
-            lastCpuDemand = switch.outputs.sumOf { it.demand }
-            lastCpuUsage = switch.outputs.sumOf { it.rate }
-            lastDemand = counters.demand
-            lastActual = counters.actual
-            lastOvercommit = counters.overcommit
-            lastInterference = counters.interference
-
-            val load = lastCpuDemand / ctx.cpus.sumOf { it.model.frequency }
-            triggerGovernors(load)
-
-            parent?.onConverge(now, delta)
-        }
-    }
+    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 e0a70926..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
@@ -35,15 +35,8 @@ public class SimFairShareHypervisorProvider : SimHypervisorProvider {
 
     override fun create(
         engine: FlowEngine,
-        parent: FlowConvergenceListener?,
+        listener: FlowConvergenceListener?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
-        listener: SimHypervisor.Listener?
-    ): SimHypervisor = SimFairShareHypervisor(
-        engine,
-        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 1b11ca6b..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.flow.FlowCounters
 
 /**
  * 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: FlowCounters
+    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 dad2cc3b..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
@@ -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(
         engine: FlowEngine,
-        parent: FlowConvergenceListener? = null,
+        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 883e0d82..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,8 +22,9 @@
 
 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.flow.FlowConvergenceListener
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.simulator.flow.mux.FlowMultiplexer
 import org.opendc.simulator.flow.mux.ForwardingFlowMultiplexer
@@ -31,12 +32,14 @@ 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(engine: FlowEngine) : SimAbstractHypervisor(engine) {
-    override fun canFit(model: MachineModel, switch: FlowMultiplexer): Boolean {
-        return switch.outputs.size - switch.inputs.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 createMultiplexer(ctx: SimMachineContext): FlowMultiplexer {
-        return ForwardingFlowMultiplexer(engine)
+    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 93921eb9..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
@@ -35,9 +35,8 @@ public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider {
 
     override fun create(
         engine: FlowEngine,
-        parent: FlowConvergenceListener?,
+        listener: FlowConvergenceListener?,
         scalingGovernor: ScalingGovernor?,
         interferenceDomain: VmInterferenceDomain?,
-        listener: SimHypervisor.Listener?
-    ): SimHypervisor = SimSpaceSharedHypervisor(engine)
+    ): 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/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
new file mode 100644
index 00000000..9db2e6ec
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt
@@ -0,0 +1,238 @@
+/*
+ * 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 kotlinx.coroutines.*
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertAll
+import org.junit.jupiter.api.assertDoesNotThrow
+import org.opendc.simulator.compute.SimBareMetalMachine
+import org.opendc.simulator.compute.kernel.cpufreq.PerformanceScalingGovernor
+import org.opendc.simulator.compute.kernel.interference.VmInterferenceGroup
+import org.opendc.simulator.compute.kernel.interference.VmInterferenceModel
+import org.opendc.simulator.compute.model.MachineModel
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingNode
+import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.power.ConstantPowerModel
+import org.opendc.simulator.compute.power.SimplePowerDriver
+import org.opendc.simulator.compute.workload.SimTraceWorkload
+import org.opendc.simulator.core.runBlockingSimulation
+import org.opendc.simulator.flow.FlowEngine
+
+/**
+ * Test suite for the [SimHypervisor] class.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+internal class SimFairShareHypervisorTest {
+    private lateinit var model: MachineModel
+
+    @BeforeEach
+    fun setUp() {
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 1)
+        model = MachineModel(
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+        )
+    }
+
+    /**
+     * Test overcommitting of resources via the hypervisor with a single VM.
+     */
+    @Test
+    fun testOvercommittedSingle() = runBlockingSimulation {
+        val duration = 5 * 60L
+        val workloadA =
+            SimTraceWorkload(
+                sequenceOf(
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
+                ),
+            )
+
+        val platform = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0)))
+        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(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" } }
+        )
+    }
+
+    /**
+     * Test overcommitting of resources via the hypervisor with two VMs.
+     */
+    @Test
+    fun testOvercommittedDual() = runBlockingSimulation {
+        val duration = 5 * 60L
+        val workloadA =
+            SimTraceWorkload(
+                sequenceOf(
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
+                ),
+            )
+        val workloadB =
+            SimTraceWorkload(
+                sequenceOf(
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3100.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 73.0, 1)
+                )
+            )
+
+        val platform = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(
+            platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
+        )
+        val hypervisor = SimFairShareHypervisor(platform, null, null, null)
+
+        launch {
+            machine.run(hypervisor)
+        }
+
+        yield()
+        coroutineScope {
+            launch {
+                val vm = hypervisor.createMachine(model)
+                vm.run(workloadA)
+                vm.close()
+            }
+            val vm = hypervisor.createMachine(model)
+            vm.run(workloadB)
+            vm.close()
+        }
+        yield()
+        machine.close()
+        yield()
+
+        assertAll(
+            { 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()) }
+        )
+    }
+
+    @Test
+    fun testMultipleCPUs() = runBlockingSimulation {
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
+        val model = MachineModel(
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+        )
+
+        val platform = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimFairShareHypervisor(platform, null, null, null)
+
+        assertDoesNotThrow {
+            launch {
+                machine.run(hypervisor)
+            }
+        }
+
+        machine.close()
+    }
+
+    @Test
+    fun testInterference() = runBlockingSimulation {
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
+        val model = MachineModel(
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+        )
+
+        val groups = listOf(
+            VmInterferenceGroup(targetLoad = 0.0, score = 0.9, members = setOf("a", "b")),
+            VmInterferenceGroup(targetLoad = 0.0, score = 0.6, members = setOf("a", "c")),
+            VmInterferenceGroup(targetLoad = 0.1, score = 0.8, members = setOf("a", "n"))
+        )
+        val interferenceModel = VmInterferenceModel(groups)
+
+        val platform = FlowEngine(coroutineContext, clock)
+        val machine = SimBareMetalMachine(
+            platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
+        )
+        val hypervisor = SimFairShareHypervisor(platform, null, null, interferenceModel.newDomain())
+
+        val duration = 5 * 60L
+        val workloadA =
+            SimTraceWorkload(
+                sequenceOf(
+                    SimTraceWorkload.Fragment(0, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
+                ),
+            )
+        val workloadB =
+            SimTraceWorkload(
+                sequenceOf(
+                    SimTraceWorkload.Fragment(0, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 3100.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 73.0, 1)
+                )
+            )
+
+        launch {
+            machine.run(hypervisor)
+        }
+
+        coroutineScope {
+            launch {
+                val vm = hypervisor.createMachine(model, "a")
+                vm.run(workloadA)
+                vm.close()
+            }
+            val vm = hypervisor.createMachine(model, "b")
+            vm.run(workloadB)
+            vm.close()
+        }
+
+        machine.close()
+    }
+}
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/SimHypervisorTest.kt
deleted file mode 100644
index 058d5d28..00000000
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt
+++ /dev/null
@@ -1,280 +0,0 @@
-/*
- * 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 kotlinx.coroutines.*
-import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.BeforeEach
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.assertAll
-import org.junit.jupiter.api.assertDoesNotThrow
-import org.opendc.simulator.compute.SimBareMetalMachine
-import org.opendc.simulator.compute.kernel.cpufreq.PerformanceScalingGovernor
-import org.opendc.simulator.compute.kernel.interference.VmInterferenceGroup
-import org.opendc.simulator.compute.kernel.interference.VmInterferenceModel
-import org.opendc.simulator.compute.model.MachineModel
-import org.opendc.simulator.compute.model.MemoryUnit
-import org.opendc.simulator.compute.model.ProcessingNode
-import org.opendc.simulator.compute.model.ProcessingUnit
-import org.opendc.simulator.compute.power.ConstantPowerModel
-import org.opendc.simulator.compute.power.SimplePowerDriver
-import org.opendc.simulator.compute.workload.SimTraceWorkload
-import org.opendc.simulator.core.runBlockingSimulation
-import org.opendc.simulator.flow.FlowEngine
-
-/**
- * Test suite for the [SimHypervisor] class.
- */
-@OptIn(ExperimentalCoroutinesApi::class)
-internal class SimHypervisorTest {
-    private lateinit var model: MachineModel
-
-    @BeforeEach
-    fun setUp() {
-        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 1)
-        model = MachineModel(
-            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
-            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
-        )
-    }
-
-    /**
-     * Test overcommitting of resources via the hypervisor with a single VM.
-     */
-    @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(
-                sequenceOf(
-                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
-                ),
-            )
-
-        val platform = FlowEngine(coroutineContext, clock)
-        val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0)))
-        val hypervisor = SimFairShareHypervisor(platform, scalingGovernor = PerformanceScalingGovernor(), listener = listener)
-
-        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(1200000, clock.millis()) { "Current time is correct" } }
-        )
-    }
-
-    /**
-     * Test overcommitting of resources via the hypervisor with two VMs.
-     */
-    @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(
-                sequenceOf(
-                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
-                ),
-            )
-        val workloadB =
-            SimTraceWorkload(
-                sequenceOf(
-                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3100.0, 1),
-                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 73.0, 1)
-                )
-            )
-
-        val platform = FlowEngine(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimFairShareHypervisor(platform, listener = listener)
-
-        launch {
-            machine.run(hypervisor)
-        }
-
-        yield()
-        coroutineScope {
-            launch {
-                val vm = hypervisor.createMachine(model)
-                vm.run(workloadA)
-                vm.close()
-            }
-            val vm = hypervisor.createMachine(model)
-            vm.run(workloadB)
-            vm.close()
-        }
-        yield()
-        machine.close()
-        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(1200000, clock.millis()) }
-        )
-    }
-
-    @Test
-    fun testMultipleCPUs() = runBlockingSimulation {
-        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
-        val model = MachineModel(
-            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
-            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
-        )
-
-        val platform = FlowEngine(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimFairShareHypervisor(platform)
-
-        assertDoesNotThrow {
-            launch {
-                machine.run(hypervisor)
-            }
-        }
-
-        machine.close()
-    }
-
-    @Test
-    fun testInterference() = runBlockingSimulation {
-        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
-        val model = MachineModel(
-            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
-            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
-        )
-
-        val groups = listOf(
-            VmInterferenceGroup(targetLoad = 0.0, score = 0.9, members = setOf("a", "b")),
-            VmInterferenceGroup(targetLoad = 0.0, score = 0.6, members = setOf("a", "c")),
-            VmInterferenceGroup(targetLoad = 0.1, score = 0.8, members = setOf("a", "n"))
-        )
-        val interferenceModel = VmInterferenceModel(groups)
-
-        val platform = FlowEngine(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            platform, model, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimFairShareHypervisor(platform, interferenceDomain = interferenceModel.newDomain())
-
-        val duration = 5 * 60L
-        val workloadA =
-            SimTraceWorkload(
-                sequenceOf(
-                    SimTraceWorkload.Fragment(0, duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
-                ),
-            )
-        val workloadB =
-            SimTraceWorkload(
-                sequenceOf(
-                    SimTraceWorkload.Fragment(0, duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 3100.0, 1),
-                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 73.0, 1)
-                )
-            )
-
-        launch {
-            machine.run(hypervisor)
-        }
-
-        coroutineScope {
-            launch {
-                val vm = hypervisor.createMachine(model, "a")
-                vm.run(workloadA)
-                vm.close()
-            }
-            val vm = hypervisor.createMachine(model, "b")
-            vm.run(workloadB)
-            vm.close()
-        }
-
-        machine.close()
-    }
-}
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 95fb6679..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
@@ -75,10 +75,8 @@ internal class SimSpaceSharedHypervisorTest {
             )
 
         val engine = FlowEngine(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            FlowEngine(coroutineContext, clock), machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimSpaceSharedHypervisor(engine)
+        val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         val vm = hypervisor.createMachine(machineModel)
@@ -99,10 +97,8 @@ internal class SimSpaceSharedHypervisorTest {
         val duration = 5 * 60L * 1000
         val workload = SimRuntimeWorkload(duration)
         val engine = FlowEngine(coroutineContext, clock)
-        val machine = SimBareMetalMachine(
-            engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-        val hypervisor = SimSpaceSharedHypervisor(engine)
+        val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -125,7 +121,7 @@ internal class SimSpaceSharedHypervisorTest {
         val machine = SimBareMetalMachine(
             engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(engine)
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -146,7 +142,7 @@ internal class SimSpaceSharedHypervisorTest {
         val machine = SimBareMetalMachine(
             engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(engine)
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -172,7 +168,7 @@ internal class SimSpaceSharedHypervisorTest {
     fun testConcurrentWorkloadFails() = runBlockingSimulation {
         val engine = FlowEngine(coroutineContext, clock)
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
-        val hypervisor = SimSpaceSharedHypervisor(engine)
+        val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
@@ -196,7 +192,7 @@ internal class SimSpaceSharedHypervisorTest {
         val machine = SimBareMetalMachine(
             interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))
         )
-        val hypervisor = SimSpaceSharedHypervisor(interpreter)
+        val hypervisor = SimSpaceSharedHypervisor(interpreter, null, null)
 
         launch { machine.run(hypervisor) }
         yield()
-- 
cgit v1.2.3