fix(simulator): Support unaligned trace fragments

3 files changed, 53 insertions, 16 deletions

diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
index 3dca1e09..393fb88d 100644
--- a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
+++ b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt
@@ -114,9 +114,9 @@ class CapelinIntegrationTest {
             { assertEquals(0, monitorResults.runningVms, "All VMs should finish after a run") },
             { assertEquals(0, monitorResults.unscheduledVms, "No VM should not be unscheduled") },
             { assertEquals(0, monitorResults.queuedVms, "No VM should not be in the queue") },
-            { assertEquals(155252275350, monitor.totalRequestedBurst) { "Incorrect requested burst" } },
-            { assertEquals(155086837649, monitor.totalGrantedBurst) { "Incorrect granted burst" } },
-            { assertEquals(155088144, monitor.totalOvercommissionedBurst) { "Incorrect overcommitted burst" } },
+            { assertEquals(155252275351, monitor.totalRequestedBurst) { "Incorrect requested burst" } },
+            { assertEquals(155086837645, monitor.totalGrantedBurst) { "Incorrect granted burst" } },
+            { assertEquals(725049, monitor.totalOvercommissionedBurst) { "Incorrect overcommitted burst" } },
             { assertEquals(0, monitor.totalInterferedBurst) { "Incorrect interfered burst" } }
         )
     }
@@ -153,7 +153,7 @@ class CapelinIntegrationTest {
         assertAll(
             { assertEquals(29454904468, monitor.totalRequestedBurst) { "Total requested work incorrect" } },
             { assertEquals(29355293349, monitor.totalGrantedBurst) { "Total granted work incorrect" } },
-            { assertEquals(99627123, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
+            { assertEquals(0, monitor.totalOvercommissionedBurst) { "Total overcommitted work incorrect" } },
             { assertEquals(0, monitor.totalInterferedBurst) { "Total interfered work incorrect" } }
         )
     }
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
index 6b420911..a985986d 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt
@@ -22,6 +22,7 @@
 
 package org.opendc.simulator.resources
 
+import org.opendc.simulator.resources.impl.SimResourceCountersImpl
 import org.opendc.simulator.resources.interference.InterferenceDomain
 import org.opendc.simulator.resources.interference.InterferenceKey
 import kotlin.math.min
@@ -53,9 +54,9 @@ public class SimResourceDistributorMaxMin(
     private val activeOutputs: MutableList<Output> = mutableListOf()
 
     /**
-     * The total amount of work requested by the output resources.
+     * The total amount of work allocated to be executed.
      */
-    private var totalRequestedWork = 0.0
+    private var totalAllocatedWork = 0.0
 
     /**
      * The total allocated speed for the output resources.
@@ -67,6 +68,13 @@ public class SimResourceDistributorMaxMin(
      */
     private var totalRequestedSpeed = 0.0
 
+    /**
+     * The resource counters of this distributor.
+     */
+    public val counters: SimResourceCounters
+        get() = _counters
+    private val _counters = SimResourceCountersImpl()
+
     /* SimResourceDistributor */
     override fun newOutput(key: InterferenceKey?): SimResourceCloseableProvider {
         val provider = Output(ctx?.capacity ?: 0.0, key)
@@ -103,6 +111,25 @@ public class SimResourceDistributorMaxMin(
     }
 
     /**
+     * Update the counters of the distributor.
+     */
+    private fun updateCounters(ctx: SimResourceControllableContext, work: Double, willOvercommit: Boolean) {
+        if (work <= 0.0) {
+            return
+        }
+
+        val counters = _counters
+        val remainingWork = ctx.remainingWork
+
+        counters.demand += work
+        counters.actual += work - remainingWork
+
+        if (willOvercommit && remainingWork > 0.0) {
+            counters.overcommit += remainingWork
+        }
+    }
+
+    /**
      * Schedule the work of the outputs.
      */
     private fun doNext(ctx: SimResourceContext): SimResourceCommand {
@@ -116,7 +143,6 @@ public class SimResourceDistributorMaxMin(
         var deadline: Long = Long.MAX_VALUE
         var availableSpeed = capacity
         var totalRequestedSpeed = 0.0
-        var totalRequestedWork = 0.0
 
         // Pull in the work of the outputs
         val outputIterator = activeOutputs.listIterator()
@@ -138,6 +164,7 @@ public class SimResourceDistributorMaxMin(
         for (output in activeOutputs) {
             val availableShare = availableSpeed / remaining--
             val grantedSpeed = min(output.allowedSpeed, availableShare)
+
             deadline = min(deadline, output.deadline)
 
             // Ignore idle computation
@@ -147,7 +174,6 @@ public class SimResourceDistributorMaxMin(
             }
 
             totalRequestedSpeed += output.limit
-            totalRequestedWork += output.work
 
             output.actualSpeed = grantedSpeed
             availableSpeed -= grantedSpeed
@@ -156,19 +182,28 @@ public class SimResourceDistributorMaxMin(
             duration = min(duration, output.work / grantedSpeed)
         }
 
+        val targetDuration = min(duration, (deadline - interpreter.clock.millis()) / 1000.0)
+        var totalRequestedWork = 0.0
+        var totalAllocatedWork = 0.0
+        for (output in activeOutputs) {
+            val work = output.work
+            val speed = output.actualSpeed
+            if (speed > 0.0) {
+                val outputDuration = work / speed
+                totalRequestedWork += work * (duration / outputDuration)
+                totalAllocatedWork += work * (targetDuration / outputDuration)
+            }
+        }
+
         assert(deadline >= interpreter.clock.millis()) { "Deadline already passed" }
 
-        this.totalRequestedWork = totalRequestedWork
         this.totalRequestedSpeed = totalRequestedSpeed
+        this.totalAllocatedWork = totalAllocatedWork
         val totalAllocatedSpeed = capacity - availableSpeed
         this.totalAllocatedSpeed = totalAllocatedSpeed
-        val totalAllocatedWork = min(
-            totalRequestedWork,
-            totalAllocatedSpeed * min((deadline - interpreter.clock.millis()) / 1000.0, duration)
-        )
 
         return if (totalAllocatedWork > 0.0 && totalAllocatedSpeed > 0.0)
-            SimResourceCommand.Consume(totalRequestedWork, totalAllocatedSpeed, deadline)
+            SimResourceCommand.Consume(totalAllocatedWork, totalAllocatedSpeed, deadline)
         else
             SimResourceCommand.Idle(deadline)
     }
@@ -265,6 +300,8 @@ public class SimResourceDistributorMaxMin(
 
         override fun onUpdate(ctx: SimResourceControllableContext, work: Double, willOvercommit: Boolean) {
             updateCounters(ctx, work, willOvercommit)
+
+            this@SimResourceDistributorMaxMin.updateCounters(ctx, work, willOvercommit)
         }
 
         override fun onFinish(ctx: SimResourceControllableContext) {
@@ -289,7 +326,7 @@ public class SimResourceDistributorMaxMin(
             }
 
             // Compute the work that was actually granted to the output.
-            return (totalRequestedWork - totalRemainingWork) * fraction * perfScore
+            return (totalAllocatedWork - totalRemainingWork) * fraction * perfScore
         }
 
         /* Comparable */
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
index ceb5a1a4..d988b70d 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt
@@ -54,7 +54,7 @@ public class SimResourceSwitchMaxMin(
      * The resource counters to track the execution metrics of all switch resources.
      */
     override val counters: SimResourceCounters
-        get() = aggregator.counters
+        get() = distributor.counters
 
     /**
      * A flag to indicate that the switch was closed.