merge: Implement multi-flow stages in simulator (#110)

This pull request introduces the new `flow2` multi-flow simulator into the OpenDC codebase
and adjust all existing modules to make use of this new simulator.

The new simulator models flow as a network of components, which can each receive flow
from (potentially) multiple other components. In the previous simulator, the framework itself
supported only single flows between components and required re-implementation of many
components to support multiplexing flows.

Initial benchmarks show performance improvements in the range 2x–4x for large scale experiments
such as the Capelin benchmarks.

## Implementation Notes :hammer_and_pick:

* Add support for multi-flow stages
* Support flow transformations
* Add forwarding flow multiplexer
* Expose metrics on FlowMultiplexer
* Re-implement network sim using flow2
* Re-implement power sim using flow2 
* Re-implement compute sim using flow2
* Optimize workload implementation of SimTrace 
* Remove old flow simulator 
* Add log4j-core dependency 

## External Dependencies :four_leaf_clover:

* N/A

## Breaking API Changes :warning:

* Removal of the `org.opendc.simulator.flow` package. You should now use
   the new flow simulator located in `org.opendc.simulator.flow2`.
* `PowerModel` interface is replaced by the `CpuPowerModel` interface.
* `PowerDriver` interface is replaced by the `SimPsu` and `SimPsuFactory` interfaces.
* Removal of `SimTraceWorkload`. Instead, create a workload from a `SimTrace` using 
  `createWorkload(offset)`.
* `ScalingGovernor` has been split in a `ScalingGovernor` and `ScalingGovernorFactory`.
* All modules in `opendc-simulator` are now written in Java. This means that default 
   parameters are not supported anymore for these modules.

3 files changed, 121 insertions, 70 deletions

diff --git a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt
index 78bd533d..8b4ebb89 100644
--- a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt
+++ b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSinkTest.kt
@@ -24,8 +24,9 @@ package org.opendc.simulator.network
 
 import io.mockk.every
 import io.mockk.mockk
-import io.mockk.spyk
 import io.mockk.verify
+import kotlinx.coroutines.yield
+import org.junit.jupiter.api.Assertions.assertAll
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertNull
@@ -33,11 +34,7 @@ import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertDoesNotThrow
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.flow.FlowConsumer
-import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSink
-import org.opendc.simulator.flow.FlowSource
-import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.flow2.FlowEngine
 import org.opendc.simulator.kotlin.runSimulation
 
 /**
@@ -46,18 +43,22 @@ import org.opendc.simulator.kotlin.runSimulation
 class SimNetworkSinkTest {
     @Test
     fun testInitialState() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
-
-        assertFalse(sink.isConnected)
-        assertNull(sink.link)
-        assertEquals(100.0, sink.capacity)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
+
+        assertAll(
+            { assertFalse(sink.isConnected) },
+            { assertNull(sink.link) },
+            { assertEquals(100.0f, sink.capacity) }
+        )
     }
 
     @Test
     fun testDisconnectIdempotent() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
 
         assertDoesNotThrow { sink.disconnect() }
         assertFalse(sink.isConnected)
@@ -65,8 +66,9 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnectCircular() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
 
         assertThrows<IllegalArgumentException> {
             sink.connect(sink)
@@ -75,8 +77,9 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnectAlreadyConnectedTarget() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
         val source = mockk<SimNetworkPort>(relaxUnitFun = true)
         every { source.isConnected } returns true
 
@@ -87,9 +90,10 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnectAlreadyConnected() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
-        val source1 = Source(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
+        val source1 = TestSource(graph)
 
         val source2 = mockk<SimNetworkPort>(relaxUnitFun = true)
 
@@ -103,41 +107,40 @@ class SimNetworkSinkTest {
 
     @Test
     fun testConnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
-        val source = spyk(Source(engine))
-        val consumer = source.consumer
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
+        val source = TestSource(graph)
 
         sink.connect(source)
 
-        assertTrue(sink.isConnected)
-        assertTrue(source.isConnected)
+        yield()
 
-        verify { source.createConsumer() }
-        verify { consumer.onStart(any(), any()) }
+        assertAll(
+            { assertTrue(sink.isConnected) },
+            { assertTrue(source.isConnected) },
+            { assertEquals(100.0f, source.outlet.capacity) }
+        )
+
+        verify { source.logic.onUpdate(any(), any()) }
     }
 
     @Test
     fun testDisconnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
-        val source = spyk(Source(engine))
-        val consumer = source.consumer
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
+        val source = TestSource(graph)
 
         sink.connect(source)
         sink.disconnect()
 
-        assertFalse(sink.isConnected)
-        assertFalse(source.isConnected)
-
-        verify { consumer.onStop(any(), any()) }
-    }
-
-    private class Source(engine: FlowEngine) : SimNetworkPort() {
-        val consumer = spyk(FixedFlowSource(Double.POSITIVE_INFINITY, utilization = 0.8))
-
-        public override fun createConsumer(): FlowSource = consumer
+        yield()
 
-        override val provider: FlowConsumer = FlowSink(engine, 0.0)
+        assertAll(
+            { assertFalse(sink.isConnected) },
+            { assertFalse(source.isConnected) },
+            { assertEquals(0.0f, source.outlet.capacity) }
+        )
     }
 }
diff --git a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt
index ecf80818..1507c4a1 100644
--- a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt
+++ b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtualTest.kt
@@ -22,16 +22,14 @@
 
 package org.opendc.simulator.network
 
-import io.mockk.spyk
 import io.mockk.verify
+import kotlinx.coroutines.yield
+import org.junit.jupiter.api.Assertions.assertAll
+import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertTrue
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.flow.FlowConsumer
-import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSink
-import org.opendc.simulator.flow.FlowSource
-import org.opendc.simulator.flow.source.FixedFlowSource
+import org.opendc.simulator.flow2.FlowEngine
 import org.opendc.simulator.kotlin.runSimulation
 
 /**
@@ -40,27 +38,32 @@ import org.opendc.simulator.kotlin.runSimulation
 class SimNetworkSwitchVirtualTest {
     @Test
     fun testConnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
-        val source = spyk(Source(engine))
-        val switch = SimNetworkSwitchVirtual(engine)
-        val consumer = source.consumer
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
+        val source = TestSource(graph)
+        val switch = SimNetworkSwitchVirtual(graph)
 
         switch.newPort().connect(sink)
         switch.newPort().connect(source)
 
-        assertTrue(sink.isConnected)
-        assertTrue(source.isConnected)
+        yield()
 
-        verify { source.createConsumer() }
-        verify { consumer.onStart(any(), any()) }
+        assertAll(
+            { assertTrue(sink.isConnected) },
+            { assertTrue(source.isConnected) },
+            { assertEquals(100.0f, source.outlet.capacity) }
+        )
+
+        verify { source.logic.onUpdate(any(), any()) }
     }
 
     @Test
     fun testConnectClosedPort() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val sink = SimNetworkSink(engine, capacity = 100.0)
-        val switch = SimNetworkSwitchVirtual(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val sink = SimNetworkSink(graph, /*capacity*/ 100.0f)
+        val switch = SimNetworkSwitchVirtual(graph)
 
         val port = switch.newPort()
         port.close()
@@ -69,12 +72,4 @@ class SimNetworkSwitchVirtualTest {
             port.connect(sink)
         }
     }
-
-    private class Source(engine: FlowEngine) : SimNetworkPort() {
-        val consumer = spyk(FixedFlowSource(Double.POSITIVE_INFINITY, utilization = 0.8))
-
-        public override fun createConsumer(): FlowSource = consumer
-
-        override val provider: FlowConsumer = FlowSink(engine, 0.0)
-    }
 }
diff --git a/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/TestSource.kt b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/TestSource.kt
new file mode 100644
index 00000000..f69db7a2
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/TestSource.kt
@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2022 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.simulator.network
+
+import io.mockk.spyk
+import org.opendc.simulator.flow2.FlowGraph
+import org.opendc.simulator.flow2.FlowStage
+import org.opendc.simulator.flow2.FlowStageLogic
+import org.opendc.simulator.flow2.InPort
+import org.opendc.simulator.flow2.Inlet
+import org.opendc.simulator.flow2.OutPort
+import org.opendc.simulator.flow2.Outlet
+
+/**
+ * A [SimNetworkPort] that acts as a test source.
+ */
+class TestSource(graph: FlowGraph) : SimNetworkPort(), FlowStageLogic {
+    val logic = spyk(this)
+    private val stage = graph.newStage(logic)
+
+    val outlet: OutPort = stage.getOutlet("out")
+    val inlet: InPort = stage.getInlet("in")
+
+    init {
+        outlet.push(80.0f)
+    }
+
+    override fun onUpdate(ctx: FlowStage, now: Long): Long = Long.MAX_VALUE
+
+    override fun getOutlet(): Outlet = outlet
+
+    override fun getInlet(): Inlet = inlet
+}