fix(simulator): Use correct flow input capacity for counters

This change fixes an issue with the FlowMultiplexer implementation where
the capacity of each flow input was equal to the capacity of all flow
outputs. Now, the user can specify the capacity of the input, which will
be used to correctly compute the active and idle time.

3 files changed, 28 insertions, 4 deletions

diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt
index 04ba7f21..a7877546 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt
@@ -62,13 +62,21 @@ public interface FlowMultiplexer {
     public val counters: FlowCounters
 
     /**
-     * Create a new input on this multiplexer.
+     * Create a new input on this multiplexer with a coupled capacity.
      *
      * @param key The key of the interference member to which the input belongs.
      */
     public fun newInput(key: InterferenceKey? = null): FlowConsumer
 
     /**
+     * Create a new input on this multiplexer with the specified [capacity].
+     *
+     * @param capacity The capacity of the input.
+     * @param key The key of the interference member to which the input belongs.
+     */
+    public fun newInput(capacity: Double, key: InterferenceKey? = null): FlowConsumer
+
+    /**
      * Remove [input] from this multiplexer.
      */
     public fun removeInput(input: FlowConsumer)
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt
index 125d10fe..b68a8baa 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt
@@ -78,6 +78,8 @@ public class ForwardingFlowMultiplexer(private val engine: FlowEngine) : FlowMul
         return input
     }
 
+    override fun newInput(capacity: Double, key: InterferenceKey?): FlowConsumer = newInput(key)
+
     override fun removeInput(input: FlowConsumer) {
         if (!_inputs.remove(input)) {
             return
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 a0fb8a4e..3d26efda 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
@@ -86,7 +86,15 @@ public class MaxMinFlowMultiplexer(
     private val scheduler = Scheduler(engine, parent)
 
     override fun newInput(key: InterferenceKey?): FlowConsumer {
-        val provider = Input(engine, scheduler, interferenceDomain, key, scheduler.capacity)
+        return newInput(isCoupled = true, Double.POSITIVE_INFINITY, key)
+    }
+
+    override fun newInput(capacity: Double, key: InterferenceKey?): FlowConsumer {
+        return newInput(isCoupled = false, capacity, key)
+    }
+
+    private fun newInput(isCoupled: Boolean, initialCapacity: Double, key: InterferenceKey?): FlowConsumer {
+        val provider = Input(engine, scheduler, interferenceDomain, key, isCoupled, initialCapacity)
         _inputs.add(provider)
         return provider
     }
@@ -206,7 +214,10 @@ public class MaxMinFlowMultiplexer(
             // Disable timers and convergence of the source if one of the output manages it
             input.shouldConsumerConverge = !hasActivationOutput
             input.enableTimers = !hasActivationOutput
-            input.capacity = capacity
+
+            if (input.isCoupled) {
+                input.capacity = capacity
+            }
 
             trigger(_clock.millis())
         }
@@ -340,7 +351,9 @@ public class MaxMinFlowMultiplexer(
             capacity = newCapacity
 
             for (input in _activeInputs) {
-                input.capacity = newCapacity
+                if (input.isCoupled) {
+                    input.capacity = newCapacity
+                }
             }
 
             // Sort outputs by their capacity
@@ -495,6 +508,7 @@ public class MaxMinFlowMultiplexer(
         private val scheduler: Scheduler,
         private val interferenceDomain: InterferenceDomain?,
         @JvmField val key: InterferenceKey?,
+        @JvmField val isCoupled: Boolean,
         initialCapacity: Double,
     ) : FlowConsumer, FlowConsumerLogic, Comparable<Input> {
         /**