simulator: Start consumers directly from workload

This change updates the SimWorkload interfaces to allow
implementations to start consumers for the machine resource providers
directly.

10 files changed, 233 insertions, 224 deletions

diff --git a/opendc-experiments/opendc-experiments-tf20/src/main/kotlin/org/opendc/experiments/tf20/core/SimTFDevice.kt b/opendc-experiments/opendc-experiments-tf20/src/main/kotlin/org/opendc/experiments/tf20/core/SimTFDevice.kt
index f345a4a5..db131e82 100644
--- a/opendc-experiments/opendc-experiments-tf20/src/main/kotlin/org/opendc/experiments/tf20/core/SimTFDevice.kt
+++ b/opendc-experiments/opendc-experiments-tf20/src/main/kotlin/org/opendc/experiments/tf20/core/SimTFDevice.kt
@@ -116,9 +116,11 @@ public class SimTFDevice(
          */
         private var activeWork: Work? = null
 
-        override fun onStart(ctx: SimMachineContext) {}
-
-        override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer = this
+        override fun onStart(ctx: SimMachineContext) {
+            for (cpu in ctx.cpus) {
+                cpu.startConsumer(this)
+            }
+        }
 
         override fun onNext(ctx: SimResourceContext): SimResourceCommand {
             val activeWork = activeWork
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
index 2b84a17c..68ecc49f 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt
@@ -22,17 +22,10 @@
 
 package org.opendc.simulator.compute
 
-import kotlinx.coroutines.coroutineScope
-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.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingUnit
-import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.resources.*
 import org.opendc.simulator.resources.SimResourceSwitch
-import java.time.Clock
 
 /**
  * Abstract implementation of the [SimHypervisor] interface.
@@ -86,117 +79,37 @@ public abstract class SimAbstractHypervisor(private val interpreter: SimResource
      * @property performanceInterferenceModel The performance interference model to utilize.
      */
     private inner class VirtualMachine(
-        override val model: SimMachineModel,
+        model: SimMachineModel,
         val performanceInterferenceModel: PerformanceInterferenceModel? = null,
-    ) : SimMachine {
-        /**
-         * A [StateFlow] representing the CPU usage of the simulated machine.
-         */
-        override val usage: MutableStateFlow<Double> = MutableStateFlow(0.0)
-
-        /**
-         *  A flag to indicate that the machine is terminated.
-         */
-        private var isTerminated = false
-
+    ) : SimAbstractMachine(interpreter, parent = null, model) {
         /**
          * The vCPUs of the machine.
          */
-        private val cpus = model.cpus.map { ProcessingUnitImpl(it, switch) }
-
-        /**
-         * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
-         */
-        override suspend fun run(workload: SimWorkload, meta: Map<String, Any>) {
-            coroutineScope {
-                require(!isTerminated) { "Machine is terminated" }
+        override val cpus = model.cpus.map { VCpu(switch.newOutput(), it) }
 
-                val ctx = object : SimMachineContext {
-                    override val cpus: List<SimProcessingUnit> = this@VirtualMachine.cpus
-
-                    override val memory: List<MemoryUnit>
-                        get() = model.memory
-
-                    override val clock: Clock
-                        get() = this@SimAbstractHypervisor.context.clock
-
-                    override val meta: Map<String, Any> = meta
-                }
-
-                interpreter.batch {
-                    workload.onStart(ctx)
-
-                    for (cpu in cpus) {
-                        launch {
-                            cpu.consume(workload.getConsumer(ctx, cpu.model))
-                        }
-                    }
-                }
-            }
-        }
-
-        /**
-         * Terminate this VM instance.
-         */
         override fun close() {
-            if (!isTerminated) {
-                isTerminated = true
+            super.close()
 
-                cpus.forEach(SimProcessingUnit::close)
-                _vms.remove(this)
-            }
+            _vms.remove(this)
         }
     }
 
     override fun onStart(ctx: SimMachineContext) {
         context = ctx
         switch = createSwitch(ctx)
-    }
 
-    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
-        val forwarder = SimResourceForwarder()
-        switch.addInput(forwarder)
-        return forwarder
+        for (cpu in ctx.cpus) {
+            switch.addInput(cpu)
+        }
     }
 
     /**
-     * The [SimProcessingUnit] of this machine.
+     * A [SimProcessingUnit] of a virtual machine.
      */
-    public inner class ProcessingUnitImpl(override val model: ProcessingUnit, switch: SimResourceSwitch) : SimProcessingUnit {
-        /**
-         * The actual resource supporting the processing unit.
-         */
-        private val source = switch.newOutput()
-
-        override val state: SimResourceState
-            get() = source.state
-
-        override val capacity: Double
-            get() = source.capacity
-
-        override val speed: Double
-            get() = source.speed
-
-        override val demand: Double
-            get() = source.demand
-
-        override val counters: SimResourceCounters
-            get() = source.counters
-
-        override fun startConsumer(consumer: SimResourceConsumer) {
-            source.startConsumer(consumer)
-        }
-
-        override fun interrupt() {
-            source.interrupt()
-        }
-
-        override fun cancel() {
-            source.cancel()
-        }
-
-        override fun close() {
-            source.close()
-        }
+    private class VCpu(
+        private val source: SimResourceProvider,
+        override val model: ProcessingUnit
+    ) : SimProcessingUnit, SimResourceProvider by source {
+        override fun toString(): String = "SimAbstractHypervisor.VCpu[model=$model]"
     }
 }
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 d40cdac5..e12ac72b 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
@@ -27,24 +27,27 @@ import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.workload.SimWorkload
-import org.opendc.simulator.resources.SimResourceInterpreter
-import org.opendc.simulator.resources.SimResourceSystem
-import org.opendc.simulator.resources.batch
-import org.opendc.simulator.resources.consume
-import java.time.Clock
+import org.opendc.simulator.resources.*
+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 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?
+    final override val parent: SimResourceSystem?,
+    final override val model: SimMachineModel
 ) : SimMachine, SimResourceSystem {
+    /**
+     * A [StateFlow] representing the CPU usage of the simulated machine.
+     */
     private val _usage = MutableStateFlow(0.0)
-    override val usage: StateFlow<Double>
+    public final override val usage: StateFlow<Double>
         get() = _usage
 
     /**
@@ -55,49 +58,66 @@ public abstract class SimAbstractMachine(
     private var _speed = doubleArrayOf()
 
     /**
-     * A flag to indicate that the machine is terminated.
-     */
-    private var isTerminated = false
-
-    /**
      * The resources allocated for this machine.
      */
     protected abstract val cpus: List<SimProcessingUnit>
 
     /**
-     * The execution context in which the workload runs.
+     * A flag to indicate that the machine is terminated.
      */
-    private inner class Context(override val meta: Map<String, Any>) : SimMachineContext {
-        override val clock: Clock
-            get() = interpreter.clock
-
-        override val cpus: List<SimProcessingUnit> = this@SimAbstractMachine.cpus
+    private var isTerminated = false
 
-        override val memory: List<MemoryUnit> = model.memory
-    }
+    /**
+     * The continuation to resume when the virtual machine workload has finished.
+     */
+    private var cont: Continuation<Unit>? = null
 
     /**
      * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished.
      */
-    override suspend fun run(workload: SimWorkload, meta: Map<String, Any>): Unit = coroutineScope {
+    override suspend fun run(workload: SimWorkload, meta: Map<String, Any>) {
         check(!isTerminated) { "Machine is terminated" }
+        check(cont == null) { "A machine cannot run concurrently" }
+
         val ctx = Context(meta)
 
         // Before the workload starts, initialize the initial power draw
         _speed = DoubleArray(model.cpus.size) { 0.0 }
         updateUsage(0.0)
 
-        interpreter.batch {
-            workload.onStart(ctx)
+        return suspendCancellableCoroutine { cont ->
+            this.cont = cont
+
+            // Cancel all cpus on cancellation
+            cont.invokeOnCancellation {
+                this.cont = null
 
+                interpreter.batch {
+                    for (cpu in cpus) {
+                        cpu.cancel()
+                    }
+                }
+            }
+
+            interpreter.batch { workload.onStart(ctx) }
+        }
+    }
+
+    override fun close() {
+        if (isTerminated) {
+            return
+        }
+
+        isTerminated = true
+        cancel()
+        interpreter.batch {
             for (cpu in cpus) {
-                val model = cpu.model
-                val consumer = workload.getConsumer(ctx, model)
-                launch { cpu.consume(consumer) }
+                cpu.close()
             }
         }
     }
 
+    /* SimResourceSystem */
     override fun onConverge(timestamp: Long) {
         val totalCapacity = model.cpus.sumOf { it.frequency }
         val cpus = cpus
@@ -117,12 +137,34 @@ public abstract class SimAbstractMachine(
         _usage.value = usage
     }
 
-    override fun close() {
-        if (isTerminated) {
-            return
+    /**
+     * Cancel the workload that is currently running on the machine.
+     */
+    private fun cancel() {
+        interpreter.batch {
+            for (cpu in cpus) {
+                cpu.cancel()
+            }
         }
 
-        isTerminated = true
-        cpus.forEach(SimProcessingUnit::close)
+        val cont = cont
+        if (cont != null) {
+            this.cont = null
+            cont.resume(Unit)
+        }
+    }
+
+    /**
+     * 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 cpus: List<SimProcessingUnit> = this@SimAbstractMachine.cpus
+
+        override val memory: List<MemoryUnit> = model.memory
+
+        override fun close() = cancel()
     }
 }
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 082719e2..7a49f29a 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
@@ -44,12 +44,14 @@ import org.opendc.simulator.resources.SimResourceInterpreter
 @OptIn(ExperimentalCoroutinesApi::class, InternalCoroutinesApi::class)
 public class SimBareMetalMachine(
     interpreter: SimResourceInterpreter,
-    override val model: SimMachineModel,
+    model: SimMachineModel,
     scalingGovernor: ScalingGovernor,
     scalingDriver: ScalingDriver,
     parent: SimResourceSystem? = null,
-) : SimAbstractMachine(interpreter, parent) {
-    override val cpus: List<SimProcessingUnit> = model.cpus.map { ProcessingUnitImpl(it) }
+) : SimAbstractMachine(interpreter, parent, model) {
+    override val cpus: List<SimProcessingUnit> = model.cpus.map { cpu ->
+        Cpu(SimResourceSource(cpu.frequency, interpreter, this@SimBareMetalMachine), cpu)
+    }
 
     /**
      * Construct the [ScalingDriver.Logic] for this machine.
@@ -81,43 +83,12 @@ public class SimBareMetalMachine(
     }
 
     /**
-     * The [SimProcessingUnit] of this machine.
+     * A [SimProcessingUnit] of a bare-metal machine.
      */
-    public inner class ProcessingUnitImpl(override val model: ProcessingUnit) : SimProcessingUnit {
-        /**
-         * The actual resource supporting the processing unit.
-         */
-        private val source = SimResourceSource(model.frequency, interpreter, this@SimBareMetalMachine)
-
-        override val state: SimResourceState
-            get() = source.state
-
-        override val capacity: Double
-            get() = source.capacity
-
-        override val speed: Double
-            get() = source.speed
-
-        override val demand: Double
-            get() = source.demand
-
-        override val counters: SimResourceCounters
-            get() = source.counters
-
-        override fun startConsumer(consumer: SimResourceConsumer) {
-            source.startConsumer(consumer)
-        }
-
-        override fun interrupt() {
-            source.interrupt()
-        }
-
-        override fun cancel() {
-            source.cancel()
-        }
-
-        override fun close() {
-            source.close()
-        }
+    private class Cpu(
+        private val source: SimResourceProvider,
+        override val model: ProcessingUnit
+    ) : SimProcessingUnit, SimResourceProvider by source {
+        override fun toString(): String = "SimBareMetalMachine.Cpu[model=$model]"
     }
 }
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 c2523a2a..5cbabc86 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
@@ -23,18 +23,18 @@
 package org.opendc.simulator.compute
 
 import org.opendc.simulator.compute.model.MemoryUnit
-import java.time.Clock
+import org.opendc.simulator.resources.SimResourceInterpreter
 
 /**
  * A simulated execution context in which a bootable image runs. This interface represents the
  * firmware interface between the running image (e.g. operating system) and the physical or virtual firmware on
  * which the image runs.
  */
-public interface SimMachineContext {
+public interface SimMachineContext : AutoCloseable {
     /**
-     * The virtual clock tracking simulation time.
+     * The resource interpreter that simulates the machine.
      */
-    public val clock: Clock
+    public val interpreter: SimResourceInterpreter
 
     /**
      * The metadata associated with the context.
@@ -50,4 +50,9 @@ public interface SimMachineContext {
      * The memory available on the machine
      */
     public val memory: List<MemoryUnit>
+
+    /**
+     * Stop the workload.
+     */
+    public override fun close()
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/util/SimWorkloadLifecycle.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/util/SimWorkloadLifecycle.kt
new file mode 100644
index 00000000..43662d93
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/util/SimWorkloadLifecycle.kt
@@ -0,0 +1,76 @@
+/*
+ * 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.util
+
+import org.opendc.simulator.compute.SimMachineContext
+import org.opendc.simulator.compute.workload.SimWorkload
+import org.opendc.simulator.resources.SimResourceConsumer
+import org.opendc.simulator.resources.SimResourceContext
+import org.opendc.simulator.resources.SimResourceEvent
+
+/**
+ * A helper class to manage the lifecycle of a [SimWorkload]
+ */
+public class SimWorkloadLifecycle(private val ctx: SimMachineContext) {
+    /**
+     * The resource consumers which represent the lifecycle of the workload.
+     */
+    private val waiting = mutableSetOf<SimResourceConsumer>()
+
+    /**
+     * Wait for the specified [consumer] to complete before ending the lifecycle of the workload.
+     */
+    public fun waitFor(consumer: SimResourceConsumer): SimResourceConsumer {
+        waiting.add(consumer)
+        return object : SimResourceConsumer by consumer {
+            override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
+                try {
+                    consumer.onEvent(ctx, event)
+                } finally {
+                    if (event == SimResourceEvent.Exit) {
+                        complete(consumer)
+                    }
+                }
+            }
+
+            override fun onFailure(ctx: SimResourceContext, cause: Throwable) {
+                try {
+                    consumer.onFailure(ctx, cause)
+                } finally {
+                    complete(consumer)
+                }
+            }
+
+            override fun toString(): String = "SimWorkloadLifecycle.Consumer[delegate=$consumer]"
+        }
+    }
+
+    /**
+     * Complete the specified [SimResourceConsumer].
+     */
+    private fun complete(consumer: SimResourceConsumer) {
+        if (waiting.remove(consumer) && waiting.isEmpty()) {
+            ctx.close()
+        }
+    }
+}
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 63c9d28c..de6832ca 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,8 +23,7 @@
 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.compute.util.SimWorkloadLifecycle
 import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
@@ -43,10 +42,11 @@ public class SimFlopsWorkload(
         require(utilization > 0.0 && utilization <= 1.0) { "Utilization must be in (0, 1]" }
     }
 
-    override fun onStart(ctx: SimMachineContext) {}
-
-    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
-        return SimWorkConsumer(flops.toDouble() / ctx.cpus.size, utilization)
+    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)))
+        }
     }
 
     override fun toString(): String = "SimFlopsWorkload(FLOPs=$flops,utilization=$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 a3420e32..318a6b49 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,8 +23,7 @@
 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.compute.util.SimWorkloadLifecycle
 import org.opendc.simulator.resources.consumer.SimWorkConsumer
 
 /**
@@ -42,11 +41,12 @@ public class SimRuntimeWorkload(
         require(utilization > 0.0 && utilization <= 1.0) { "Utilization must be in (0, 1]" }
     }
 
-    override fun onStart(ctx: SimMachineContext) {}
-
-    override fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer {
-        val limit = cpu.frequency * utilization
-        return SimWorkConsumer((limit / 1000) * duration, utilization)
+    override fun onStart(ctx: SimMachineContext) {
+        val lifecycle = SimWorkloadLifecycle(ctx)
+        for (cpu in ctx.cpus) {
+            val limit = cpu.capacity * utilization
+            cpu.startConsumer(lifecycle.waitFor(SimWorkConsumer((limit / 1000) * duration, utilization)))
+        }
     }
 
     override fun toString(): String = "SimRuntimeWorkload(duration=$duration,utilization=$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 ffb332d1..6929f4d2 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,6 +24,7 @@ package org.opendc.simulator.compute.workload
 
 import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.compute.model.ProcessingUnit
+import org.opendc.simulator.compute.util.SimWorkloadLifecycle
 import org.opendc.simulator.resources.SimResourceCommand
 import org.opendc.simulator.resources.SimResourceConsumer
 import org.opendc.simulator.resources.SimResourceContext
@@ -44,33 +45,12 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>) : SimWorkloa
 
         barrier = SimConsumerBarrier(ctx.cpus.size)
         fragment = nextFragment()
-        offset = ctx.clock.millis()
-    }
-
-    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 usage = fragment.usage / fragment.cores
-                val work = (fragment.duration / 1000) * usage
-                val deadline = offset + fragment.duration
-
-                assert(deadline >= now) { "Deadline already passed" }
-
-                val cmd =
-                    if (cpu.id < fragment.cores && work > 0.0)
-                        SimResourceCommand.Consume(work, usage, deadline)
-                    else
-                        SimResourceCommand.Idle(deadline)
+        offset = ctx.interpreter.clock.millis()
 
-                if (barrier.enter()) {
-                    this@SimTraceWorkload.fragment = nextFragment()
-                    this@SimTraceWorkload.offset += fragment.duration
-                }
+        val lifecycle = SimWorkloadLifecycle(ctx)
 
-                return cmd
-            }
+        for (cpu in ctx.cpus) {
+            cpu.startConsumer(lifecycle.waitFor(Consumer(cpu.model)))
         }
     }
 
@@ -87,6 +67,31 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>) : SimWorkloa
         }
     }
 
+    private inner class Consumer(val cpu: ProcessingUnit) : SimResourceConsumer {
+        override fun onNext(ctx: SimResourceContext): SimResourceCommand {
+            val now = ctx.clock.millis()
+            val fragment = fragment ?: return SimResourceCommand.Exit
+            val usage = fragment.usage / fragment.cores
+            val work = (fragment.duration / 1000) * usage
+            val deadline = offset + fragment.duration
+
+            assert(deadline >= now) { "Deadline already passed" }
+
+            val cmd =
+                if (cpu.id < fragment.cores && work > 0.0)
+                    SimResourceCommand.Consume(work, usage, deadline)
+                else
+                    SimResourceCommand.Idle(deadline)
+
+            if (barrier.enter()) {
+                this@SimTraceWorkload.fragment = nextFragment()
+                this@SimTraceWorkload.offset += fragment.duration
+            }
+
+            return cmd
+        }
+    }
+
     /**
      * A fragment of the workload.
      */
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt
index bdc12bb5..b80665fa 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkload.kt
@@ -23,8 +23,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.SimResourceConsumer
 
 /**
  * A model that characterizes the runtime behavior of some particular workload.
@@ -35,11 +33,8 @@ import org.opendc.simulator.resources.SimResourceConsumer
 public interface SimWorkload {
     /**
      * This method is invoked when the workload is started.
+     *
+     * @param ctx The execution context in which the machine runs.
      */
     public fun onStart(ctx: SimMachineContext)
-
-    /**
-     * Obtain the resource consumer for the specified processing unit.
-     */
-    public fun getConsumer(ctx: SimMachineContext, cpu: ProcessingUnit): SimResourceConsumer
 }