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

11 files changed, 461 insertions, 309 deletions

diff --git a/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkLink.java b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkLink.java
new file mode 100644
index 00000000..1ea9cb0e
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkLink.java
@@ -0,0 +1,77 @@
+/*
+ * 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;
+
+/**
+ * A physical bidirectional communication link between two [SimNetworkPort]s.
+ */
+public final class SimNetworkLink {
+    private final SimNetworkPort left;
+    private final SimNetworkPort right;
+
+    SimNetworkLink(SimNetworkPort left, SimNetworkPort right) {
+        this.left = left;
+        this.right = right;
+    }
+
+    /**
+     * Determine whether the specified <code>port</code> participates in this network link.
+     *
+     * @return <code>true</code> if the port participates in this link, <code>false</code> otherwise.
+     */
+    public boolean contains(SimNetworkPort port) {
+        return port == left || port == right;
+    }
+
+    /**
+     * Obtain the opposite port to which the specified <code>port</code> is connected through this link.
+     */
+    public SimNetworkPort opposite(SimNetworkPort port) {
+        if (port == left) {
+            return right;
+        } else if (port == right) {
+            return left;
+        }
+
+        throw new IllegalArgumentException("Invalid port given");
+    }
+
+    /**
+     * Return the first port of the link.
+     */
+    public SimNetworkPort getLeft() {
+        return left;
+    }
+
+    /**
+     * Return the second port of the link.
+     */
+    public SimNetworkPort getRight() {
+        return right;
+    }
+
+    @Override
+    public String toString() {
+        return "SimNetworkLink[left=" + left + ",right=" + right + "]";
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkPort.java b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkPort.java
new file mode 100644
index 00000000..b5e09b9b
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkPort.java
@@ -0,0 +1,110 @@
+/*
+ * 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 org.opendc.simulator.flow2.Inlet;
+import org.opendc.simulator.flow2.Outlet;
+
+/**
+ * A network port allows network devices to be connected to network through links.
+ */
+public abstract class SimNetworkPort {
+    SimNetworkLink link;
+
+    /**
+     * Determine whether the network port is connected to another port.
+     *
+     * @return <code>true</code> if the network port is connected, <code>false</code> otherwise.
+     */
+    public boolean isConnected() {
+        return link != null;
+    }
+
+    /**
+     * Return network link which connects this port to another port.
+     */
+    public SimNetworkLink getLink() {
+        return link;
+    }
+
+    /**
+     * Connect this port to the specified <code>port</code>.
+     */
+    public void connect(SimNetworkPort port) {
+        if (port == this) {
+            throw new IllegalArgumentException("Circular reference");
+        }
+        if (isConnected()) {
+            throw new IllegalStateException("Port already connected");
+        }
+        if (port.isConnected()) {
+            throw new IllegalStateException("Target port already connected");
+        }
+
+        final SimNetworkLink link = new SimNetworkLink(this, port);
+        this.link = link;
+        port.link = link;
+
+        // Start bidirectional flow channel between the two ports
+        final Outlet outlet = getOutlet();
+        final Inlet inlet = getInlet();
+
+        outlet.getGraph().connect(outlet, port.getInlet());
+        inlet.getGraph().connect(port.getOutlet(), inlet);
+    }
+
+    /**
+     * Disconnect the current network link if it exists.
+     */
+    public void disconnect() {
+        final SimNetworkLink link = this.link;
+        if (link == null) {
+            return;
+        }
+
+        final SimNetworkPort opposite = link.opposite(this);
+        this.link = null;
+        opposite.link = null;
+
+        final Outlet outlet = getOutlet();
+        final Inlet inlet = getInlet();
+
+        outlet.getGraph().disconnect(outlet);
+        inlet.getGraph().disconnect(inlet);
+    }
+
+    /**
+     * Return the {@link Outlet} representing the outgoing traffic of this port.
+     */
+    protected abstract Outlet getOutlet();
+
+    /**
+     * An [Inlet] representing the ingoing traffic of this port.
+     */
+    protected abstract Inlet getInlet();
+
+    @Override
+    public String toString() {
+        return "SimNetworkPort[isConnected=" + isConnected() + "]";
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSink.java b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSink.java
new file mode 100644
index 00000000..f8918328
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSink.java
@@ -0,0 +1,70 @@
+/*
+ * 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 org.opendc.simulator.flow2.FlowGraph;
+import org.opendc.simulator.flow2.Inlet;
+import org.opendc.simulator.flow2.Outlet;
+import org.opendc.simulator.flow2.sink.SimpleFlowSink;
+import org.opendc.simulator.flow2.source.EmptyFlowSource;
+
+/**
+ * A network sink which discards all received traffic and does not generate any traffic itself.
+ */
+public final class SimNetworkSink extends SimNetworkPort {
+    private final EmptyFlowSource source;
+    private final SimpleFlowSink sink;
+
+    /**
+     * Construct a {@link SimNetworkSink} instance.
+     *
+     * @param graph The {@link FlowGraph} to which the sink belongs.
+     * @param capacity The capacity of the sink in terms of processed data.
+     */
+    public SimNetworkSink(FlowGraph graph, float capacity) {
+        this.source = new EmptyFlowSource(graph);
+        this.sink = new SimpleFlowSink(graph, capacity);
+    }
+
+    /**
+     * Return the capacity of the sink.
+     */
+    public float getCapacity() {
+        return sink.getCapacity();
+    }
+
+    @Override
+    protected Outlet getOutlet() {
+        return source.getOutput();
+    }
+
+    @Override
+    protected Inlet getInlet() {
+        return sink.getInput();
+    }
+
+    @Override
+    public String toString() {
+        return "SimNetworkSink[capacity=" + getCapacity() + "]";
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitch.kt b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSwitch.java
index 7dc249ab..b05dc53d 100644
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitch.kt
+++ b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSwitch.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021 AtLarge Research
+ * 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
@@ -20,14 +20,16 @@
  * SOFTWARE.
  */
 
-package org.opendc.simulator.network
+package org.opendc.simulator.network;
+
+import java.util.List;
 
 /**
  * A network device connects devices on a network by switching the traffic over its ports.
  */
 public interface SimNetworkSwitch {
     /**
-     * The ports of the switch.
+     * Return the ports of the switch.
      */
-    public val ports: List<SimNetworkPort>
+    List<SimNetworkPort> getPorts();
 }
diff --git a/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSwitchVirtual.java b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSwitchVirtual.java
new file mode 100644
index 00000000..a94bf799
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-network/src/main/java/org/opendc/simulator/network/SimNetworkSwitchVirtual.java
@@ -0,0 +1,107 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.opendc.simulator.flow2.FlowGraph;
+import org.opendc.simulator.flow2.Inlet;
+import org.opendc.simulator.flow2.Outlet;
+import org.opendc.simulator.flow2.mux.FlowMultiplexer;
+import org.opendc.simulator.flow2.mux.MaxMinFlowMultiplexer;
+
+/**
+ * A {@link SimNetworkSwitch} that can support new networking ports on demand.
+ */
+public final class SimNetworkSwitchVirtual implements SimNetworkSwitch {
+    private final List<Port> ports = new ArrayList<>();
+
+    /**
+     * The {@link MaxMinFlowMultiplexer} to actually perform the switching.
+     */
+    private final MaxMinFlowMultiplexer mux;
+
+    /**
+     * Construct a {@link SimNetworkSwitchVirtual} instance.
+     *
+     * @param graph The {@link FlowGraph} to drive the simulation.
+     */
+    public SimNetworkSwitchVirtual(FlowGraph graph) {
+        this.mux = new MaxMinFlowMultiplexer(graph);
+    }
+
+    /**
+     * Open a new port on the switch.
+     */
+    public Port newPort() {
+        final Port port = new Port(mux);
+        ports.add(port);
+        return port;
+    }
+
+    @Override
+    public List<SimNetworkPort> getPorts() {
+        return Collections.unmodifiableList(ports);
+    }
+
+    /**
+     * A port on the network switch.
+     */
+    public class Port extends SimNetworkPort implements AutoCloseable {
+        private final FlowMultiplexer mux;
+        private final Inlet inlet;
+        private final Outlet outlet;
+        private boolean isClosed;
+
+        private Port(FlowMultiplexer mux) {
+            this.mux = mux;
+            this.inlet = mux.newInput();
+            this.outlet = mux.newOutput();
+        }
+
+        @Override
+        protected Outlet getOutlet() {
+            if (isClosed) {
+                throw new IllegalStateException("Port is closed");
+            }
+            return outlet;
+        }
+
+        @Override
+        protected Inlet getInlet() {
+            if (isClosed) {
+                throw new IllegalStateException("Port is closed");
+            }
+            return inlet;
+        }
+
+        @Override
+        public void close() {
+            isClosed = true;
+            mux.releaseInput(inlet);
+            mux.releaseOutput(outlet);
+            ports.remove(this);
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkLink.kt b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkLink.kt
deleted file mode 100644
index 67562640..00000000
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkLink.kt
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Copyright (c) 2021 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
-
-/**
- * A physical bi-directional communication link between two [SimNetworkPort]s.
- *
- * @param left The first port of the link.
- * @param right The second port of the link.
- */
-public class SimNetworkLink(public val left: SimNetworkPort, public val right: SimNetworkPort) {
-    /**
-     * Determine whether the specified [port] participates in this network link.
-     */
-    public operator fun contains(port: SimNetworkPort): Boolean = port == left || port == right
-
-    /**
-     * Obtain the opposite port to which the specified [port] is connected through this link.
-     */
-    public fun opposite(port: SimNetworkPort): SimNetworkPort {
-        return when (port) {
-            left -> right
-            right -> left
-            else -> throw IllegalArgumentException("Invalid port given")
-        }
-    }
-
-    override fun toString(): String = "SimNetworkLink[left=$left,right=$right]"
-}
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt
deleted file mode 100644
index 4b66d5cf..00000000
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkPort.kt
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Copyright (c) 2021 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 org.opendc.simulator.flow.FlowConsumer
-import org.opendc.simulator.flow.FlowSource
-
-/**
- * A network port allows network devices to be connected to network through links.
- */
-public abstract class SimNetworkPort {
-    /**
-     * A flag to indicate that the network port is connected to another port.
-     */
-    public val isConnected: Boolean
-        get() = _link != null
-
-    /**
-     * The network link which connects this port to another port.
-     */
-    public val link: SimNetworkLink?
-        get() = _link
-    private var _link: SimNetworkLink? = null
-
-    /**
-     * Connect this port to the specified [port].
-     */
-    public fun connect(port: SimNetworkPort) {
-        require(port !== this) { "Circular reference" }
-        check(!isConnected) { "Port already connected" }
-        check(!port.isConnected) { "Target port already connected" }
-
-        val link = SimNetworkLink(this, port)
-        _link = link
-        port._link = link
-
-        // Start bi-directional flow channel between the two ports
-        try {
-            provider.startConsumer(port.createConsumer())
-            port.provider.startConsumer(createConsumer())
-        } catch (e: Throwable) {
-            disconnect()
-            throw e
-        }
-    }
-
-    /**
-     * Disconnect the current network link if it exists.
-     */
-    public fun disconnect() {
-        val link = _link ?: return
-        val opposite = link.opposite(this)
-        _link = null
-        opposite._link = null
-
-        provider.cancel()
-        opposite.provider.cancel()
-    }
-
-    /**
-     * Create a [FlowSource] which generates the outgoing traffic of this port.
-     */
-    protected abstract fun createConsumer(): FlowSource
-
-    /**
-     * The [FlowConsumer] which processes the ingoing traffic of this port.
-     */
-    protected abstract val provider: FlowConsumer
-
-    override fun toString(): String = "SimNetworkPort[isConnected=$isConnected]"
-}
diff --git a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt b/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt
deleted file mode 100644
index c59c44f1..00000000
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSwitchVirtual.kt
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Copyright (c) 2021 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 org.opendc.simulator.flow.FlowConsumer
-import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSource
-import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
-
-/**
- * A [SimNetworkSwitch] that can support new networking ports on demand.
- */
-public class SimNetworkSwitchVirtual(private val engine: FlowEngine) : SimNetworkSwitch {
-    /**
-     * The ports of this switch.
-     */
-    override val ports: List<Port>
-        get() = _ports
-    private val _ports = mutableListOf<Port>()
-
-    /**
-     * The [MaxMinFlowMultiplexer] to actually perform the switching.
-     */
-    private val mux = MaxMinFlowMultiplexer(engine)
-
-    /**
-     * Open a new port on the switch.
-     */
-    public fun newPort(): Port {
-        val port = Port()
-        _ports.add(port)
-        return port
-    }
-
-    /**
-     * A port on the network switch.
-     */
-    public inner class Port : SimNetworkPort(), AutoCloseable {
-        /**
-         * A flag to indicate that this virtual port was removed from the switch.
-         */
-        private var isClosed: Boolean = false
-
-        override val provider: FlowConsumer
-            get() = _provider
-        private val _provider = mux.newInput()
-
-        private val _source = mux.newOutput()
-
-        override fun createConsumer(): FlowSource = _source
-
-        override fun close() {
-            isClosed = true
-            mux.removeInput(_provider)
-            _ports.remove(this)
-        }
-    }
-}
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/main/kotlin/org/opendc/simulator/network/SimNetworkSink.kt b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/TestSource.kt
index 684b4a14..f69db7a2 100644
--- a/opendc-simulator/opendc-simulator-network/src/main/kotlin/org/opendc/simulator/network/SimNetworkSink.kt
+++ b/opendc-simulator/opendc-simulator-network/src/test/kotlin/org/opendc/simulator/network/TestSource.kt
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021 AtLarge Research
+ * 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
@@ -22,26 +22,32 @@
 
 package org.opendc.simulator.network
 
-import org.opendc.simulator.flow.FlowConnection
-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 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 network sink which discards all received traffic and does not generate any traffic itself.
+ * A [SimNetworkPort] that acts as a test source.
  */
-public class SimNetworkSink(
-    engine: FlowEngine,
-    public val capacity: Double
-) : SimNetworkPort() {
-    override fun createConsumer(): FlowSource = object : FlowSource {
-        override fun onPull(conn: FlowConnection, now: Long): Long = Long.MAX_VALUE
-
-        override fun toString(): String = "SimNetworkSink.Consumer"
+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 val provider: FlowConsumer = FlowSink(engine, capacity)
+    override fun onUpdate(ctx: FlowStage, now: Long): Long = Long.MAX_VALUE
+
+    override fun getOutlet(): Outlet = outlet
 
-    override fun toString(): String = "SimNetworkSink[capacity=$capacity]"
+    override fun getInlet(): Inlet = inlet
 }