merge: Execute trace fragments based on timestamps

This pull request updates the simulator to execute workload traces based on the fragment's timestamps. 
This means that traces will execute their timestamps at the correct time.

* Execute traces based on timestamps
* Record overcommit only after deadline exceeded
* Support trace fragments with zero cores available
* Support unaligned trace fragments

**Breaking API Changes**

* `SimTraceWorkload.Fragment` now requires a `timestamp` parameter.

7 files changed, 67 insertions, 24 deletions

diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
index 8a24b3e7..00648876 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt
@@ -112,8 +112,8 @@ public abstract class SimAbstractResourceAggregator(
                     doFinish()
                 }
 
-                override fun onUpdate(ctx: SimResourceControllableContext, work: Double) {
-                    updateCounters(ctx, work)
+                override fun onUpdate(ctx: SimResourceControllableContext, work: Double, willOvercommit: Boolean) {
+                    updateCounters(ctx, work, willOvercommit)
                 }
 
                 override fun getConsumedWork(ctx: SimResourceControllableContext, work: Double, speed: Double, duration: Long): Double {
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt
index 860c50ee..4e8e803a 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceProvider.kt
@@ -87,7 +87,7 @@ public abstract class SimAbstractResourceProvider(
     /**
      * Update the counters of the resource provider.
      */
-    protected fun updateCounters(ctx: SimResourceContext, work: Double) {
+    protected fun updateCounters(ctx: SimResourceContext, work: Double, willOvercommit: Boolean) {
         if (work <= 0.0) {
             return
         }
@@ -96,7 +96,10 @@ public abstract class SimAbstractResourceProvider(
         val remainingWork = ctx.remainingWork
         counters.demand += work
         counters.actual += work - remainingWork
-        counters.overcommit += remainingWork
+
+        if (willOvercommit && remainingWork > 0.0) {
+            counters.overcommit += remainingWork
+        }
     }
 
     final override fun startConsumer(consumer: SimResourceConsumer) {
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 398797cf..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,18 +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.totalAllocatedSpeed = capacity - availableSpeed
-        val totalAllocatedWork = min(
-            totalRequestedWork,
-            totalAllocatedSpeed * min((deadline - interpreter.clock.millis()) / 1000.0, duration)
-        )
+        this.totalAllocatedWork = totalAllocatedWork
+        val totalAllocatedSpeed = capacity - availableSpeed
+        this.totalAllocatedSpeed = totalAllocatedSpeed
 
         return if (totalAllocatedWork > 0.0 && totalAllocatedSpeed > 0.0)
-            SimResourceCommand.Consume(totalRequestedWork, totalAllocatedSpeed, deadline)
+            SimResourceCommand.Consume(totalAllocatedWork, totalAllocatedSpeed, deadline)
         else
             SimResourceCommand.Idle(deadline)
     }
@@ -262,8 +298,10 @@ public class SimResourceDistributorMaxMin(
             return Long.MAX_VALUE
         }
 
-        override fun onUpdate(ctx: SimResourceControllableContext, work: Double) {
-            updateCounters(ctx, work)
+        override fun onUpdate(ctx: SimResourceControllableContext, work: Double, willOvercommit: Boolean) {
+            updateCounters(ctx, work, willOvercommit)
+
+            this@SimResourceDistributorMaxMin.updateCounters(ctx, work, willOvercommit)
         }
 
         override fun onFinish(ctx: SimResourceControllableContext) {
@@ -288,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/SimResourceProviderLogic.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProviderLogic.kt
index 17045557..2fe1b00f 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProviderLogic.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProviderLogic.kt
@@ -52,8 +52,9 @@ public interface SimResourceProviderLogic {
      *
      * @param ctx The context in which the provider runs.
      * @param work The amount of work that was requested by the resource consumer.
+     * @param willOvercommit A flag to indicate that the remaining work is overcommitted.
      */
-    public fun onUpdate(ctx: SimResourceControllableContext, work: Double) {}
+    public fun onUpdate(ctx: SimResourceControllableContext, work: Double, willOvercommit: Boolean) {}
 
     /**
      * This method is invoked when the resource consumer has finished.
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt
index 2f70e3cc..2d53198a 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt
@@ -51,8 +51,8 @@ public class SimResourceSource(
                 }
             }
 
-            override fun onUpdate(ctx: SimResourceControllableContext, work: Double) {
-                updateCounters(ctx, work)
+            override fun onUpdate(ctx: SimResourceControllableContext, work: Double, willOvercommit: Boolean) {
+                updateCounters(ctx, work, willOvercommit)
             }
 
             override fun onFinish(ctx: SimResourceControllableContext) {
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.
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/impl/SimResourceContextImpl.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/impl/SimResourceContextImpl.kt
index 98fad068..b79998a3 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/impl/SimResourceContextImpl.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/impl/SimResourceContextImpl.kt
@@ -202,17 +202,18 @@ internal class SimResourceContextImpl(
                 val isInterrupted = _flag and FLAG_INTERRUPT != 0
                 val remainingWork = getRemainingWork(timestamp)
                 val isConsume = _limit > 0.0
+                val reachedDeadline = _deadline <= timestamp
 
                 // Update the resource counters only if there is some progress
                 if (timestamp > _timestamp) {
-                    logic.onUpdate(this, _work)
+                    logic.onUpdate(this, _work, reachedDeadline)
                 }
 
                 // We should only continue processing the next command if:
                 // 1. The resource consumption was finished.
                 // 2. The resource capacity cannot satisfy the demand.
                 // 3. The resource consumer should be interrupted (e.g., someone called .interrupt())
-                if ((isConsume && remainingWork == 0.0) || _deadline <= timestamp || isInterrupted) {
+                if ((isConsume && remainingWork == 0.0) || reachedDeadline || isInterrupted) {
                     when (val command = consumer.onNext(this)) {
                         is SimResourceCommand.Idle -> interpretIdle(timestamp, command.deadline)
                         is SimResourceCommand.Consume -> interpretConsume(timestamp, command.work, command.limit, command.deadline)