refactor(simulator): Simplify max-min aggregator implementation

This change simplifies the implementation of the
SimResourceAggregatorMaxMin class by utilizing the new push method.
This approach should offer better performance than the previous version,
since we can directly push changes to the source.

2 files changed, 16 insertions, 80 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 8e0eb5f8..d064d7fa 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
@@ -32,7 +32,7 @@ public abstract class SimAbstractResourceAggregator(
     /**
      * This method is invoked when the resource consumer consumes resources.
      */
-    protected abstract fun doConsume(limit: Double, duration: Long)
+    protected abstract fun doConsume(limit: Double)
 
     /**
      * This method is invoked when the resource consumer finishes processing.
@@ -42,12 +42,12 @@ public abstract class SimAbstractResourceAggregator(
     /**
      * This method is invoked when an input context is started.
      */
-    protected abstract fun onInputStarted(input: Input)
+    protected abstract fun onInputStarted(input: SimResourceContext)
 
     /**
      * This method is invoked when an input is stopped.
      */
-    protected abstract fun onInputFinished(input: Input)
+    protected abstract fun onInputFinished(input: SimResourceContext)
 
     /* SimResourceAggregator */
     override fun addInput(input: SimResourceProvider) {
@@ -95,7 +95,7 @@ public abstract class SimAbstractResourceAggregator(
             return object : SimResourceProviderLogic {
 
                 override fun onConsume(ctx: SimResourceControllableContext, now: Long, limit: Double, duration: Long): Long {
-                    doConsume(limit, duration)
+                    doConsume(limit)
                     return super.onConsume(ctx, now, limit, duration)
                 }
 
@@ -113,90 +113,25 @@ public abstract class SimAbstractResourceAggregator(
                 }
             }
         }
-
-        /**
-         * Flush the progress of the output if possible.
-         */
-        fun flush() {
-            ctx?.flush()
-        }
-    }
-
-    /**
-     * An input for the resource aggregator.
-     */
-    public interface Input : AutoCloseable {
-        /**
-         * The [SimResourceContext] associated with the input.
-         */
-        public val ctx: SimResourceContext
-
-        /**
-         * Push to this input with the specified [limit] and [duration].
-         */
-        public fun push(limit: Double, duration: Long)
-
-        /**
-         * Close the input for further input.
-         */
-        public override fun close()
     }
 
     /**
      * An internal [SimResourceConsumer] implementation for aggregator inputs.
      */
-    private inner class Consumer : Input, SimResourceConsumer {
+    private inner class Consumer : SimResourceConsumer {
         /**
          * The resource context associated with the input.
          */
-        override val ctx: SimResourceContext
-            get() = _ctx!!
         private var _ctx: SimResourceContext? = null
 
-        /**
-         * The resource command to run next.
-         */
-        private var _duration: Long = Long.MAX_VALUE
-
-        /**
-         * A flag to indicate that the consumer should flush.
-         */
-        private var _isPushed = false
-
         private fun updateCapacity() {
             // Adjust capacity of output resource
             _output.capacity = _inputConsumers.sumOf { it._ctx?.capacity ?: 0.0 }
         }
 
-        /* Input */
-        override fun push(limit: Double, duration: Long) {
-            _duration = duration
-            val ctx = _ctx
-            if (ctx != null) {
-                ctx.push(limit)
-                ctx.interrupt()
-            }
-            _isPushed = true
-        }
-
-        override fun close() {
-            _duration = Long.MAX_VALUE
-            _isPushed = true
-            _ctx?.close()
-        }
-
         /* SimResourceConsumer */
         override fun onNext(ctx: SimResourceContext, now: Long, delta: Long): Long {
-            var next = _duration
-
-            if (!_isPushed) {
-                _output.flush()
-                next = _duration
-            }
-
-            _isPushed = false
-            _duration = Long.MAX_VALUE
-            return next
+            return Long.MAX_VALUE
         }
 
         override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {
@@ -205,10 +140,10 @@ public abstract class SimAbstractResourceAggregator(
                     _ctx = ctx
                     updateCapacity()
 
-                    onInputStarted(this)
+                    onInputStarted(ctx)
                 }
                 SimResourceEvent.Capacity -> updateCapacity()
-                SimResourceEvent.Exit -> onInputFinished(this)
+                SimResourceEvent.Exit -> onInputFinished(ctx)
                 else -> {}
             }
         }
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt
index b258a368..f131ac6c 100644
--- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt
+++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt
@@ -29,19 +29,20 @@ public class SimResourceAggregatorMaxMin(
     interpreter: SimResourceInterpreter,
     parent: SimResourceSystem? = null
 ) : SimAbstractResourceAggregator(interpreter, parent) {
-    private val consumers = mutableListOf<Input>()
+    private val consumers = mutableListOf<SimResourceContext>()
 
-    override fun doConsume(limit: Double, duration: Long) {
+    override fun doConsume(limit: Double) {
         // Sort all consumers by their capacity
-        consumers.sortWith(compareBy { it.ctx.capacity })
+        consumers.sortWith(compareBy { it.capacity })
 
         // Divide the requests over the available capacity of the input resources fairly
         for (input in consumers) {
-            val inputCapacity = input.ctx.capacity
+            val inputCapacity = input.capacity
             val fraction = inputCapacity / capacity
             val grantedSpeed = limit * fraction
 
-            input.push(grantedSpeed, duration)
+            input.push(grantedSpeed)
+            input.interrupt()
         }
     }
 
@@ -53,11 +54,11 @@ public class SimResourceAggregatorMaxMin(
         }
     }
 
-    override fun onInputStarted(input: Input) {
+    override fun onInputStarted(input: SimResourceContext) {
         consumers.add(input)
     }
 
-    override fun onInputFinished(input: Input) {
+    override fun onInputFinished(input: SimResourceContext) {
         consumers.remove(input)
     }
 }