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.

4 files changed, 48 insertions, 20 deletions

diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt
index 90bf5e25..eda59d2d 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt
@@ -144,7 +144,7 @@ public abstract class SimAbstractHypervisor(
      */
     private inner class VirtualMachine(
         model: MachineModel,
-        private val interferenceId: String? = null
+        interferenceId: String? = null
     ) : SimAbstractMachine(engine, parent = null, model), SimVirtualMachine {
         /**
          * The interference key of this virtual machine.
@@ -154,24 +154,14 @@ public abstract class SimAbstractHypervisor(
         /**
          * The vCPUs of the machine.
          */
-        override val cpus = model.cpus.map { VCpu(mux, mux.newInput(interferenceKey), it) }
+        override val cpus = model.cpus.map { cpu -> VCpu(mux, mux.newInput(cpu.frequency, interferenceKey), cpu) }
 
         /**
          * The resource counters associated with the hypervisor.
          */
         override val counters: SimHypervisorCounters
             get() = _counters
-        private val _counters = object : SimHypervisorCounters {
-            private val d = cpus.size / cpus.sumOf { it.model.frequency } * 1000
-
-            override val cpuActiveTime: Long
-                get() = (cpus.sumOf { it.counters.actual } * d).roundToLong()
-            override val cpuIdleTime: Long
-                get() = (cpus.sumOf { it.counters.actual + it.counters.remaining } * d).roundToLong()
-            override val cpuStealTime: Long
-                get() = (cpus.sumOf { it.counters.demand - it.counters.actual } * d).roundToLong()
-            override val cpuLostTime: Long = (cpus.sumOf { it.counters.interference } * d).roundToLong()
-        }
+        private val _counters = VmCountersImpl(cpus)
 
         /**
          * The CPU capacity of the hypervisor in MHz.
@@ -235,9 +225,7 @@ public abstract class SimAbstractHypervisor(
     ) : SimProcessingUnit, FlowConsumer by source {
         override var capacity: Double
             get() = source.capacity
-            set(_) {
-                // Ignore capacity changes
-            }
+            set(_) = TODO("Capacity changes on vCPU not supported")
 
         override fun toString(): String = "SimAbstractHypervisor.VCpu[model=$model]"
 
@@ -311,4 +299,20 @@ public abstract class SimAbstractHypervisor(
             cpuTime[3] += (interferenceDelta * d).roundToLong()
         }
     }
+
+    /**
+     * A [SimHypervisorCounters] implementation for a virtual machine.
+     */
+    private class VmCountersImpl(private val cpus: List<VCpu>) : SimHypervisorCounters {
+        private val d = cpus.size / cpus.sumOf { it.model.frequency } * 1000
+
+        override val cpuActiveTime: Long
+            get() = (cpus.sumOf { it.counters.actual } * d).roundToLong()
+        override val cpuIdleTime: Long
+            get() = (cpus.sumOf { it.counters.actual + it.counters.remaining } * d).roundToLong()
+        override val cpuStealTime: Long
+            get() = (cpus.sumOf { it.counters.demand - it.counters.actual } * d).roundToLong()
+        override val cpuLostTime: Long
+            get() = (cpus.sumOf { it.counters.interference } * d).roundToLong()
+    }
 }
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> {
         /**