perf(simulator): Skip fair-share algorithm if capacity remaining

This change updates the MaxMinFlowMultiplexer implementation to skip the
fair-share algorithm in case the total demand is lower than the
available capacity. In this case, no re-division of capacity is
necessary.

2 files changed, 40 insertions, 22 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 337d68bf..e34c5bdc 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
@@ -213,7 +213,7 @@ class CapelinIntegrationTest {
             { assertEquals(6013515, this@CapelinIntegrationTest.exporter.idleTime) { "Idle time incorrect" } },
             { assertEquals(14724500, this@CapelinIntegrationTest.exporter.activeTime) { "Active time incorrect" } },
             { assertEquals(12530742, this@CapelinIntegrationTest.exporter.stealTime) { "Steal time incorrect" } },
-            { assertEquals(481270, this@CapelinIntegrationTest.exporter.lostTime) { "Lost time incorrect" } }
+            { assertEquals(481251, this@CapelinIntegrationTest.exporter.lostTime) { "Lost time incorrect" } }
         )
     }
 
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt
index 97059e93..9131eb54 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt
@@ -372,6 +372,8 @@ public class MaxMinFlowMultiplexer(
 
             val capacity = capacity
             var availableCapacity = capacity
+            var deadline = Long.MAX_VALUE
+            var demand = 0.0
 
             // Pull in the work of the outputs
             val inputIterator = activeInputs.listIterator()
@@ -382,32 +384,36 @@ public class MaxMinFlowMultiplexer(
                 if (!input.isActive) {
                     input.actualRate = 0.0
                     inputIterator.remove()
+                } else {
+                    demand += input.limit
+                    deadline = min(deadline, input.deadline)
                 }
             }
 
-            var demand = 0.0
-            var deadline = Long.MAX_VALUE
+            val rate = if (demand > capacity) {
+                // If the demand is higher than the capacity, we need use max-min fair sharing to distribute the
+                // constrained capacity across the inputs.
 
-            // Sort in-place the inputs based on their pushed flow.
-            // Profiling shows that it is faster than maintaining some kind of sorted set.
-            activeInputs.sort()
+                // Sort in-place the inputs based on their pushed flow.
+                // Profiling shows that it is faster than maintaining some kind of sorted set.
+                activeInputs.sort()
 
-            // Divide the available output capacity fairly over the inputs using max-min fair sharing
-            val size = activeInputs.size
-            for (i in activeInputs.indices) {
-                val input = activeInputs[i]
-                val availableShare = availableCapacity / (size - i)
-                val grantedRate = min(input.allowedRate, availableShare)
+                // Divide the available output capacity fairly over the inputs using max-min fair sharing
+                val size = activeInputs.size
+                for (i in activeInputs.indices) {
+                    val input = activeInputs[i]
+                    val availableShare = availableCapacity / (size - i)
+                    val grantedRate = min(input.allowedRate, availableShare)
 
-                demand += input.limit
-                deadline = min(deadline, input.deadline)
-                availableCapacity -= grantedRate
+                    availableCapacity -= grantedRate
+                    input.actualRate = grantedRate
+                }
 
-                input.actualRate = grantedRate
+                capacity - availableCapacity
+            } else {
+                demand
             }
 
-            val rate = capacity - availableCapacity
-
             this.demand = demand
             this.rate = rate
 
@@ -468,16 +474,25 @@ public class MaxMinFlowMultiplexer(
         @JvmField var actualRate: Double = 0.0
 
         /**
+         * The processing rate that is allowed by the model constraints.
+         */
+        @JvmField var allowedRate: Double = 0.0
+
+        /**
          * The deadline of the input.
          */
         val deadline: Long
             get() = ctx?.deadline ?: Long.MAX_VALUE
 
         /**
-         * The processing rate that is allowed by the model constraints.
+         * The capacity of the input.
          */
-        val allowedRate: Double
-            get() = min(capacity, limit)
+        override var capacity: Double
+            get() = super.capacity
+            set(value) {
+                allowedRate = min(limit, value)
+                super.capacity = value
+            }
 
         /**
          * A flag to enable timers for the input.
@@ -530,8 +545,10 @@ public class MaxMinFlowMultiplexer(
         ) {
             doUpdateCounters(delta)
 
-            actualRate = 0.0
+            val allowed = min(rate, capacity)
             limit = rate
+            actualRate = allowed
+            allowedRate = allowed
 
             scheduler.trigger(now)
         }
@@ -541,6 +558,7 @@ public class MaxMinFlowMultiplexer(
 
             limit = 0.0
             actualRate = 0.0
+            allowedRate = 0.0
 
             scheduler.deregisterInput(this, now)