merge: Support convergence of space shared hypervisor

This change addresses an issue with the SimSpaceSharedHypervisor 
implementation where it did not emit convergence events 
due to missing implementation. 
This caused issues with users of this class trying to obtain usage data, 
which depended on these events being emitted.

* Support `FlowConvergenceListener` in `FlowForwarder` and `ForwardingFlowMux`
* Support `FlowConvergenceListener` in `SimSpaceSharedHypervisor`.

## Breaking API Changes :warning:

* N/A

4 files changed, 29 insertions, 5 deletions

diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 98521e71..6f3c561c 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -43,6 +43,7 @@ jobs:
           github_token: ${{ secrets.GITHUB_TOKEN }}
       - name: Upload artifacts
         if: always()
+        continue-on-error: true # See https://github.com/actions/upload-artifact/issues/270
         uses: actions/upload-artifact@v2
         with:
           name: reports-${{ matrix.os }}-jdk${{ matrix.java }}
diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt
index 82f8df38..3f3bf6ad 100644
--- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt
+++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt
@@ -37,7 +37,7 @@ public class SimSpaceSharedHypervisor(
     listener: FlowConvergenceListener?,
     scalingGovernor: ScalingGovernor?,
 ) : SimAbstractHypervisor(engine, listener, scalingGovernor) {
-    override val mux: FlowMultiplexer = ForwardingFlowMultiplexer(engine)
+    override val mux: FlowMultiplexer = ForwardingFlowMultiplexer(engine, this)
 
     override fun canFit(model: MachineModel): Boolean {
         return mux.outputs.size - mux.inputs.size >= model.cpus.size
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt
index e3bdd7ba..a5663293 100644
--- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt
@@ -31,9 +31,14 @@ import kotlin.math.max
  * A class that acts as a [FlowSource] and [FlowConsumer] at the same time.
  *
  * @param engine The [FlowEngine] the forwarder runs in.
+ * @param listener The convergence lister to use.
  * @param isCoupled A flag to indicate that the transformer will exit when the resource consumer exits.
  */
-public class FlowForwarder(private val engine: FlowEngine, private val isCoupled: Boolean = false) : FlowSource, FlowConsumer, AutoCloseable {
+public class FlowForwarder(
+    private val engine: FlowEngine,
+    private val listener: FlowConvergenceListener? = null,
+    private val isCoupled: Boolean = false
+) : FlowSource, FlowConsumer, AutoCloseable {
     /**
      * The logging instance of this connection.
      */
@@ -153,7 +158,7 @@ public class FlowForwarder(private val engine: FlowEngine, private val isCoupled
     override fun onStart(conn: FlowConnection, now: Long) {
         _innerCtx = conn
 
-        if (_ctx.shouldSourceConverge) {
+        if (listener != null || _ctx.shouldSourceConverge) {
             conn.shouldSourceConverge = true
         }
     }
@@ -196,6 +201,7 @@ public class FlowForwarder(private val engine: FlowEngine, private val isCoupled
     override fun onConverge(conn: FlowConnection, now: Long, delta: Long) {
         try {
             delegate?.onConverge(this._ctx, now, delta)
+            listener?.onConverge(now, delta)
         } catch (cause: Throwable) {
             logger.error(cause) { "Uncaught exception" }
 
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 b68a8baa..eff111b8 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
@@ -25,6 +25,7 @@ package org.opendc.simulator.flow.mux
 import org.opendc.simulator.flow.*
 import org.opendc.simulator.flow.interference.InterferenceKey
 import java.util.ArrayDeque
+import kotlin.math.max
 
 /**
  * A [FlowMultiplexer] implementation that allocates inputs to the outputs of the multiplexer exclusively. This means
@@ -32,8 +33,12 @@ import java.util.ArrayDeque
  * inputs as outputs.
  *
  * @param engine The [FlowEngine] driving the simulation.
+ * @param listener The convergence listener of the multiplexer.
  */
-public class ForwardingFlowMultiplexer(private val engine: FlowEngine) : FlowMultiplexer {
+public class ForwardingFlowMultiplexer(
+    private val engine: FlowEngine,
+    private val listener: FlowConvergenceListener? = null
+) : FlowMultiplexer, FlowConvergenceListener {
     override val inputs: Set<FlowConsumer>
         get() = _inputs
     private val _inputs = mutableSetOf<Input>()
@@ -91,7 +96,7 @@ public class ForwardingFlowMultiplexer(private val engine: FlowEngine) : FlowMul
     }
 
     override fun newOutput(): FlowSource {
-        val forwarder = FlowForwarder(engine)
+        val forwarder = FlowForwarder(engine, this)
         val output = Output(forwarder)
 
         _outputs += output
@@ -130,6 +135,18 @@ public class ForwardingFlowMultiplexer(private val engine: FlowEngine) : FlowMul
         clearInputs()
     }
 
+    private var _lastConverge = Long.MAX_VALUE
+
+    override fun onConverge(now: Long, delta: Long) {
+        val listener = listener
+        if (listener != null) {
+            val lastConverge = _lastConverge
+            _lastConverge = now
+            val duration = max(0, now - lastConverge)
+            listener.onConverge(now, duration)
+        }
+    }
+
     /**
      * An input on the multiplexer.
      */