perf(simulator): Do not use set for tracking convergence

This change removes the use of a HashSet for tracking the flow
connections that can converge. A HashSet requires an allocation for
every addition, which caused a significant overhead. The new approach
using an ArrayDeque should not allocate any memory.

2 files changed, 25 insertions, 10 deletions

diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt
index f62528ed..a4d82a3d 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt
@@ -95,6 +95,11 @@ internal class FlowConsumerContextImpl(
     private var _isImmediateUpdateScheduled = false
 
     /**
+     * A flag that indicates to the [FlowEngine] that the context is already enqueued to converge.
+     */
+    private var _willConverge: Boolean = false
+
+    /**
      * The timestamp of calls to the callbacks.
      */
     private var _lastPull: Long = Long.MIN_VALUE // Last call to `onPull`
@@ -177,12 +182,18 @@ internal class FlowConsumerContextImpl(
     }
 
     /**
-     * Update the state of the resource context.
+     * Update the state of the flow connection.
+     *
+     * @param now The current virtual timestamp.
+     * @return A flag to indicate whether the connection has already been updated before convergence.
      */
-    fun doUpdate(now: Long) {
+    fun doUpdate(now: Long): Boolean {
+        val willConverge = _willConverge
+        _willConverge = true
+
         val oldState = _state
         if (oldState != State.Active) {
-            return
+            return willConverge
         }
 
         _isUpdateActive = true
@@ -238,6 +249,8 @@ internal class FlowConsumerContextImpl(
         } finally {
             _isUpdateActive = false
         }
+
+        return willConverge
     }
 
     /**
@@ -270,6 +283,7 @@ internal class FlowConsumerContextImpl(
     fun onConverge(timestamp: Long) {
         val delta = max(0, timestamp - _lastConvergence)
         _lastConvergence = timestamp
+        _willConverge = false
 
         try {
             if (_state == State.Active) {
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt
index 1a50da2c..5f15fbed 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt
@@ -63,7 +63,7 @@ internal class FlowEngineImpl(private val context: CoroutineContext, override va
     /**
      * The systems that have been visited during the engine cycle.
      */
-    private val visited = linkedSetOf<FlowConsumerContextImpl>()
+    private val visited = ArrayDeque<FlowConsumerContextImpl>()
 
     /**
      * The index in the batch stack.
@@ -80,8 +80,9 @@ internal class FlowEngineImpl(private val context: CoroutineContext, override va
      * Update the specified [ctx] synchronously.
      */
     fun scheduleSync(now: Long, ctx: FlowConsumerContextImpl) {
-        ctx.doUpdate(now)
-        visited.add(ctx)
+        if (!ctx.doUpdate(now)) {
+            visited.add(ctx)
+        }
 
         // In-case the engine is already running in the call-stack, return immediately. The changes will be picked
         // up by the active engine.
@@ -192,8 +193,9 @@ internal class FlowEngineImpl(private val context: CoroutineContext, override va
             ctx.pruneTimers(now)
 
             if (ctx.shouldUpdate(now)) {
-                ctx.doUpdate(now)
-                visited.add(ctx)
+                if (!ctx.doUpdate(now)) {
+                    visited.add(ctx)
+                }
             } else {
                 ctx.tryReschedule(now)
             }
@@ -206,8 +208,7 @@ internal class FlowEngineImpl(private val context: CoroutineContext, override va
             while (true) {
                 val ctx = queue.poll() ?: break
 
-                if (ctx.shouldUpdate(now)) {
-                    ctx.doUpdate(now)
+                if (ctx.shouldUpdate(now) && !ctx.doUpdate(now)) {
                     visited.add(ctx)
                 }
             }