merge: Support snapshotting simulated workloads (#113)

This pull request adds support for snapshotting simulated workloads in OpenDC, which
serves as the basis for virtual machine migration/suspension support.

Part of #32 

## Implementation Notes :hammer_and_pick:

* Support synchronous update of FlowStage
* Report exceptions in onStop as suppressed 
* Add support for snapshotting workloads

2 files changed, 201 insertions, 11 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 266839bd..2acf6ec7 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
@@ -22,6 +22,8 @@
 
 package org.opendc.simulator.compute
 
+import io.mockk.every
+import io.mockk.mockk
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.coroutineScope
@@ -175,6 +177,8 @@ class SimMachineTest {
             }
 
             override fun onStop(ctx: SimMachineContext) {}
+
+            override fun snapshot(): SimWorkload = TODO()
         })
     }
 
@@ -195,6 +199,8 @@ class SimMachineTest {
             }
 
             override fun onStop(ctx: SimMachineContext) {}
+
+            override fun snapshot(): SimWorkload = TODO()
         })
     }
 
@@ -215,6 +221,8 @@ class SimMachineTest {
             }
 
             override fun onStop(ctx: SimMachineContext) {}
+
+            override fun snapshot(): SimWorkload = TODO()
         })
 
         assertEquals(1000, clock.millis())
@@ -241,6 +249,8 @@ class SimMachineTest {
             }
 
             override fun onStop(ctx: SimMachineContext) {}
+
+            override fun snapshot(): SimWorkload = TODO()
         })
 
         assertEquals(40, clock.millis())
@@ -264,6 +274,8 @@ class SimMachineTest {
             }
 
             override fun onStop(ctx: SimMachineContext) {}
+
+            override fun snapshot(): SimWorkload = TODO()
         })
 
         assertEquals(4000, clock.millis())
@@ -287,6 +299,8 @@ class SimMachineTest {
             }
 
             override fun onStop(ctx: SimMachineContext) {}
+
+            override fun snapshot(): SimWorkload = TODO()
         })
 
         assertEquals(4000, clock.millis())
@@ -334,4 +348,71 @@ class SimMachineTest {
             }
         }
     }
+
+    @Test
+    fun testCatchStartFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workload = mockk<SimWorkload>()
+        every { workload.onStart(any()) } throws IllegalStateException()
+
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+    }
+
+    @Test
+    fun testCatchStopFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workload = mockk<SimWorkload>()
+        every { workload.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown() }
+        every { workload.onStop(any()) } throws IllegalStateException()
+
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+    }
+
+    @Test
+    fun testCatchShutdownFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workload = mockk<SimWorkload>()
+        every { workload.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) }
+
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+    }
+
+    @Test
+    fun testCatchNestedFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workload = mockk<SimWorkload>()
+        every { workload.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) }
+        every { workload.onStop(any()) } throws IllegalStateException()
+
+        val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+        assertEquals(1, exc.cause!!.suppressedExceptions.size)
+    }
 }
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt
index 6bf05f65..d0b0efaa 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimChainWorkloadTest.kt
@@ -25,10 +25,14 @@ package org.opendc.simulator.compute.workload
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.spyk
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
 import org.opendc.simulator.compute.SimBareMetalMachine
+import org.opendc.simulator.compute.SimMachineContext
 import org.opendc.simulator.compute.model.MachineModel
 import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.model.ProcessingNode
@@ -65,8 +69,8 @@ class SimChainWorkloadTest {
 
         val workload =
             SimWorkloads.chain(
-                SimRuntimeWorkload(1000, 1.0),
-                SimRuntimeWorkload(1000, 1.0)
+                SimWorkloads.runtime(1000, 1.0),
+                SimWorkloads.runtime(1000, 1.0)
             )
 
         machine.runWorkload(workload)
@@ -91,10 +95,10 @@ class SimChainWorkloadTest {
         val workload =
             SimWorkloads.chain(
                 workloadA,
-                SimRuntimeWorkload(1000, 1.0)
+                SimWorkloads.runtime(1000, 1.0)
             )
 
-        machine.runWorkload(workload)
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
 
         assertEquals(0, clock.millis())
     }
@@ -115,12 +119,12 @@ class SimChainWorkloadTest {
 
         val workload =
             SimWorkloads.chain(
-                SimRuntimeWorkload(1000, 1.0),
+                SimWorkloads.runtime(1000, 1.0),
                 workloadA,
-                SimRuntimeWorkload(1000, 1.0)
+                SimWorkloads.runtime(1000, 1.0)
             )
 
-        machine.runWorkload(workload)
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
 
         assertEquals(1000, clock.millis())
     }
@@ -141,10 +145,10 @@ class SimChainWorkloadTest {
         val workload =
             SimWorkloads.chain(
                 workloadA,
-                SimRuntimeWorkload(1000, 1.0)
+                SimWorkloads.runtime(1000, 1.0)
             )
 
-        machine.runWorkload(workload)
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
 
         assertEquals(1000, clock.millis())
     }
@@ -164,13 +168,118 @@ class SimChainWorkloadTest {
 
         val workload =
             SimWorkloads.chain(
+                SimWorkloads.runtime(1000, 1.0),
+                workloadA,
+                SimWorkloads.runtime(1000, 1.0)
+            )
+
+        assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+
+        assertEquals(2000, clock.millis())
+    }
+
+    @Test
+    fun testStartAndStopFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workloadA = mockk<SimWorkload>()
+        every { workloadA.onStart(any()) } throws IllegalStateException()
+        every { workloadA.onStop(any()) } throws IllegalStateException()
+
+        val workload =
+            SimWorkloads.chain(
+                SimRuntimeWorkload(1000, 1.0),
+                workloadA
+            )
+
+        val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+
+        assertEquals(2, exc.cause!!.suppressedExceptions.size)
+        assertEquals(1000, clock.millis())
+    }
+
+    @Test
+    fun testShutdownAndStopFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workloadA = mockk<SimWorkload>()
+        every { workloadA.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) }
+        every { workloadA.onStop(any()) } throws IllegalStateException()
+
+        val workload =
+            SimWorkloads.chain(
+                SimRuntimeWorkload(1000, 1.0),
+                workloadA
+            )
+
+        val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+
+        assertEquals(1, exc.cause!!.suppressedExceptions.size)
+        assertEquals(1000, clock.millis())
+    }
+
+    @Test
+    fun testShutdownAndStartFailure() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(
+            graph,
+            machineModel
+        )
+
+        val workloadA = mockk<SimWorkload>(relaxUnitFun = true)
+        every { workloadA.onStart(any()) } answers { (it.invocation.args[0] as SimMachineContext).shutdown(IllegalStateException()) }
+
+        val workloadB = mockk<SimWorkload>(relaxUnitFun = true)
+        every { workloadB.onStart(any()) } throws IllegalStateException()
+
+        val workload =
+            SimWorkloads.chain(
                 SimRuntimeWorkload(1000, 1.0),
                 workloadA,
-                SimRuntimeWorkload(1000, 1.0)
+                workloadB
             )
 
-        machine.runWorkload(workload)
+        val exc = assertThrows<IllegalStateException> { machine.runWorkload(workload) }
+        assertEquals(1, exc.cause!!.suppressedExceptions.size)
+        assertEquals(1000, clock.millis())
+    }
+
+    @Test
+    fun testSnapshot() = runSimulation {
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+
+        val machine = SimBareMetalMachine.create(graph, machineModel)
+        val workload =
+            SimWorkloads.chain(
+                SimWorkloads.runtime(1000, 1.0),
+                SimWorkloads.runtime(1000, 1.0)
+            )
+
+        val job = launch { machine.runWorkload(workload) }
+        delay(500L)
+        val snapshot = workload.snapshot()
+
+        job.join()
 
         assertEquals(2000, clock.millis())
+
+        machine.runWorkload(snapshot)
+
+        assertEquals(3500, clock.millis())
     }
 }