refactor(simulator): Execute traces based on timestamps

This change refactors the trace workload in the OpenDC simulator to
track execute a fragment based on the fragment's timestamp. This makes
sure that the trace is replayed identically to the original execution.

7 files changed, 208 insertions, 67 deletions

diff --git a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
index 8f60bf05..30797089 100644
--- a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt
@@ -70,14 +70,14 @@ class SimMachineBenchmarks {
         @Setup
         fun setUp() {
             trace = sequenceOf(
-                SimTraceWorkload.Fragment(1000, 28.0, 1),
-                SimTraceWorkload.Fragment(1000, 3500.0, 1),
-                SimTraceWorkload.Fragment(1000, 0.0, 1),
-                SimTraceWorkload.Fragment(1000, 183.0, 1),
-                SimTraceWorkload.Fragment(1000, 400.0, 1),
-                SimTraceWorkload.Fragment(1000, 100.0, 1),
-                SimTraceWorkload.Fragment(1000, 3000.0, 1),
-                SimTraceWorkload.Fragment(1000, 4500.0, 1),
+                SimTraceWorkload.Fragment(0, 1000, 28.0, 1),
+                SimTraceWorkload.Fragment(1000, 1000, 3500.0, 1),
+                SimTraceWorkload.Fragment(2000, 1000, 0.0, 1),
+                SimTraceWorkload.Fragment(3000, 1000, 183.0, 1),
+                SimTraceWorkload.Fragment(4000, 1000, 400.0, 1),
+                SimTraceWorkload.Fragment(5000, 1000, 100.0, 1),
+                SimTraceWorkload.Fragment(6000, 1000, 3000.0, 1),
+                SimTraceWorkload.Fragment(7000, 1000, 4500.0, 1),
             )
         }
     }
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 139c66e0..f416643e 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
@@ -116,7 +116,6 @@ public abstract class SimAbstractMachine(
             // Cancel all cpus on cancellation
             cont.invokeOnCancellation {
                 this.cont = null
-
                 interpreter.batch {
                     for (cpu in cpus) {
                         cpu.cancel()
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 622bcd4d..fc49f357 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
@@ -27,25 +27,19 @@ import org.opendc.simulator.compute.model.ProcessingUnit
 import org.opendc.simulator.resources.SimResourceCommand
 import org.opendc.simulator.resources.SimResourceConsumer
 import org.opendc.simulator.resources.SimResourceContext
-import org.opendc.simulator.resources.consumer.SimConsumerBarrier
 
 /**
  * A [SimWorkload] that replays a workload trace consisting of multiple fragments, each indicating the resource
  * consumption for some period of time.
+ *
+ * @param trace The trace of fragments to use.
+ * @param offset The offset for the timestamps.
  */
-public class SimTraceWorkload(public val trace: Sequence<Fragment>) : SimWorkload {
-    private var offset = Long.MIN_VALUE
+public class SimTraceWorkload(public val trace: Sequence<Fragment>, private val offset: Long = 0L) : SimWorkload {
     private val iterator = trace.iterator()
     private var fragment: Fragment? = null
-    private lateinit var barrier: SimConsumerBarrier
 
     override fun onStart(ctx: SimMachineContext) {
-        check(offset == Long.MIN_VALUE) { "Workload does not support re-use" }
-
-        barrier = SimConsumerBarrier(ctx.cpus.size)
-        fragment = nextFragment()
-        offset = ctx.interpreter.clock.millis()
-
         val lifecycle = SimWorkloadLifecycle(ctx)
 
         for (cpu in ctx.cpus) {
@@ -56,43 +50,59 @@ public class SimTraceWorkload(public val trace: Sequence<Fragment>) : SimWorkloa
     override fun toString(): String = "SimTraceWorkload"
 
     /**
-     * Obtain the next fragment.
+     * Obtain the fragment with a timestamp equal or greater than [now].
      */
-    private fun nextFragment(): Fragment? {
-        return if (iterator.hasNext()) {
-            iterator.next()
-        } else {
-            null
+    private fun pullFragment(now: Long): Fragment? {
+        var fragment = fragment
+        if (fragment != null && !fragment.isExpired(now)) {
+            return fragment
+        }
+
+        while (iterator.hasNext()) {
+            fragment = iterator.next()
+            if (!fragment.isExpired(now)) {
+                this.fragment = fragment
+                return fragment
+            }
         }
+
+        this.fragment = null
+        return null
+    }
+
+    /**
+     * Determine if the specified [Fragment] is expired, i.e., it has already passed.
+     */
+    private fun Fragment.isExpired(now: Long): Boolean {
+        val timestamp = this.timestamp + offset
+        return now >= timestamp + duration
     }
 
     private inner class Consumer(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
+            val fragment = pullFragment(now) ?: return SimResourceCommand.Exit
+            val timestamp = fragment.timestamp + offset
 
-            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
+            // Fragment is in the future
+            if (timestamp > now) {
+                return SimResourceCommand.Idle(timestamp)
             }
 
-            return cmd
+            val usage = fragment.usage / fragment.cores
+            val deadline = timestamp + fragment.duration
+            val duration = deadline - now
+            val work = duration * usage / 1000
+
+            return if (cpu.id < fragment.cores && work > 0.0)
+                SimResourceCommand.Consume(work, usage, deadline)
+            else
+                SimResourceCommand.Idle(deadline)
         }
     }
 
     /**
      * A fragment of the workload.
      */
-    public data class Fragment(val duration: Long, val usage: Double, val cores: Int)
+    public data class Fragment(val timestamp: Long, val duration: Long, val usage: Double, val cores: Int)
 }
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 a6d955ca..19808a77 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
@@ -44,7 +44,6 @@ import org.opendc.simulator.resources.consumer.SimWorkConsumer
 /**
  * Test suite for the [SimBareMetalMachine] class.
  */
-@OptIn(ExperimentalCoroutinesApi::class)
 class SimMachineTest {
     private lateinit var machineModel: MachineModel
 
diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt
index a61cba8d..afc4c949 100644
--- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorTest.kt
@@ -91,10 +91,10 @@ internal class SimHypervisorTest {
         val workloadA =
             SimTraceWorkload(
                 sequenceOf(
-                    SimTraceWorkload.Fragment(duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 183.0, 1)
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
                 ),
             )
 
@@ -154,19 +154,19 @@ internal class SimHypervisorTest {
         val workloadA =
             SimTraceWorkload(
                 sequenceOf(
-                    SimTraceWorkload.Fragment(duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 183.0, 1)
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
                 ),
             )
         val workloadB =
             SimTraceWorkload(
                 sequenceOf(
-                    SimTraceWorkload.Fragment(duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 3100.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 73.0, 1)
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3100.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 73.0, 1)
                 )
             )
 
@@ -251,19 +251,19 @@ internal class SimHypervisorTest {
         val workloadA =
             SimTraceWorkload(
                 sequenceOf(
-                    SimTraceWorkload.Fragment(duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 183.0, 1)
+                    SimTraceWorkload.Fragment(0, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
                 ),
             )
         val workloadB =
             SimTraceWorkload(
                 sequenceOf(
-                    SimTraceWorkload.Fragment(duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 3100.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 73.0, 1)
+                    SimTraceWorkload.Fragment(0, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 3100.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 73.0, 1)
                 )
             )
 
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 7c77b283..80496992 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
@@ -71,10 +71,10 @@ internal class SimSpaceSharedHypervisorTest {
         val workloadA =
             SimTraceWorkload(
                 sequenceOf(
-                    SimTraceWorkload.Fragment(duration * 1000, 28.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 3500.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 0.0, 1),
-                    SimTraceWorkload.Fragment(duration * 1000, 183.0, 1)
+                    SimTraceWorkload.Fragment(0, duration * 1000, 28.0, 1),
+                    SimTraceWorkload.Fragment(duration * 1000, duration * 1000, 3500.0, 1),
+                    SimTraceWorkload.Fragment(duration * 2000, duration * 1000, 0.0, 1),
+                    SimTraceWorkload.Fragment(duration * 3000, duration * 1000, 183.0, 1)
                 ),
             )
 
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
new file mode 100644
index 00000000..39c1eb5a
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkloadTest.kt
@@ -0,0 +1,133 @@
+/*
+ * 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.workload
+
+import kotlinx.coroutines.delay
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+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.core.runBlockingSimulation
+import org.opendc.simulator.resources.SimResourceInterpreter
+
+/**
+ * Test suite for the [SimTraceWorkloadTest] class.
+ */
+class SimTraceWorkloadTest {
+    private lateinit var machineModel: MachineModel
+
+    @BeforeEach
+    fun setUp() {
+        val cpuNode = ProcessingNode("Intel", "Xeon", "amd64", 2)
+
+        machineModel = MachineModel(
+            cpus = List(cpuNode.coreCount) { ProcessingUnit(cpuNode, it, 1000.0) },
+            memory = List(4) { MemoryUnit("Crucial", "MTA18ASF4G72AZ-3G2B1", 3200.0, 32_000) }
+        )
+    }
+
+    @Test
+    fun testSmoke() = runBlockingSimulation {
+        val machine = SimBareMetalMachine(
+            SimResourceInterpreter(coroutineContext, clock),
+            machineModel,
+            SimplePowerDriver(ConstantPowerModel(0.0))
+        )
+
+        val workload = SimTraceWorkload(
+            sequenceOf(
+                SimTraceWorkload.Fragment(0, 1000, 2 * 28.0, 2),
+                SimTraceWorkload.Fragment(1000, 1000, 2 * 3100.0, 2),
+                SimTraceWorkload.Fragment(2000, 1000, 0.0, 2),
+                SimTraceWorkload.Fragment(3000, 1000, 2 * 73.0, 2)
+            ),
+            offset = 0
+        )
+
+        try {
+            machine.run(workload)
+
+            assertEquals(4000, clock.millis())
+        } finally {
+            machine.close()
+        }
+    }
+
+    @Test
+    fun testOffset() = runBlockingSimulation {
+        val machine = SimBareMetalMachine(
+            SimResourceInterpreter(coroutineContext, clock),
+            machineModel,
+            SimplePowerDriver(ConstantPowerModel(0.0))
+        )
+
+        val workload = SimTraceWorkload(
+            sequenceOf(
+                SimTraceWorkload.Fragment(0, 1000, 2 * 28.0, 2),
+                SimTraceWorkload.Fragment(1000, 1000, 2 * 3100.0, 2),
+                SimTraceWorkload.Fragment(2000, 1000, 0.0, 2),
+                SimTraceWorkload.Fragment(3000, 1000, 2 * 73.0, 2)
+            ),
+            offset = 1000
+        )
+
+        try {
+            machine.run(workload)
+
+            assertEquals(5000, clock.millis())
+        } finally {
+            machine.close()
+        }
+    }
+
+    @Test
+    fun testSkipFragment() = runBlockingSimulation {
+        val machine = SimBareMetalMachine(
+            SimResourceInterpreter(coroutineContext, clock),
+            machineModel,
+            SimplePowerDriver(ConstantPowerModel(0.0))
+        )
+
+        val workload = SimTraceWorkload(
+            sequenceOf(
+                SimTraceWorkload.Fragment(0, 1000, 2 * 28.0, 2),
+                SimTraceWorkload.Fragment(1000, 1000, 2 * 3100.0, 2),
+                SimTraceWorkload.Fragment(2000, 1000, 0.0, 2),
+                SimTraceWorkload.Fragment(3000, 1000, 2 * 73.0, 2)
+            ),
+            offset = 0
+        )
+
+        try {
+            delay(1000L)
+            machine.run(workload)
+
+            assertEquals(4000, clock.millis())
+        } finally {
+            machine.close()
+        }
+    }
+}