refactor(sim/net): Re-implement network sim using flow2

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
+}