simulator: Add uniform resource consumption model (v2)

This is the second pull request in the series of pull requests to add a uniform resource consumption model to OpenDC. This pull request focusses on adding dynamic capacity negotiation and propagation between resource consumer and resource provider:

* The generic resource constraint is removed from the interfaces of `opendc-simulator-resources`. Users of the API are expected to use the untyped variants where only the capacity needs to be specified. Users are expected to build higher-level abstractions on top of these interface to represent actual resources (e.g., CPU, disk or network).
* Added benchmarks for the most important implementations of `opendc-simulator-resources`. This allows us to quantify the effects of changes on the runtime.
* The `SimResourceSwitchMaxMin` has been split into a `SimResourceAggregatorMaxMin` and `SimResourceDistributorMaxMin` which respectively aggregate input resources and distribute output resources using max-min fair sharing.
* The `SimResourceConsumer` interface has a new method for receiving capacity change events: `onCapacityChanged(ctx, isThrottled)`

**Breaking API Changes**

*  All interfaces in `opendc-simulator-resources`.

17 files changed, 108 insertions, 165 deletions

diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
index a99b082a..81d09f12 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
@@ -27,12 +27,11 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.launch
 import org.opendc.simulator.compute.interference.PerformanceInterferenceModel
-import org.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.*
 import java.time.Clock
-import kotlin.coroutines.CoroutineContext
 
 /**
  * Abstract implementation of the [SimHypervisor] interface.
@@ -46,7 +45,7 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
     /**
      * The resource switch to use.
      */
-    private lateinit var switch: SimResourceSwitch<SimProcessingUnit>
+    private lateinit var switch: SimResourceSwitch
 
     /**
      * The virtual machines running on this hypervisor.
@@ -58,12 +57,12 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
     /**
      * Construct the [SimResourceSwitch] implementation that performs the actual scheduling of the CPUs.
      */
-    public abstract fun createSwitch(ctx: SimMachineContext): SimResourceSwitch<SimProcessingUnit>
+    public abstract fun createSwitch(ctx: SimMachineContext): SimResourceSwitch
 
     /**
      * Check whether the specified machine model fits on this hypervisor.
      */
-    public abstract fun canFit(model: SimMachineModel, switch: SimResourceSwitch<SimProcessingUnit>): Boolean
+    public abstract fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean
 
     override fun canFit(model: SimMachineModel): Boolean {
         return canFit(model, switch)
@@ -102,7 +101,7 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
         /**
          * The vCPUs of the machine.
          */
-        private val cpus: Map<SimProcessingUnit, SimResourceProvider<SimProcessingUnit>> = model.cpus.associateWith { switch.addOutput(it) }
+        private val cpus: Map<ProcessingUnit, SimResourceProvider> = model.cpus.associateWith { switch.addOutput(it.frequency) }
 
         /**
          * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
@@ -112,19 +111,19 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
                 require(!isTerminated) { "Machine is terminated" }
 
                 val ctx = object : SimMachineContext {
-                    override val cpus: List<SimProcessingUnit>
+                    override val cpus: List<ProcessingUnit>
                         get() = model.cpus
 
-                    override val memory: List<SimMemoryUnit>
+                    override val memory: List<MemoryUnit>
                         get() = model.memory
 
                     override val clock: Clock
                         get() = this@SimAbstractHypervisor.context.clock
 
-                    override val meta: Map<String, Any> = meta + mapOf("coroutine-context" to context.meta["coroutine-context"] as CoroutineContext)
+                    override val meta: Map<String, Any> = meta
 
-                    override fun interrupt(resource: SimResource) {
-                        requireNotNull(this@VirtualMachine.cpus[resource]).interrupt()
+                    override fun interrupt(cpu: ProcessingUnit) {
+                        requireNotNull(this@VirtualMachine.cpus[cpu]).interrupt()
                     }
                 }
 
@@ -156,8 +155,8 @@ public abstract class SimAbstractHypervisor : SimHypervisor {
         switch = createSwitch(ctx)
     }
 
-    override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit> {
-        val forwarder = SimResourceForwarder(cpu)
+    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
+        val forwarder = SimResourceForwarder()
         switch.addInput(forwarder)
         return forwarder
     }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
index 39ae34fe..1c0f94fd 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt
@@ -27,12 +27,12 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import org.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.resources.SimResource
 import org.opendc.simulator.resources.SimResourceProvider
 import org.opendc.simulator.resources.SimResourceSource
+import org.opendc.simulator.resources.consume
 import java.time.Clock
 import kotlin.coroutines.CoroutineContext
 
@@ -64,24 +64,24 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
     /**
      * The resources allocated for this machine.
      */
-    protected abstract val resources: Map<SimProcessingUnit, SimResourceSource<SimProcessingUnit>>
+    protected abstract val resources: Map<ProcessingUnit, SimResourceSource>
 
     /**
      * The execution context in which the workload runs.
      */
     private inner class Context(
-        val sources: Map<SimProcessingUnit, SimResourceProvider<SimProcessingUnit>>,
+        val sources: Map<ProcessingUnit, SimResourceProvider>,
         override val meta: Map<String, Any>
     ) : SimMachineContext {
         override val clock: Clock
             get() = this@SimAbstractMachine.clock
 
-        override val cpus: List<SimProcessingUnit> = model.cpus
+        override val cpus: List<ProcessingUnit> = model.cpus
 
-        override val memory: List<SimMemoryUnit> = model.memory
+        override val memory: List<MemoryUnit> = model.memory
 
-        override fun interrupt(resource: SimResource) {
-            checkNotNull(sources[resource]) { "Invalid resource" }.interrupt()
+        override fun interrupt(cpu: ProcessingUnit) {
+            checkNotNull(sources[cpu]) { "Invalid resource" }.interrupt()
         }
     }
 
@@ -91,7 +91,7 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
     override suspend fun run(workload: SimWorkload, meta: Map<String, Any>): Unit = withContext(context) {
         val resources = resources
         require(!isTerminated) { "Machine is terminated" }
-        val ctx = Context(resources, meta + mapOf("coroutine-context" to context))
+        val ctx = Context(resources, meta)
         val totalCapacity = model.cpus.sumByDouble { it.frequency }
 
         _speed = MutableList(model.cpus.size) { 0.0 }
@@ -102,7 +102,7 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
             val consumer = workload.getConsumer(ctx, cpu)
             val job = source.speed
                 .onEach {
-                    _speed[cpu.id] = source.speed.value
+                    _speed[cpu.id] = it
                     _usage.value = _speed.sum() / totalCapacity
                 }
                 .launchIn(this)
@@ -116,9 +116,8 @@ public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine
 
     override fun close() {
         if (!isTerminated) {
-            resources.forEach { (_, provider) -> provider.close() }
-        } else {
             isTerminated = true
+            resources.forEach { (_, provider) -> provider.close() }
         }
     }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
index 79982ea8..19479719 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt
@@ -23,7 +23,7 @@
 package org.opendc.simulator.compute
 
 import kotlinx.coroutines.*
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.resources.*
 import org.opendc.utils.TimerScheduler
 import java.time.Clock
@@ -57,8 +57,8 @@ public class SimBareMetalMachine(
      */
     private val scheduler = TimerScheduler<Any>(this.context, clock)
 
-    override val resources: Map<SimProcessingUnit, SimResourceSource<SimProcessingUnit>> =
-        model.cpus.associateWith { SimResourceSource(it, clock, scheduler) }
+    override val resources: Map<ProcessingUnit, SimResourceSource> =
+        model.cpus.associateWith { SimResourceSource(it.frequency, clock, scheduler) }
 
     override fun close() {
         super.close()
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
index bb97192d..fa677de9 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt
@@ -22,10 +22,8 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.compute.model.SimProcessingUnit
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.*
-import kotlin.coroutines.CoroutineContext
 
 /**
  * A [SimHypervisor] that distributes the computing requirements of multiple [SimWorkload] on a single
@@ -35,15 +33,14 @@ import kotlin.coroutines.CoroutineContext
  */
 public class SimFairShareHypervisor(private val listener: SimHypervisor.Listener? = null) : SimAbstractHypervisor() {
 
-    override fun canFit(model: SimMachineModel, switch: SimResourceSwitch<SimProcessingUnit>): Boolean = true
+    override fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean = true
 
-    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch<SimProcessingUnit> {
+    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
         return SimResourceSwitchMaxMin(
             ctx.clock,
-            ctx.meta["coroutine-context"] as CoroutineContext,
-            object : SimResourceSwitchMaxMin.Listener<SimProcessingUnit> {
+            object : SimResourceSwitchMaxMin.Listener {
                 override fun onSliceFinish(
-                    switch: SimResourceSwitchMaxMin<SimProcessingUnit>,
+                    switch: SimResourceSwitchMaxMin,
                     requestedWork: Long,
                     grantedWork: Long,
                     overcommittedWork: Long,
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt
index cff70826..85404e6e 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt
@@ -22,9 +22,8 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingUnit
-import org.opendc.simulator.resources.SimResource
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingUnit
 import java.time.Clock
 
 /**
@@ -46,17 +45,17 @@ public interface SimMachineContext {
     /**
      * The CPUs available on the machine.
      */
-    public val cpus: List<SimProcessingUnit>
+    public val cpus: List<ProcessingUnit>
 
     /**
      * The memory available on the machine
      */
-    public val memory: List<SimMemoryUnit>
+    public val memory: List<MemoryUnit>
 
     /**
-     * Interrupt the specified [resource].
+     * Interrupt the specified [cpu].
      *
      * @throws IllegalArgumentException if the resource does not belong to this execution context.
      */
-    public fun interrupt(resource: SimResource)
+    public fun interrupt(cpu: ProcessingUnit)
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineModel.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineModel.kt
index d6bf0e99..2b414540 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineModel.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineModel.kt
@@ -22,8 +22,8 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingUnit
 
 /**
  * A description of the physical or virtual machine on which a bootable image runs.
@@ -31,4 +31,4 @@ import org.opendc.simulator.compute.model.SimProcessingUnit
  * @property cpus The list of processing units available to the image.
  * @property memory The list of memory units available to the image.
  */
-public data class SimMachineModel(public val cpus: List<SimProcessingUnit>, public val memory: List<SimMemoryUnit>)
+public data class SimMachineModel(public val cpus: List<ProcessingUnit>, public val memory: List<MemoryUnit>)
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
index 2001a230..fd8e546f 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt
@@ -22,19 +22,17 @@
 
 package org.opendc.simulator.compute
 
-import org.opendc.simulator.compute.model.SimProcessingUnit
 import org.opendc.simulator.resources.*
-import kotlin.coroutines.CoroutineContext
 
 /**
  * A [SimHypervisor] that allocates its sub-resources exclusively for the virtual machine that it hosts.
  */
 public class SimSpaceSharedHypervisor : SimAbstractHypervisor() {
-    override fun canFit(model: SimMachineModel, switch: SimResourceSwitch<SimProcessingUnit>): Boolean {
+    override fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean {
         return switch.inputs.size - switch.outputs.size >= model.cpus.size
     }
 
-    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch<SimProcessingUnit> {
-        return SimResourceSwitchExclusive(ctx.meta["coroutine-context"] as CoroutineContext)
+    override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch {
+        return SimResourceSwitchExclusive()
     }
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/SimMemoryUnit.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/MemoryUnit.kt
index 49745868..bcbde5b1 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/SimMemoryUnit.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/MemoryUnit.kt
@@ -22,8 +22,6 @@
 
 package org.opendc.simulator.compute.model
 
-import org.opendc.simulator.resources.SimResource
-
 /**
  * A memory unit of a compute resource, either virtual or physical.
  *
@@ -32,12 +30,9 @@ import org.opendc.simulator.resources.SimResource
  * @property speed The access speed of the memory in MHz.
  * @property size The size of the memory unit in MBs.
  */
-public data class SimMemoryUnit(
+public data class MemoryUnit(
     public val vendor: String,
     public val modelName: String,
     public val speed: Double,
     public val size: Long
-) : SimResource {
-    override val capacity: Double
-        get() = speed
-}
+)
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/SimProcessingNode.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/ProcessingNode.kt
index 4022ecb3..58ed816c 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/SimProcessingNode.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/ProcessingNode.kt
@@ -30,7 +30,7 @@ package org.opendc.simulator.compute.model
  * @property arch The micro-architecture of the processor node.
  * @property coreCount The number of logical CPUs in the processor node.
  */
-public data class SimProcessingNode(
+public data class ProcessingNode(
     public val vendor: String,
     public val arch: String,
     public val modelName: String,
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/SimProcessingUnit.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/ProcessingUnit.kt
index 1c989254..415e95e6 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/SimProcessingUnit.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/model/ProcessingUnit.kt
@@ -22,8 +22,6 @@
 
 package org.opendc.simulator.compute.model
 
-import org.opendc.simulator.resources.SimResource
-
 /**
  * A single logical compute unit of processor node, either virtual or physical.
  *
@@ -31,11 +29,8 @@ import org.opendc.simulator.resources.SimResource
  * @property id The identifier of the CPU core within the processing node.
  * @property frequency The clock rate of the CPU in MHz.
  */
-public data class SimProcessingUnit(
-    public val node: SimProcessingNode,
+public data class ProcessingUnit(
+    public val node: ProcessingNode,
     public val id: Int,
     public val frequency: Double
-) : SimResource {
-    override val capacity: Double
-        get() = frequency
-}
+)
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
index 9b47821e..63c9d28c 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimFlopsWorkload.kt
@@ -23,10 +23,9 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.compute.model.SimProcessingUnit
-import org.opendc.simulator.resources.SimResourceCommand
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
  * A [SimWorkload] that models applications as a static number of floating point operations ([flops]) executed on
@@ -46,30 +45,8 @@ public class SimFlopsWorkload(
 
     override fun onStart(ctx: SimMachineContext) {}
 
-    override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit> {
-        return CpuConsumer(ctx)
-    }
-
-    private inner class CpuConsumer(private val machine: SimMachineContext) : SimResourceConsumer<SimProcessingUnit> {
-        override fun onStart(ctx: SimResourceContext<SimProcessingUnit>): SimResourceCommand {
-            val limit = ctx.resource.frequency * utilization
-            val work = flops.toDouble() / machine.cpus.size
-
-            return if (work > 0.0) {
-                SimResourceCommand.Consume(work, limit)
-            } else {
-                SimResourceCommand.Exit
-            }
-        }
-
-        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, remainingWork: Double): SimResourceCommand {
-            return if (remainingWork > 0.0) {
-                val limit = ctx.resource.frequency * utilization
-                return SimResourceCommand.Consume(remainingWork, limit)
-            } else {
-                SimResourceCommand.Exit
-            }
-        }
+    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
+        return SimWorkConsumer(flops.toDouble() / ctx.cpus.size, utilization)
     }
 
     override fun toString(): String = "SimFlopsWorkload(FLOPs=$flops,utilization=$utilization)"
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
index 313b6ed5..a3420e32 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimRuntimeWorkload.kt
@@ -23,10 +23,9 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.compute.model.SimProcessingUnit
-import org.opendc.simulator.resources.SimResourceCommand
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.resources.SimResourceConsumer
-import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
  * A [SimWorkload] that models application execution as a single duration.
@@ -45,25 +44,9 @@ public class SimRuntimeWorkload(
 
     override fun onStart(ctx: SimMachineContext) {}
 
-    override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit> {
-        return CpuConsumer()
-    }
-
-    private inner class CpuConsumer : SimResourceConsumer<SimProcessingUnit> {
-        override fun onStart(ctx: SimResourceContext<SimProcessingUnit>): SimResourceCommand {
-            val limit = ctx.resource.frequency * utilization
-            val work = (limit / 1000) * duration
-            return SimResourceCommand.Consume(work, limit)
-        }
-
-        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, remainingWork: Double): SimResourceCommand {
-            return if (remainingWork > 0.0) {
-                val limit = ctx.resource.frequency * utilization
-                SimResourceCommand.Consume(remainingWork, limit)
-            } else {
-                SimResourceCommand.Exit
-            }
-        }
+    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
+        val limit = cpu.frequency * utilization
+        return SimWorkConsumer((limit / 1000) * duration, utilization)
     }
 
     override fun toString(): String = "SimRuntimeWorkload(duration=$duration,utilization=$utilization)"
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
index 31f58a0f..2442d748 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt
@@ -23,7 +23,7 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.compute.model.SimProcessingUnit
+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
@@ -45,35 +45,29 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>) : SimWorkloa
         offset = ctx.clock.millis()
     }
 
-    override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit> {
-        return CpuConsumer()
-    }
-
-    private inner class CpuConsumer : SimResourceConsumer<SimProcessingUnit> {
-        override fun onStart(ctx: SimResourceContext<SimProcessingUnit>): SimResourceCommand {
-            return onNext(ctx, 0.0)
-        }
+    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
+        return object : SimResourceConsumer {
+            override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+                val now = ctx.clock.millis()
+                val fragment = fragment ?: return SimResourceCommand.Exit
+                val work = (fragment.duration / 1000) * fragment.usage
+                val deadline = offset + fragment.duration
 
-        override fun onNext(ctx: SimResourceContext<SimProcessingUnit>, remainingWork: Double): SimResourceCommand {
-            val now = ctx.clock.millis()
-            val fragment = fragment ?: return SimResourceCommand.Exit
-            val work = (fragment.duration / 1000) * fragment.usage
-            val deadline = offset + fragment.duration
+                assert(deadline >= now) { "Deadline already passed" }
 
-            assert(deadline >= now) { "Deadline already passed" }
+                val cmd =
+                    if (cpu.id < fragment.cores && work > 0.0)
+                        SimResourceCommand.Consume(work, fragment.usage, deadline)
+                    else
+                        SimResourceCommand.Idle(deadline)
 
-            val cmd =
-                if (ctx.resource.id < fragment.cores && work > 0.0)
-                    SimResourceCommand.Consume(work, fragment.usage, deadline)
-                else
-                    SimResourceCommand.Idle(deadline)
+                if (barrier.enter()) {
+                    this@SimTraceWorkload.fragment = nextFragment()
+                    this@SimTraceWorkload.offset += fragment.duration
+                }
 
-            if (barrier.enter()) {
-                this@SimTraceWorkload.fragment = nextFragment()
-                this@SimTraceWorkload.offset += fragment.duration
+                return cmd
             }
-
-            return cmd
         }
     }
 
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt
index 60661e23..bdc12bb5 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt
@@ -23,7 +23,7 @@
 package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.resources.SimResourceConsumer
 
 /**
@@ -41,5 +41,5 @@ public interface SimWorkload {
     /**
      * Obtain the resource consumer for the specified processing unit.
      */
-    public fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer<SimProcessingUnit>
+    public fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer
 }
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimHypervisorTest.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimHypervisorTest.kt
index 4ac8cf63..5773b325 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimHypervisorTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimHypervisorTest.kt
@@ -24,6 +24,7 @@ package org.opendc.simulator.compute
 
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runBlockingTest
 import kotlinx.coroutines.yield
@@ -31,9 +32,9 @@ 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.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingNode
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingNode
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimTraceWorkload
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 
@@ -46,10 +47,10 @@ internal class SimHypervisorTest {
 
     @BeforeEach
     fun setUp() {
-        val cpuNode = SimProcessingNode("Intel", "Xeon", "amd64", 1)
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 1)
         model = SimMachineModel(
-            cpus = List(cpuNode.coreCount) { SimProcessingUnit(cpuNode, it, 3200.0) },
-            memory = List(4) { SimMemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
     }
 
@@ -98,15 +99,21 @@ internal class SimHypervisorTest {
             println("Hypervisor finished")
         }
         yield()
-        hypervisor.createMachine(model).run(workloadA)
+        val vm = hypervisor.createMachine(model)
+        val res = mutableListOf<Double>()
+        val job = launch { machine.usage.toList(res) }
+
+        vm.run(workloadA)
         yield()
+        job.cancel()
         machine.close()
 
         assertAll(
             { assertEquals(1113300, listener.totalRequestedWork, "Requested Burst does not match") },
             { assertEquals(1023300, listener.totalGrantedWork, "Granted Burst does not match") },
             { assertEquals(90000, listener.totalOvercommittedWork, "Overcommissioned Burst does not match") },
-            { assertEquals(1200000, currentTime) }
+            { assertEquals(listOf(0.0, 0.00875, 1.0, 0.0, 0.0571875, 0.0), res) { "VM usage is correct" } },
+            { assertEquals(1200000, currentTime) { "Current time is correct" } }
         )
     }
 
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
index 6adc41d0..071bdf77 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimMachineTest.kt
@@ -28,9 +28,9 @@ import kotlinx.coroutines.test.runBlockingTest
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
-import org.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingNode
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingNode
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimFlopsWorkload
 import org.opendc.simulator.utils.DelayControllerClockAdapter
 
@@ -43,11 +43,11 @@ class SimMachineTest {
 
     @BeforeEach
     fun setUp() {
-        val cpuNode = SimProcessingNode("Intel", "Xeon", "amd64", 2)
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
 
         machineModel = SimMachineModel(
-            cpus = List(cpuNode.coreCount) { SimProcessingUnit(cpuNode, it, 1000.0) },
-            memory = List(4) { SimMemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 1000.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
     }
 
@@ -76,7 +76,7 @@ class SimMachineTest {
 
         try {
             machine.run(SimFlopsWorkload(2_000, utilization = 1.0))
-
+            yield()
             job.cancel()
             assertEquals(listOf(0.0, 0.5, 1.0, 0.5, 0.0), res) { "Machine is fully utilized" }
         } finally {
diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorTest.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorTest.kt
index 8428a0a7..fb0523af 100644
--- a/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorTest.kt
+++ b/simulator/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorTest.kt
@@ -31,9 +31,9 @@ import org.junit.jupiter.api.Assertions.*
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.compute.model.SimMemoryUnit
-import org.opendc.simulator.compute.model.SimProcessingNode
-import org.opendc.simulator.compute.model.SimProcessingUnit
+import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.model.ProcessingNode
+import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.compute.workload.SimFlopsWorkload
 import org.opendc.simulator.compute.workload.SimRuntimeWorkload
 import org.opendc.simulator.compute.workload.SimTraceWorkload
@@ -48,10 +48,10 @@ internal class SimSpaceSharedHypervisorTest {
 
     @BeforeEach
     fun setUp() {
-        val cpuNode = SimProcessingNode("Intel", "Xeon", "amd64", 1)
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 1)
         machineModel = SimMachineModel(
-            cpus = List(cpuNode.coreCount) { SimProcessingUnit(cpuNode, it, 3200.0) },
-            memory = List(4) { SimMemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 3200.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
         )
     }