refactor(sim/power): Re-implement power sim using flow2

This change updates the `opendc-simulator-power` module to use the new
flow simulation framework in OpenDC (named flow2 for now).

4 files changed, 185 insertions, 117 deletions

diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt
index 29c50d3f..6adb0548 100644
--- a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPduTest.kt
@@ -22,14 +22,11 @@
 
 package org.opendc.simulator.power
 
-import io.mockk.spyk
-import io.mockk.verify
+import kotlinx.coroutines.yield
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.assertThrows
-import org.opendc.simulator.flow.FlowEngine
-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
 
 /**
@@ -38,82 +35,93 @@ import org.opendc.simulator.kotlin.runSimulation
 internal class SimPduTest {
     @Test
     fun testZeroOutlets() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val pdu = SimPdu(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+        val pdu = SimPdu(graph)
         source.connect(pdu)
 
-        assertEquals(0.0, source.powerDraw)
+        yield()
+
+        assertEquals(0.0f, source.powerDraw)
     }
 
     @Test
     fun testSingleOutlet() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val pdu = SimPdu(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+        val pdu = SimPdu(graph)
         source.connect(pdu)
-        pdu.newOutlet().connect(SimpleInlet())
+        pdu.newOutlet().connect(TestInlet(graph))
+
+        yield()
 
-        assertEquals(50.0, source.powerDraw)
+        assertEquals(100.0f, source.powerDraw)
     }
 
     @Test
     fun testDoubleOutlet() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val pdu = SimPdu(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 200.0f)
+        val pdu = SimPdu(graph)
         source.connect(pdu)
 
-        pdu.newOutlet().connect(SimpleInlet())
-        pdu.newOutlet().connect(SimpleInlet())
+        pdu.newOutlet().connect(TestInlet(graph))
+        pdu.newOutlet().connect(TestInlet(graph))
+
+        yield()
 
-        assertEquals(100.0, source.powerDraw)
+        assertEquals(200.0f, source.powerDraw)
     }
 
     @Test
     fun testDisconnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val pdu = SimPdu(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 300.0f)
+        val pdu = SimPdu(graph)
         source.connect(pdu)
-        val consumer = spyk(FixedFlowSource(100.0, utilization = 1.0))
-        val inlet = object : SimPowerInlet() {
-            override fun createSource(): FlowSource = consumer
-        }
 
         val outlet = pdu.newOutlet()
-        outlet.connect(inlet)
+        outlet.connect(TestInlet(graph))
         outlet.disconnect()
 
-        verify { consumer.onStop(any(), any()) }
+        yield()
+
+        assertEquals(0.0f, source.powerDraw)
     }
 
     @Test
     fun testLoss() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 500.0f)
         // https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
-        val pdu = SimPdu(engine, idlePower = 1.5, lossCoefficient = 0.015)
+        val pdu = SimPdu(graph, /*idlePower*/ 1.5f, /*lossCoefficient*/ 0.015f)
         source.connect(pdu)
-        pdu.newOutlet().connect(SimpleInlet())
-        assertEquals(89.0, source.powerDraw, 0.01)
+        pdu.newOutlet().connect(TestInlet(graph))
+
+        yield()
+
+        assertEquals(251.5f, source.powerDraw, 0.01f)
     }
 
     @Test
     fun testOutletClose() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val pdu = SimPdu(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+        val pdu = SimPdu(graph)
         source.connect(pdu)
         val outlet = pdu.newOutlet()
         outlet.close()
 
+        yield()
+
         assertThrows<IllegalStateException> {
-            outlet.connect(SimpleInlet())
+            outlet.connect(TestInlet(graph))
         }
     }
-
-    class SimpleInlet : SimPowerInlet() {
-        override fun createSource(): FlowSource = FixedFlowSource(100.0, utilization = 0.5)
-    }
 }
diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt
index 963ba710..03b8182c 100644
--- a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimPowerSourceTest.kt
@@ -24,8 +24,8 @@ package org.opendc.simulator.power
 
 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,9 +33,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.FlowEngine
-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
 
 /**
@@ -44,18 +42,24 @@ import org.opendc.simulator.kotlin.runSimulation
 internal class SimPowerSourceTest {
     @Test
     fun testInitialState() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
 
-        assertFalse(source.isConnected)
-        assertNull(source.inlet)
-        assertEquals(100.0, source.capacity)
+        yield()
+
+        assertAll(
+            { assertFalse(source.isConnected) },
+            { assertNull(source.inlet) },
+            { assertEquals(100.0f, source.capacity) }
+        )
     }
 
     @Test
     fun testDisconnectIdempotent() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
 
         assertDoesNotThrow { source.disconnect() }
         assertFalse(source.isConnected)
@@ -63,44 +67,51 @@ internal class SimPowerSourceTest {
 
     @Test
     fun testConnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val inlet = SimpleInlet()
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+        val inlet = TestInlet(graph)
 
         source.connect(inlet)
 
-        assertTrue(source.isConnected)
-        assertEquals(inlet, source.inlet)
-        assertTrue(inlet.isConnected)
-        assertEquals(source, inlet.outlet)
-        assertEquals(100.0, source.powerDraw)
+        yield()
+
+        assertAll(
+            { assertTrue(source.isConnected) },
+            { assertEquals(inlet, source.inlet) },
+            { assertTrue(inlet.isConnected) },
+            { assertEquals(source, inlet.outlet) },
+            { assertEquals(100.0f, source.powerDraw) }
+        )
     }
 
     @Test
     fun testDisconnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val consumer = spyk(FixedFlowSource(100.0, utilization = 1.0))
-        val inlet = object : SimPowerInlet() {
-            override fun createSource(): FlowSource = consumer
-        }
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+        val inlet = TestInlet(graph)
 
         source.connect(inlet)
         source.disconnect()
 
-        verify { consumer.onStop(any(), any()) }
+        yield()
+
+        assertEquals(0.0f, inlet.flowOutlet.capacity)
     }
 
     @Test
     fun testDisconnectAssertion() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+
         val inlet = mockk<SimPowerInlet>(relaxUnitFun = true)
         every { inlet.isConnected } returns false
-        every { inlet._outlet } returns null
-        every { inlet.createSource() } returns FixedFlowSource(100.0, utilization = 1.0)
+        every { inlet.flowOutlet } returns TestInlet(graph).flowOutlet
 
         source.connect(inlet)
+        inlet.outlet = null
 
         assertThrows<AssertionError> {
             source.disconnect()
@@ -109,13 +120,14 @@ internal class SimPowerSourceTest {
 
     @Test
     fun testOutletAlreadyConnected() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val inlet = SimpleInlet()
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
+        val inlet = TestInlet(graph)
 
         source.connect(inlet)
         assertThrows<IllegalStateException> {
-            source.connect(SimpleInlet())
+            source.connect(TestInlet(graph))
         }
 
         assertEquals(inlet, source.inlet)
@@ -123,8 +135,9 @@ internal class SimPowerSourceTest {
 
     @Test
     fun testInletAlreadyConnected() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 100.0f)
         val inlet = mockk<SimPowerInlet>(relaxUnitFun = true)
         every { inlet.isConnected } returns true
 
@@ -132,8 +145,4 @@ internal class SimPowerSourceTest {
             source.connect(inlet)
         }
     }
-
-    class SimpleInlet : SimPowerInlet() {
-        override fun createSource(): FlowSource = FixedFlowSource(100.0, utilization = 1.0)
-    }
 }
diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt
index 2b2921d7..d984a8cb 100644
--- a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/SimUpsTest.kt
@@ -22,14 +22,11 @@
 
 package org.opendc.simulator.power
 
-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.Test
-import org.opendc.simulator.flow.FlowEngine
-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
 
 /**
@@ -38,64 +35,70 @@ import org.opendc.simulator.kotlin.runSimulation
 internal class SimUpsTest {
     @Test
     fun testSingleInlet() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
-        val ups = SimUps(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 200.0f)
+        val ups = SimUps(graph)
         source.connect(ups.newInlet())
-        ups.connect(SimpleInlet())
+        ups.connect(TestInlet(graph))
 
-        assertEquals(50.0, source.powerDraw)
+        yield()
+
+        assertEquals(100.0f, source.powerDraw)
     }
 
     @Test
     fun testDoubleInlet() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source1 = SimPowerSource(engine, capacity = 100.0)
-        val source2 = SimPowerSource(engine, capacity = 100.0)
-        val ups = SimUps(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source1 = SimPowerSource(graph, /*capacity*/ 200.0f)
+        val source2 = SimPowerSource(graph, /*capacity*/ 200.0f)
+        val ups = SimUps(graph)
         source1.connect(ups.newInlet())
         source2.connect(ups.newInlet())
 
-        ups.connect(SimpleInlet())
+        ups.connect(TestInlet(graph))
+
+        yield()
 
         assertAll(
-            { assertEquals(50.0, source1.powerDraw) },
-            { assertEquals(50.0, source2.powerDraw) }
+            { assertEquals(50.0f, source1.powerDraw) },
+            { assertEquals(50.0f, source2.powerDraw) }
         )
     }
 
     @Test
     fun testLoss() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source = SimPowerSource(engine, capacity = 100.0)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source = SimPowerSource(graph, /*capacity*/ 500.0f)
         // https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
-        val ups = SimUps(engine, idlePower = 4.0, lossCoefficient = 0.05)
+        val ups = SimUps(graph, /*idlePower*/ 4.0f, /*lossCoefficient*/ 0.05f)
         source.connect(ups.newInlet())
-        ups.connect(SimpleInlet())
+        ups.connect(TestInlet(graph))
+
+        yield()
 
-        assertEquals(56.5, source.powerDraw)
+        assertEquals(108.99f, source.powerDraw, 0.01f)
     }
 
     @Test
     fun testDisconnect() = runSimulation {
-        val engine = FlowEngine(coroutineContext, clock)
-        val source1 = SimPowerSource(engine, capacity = 100.0)
-        val source2 = SimPowerSource(engine, capacity = 100.0)
-        val ups = SimUps(engine)
+        val engine = FlowEngine.create(coroutineContext, clock)
+        val graph = engine.newGraph()
+        val source1 = SimPowerSource(graph, /*capacity*/ 200.0f)
+        val source2 = SimPowerSource(graph, /*capacity*/ 200.0f)
+        val ups = SimUps(graph)
         source1.connect(ups.newInlet())
         source2.connect(ups.newInlet())
-        val consumer = spyk(FixedFlowSource(100.0, utilization = 1.0))
-        val inlet = object : SimPowerInlet() {
-            override fun createSource(): FlowSource = consumer
-        }
+
+        val inlet = TestInlet(graph)
 
         ups.connect(inlet)
         ups.disconnect()
 
-        verify { consumer.onStop(any(), any()) }
-    }
+        yield()
 
-    class SimpleInlet : SimPowerInlet() {
-        override fun createSource(): FlowSource = FixedFlowSource(100.0, utilization = 0.5)
+        assertEquals(0.0f, inlet.flowOutlet.capacity)
     }
 }
diff --git a/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/TestInlet.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/TestInlet.kt
new file mode 100644
index 00000000..7ba12ed9
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/TestInlet.kt
@@ -0,0 +1,48 @@
+/*
+ * 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.power
+
+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.Outlet
+
+/**
+ * A test inlet.
+ */
+class TestInlet(graph: FlowGraph) : SimPowerInlet(), FlowStageLogic {
+    val logic = spyk(this)
+    private val stage = graph.newStage(logic)
+    val flowOutlet = stage.getOutlet("out")
+
+    init {
+        flowOutlet.push(100.0f)
+    }
+
+    override fun onUpdate(ctx: FlowStage, now: Long): Long = Long.MAX_VALUE
+
+    override fun getFlowOutlet(): Outlet {
+        return flowOutlet
+    }
+}