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).

12 files changed, 611 insertions, 428 deletions

diff --git a/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPdu.java b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPdu.java
new file mode 100644
index 00000000..8790a2d7
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPdu.java
@@ -0,0 +1,141 @@
+/*
+ * 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 org.jetbrains.annotations.NotNull;
+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;
+import org.opendc.simulator.flow2.util.FlowTransform;
+import org.opendc.simulator.flow2.util.FlowTransformer;
+
+/**
+ * A model of a Power Distribution Unit (PDU).
+ */
+public final class SimPdu extends SimPowerInlet {
+    /**
+     * The {@link FlowMultiplexer} that distributes the electricity over the PDU outlets.
+     */
+    private final MaxMinFlowMultiplexer mux;
+
+    /**
+     * A {@link FlowTransformer} that applies the power loss to the PDU's power inlet.
+     */
+    private final FlowTransformer transformer;
+
+    /**
+     * Construct a {@link SimPdu} instance.
+     *
+     * @param graph The underlying {@link FlowGraph} to which the PDU belongs.
+     * @param idlePower The idle power consumption of the PDU independent of the load on the PDU.
+     * @param lossCoefficient The coefficient for the power loss of the PDU proportional to the square load.
+     */
+    public SimPdu(FlowGraph graph, float idlePower, float lossCoefficient) {
+        this.mux = new MaxMinFlowMultiplexer(graph);
+        this.transformer = new FlowTransformer(graph, new FlowTransform() {
+            @Override
+            public float apply(float value) {
+                // See https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
+                return value * (lossCoefficient * value + 1) + idlePower;
+            }
+
+            @Override
+            public float applyInverse(float value) {
+                float c = lossCoefficient;
+                if (c != 0.f) {
+                    return (float) (1 + Math.sqrt(4 * value * c - 4 * idlePower * c + 1)) / (2 * c);
+                } else {
+                    return value - idlePower;
+                }
+            }
+        });
+
+        graph.connect(mux.newOutput(), transformer.getInput());
+    }
+
+    /**
+     * Construct a {@link SimPdu} instance without any loss.
+     *
+     * @param graph The underlying {@link FlowGraph} to which the PDU belongs.
+     */
+    public SimPdu(FlowGraph graph) {
+        this(graph, 0.f, 0.f);
+    }
+
+    /**
+     * Create a new PDU outlet.
+     */
+    public PowerOutlet newOutlet() {
+        return new PowerOutlet(mux);
+    }
+
+    @NotNull
+    @Override
+    public Outlet getFlowOutlet() {
+        return transformer.getOutput();
+    }
+
+    @Override
+    public String toString() {
+        return "SimPdu";
+    }
+
+    /**
+     * A PDU outlet.
+     */
+    public static final class PowerOutlet extends SimPowerOutlet implements AutoCloseable {
+        private final FlowMultiplexer mux;
+        private final Inlet inlet;
+        private boolean isClosed;
+
+        private PowerOutlet(FlowMultiplexer mux) {
+            this.mux = mux;
+            this.inlet = mux.newInput();
+        }
+
+        /**
+         * Remove the outlet from the PDU.
+         */
+        @Override
+        public void close() {
+            isClosed = true;
+            mux.releaseInput(inlet);
+        }
+
+        @Override
+        public String toString() {
+            return "SimPdu.Outlet";
+        }
+
+        @NotNull
+        @Override
+        protected Inlet getFlowInlet() {
+            if (isClosed) {
+                throw new IllegalStateException("Outlet is closed");
+            }
+            return inlet;
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPowerInlet.kt b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerInlet.java
index de587b7f..a6e167c2 100644
--- a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPowerInlet.kt
+++ b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerInlet.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,29 +20,34 @@
  * SOFTWARE.
  */
 
-package org.opendc.simulator.power
+package org.opendc.simulator.power;
 
-import org.opendc.simulator.flow.FlowSource
+import org.opendc.simulator.flow2.Outlet;
 
 /**
  * An abstract inlet that consumes electricity from a power outlet.
  */
 public abstract class SimPowerInlet {
+    SimPowerOutlet outlet;
+
     /**
-     * A flag to indicate that the inlet is currently connected to an outlet.
+     * Determine whether the inlet is connected to a {@link SimPowerOutlet}.
+     *
+     * @return <code>true</code> if the inlet is connected to an outlet, <code>false</code> otherwise.
      */
-    public val isConnected: Boolean
-        get() = _outlet != null
+    public boolean isConnected() {
+        return outlet != null;
+    }
 
     /**
-     * The [SimPowerOutlet] to which the inlet is connected.
+     * Return the {@link SimPowerOutlet} to which the inlet is connected.
      */
-    public val outlet: SimPowerOutlet?
-        get() = _outlet
-    internal var _outlet: SimPowerOutlet? = null
+    public SimPowerOutlet getOutlet() {
+        return outlet;
+    }
 
     /**
-     * Create a [FlowSource] which represents the consumption of electricity from the power outlet.
+     * Return the flow {@link Outlet} that models the consumption of a power inlet as flow output.
      */
-    public abstract fun createSource(): FlowSource
+    protected abstract Outlet getFlowOutlet();
 }
diff --git a/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerOutlet.java b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerOutlet.java
new file mode 100644
index 00000000..e33d35d0
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerOutlet.java
@@ -0,0 +1,91 @@
+/*
+ * 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 org.opendc.simulator.flow2.Inlet;
+import org.opendc.simulator.flow2.Outlet;
+
+/**
+ * An abstract outlet that provides a source of electricity for datacenter components.
+ */
+public abstract class SimPowerOutlet {
+    private SimPowerInlet inlet;
+
+    /**
+     * Determine whether the outlet is connected to a {@link SimPowerInlet}.
+     *
+     * @return <code>true</code> if the outlet is connected to an inlet, <code>false</code> otherwise.
+     */
+    public boolean isConnected() {
+        return inlet != null;
+    }
+
+    /**
+     * Return the {@link SimPowerInlet} to which the outlet is connected.
+     */
+    public SimPowerInlet getInlet() {
+        return inlet;
+    }
+
+    /**
+     * Connect the specified power [inlet] to this outlet.
+     *
+     * @param inlet The inlet to connect to the outlet.
+     */
+    public void connect(SimPowerInlet inlet) {
+        if (isConnected()) {
+            throw new IllegalStateException("Outlet already connected");
+        }
+        if (inlet.isConnected()) {
+            throw new IllegalStateException("Inlet already connected");
+        }
+
+        this.inlet = inlet;
+        this.inlet.outlet = this;
+
+        final Inlet flowInlet = getFlowInlet();
+        final Outlet flowOutlet = inlet.getFlowOutlet();
+
+        flowInlet.getGraph().connect(flowOutlet, flowInlet);
+    }
+
+    /**
+     * Disconnect the connected power outlet from this inlet
+     */
+    public void disconnect() {
+        SimPowerInlet inlet = this.inlet;
+        if (inlet != null) {
+            this.inlet = null;
+            assert inlet.outlet == this : "Inlet state incorrect";
+            inlet.outlet = null;
+
+            final Inlet flowInlet = getFlowInlet();
+            flowInlet.getGraph().disconnect(flowInlet);
+        }
+    }
+
+    /**
+     * Return the flow {@link Inlet} that models the consumption of a power outlet as flow input.
+     */
+    protected abstract Inlet getFlowInlet();
+}
diff --git a/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerSource.java b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerSource.java
new file mode 100644
index 00000000..a2d62c48
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimPowerSource.java
@@ -0,0 +1,71 @@
+/*
+ * 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 org.opendc.simulator.flow2.FlowGraph;
+import org.opendc.simulator.flow2.Inlet;
+import org.opendc.simulator.flow2.sink.SimpleFlowSink;
+
+/**
+ * A {@link SimPowerOutlet} that represents a source of electricity with a maximum capacity.
+ */
+public final class SimPowerSource extends SimPowerOutlet {
+    /**
+     * The resource source that drives this power source.
+     */
+    private final SimpleFlowSink sink;
+
+    /**
+     * Construct a {@link SimPowerSource} instance.
+     *
+     * @param graph The underlying {@link FlowGraph} to which the power source belongs.
+     * @param capacity The maximum amount of power provided by the source.
+     */
+    public SimPowerSource(FlowGraph graph, float capacity) {
+        this.sink = new SimpleFlowSink(graph, capacity);
+    }
+
+    /**
+     * Return the capacity of the power source.
+     */
+    public float getCapacity() {
+        return sink.getCapacity();
+    }
+
+    /**
+     * Return the power draw at this instant.
+     */
+    public float getPowerDraw() {
+        return sink.getRate();
+    }
+
+    @Override
+    protected Inlet getFlowInlet() {
+        return sink.getInput();
+    }
+
+    @Override
+    public String toString() {
+        return "SimPowerSource";
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimUps.java b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimUps.java
new file mode 100644
index 00000000..df7508d9
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-power/src/main/java/org/opendc/simulator/power/SimUps.java
@@ -0,0 +1,137 @@
+/*
+ * 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 org.jetbrains.annotations.NotNull;
+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;
+import org.opendc.simulator.flow2.util.FlowTransform;
+import org.opendc.simulator.flow2.util.FlowTransformer;
+
+/**
+ * A model of an Uninterruptible Power Supply (UPS).
+ * <p>
+ * This model aggregates multiple power sources into a single source in order to ensure that power is always available.
+ */
+public final class SimUps extends SimPowerOutlet {
+    /**
+     * The {@link FlowMultiplexer} that distributes the electricity over the PDU outlets.
+     */
+    private final MaxMinFlowMultiplexer mux;
+
+    /**
+     * A {@link FlowTransformer} that applies the power loss to the PDU's power inlet.
+     */
+    private final FlowTransformer transformer;
+
+    /**
+     * Construct a {@link SimUps} instance.
+     *
+     * @param graph The underlying {@link FlowGraph} to which the UPS belongs.
+     * @param idlePower The idle power consumption of the UPS independent of the load.
+     * @param lossCoefficient The coefficient for the power loss of the UPS proportional to the load.
+     */
+    public SimUps(FlowGraph graph, float idlePower, float lossCoefficient) {
+        this.mux = new MaxMinFlowMultiplexer(graph);
+        this.transformer = new FlowTransformer(graph, new FlowTransform() {
+            @Override
+            public float apply(float value) {
+                // See https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
+                return value * (lossCoefficient + 1) + idlePower;
+            }
+
+            @Override
+            public float applyInverse(float value) {
+                return (value - idlePower) / (lossCoefficient + 1);
+            }
+        });
+
+        graph.connect(transformer.getOutput(), mux.newInput());
+    }
+
+    /**
+     * Construct a {@link SimUps} instance without any loss.
+     *
+     * @param graph The underlying {@link FlowGraph} to which the UPS belongs.
+     */
+    public SimUps(FlowGraph graph) {
+        this(graph, 0.f, 0.f);
+    }
+
+    /**
+     * Create a new UPS inlet.
+     */
+    public PowerInlet newInlet() {
+        return new PowerInlet(mux);
+    }
+
+    @Override
+    protected Inlet getFlowInlet() {
+        return transformer.getInput();
+    }
+
+    @Override
+    public String toString() {
+        return "SimUps";
+    }
+
+    /**
+     * A UPS inlet.
+     */
+    public static final class PowerInlet extends SimPowerInlet implements AutoCloseable {
+        private final FlowMultiplexer mux;
+        private final Outlet outlet;
+        private boolean isClosed;
+
+        private PowerInlet(FlowMultiplexer mux) {
+            this.mux = mux;
+            this.outlet = mux.newOutput();
+        }
+
+        /**
+         * Remove the inlet from the PDU.
+         */
+        @Override
+        public void close() {
+            isClosed = true;
+            mux.releaseOutput(outlet);
+        }
+
+        @Override
+        public String toString() {
+            return "SimPdu.Inlet";
+        }
+
+        @NotNull
+        @Override
+        protected Outlet getFlowOutlet() {
+            if (isClosed) {
+                throw new IllegalStateException("Inlet is closed");
+            }
+            return outlet;
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt b/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt
deleted file mode 100644
index c4076310..00000000
--- a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPdu.kt
+++ /dev/null
@@ -1,95 +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.power
-
-import org.opendc.simulator.flow.FlowConsumer
-import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowMapper
-import org.opendc.simulator.flow.FlowSource
-import org.opendc.simulator.flow.mux.FlowMultiplexer
-import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
-
-/**
- * A model of a Power Distribution Unit (PDU).
- *
- * @param engine The underlying [FlowEngine] to drive the simulation under the hood.
- * @param idlePower The idle power consumption of the PDU independent of the load on the PDU.
- * @param lossCoefficient The coefficient for the power loss of the PDU proportional to the square load.
- */
-public class SimPdu(
-    engine: FlowEngine,
-    private val idlePower: Double = 0.0,
-    private val lossCoefficient: Double = 0.0
-) : SimPowerInlet() {
-    /**
-     * The [FlowMultiplexer] that distributes the electricity over the PDU outlets.
-     */
-    private val mux = MaxMinFlowMultiplexer(engine)
-
-    /**
-     * The [FlowForwarder] that represents the input of the PDU.
-     */
-    private val output = mux.newOutput()
-
-    /**
-     * Create a new PDU outlet.
-     */
-    public fun newOutlet(): Outlet = Outlet(mux, mux.newInput())
-
-    override fun createSource(): FlowSource = FlowMapper(output) { _, rate ->
-        val loss = computePowerLoss(rate)
-        rate + loss
-    }
-
-    override fun toString(): String = "SimPdu"
-
-    /**
-     * Compute the power loss that occurs in the PDU.
-     */
-    private fun computePowerLoss(load: Double): Double {
-        // See https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
-        return idlePower + lossCoefficient * (load * load)
-    }
-
-    /**
-     * A PDU outlet.
-     */
-    public class Outlet(private val switch: FlowMultiplexer, private val provider: FlowConsumer) : SimPowerOutlet(), AutoCloseable {
-        override fun onConnect(inlet: SimPowerInlet) {
-            provider.startConsumer(inlet.createSource())
-        }
-
-        override fun onDisconnect(inlet: SimPowerInlet) {
-            provider.cancel()
-        }
-
-        /**
-         * Remove the outlet from the PDU.
-         */
-        override fun close() {
-            switch.removeInput(provider)
-        }
-
-        override fun toString(): String = "SimPdu.Outlet"
-    }
-}
diff --git a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPowerOutlet.kt b/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPowerOutlet.kt
deleted file mode 100644
index 72f52acc..00000000
--- a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPowerOutlet.kt
+++ /dev/null
@@ -1,80 +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.power
-
-/**
- * An abstract outlet that provides a source of electricity for datacenter components.
- */
-public abstract class SimPowerOutlet {
-    /**
-     * A flag to indicate that the inlet is currently connected to an outlet.
-     */
-    public val isConnected: Boolean
-        get() = _inlet != null
-
-    /**
-     * The inlet that is connected to this outlet currently.
-     */
-    public val inlet: SimPowerInlet?
-        get() = _inlet
-    private var _inlet: SimPowerInlet? = null
-
-    /**
-     * Connect the specified power [inlet] to this outlet.
-     *
-     * @param inlet The inlet to connect to the outlet.
-     */
-    public fun connect(inlet: SimPowerInlet) {
-        check(!isConnected) { "Outlet already connected" }
-        check(!inlet.isConnected) { "Inlet already connected" }
-
-        _inlet = inlet
-        inlet._outlet = this
-
-        onConnect(inlet)
-    }
-
-    /**
-     * Disconnect the connected power outlet from this inlet
-     */
-    public fun disconnect() {
-        val inlet = _inlet
-        if (inlet != null) {
-            _inlet = null
-            assert(inlet._outlet == this) { "Inlet state incorrect" }
-            inlet._outlet = null
-
-            onDisconnect(inlet)
-        }
-    }
-
-    /**
-     * This method is invoked when an inlet is connected to the outlet.
-     */
-    protected abstract fun onConnect(inlet: SimPowerInlet)
-
-    /**
-     * This method is invoked when an inlet is disconnected from the outlet.
-     */
-    protected abstract fun onDisconnect(inlet: SimPowerInlet)
-}
diff --git a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimUps.kt b/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimUps.kt
deleted file mode 100644
index 0431d3cf..00000000
--- a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimUps.kt
+++ /dev/null
@@ -1,101 +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.power
-
-import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowForwarder
-import org.opendc.simulator.flow.FlowMapper
-import org.opendc.simulator.flow.FlowSource
-import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer
-
-/**
- * A model of an Uninterruptible Power Supply (UPS).
- *
- * This model aggregates multiple power sources into a single source in order to ensure that power is always available.
- *
- * @param engine The underlying [FlowEngine] to drive the simulation under the hood.
- * @param idlePower The idle power consumption of the UPS independent of the load.
- * @param lossCoefficient The coefficient for the power loss of the UPS proportional to the load.
- */
-public class SimUps(
-    private val engine: FlowEngine,
-    private val idlePower: Double = 0.0,
-    private val lossCoefficient: Double = 0.0
-) : SimPowerOutlet() {
-    /**
-     * The resource aggregator used to combine the input sources.
-     */
-    private val mux = MaxMinFlowMultiplexer(engine)
-
-    /**
-     * The [FlowConsumer] that represents the output of the UPS.
-     */
-    private val provider = mux.newInput()
-
-    /**
-     * Create a new UPS outlet.
-     */
-    public fun newInlet(): SimPowerInlet {
-        val forward = FlowForwarder(engine, isCoupled = true)
-        forward.startConsumer(mux.newOutput())
-        return Inlet(forward)
-    }
-
-    override fun onConnect(inlet: SimPowerInlet) {
-        val source = inlet.createSource()
-        val mapper = FlowMapper(source) { _, rate ->
-            val loss = computePowerLoss(rate)
-            rate + loss
-        }
-
-        provider.startConsumer(mapper)
-    }
-
-    override fun onDisconnect(inlet: SimPowerInlet) {
-        provider.cancel()
-    }
-
-    /**
-     * Compute the power loss that occurs in the UPS.
-     */
-    private fun computePowerLoss(load: Double): Double {
-        // See https://download.schneider-electric.com/files?p_Doc_Ref=SPD_NRAN-66CK3D_EN
-        return idlePower + lossCoefficient * load
-    }
-
-    /**
-     * A UPS inlet.
-     */
-    public inner class Inlet(private val forwarder: FlowForwarder) : SimPowerInlet(), AutoCloseable {
-        override fun createSource(): FlowSource = forwarder
-
-        /**
-         * Remove the inlet from the PSU.
-         */
-        override fun close() {
-            forwarder.close()
-        }
-
-        override fun toString(): String = "SimPsu.Inlet"
-    }
-}
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/main/kotlin/org/opendc/simulator/power/SimPowerSource.kt b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/TestInlet.kt
index 07e9f52e..7ba12ed9 100644
--- a/opendc-simulator/opendc-simulator-power/src/main/kotlin/org/opendc/simulator/power/SimPowerSource.kt
+++ b/opendc-simulator/opendc-simulator-power/src/test/kotlin/org/opendc/simulator/power/TestInlet.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,33 +22,27 @@
 
 package org.opendc.simulator.power
 
-import org.opendc.simulator.flow.FlowEngine
-import org.opendc.simulator.flow.FlowSink
+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 [SimPowerOutlet] that represents a source of electricity.
- *
- * @param engine The underlying [FlowEngine] to drive the simulation under the hood.
+ * A test inlet.
  */
-public class SimPowerSource(engine: FlowEngine, public val capacity: Double) : SimPowerOutlet() {
-    /**
-     * The resource source that drives this power source.
-     */
-    private val source = FlowSink(engine, capacity)
-
-    /**
-     * The power draw at this instant.
-     */
-    public val powerDraw: Double
-        get() = source.rate
+class TestInlet(graph: FlowGraph) : SimPowerInlet(), FlowStageLogic {
+    val logic = spyk(this)
+    private val stage = graph.newStage(logic)
+    val flowOutlet = stage.getOutlet("out")
 
-    override fun onConnect(inlet: SimPowerInlet) {
-        source.startConsumer(inlet.createSource())
+    init {
+        flowOutlet.push(100.0f)
     }
 
-    override fun onDisconnect(inlet: SimPowerInlet) {
-        source.cancel()
-    }
+    override fun onUpdate(ctx: FlowStage, now: Long): Long = Long.MAX_VALUE
 
-    override fun toString(): String = "SimPowerSource"
+    override fun getFlowOutlet(): Outlet {
+        return flowOutlet
+    }
 }