refactor(simulator): Support running workloads without coroutines

This change updates the SimMachine interface to drop the coroutine
requirement for running a workload on a machines. Users can now
asynchronously start a workload and receive notifications via the
workload callbacks.

Users still have the possibility to suspend execution during workload
execution by using the new `runWorkload` method, which is implemented on
top of the new `startWorkload` primitive.

4 files changed, 150 insertions, 204 deletions

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 0bb24ed8..644eb497 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
@@ -65,14 +65,10 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(SimFlopsWorkload(2_000, utilization = 1.0))
+        machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0))
 
-            // Two cores execute 1000 MFlOps per second (1000 ms)
-            assertEquals(1000, clock.millis())
-        } finally {
-            machine.close()
-        }
+        // Two cores execute 1000 MFlOps per second (1000 ms)
+        assertEquals(1000, clock.millis())
     }
 
     @Test
@@ -88,14 +84,10 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(SimFlopsWorkload(2_000, utilization = 1.0))
+        machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0))
 
-            // Two sockets with two cores execute 2000 MFlOps per second (500 ms)
-            assertEquals(500, clock.millis())
-        } finally {
-            machine.close()
-        }
+        // Two sockets with two cores execute 2000 MFlOps per second (500 ms)
+        assertEquals(500, clock.millis())
     }
 
     @Test
@@ -109,16 +101,12 @@ class SimMachineTest {
         val source = SimPowerSource(engine, capacity = 1000.0)
         source.connect(machine.psu)
 
-        try {
-            coroutineScope {
-                launch { machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) }
-                assertAll(
-                    { assertEquals(100.0, machine.psu.powerDraw) },
-                    { assertEquals(100.0, source.powerDraw) }
-                )
-            }
-        } finally {
-            machine.close()
+        coroutineScope {
+            launch { machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) }
+            assertAll(
+                { assertEquals(100.0, machine.psu.powerDraw) },
+                { assertEquals(100.0, source.powerDraw) }
+            )
         }
     }
 
@@ -130,22 +118,20 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(object : SimWorkload {
-                override fun onStart(ctx: SimMachineContext) {
-                    val cpu = ctx.cpus[0]
-
-                    cpu.capacity = cpu.model.frequency + 1000.0
-                    assertEquals(cpu.model.frequency, cpu.capacity)
-                    cpu.capacity = -1.0
-                    assertEquals(0.0, cpu.capacity)
-
-                    ctx.close()
-                }
-            })
-        } finally {
-            machine.close()
-        }
+        machine.runWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                val cpu = ctx.cpus[0]
+
+                cpu.capacity = cpu.model.frequency + 1000.0
+                assertEquals(cpu.model.frequency, cpu.capacity)
+                cpu.capacity = -1.0
+                assertEquals(0.0, cpu.capacity)
+
+                ctx.close()
+            }
+
+            override fun onStop(ctx: SimMachineContext) {}
+        })
     }
 
     @Test
@@ -156,16 +142,14 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(object : SimWorkload {
-                override fun onStart(ctx: SimMachineContext) {
-                    assertEquals(32_000 * 4.0, ctx.memory.capacity)
-                    ctx.close()
-                }
-            })
-        } finally {
-            machine.close()
-        }
+        machine.runWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                assertEquals(32_000 * 4.0, ctx.memory.capacity)
+                ctx.close()
+            }
+
+            override fun onStop(ctx: SimMachineContext) {}
+        })
     }
 
     @Test
@@ -176,18 +160,16 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(object : SimWorkload {
-                override fun onStart(ctx: SimMachineContext) {
-                    val lifecycle = SimWorkloadLifecycle(ctx)
-                    ctx.memory.startConsumer(lifecycle.waitFor(FixedFlowSource(ctx.memory.capacity, utilization = 0.8)))
-                }
-            })
-
-            assertEquals(1250, clock.millis())
-        } finally {
-            machine.close()
-        }
+        machine.runWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                val lifecycle = SimWorkloadLifecycle(ctx)
+                ctx.memory.startConsumer(lifecycle.waitFor(FixedFlowSource(ctx.memory.capacity, utilization = 0.8)))
+            }
+
+            override fun onStop(ctx: SimMachineContext) {}
+        })
+
+        assertEquals(1250, clock.millis())
     }
 
     @Test
@@ -202,19 +184,17 @@ class SimMachineTest {
         val adapter = (machine.peripherals[0] as SimNetworkAdapter)
         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(FixedFlowSource(iface.bandwidth, utilization = 0.8)))
-                }
-            })
-
-            assertEquals(1250, clock.millis())
-        } finally {
-            machine.close()
-        }
+        machine.runWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                val lifecycle = SimWorkloadLifecycle(ctx)
+                val iface = ctx.net[0]
+                iface.tx.startConsumer(lifecycle.waitFor(FixedFlowSource(iface.bandwidth, utilization = 0.8)))
+            }
+
+            override fun onStop(ctx: SimMachineContext) {}
+        })
+
+        assertEquals(1250, clock.millis())
     }
 
     @Test
@@ -226,19 +206,17 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(object : SimWorkload {
-                override fun onStart(ctx: SimMachineContext) {
-                    val lifecycle = SimWorkloadLifecycle(ctx)
-                    val disk = ctx.storage[0]
-                    disk.read.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.read.capacity, utilization = 0.8)))
-                }
-            })
-
-            assertEquals(1250, clock.millis())
-        } finally {
-            machine.close()
-        }
+        machine.runWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                val lifecycle = SimWorkloadLifecycle(ctx)
+                val disk = ctx.storage[0]
+                disk.read.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.read.capacity, utilization = 0.8)))
+            }
+
+            override fun onStop(ctx: SimMachineContext) {}
+        })
+
+        assertEquals(1250, clock.millis())
     }
 
     @Test
@@ -250,19 +228,17 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            machine.run(object : SimWorkload {
-                override fun onStart(ctx: SimMachineContext) {
-                    val lifecycle = SimWorkloadLifecycle(ctx)
-                    val disk = ctx.storage[0]
-                    disk.write.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.write.capacity, utilization = 0.8)))
-                }
-            })
-
-            assertEquals(1250, clock.millis())
-        } finally {
-            machine.close()
-        }
+        machine.runWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                val lifecycle = SimWorkloadLifecycle(ctx)
+                val disk = ctx.storage[0]
+                disk.write.startConsumer(lifecycle.waitFor(FixedFlowSource(disk.write.capacity, utilization = 0.8)))
+            }
+
+            override fun onStop(ctx: SimMachineContext) {}
+        })
+
+        assertEquals(1250, clock.millis())
     }
 
     @Test
@@ -275,13 +251,11 @@ class SimMachineTest {
 
         try {
             coroutineScope {
-                launch { machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) }
+                launch { machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0)) }
                 cancel()
             }
         } catch (_: CancellationException) {
             // Ignore
-        } finally {
-            machine.close()
         }
 
         assertEquals(0, clock.millis())
@@ -295,31 +269,14 @@ class SimMachineTest {
             SimplePowerDriver(ConstantPowerModel(0.0))
         )
 
-        try {
-            coroutineScope {
-                launch {
-                    machine.run(SimFlopsWorkload(2_000, utilization = 1.0))
-                }
+        coroutineScope {
+            launch {
+                machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0))
+            }
 
-                assertThrows<IllegalStateException> {
-                    machine.run(SimFlopsWorkload(2_000, utilization = 1.0))
-                }
+            assertThrows<IllegalStateException> {
+                machine.runWorkload(SimFlopsWorkload(2_000, utilization = 1.0))
             }
-        } finally {
-            machine.close()
         }
     }
-
-    @Test
-    fun testClose() = runBlockingSimulation {
-        val machine = SimBareMetalMachine(
-            FlowEngine(coroutineContext, clock),
-            machineModel,
-            SimplePowerDriver(ConstantPowerModel(0.0))
-        )
-
-        machine.close()
-        assertDoesNotThrow { machine.close() }
-        assertThrows<IllegalStateException> { machine.run(SimFlopsWorkload(2_000, utilization = 1.0)) }
-    }
 }
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
index b7f5bf8e..91855e8d 100644
--- 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
@@ -37,6 +37,7 @@ 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.runWorkload
 import org.opendc.simulator.compute.workload.SimTrace
 import org.opendc.simulator.compute.workload.SimTraceFragment
 import org.opendc.simulator.compute.workload.SimTraceWorkload
@@ -80,16 +81,16 @@ internal class SimFairShareHypervisorTest {
         val hypervisor = SimFairShareHypervisor(platform, null, PerformanceScalingGovernor(), null)
 
         launch {
-            machine.run(hypervisor)
+            machine.runWorkload(hypervisor)
             println("Hypervisor finished")
         }
         yield()
 
-        val vm = hypervisor.createMachine(model)
-        vm.run(workloadA)
+        val vm = hypervisor.newMachine(model)
+        vm.runWorkload(workloadA)
 
         yield()
-        machine.close()
+        machine.cancel()
 
         assertAll(
             { assertEquals(319781, hypervisor.counters.cpuActiveTime, "Active time does not match") },
@@ -131,22 +132,22 @@ internal class SimFairShareHypervisorTest {
         val hypervisor = SimFairShareHypervisor(platform, null, null, null)
 
         launch {
-            machine.run(hypervisor)
+            machine.runWorkload(hypervisor)
         }
 
         yield()
         coroutineScope {
             launch {
-                val vm = hypervisor.createMachine(model)
-                vm.run(workloadA)
-                vm.close()
+                val vm = hypervisor.newMachine(model)
+                vm.runWorkload(workloadA)
+                hypervisor.removeMachine(vm)
             }
-            val vm = hypervisor.createMachine(model)
-            vm.run(workloadB)
-            vm.close()
+            val vm = hypervisor.newMachine(model)
+            vm.runWorkload(workloadB)
+            hypervisor.removeMachine(vm)
         }
         yield()
-        machine.close()
+        machine.cancel()
         yield()
 
         assertAll(
@@ -171,11 +172,11 @@ internal class SimFairShareHypervisorTest {
 
         assertDoesNotThrow {
             launch {
-                machine.run(hypervisor)
+                machine.runWorkload(hypervisor)
             }
         }
 
-        machine.close()
+        machine.cancel()
     }
 
     @Test
@@ -219,20 +220,20 @@ internal class SimFairShareHypervisorTest {
             )
 
         launch {
-            machine.run(hypervisor)
+            machine.runWorkload(hypervisor)
         }
 
         coroutineScope {
             launch {
-                val vm = hypervisor.createMachine(model, "a")
-                vm.run(workloadA)
-                vm.close()
+                val vm = hypervisor.newMachine(model, "a")
+                vm.runWorkload(workloadA)
+                hypervisor.removeMachine(vm)
             }
-            val vm = hypervisor.createMachine(model, "b")
-            vm.run(workloadB)
-            vm.close()
+            val vm = hypervisor.newMachine(model, "b")
+            vm.runWorkload(workloadB)
+            hypervisor.removeMachine(vm)
         }
 
-        machine.close()
+        machine.cancel()
     }
 }
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 02d308ff..823a0ae3 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
@@ -36,6 +36,7 @@ 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.runWorkload
 import org.opendc.simulator.compute.workload.*
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
@@ -76,13 +77,13 @@ internal class SimSpaceSharedHypervisorTest {
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
         val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
-        launch { machine.run(hypervisor) }
-        val vm = hypervisor.createMachine(machineModel)
-        vm.run(workloadA)
+        launch { machine.runWorkload(hypervisor) }
+        val vm = hypervisor.newMachine(machineModel)
+        vm.runWorkload(workloadA)
         yield()
 
-        vm.close()
-        machine.close()
+        hypervisor.removeMachine(vm)
+        machine.cancel()
 
         assertEquals(5 * 60L * 4000, clock.millis()) { "Took enough time" }
     }
@@ -98,12 +99,13 @@ internal class SimSpaceSharedHypervisorTest {
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
         val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
-        launch { machine.run(hypervisor) }
+        launch { machine.runWorkload(hypervisor) }
         yield()
-        val vm = hypervisor.createMachine(machineModel)
-        vm.run(workload)
-        vm.close()
-        machine.close()
+        val vm = hypervisor.newMachine(machineModel)
+        vm.runWorkload(workload)
+        hypervisor.removeMachine(vm)
+
+        machine.cancel()
 
         assertEquals(duration, clock.millis()) { "Took enough time" }
     }
@@ -121,11 +123,11 @@ internal class SimSpaceSharedHypervisorTest {
         )
         val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
-        launch { machine.run(hypervisor) }
+        launch { machine.runWorkload(hypervisor) }
         yield()
-        val vm = hypervisor.createMachine(machineModel)
-        vm.run(workload)
-        machine.close()
+        val vm = hypervisor.newMachine(machineModel)
+        vm.runWorkload(workload)
+        machine.cancel()
 
         assertEquals(duration, clock.millis()) { "Took enough time" }
     }
@@ -142,19 +144,20 @@ internal class SimSpaceSharedHypervisorTest {
         )
         val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
-        launch { machine.run(hypervisor) }
+        launch { machine.runWorkload(hypervisor) }
         yield()
 
-        val vm = hypervisor.createMachine(machineModel)
-        vm.run(SimRuntimeWorkload(duration))
-        vm.close()
+        val vm = hypervisor.newMachine(machineModel)
+        vm.runWorkload(SimRuntimeWorkload(duration))
+        hypervisor.removeMachine(vm)
 
         yield()
 
-        val vm2 = hypervisor.createMachine(machineModel)
-        vm2.run(SimRuntimeWorkload(duration))
-        vm2.close()
-        machine.close()
+        val vm2 = hypervisor.newMachine(machineModel)
+        vm2.runWorkload(SimRuntimeWorkload(duration))
+        hypervisor.removeMachine(vm2)
+
+        machine.cancel()
 
         assertEquals(duration * 2, clock.millis()) { "Took enough time" }
     }
@@ -168,17 +171,17 @@ internal class SimSpaceSharedHypervisorTest {
         val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)))
         val hypervisor = SimSpaceSharedHypervisor(engine, null, null)
 
-        launch { machine.run(hypervisor) }
+        launch { machine.runWorkload(hypervisor) }
         yield()
 
-        hypervisor.createMachine(machineModel)
+        hypervisor.newMachine(machineModel)
 
         assertAll(
             { assertFalse(hypervisor.canFit(machineModel)) },
-            { assertThrows<IllegalArgumentException> { hypervisor.createMachine(machineModel) } }
+            { assertThrows<IllegalArgumentException> { hypervisor.newMachine(machineModel) } }
         )
 
-        machine.close()
+        machine.cancel()
     }
 
     /**
@@ -192,16 +195,16 @@ internal class SimSpaceSharedHypervisorTest {
         )
         val hypervisor = SimSpaceSharedHypervisor(interpreter, null, null)
 
-        launch { machine.run(hypervisor) }
+        launch { machine.runWorkload(hypervisor) }
         yield()
 
-        hypervisor.createMachine(machineModel).close()
+        hypervisor.removeMachine(hypervisor.newMachine(machineModel))
 
         assertAll(
             { assertTrue(hypervisor.canFit(machineModel)) },
-            { assertDoesNotThrow { hypervisor.createMachine(machineModel) } }
+            { assertDoesNotThrow { hypervisor.newMachine(machineModel) } }
         )
 
-        machine.close()
+        machine.cancel()
     }
 }
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 574860e8..aa91984a 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
@@ -30,6 +30,7 @@ import org.opendc.simulator.compute.SimBareMetalMachine
 import org.opendc.simulator.compute.model.*
 import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.SimplePowerDriver
+import org.opendc.simulator.compute.runWorkload
 import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 
@@ -67,13 +68,9 @@ class SimTraceWorkloadTest {
             offset = 0
         )
 
-        try {
-            machine.run(workload)
+        machine.runWorkload(workload)
 
-            assertEquals(4000, clock.millis())
-        } finally {
-            machine.close()
-        }
+        assertEquals(4000, clock.millis())
     }
 
     @Test
@@ -94,13 +91,9 @@ class SimTraceWorkloadTest {
             offset = 1000
         )
 
-        try {
-            machine.run(workload)
+        machine.runWorkload(workload)
 
-            assertEquals(5000, clock.millis())
-        } finally {
-            machine.close()
-        }
+        assertEquals(5000, clock.millis())
     }
 
     @Test
@@ -121,14 +114,10 @@ class SimTraceWorkloadTest {
             offset = 0
         )
 
-        try {
-            delay(1000L)
-            machine.run(workload)
+        delay(1000L)
+        machine.runWorkload(workload)
 
-            assertEquals(4000, clock.millis())
-        } finally {
-            machine.close()
-        }
+        assertEquals(4000, clock.millis())
     }
 
     @Test
@@ -149,12 +138,8 @@ class SimTraceWorkloadTest {
             offset = 0
         )
 
-        try {
-            machine.run(workload)
+        machine.runWorkload(workload)
 
-            assertEquals(4000, clock.millis())
-        } finally {
-            machine.close()
-        }
+        assertEquals(4000, clock.millis())
     }
 }