merge: Migrate to flow-based simulation for low-level models

This pull request converts the `opendc-simulator-resources` module into a flow
simulator and adapts the existing low-level models (e.g., CPU, network, disk)
to this new flow simulator.

The flow simulator works differently from the uniform resource consumption
model, in that it models flow through a system of connections, as opposed to
resource consumptions. Concretely, this means that while in the uniform
resource consumption model, consumptions with the same usage are propagated to
the resources, in the flow simulator, only changes to the flow in the system
are propagated. 

Overall, this leads to less updates in the system and therefore higher
performance. The benchmarks shows that the new implementation obtains more than
double the performance of the old implementation. We have focused in the new
implementation on reducing the amount of work and memory
allocations/loads/stores per updates.

* Migrate from kotlinx-benchmark to jmh-gradle (for better profiling support)
* Use longer traces for benchmarks (to prevent measuring the benchmark
overhead)
* Use direct field access for perf-sensitive code
* Combine work and deadline to duration
* Add support for pushing flow from context (to eliminate the allocation for
every `SimResourceCommand`)
* Reduce memory allocations in SimResourceInterpreter, by revamping the way
timers are allocated.
* Simplify max-min aggregator implementation (by utilizing the new push
mechanism) 
* Invoke consumer callback on every invalidation (in order to propagate changes
downstream)
* Lazily push changes to resource context (by not updating the flow rate
immediately after a push, but only after an update)
* Remove onUpdate callback 
* Merge distributor and aggregator into switch 
* Separate push and pull flags 
* Remove failure callback from FlowSource 
* Create separate callbacks for remaining events 
* Make convergence callback optional
* Reduce field accesses in FlowConsumerContextImpl
* Optimize hot path in SimTraceWorkload
* Expose CPU time counters directly on hypervisor 
* Optimize telemetry collection

**Breaking API Changes**

* The entire `opendc-simulator-resources` module has been replaced by the
`opendc-simulator-flow` module.
* `SimHypervisor.Listener` has been removed in favour of a new interface that
exposes the performance counters of the hypervisor directly. To listen for
convergence, use `FlowConvergenceListener`.

25 files changed, 2903 insertions, 0 deletions

diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/AbstractFlowConsumer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/AbstractFlowConsumer.kt
new file mode 100644
index 00000000..b02426e3
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/AbstractFlowConsumer.kt
@@ -0,0 +1,147 @@
+/*
+ * 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.flow
+
+import org.opendc.simulator.flow.internal.FlowCountersImpl
+
+/**
+ * Abstract implementation of the [FlowConsumer] which can be re-used by other implementations.
+ */
+public abstract class AbstractFlowConsumer(private val engine: FlowEngine, initialCapacity: Double) : FlowConsumer {
+    /**
+     * A flag to indicate that the flow consumer is active.
+     */
+    public override val isActive: Boolean
+        get() = ctx != null
+
+    /**
+     * The capacity of the consumer.
+     */
+    public override var capacity: Double = initialCapacity
+        set(value) {
+            field = value
+            ctx?.capacity = value
+        }
+
+    /**
+     * The current processing rate of the consumer.
+     */
+    public override val rate: Double
+        get() = ctx?.rate ?: 0.0
+
+    /**
+     * The flow processing rate demand at this instant.
+     */
+    public override val demand: Double
+        get() = ctx?.demand ?: 0.0
+
+    /**
+     * The flow counters to track the flow metrics of the consumer.
+     */
+    public override val counters: FlowCounters
+        get() = _counters
+    private val _counters = FlowCountersImpl()
+
+    /**
+     * The [FlowConsumerContext] that is currently running.
+     */
+    protected var ctx: FlowConsumerContext? = null
+        private set
+
+    /**
+     * Construct the [FlowConsumerLogic] instance for a new source.
+     */
+    protected abstract fun createLogic(): FlowConsumerLogic
+
+    /**
+     * Start the specified [FlowConsumerContext].
+     */
+    protected open fun start(ctx: FlowConsumerContext) {
+        ctx.start()
+    }
+
+    /**
+     * The previous demand for the consumer.
+     */
+    private var _previousDemand = 0.0
+    private var _previousCapacity = 0.0
+
+    /**
+     * Update the counters of the flow consumer.
+     */
+    protected fun updateCounters(ctx: FlowConnection, delta: Long) {
+        val demand = _previousDemand
+        val capacity = _previousCapacity
+
+        _previousDemand = ctx.demand
+        _previousCapacity = ctx.capacity
+
+        if (delta <= 0) {
+            return
+        }
+
+        val counters = _counters
+        val deltaS = delta / 1000.0
+        val total = demand * deltaS
+        val work = capacity * deltaS
+        val actualWork = ctx.rate * deltaS
+
+        counters.demand += work
+        counters.actual += actualWork
+        counters.remaining += (total - actualWork)
+    }
+
+    /**
+     * Update the counters of the flow consumer.
+     */
+    protected fun updateCounters(demand: Double, actual: Double, remaining: Double) {
+        val counters = _counters
+        counters.demand += demand
+        counters.actual += actual
+        counters.remaining += remaining
+    }
+
+    final override fun startConsumer(source: FlowSource) {
+        check(ctx == null) { "Consumer is in invalid state" }
+        val ctx = engine.newContext(source, createLogic())
+
+        ctx.capacity = capacity
+        this.ctx = ctx
+
+        start(ctx)
+    }
+
+    final override fun pull() {
+        ctx?.pull()
+    }
+
+    final override fun cancel() {
+        val ctx = ctx
+        if (ctx != null) {
+            this.ctx = null
+            ctx.close()
+        }
+    }
+
+    override fun toString(): String = "AbstractFlowConsumer[capacity=$capacity]"
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConnection.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConnection.kt
new file mode 100644
index 00000000..c327e1e9
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConnection.kt
@@ -0,0 +1,65 @@
+/*
+ * 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.flow
+
+/**
+ * An active connection between a [FlowSource] and [FlowConsumer].
+ */
+public interface FlowConnection : AutoCloseable {
+    /**
+     * The capacity of the connection.
+     */
+    public val capacity: Double
+
+    /**
+     * The flow rate over the connection.
+     */
+    public val rate: Double
+
+    /**
+     * The flow demand of the source.
+     */
+    public val demand: Double
+
+    /**
+     * A flag to control whether [FlowSource.onConverge] should be invoked for this source.
+     */
+    public var shouldSourceConverge: Boolean
+
+    /**
+     * Pull the source.
+     */
+    public fun pull()
+
+    /**
+     * Push the given flow [rate] over this connection.
+     *
+     * @param rate The rate of the flow to push.
+     */
+    public fun push(rate: Double)
+
+    /**
+     * Disconnect the consumer from its source.
+     */
+    public override fun close()
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumer.kt
new file mode 100644
index 00000000..4685a755
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumer.kt
@@ -0,0 +1,131 @@
+/*
+ * 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.flow
+
+import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+
+/**
+ * A consumer of a [FlowSource].
+ */
+public interface FlowConsumer {
+    /**
+     * A flag to indicate that the consumer is currently consuming a [FlowSource].
+     */
+    public val isActive: Boolean
+
+    /**
+     * The flow capacity of this consumer.
+     */
+    public val capacity: Double
+
+    /**
+     * The current flow rate of the consumer.
+     */
+    public val rate: Double
+
+    /**
+     * The current flow demand.
+     */
+    public val demand: Double
+
+    /**
+     * The flow counters to track the flow metrics of the consumer.
+     */
+    public val counters: FlowCounters
+
+    /**
+     * Start consuming the specified [source].
+     *
+     * @throws IllegalStateException if the consumer is already active.
+     */
+    public fun startConsumer(source: FlowSource)
+
+    /**
+     * Ask the consumer to pull its source.
+     *
+     * If the consumer is not active, this operation will be a no-op.
+     */
+    public fun pull()
+
+    /**
+     * Disconnect the consumer from its source.
+     *
+     * If the consumer is not active, this operation will be a no-op.
+     */
+    public fun cancel()
+}
+
+/**
+ * Consume the specified [source] and suspend execution until the source is fully consumed or failed.
+ */
+public suspend fun FlowConsumer.consume(source: FlowSource) {
+    return suspendCancellableCoroutine { cont ->
+        startConsumer(object : FlowSource {
+            override fun onStart(conn: FlowConnection, now: Long) {
+                try {
+                    source.onStart(conn, now)
+                } catch (cause: Throwable) {
+                    cont.resumeWithException(cause)
+                    throw cause
+                }
+            }
+
+            override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+                try {
+                    source.onStop(conn, now, delta)
+
+                    if (!cont.isCompleted) {
+                        cont.resume(Unit)
+                    }
+                } catch (cause: Throwable) {
+                    cont.resumeWithException(cause)
+                    throw cause
+                }
+            }
+
+            override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+                return try {
+                    source.onPull(conn, now, delta)
+                } catch (cause: Throwable) {
+                    cont.resumeWithException(cause)
+                    throw cause
+                }
+            }
+
+            override fun onConverge(conn: FlowConnection, now: Long, delta: Long) {
+                try {
+                    source.onConverge(conn, now, delta)
+                } catch (cause: Throwable) {
+                    cont.resumeWithException(cause)
+                    throw cause
+                }
+            }
+
+            override fun toString(): String = "SuspendingFlowSource"
+        })
+
+        cont.invokeOnCancellation { cancel() }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerContext.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerContext.kt
new file mode 100644
index 00000000..15f9b93b
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerContext.kt
@@ -0,0 +1,50 @@
+/*
+ * 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.flow
+
+/**
+ * A controllable [FlowConnection].
+ *
+ * This interface is used by [FlowConsumer]s to control the connection between it and the source.
+ */
+public interface FlowConsumerContext : FlowConnection {
+    /**
+     * The capacity of the connection.
+     */
+    public override var capacity: Double
+
+    /**
+     * A flag to control whether [FlowConsumerLogic.onConverge] should be invoked for the consumer.
+     */
+    public var shouldConsumerConverge: Boolean
+
+    /**
+     * Start the flow over the connection.
+     */
+    public fun start()
+
+    /**
+     * Synchronously flush the changes of the connection.
+     */
+    public fun flush()
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerLogic.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerLogic.kt
new file mode 100644
index 00000000..50fbc9c7
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerLogic.kt
@@ -0,0 +1,60 @@
+/*
+ * 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.flow
+
+/**
+ * A collection of callbacks associated with a [FlowConsumer].
+ */
+public interface FlowConsumerLogic {
+    /**
+     * This method is invoked when a [FlowSource] changes the rate of flow to this consumer.
+     *
+     * @param ctx The context in which the provider runs.
+     * @param now The virtual timestamp in milliseconds at which the update is occurring.
+     * @param delta The virtual duration between this call and the last call to [onPush] in milliseconds.
+     * @param rate The requested processing rate of the source.
+     */
+    public fun onPush(ctx: FlowConsumerContext, now: Long, delta: Long, rate: Double) {}
+
+    /**
+     * This method is invoked when the flow graph has converged into a steady-state system.
+     *
+     * Make sure to enable [FlowConsumerContext.shouldSourceConverge] if you need this callback. By default, this method
+     * will not be invoked.
+     *
+     * @param ctx The context in which the provider runs.
+     * @param now The virtual timestamp in milliseconds at which the system converged.
+     * @param delta The virtual duration between this call and the last call to [onConverge] in milliseconds.
+     */
+    public fun onConverge(ctx: FlowConsumerContext, now: Long, delta: Long) {}
+
+    /**
+     * This method is invoked when the [FlowSource] completed or failed.
+     *
+     * @param ctx The context in which the provider runs.
+     * @param now The virtual timestamp in milliseconds at which the provider finished.
+     * @param delta The virtual duration between this call and the last call to [onPush] in milliseconds.
+     * @param cause The cause of the failure or `null` if the source completed.
+     */
+    public fun onFinish(ctx: FlowConsumerContext, now: Long, delta: Long, cause: Throwable?) {}
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConvergenceListener.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConvergenceListener.kt
new file mode 100644
index 00000000..d1afda6f
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConvergenceListener.kt
@@ -0,0 +1,36 @@
+/*
+ * 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.flow
+
+/**
+ * A listener interface for when a flow stage has converged into a steady-state.
+ */
+public interface FlowConvergenceListener {
+    /**
+     * This method is invoked when the system has converged to a steady-state.
+     *
+     * @param now The timestamp at which the system converged.
+     * @param delta The virtual duration between this call and the last call to [onConverge] in milliseconds.
+     */
+    public fun onConverge(now: Long, delta: Long) {}
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowCounters.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowCounters.kt
new file mode 100644
index 00000000..a717ae6e
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowCounters.kt
@@ -0,0 +1,53 @@
+/*
+ * 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.flow
+
+/**
+ * An interface that tracks cumulative counts of the flow accumulation over a stage.
+ */
+public interface FlowCounters {
+    /**
+     * The accumulated flow that a source wanted to push over the connection.
+     */
+    public val demand: Double
+
+    /**
+     * The accumulated flow that was actually transferred over the connection.
+     */
+    public val actual: Double
+
+    /**
+     * The amount of capacity that was not utilized.
+     */
+    public val remaining: Double
+
+    /**
+     * The accumulated flow lost due to interference between sources.
+     */
+    public val interference: Double
+
+    /**
+     * Reset the flow counters.
+     */
+    public fun reset()
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowEngine.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowEngine.kt
new file mode 100644
index 00000000..65224827
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowEngine.kt
@@ -0,0 +1,95 @@
+/*
+ * 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.flow
+
+import org.opendc.simulator.flow.internal.FlowEngineImpl
+import java.time.Clock
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * A [FlowEngine] is responsible for managing the interaction between [FlowSource]s and [FlowConsumer]s.
+ *
+ * The engine centralizes the scheduling logic of state updates of flow connections, allowing update propagation
+ * to happen more efficiently. and overall, reducing the work necessary to transition into a steady state.
+ */
+public interface FlowEngine {
+    /**
+     * The virtual [Clock] associated with this engine.
+     */
+    public val clock: Clock
+
+    /**
+     * Create a new [FlowConsumerContext] with the given [provider].
+     *
+     * @param consumer The consumer logic.
+     * @param provider The logic of the resource provider.
+     */
+    public fun newContext(consumer: FlowSource, provider: FlowConsumerLogic): FlowConsumerContext
+
+    /**
+     * Start batching the execution of resource updates until [popBatch] is called.
+     *
+     * This method is useful if you want to propagate multiple resources updates (e.g., starting multiple CPUs
+     * simultaneously) in a single state update.
+     *
+     * Multiple calls to this method requires the same number of [popBatch] calls in order to properly flush the
+     * resource updates. This allows nested calls to [pushBatch], but might cause issues if [popBatch] is not called
+     * the same amount of times. To simplify batching, see [batch].
+     */
+    public fun pushBatch()
+
+    /**
+     * Stop the batching of resource updates and run the interpreter on the batch.
+     *
+     * Note that method will only flush the event once the first call to [pushBatch] has received a [popBatch] call.
+     */
+    public fun popBatch()
+
+    public companion object {
+        /**
+         * Construct a new [FlowEngine] implementation.
+         *
+         * @param context The coroutine context to use.
+         * @param clock The virtual simulation clock.
+         */
+        @JvmStatic
+        @JvmName("create")
+        public operator fun invoke(context: CoroutineContext, clock: Clock): FlowEngine {
+            return FlowEngineImpl(context, clock)
+        }
+    }
+}
+
+/**
+ * Batch the execution of several interrupts into a single call.
+ *
+ * This method is useful if you want to propagate the start of multiple resources (e.g., CPUs) in a single update.
+ */
+public inline fun FlowEngine.batch(block: () -> Unit) {
+    try {
+        pushBatch()
+        block()
+    } finally {
+        popBatch()
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt
new file mode 100644
index 00000000..7eaaf6c2
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt
@@ -0,0 +1,252 @@
+/*
+ * 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.flow
+
+import mu.KotlinLogging
+import org.opendc.simulator.flow.internal.FlowCountersImpl
+import kotlin.math.max
+
+/**
+ * A class that acts as a [FlowSource] and [FlowConsumer] at the same time.
+ *
+ * @param engine The [FlowEngine] the forwarder runs in.
+ * @param isCoupled A flag to indicate that the transformer will exit when the resource consumer exits.
+ */
+public class FlowForwarder(private val engine: FlowEngine, private val isCoupled: Boolean = false) : FlowSource, FlowConsumer, AutoCloseable {
+    /**
+     * The logging instance of this connection.
+     */
+    private val logger = KotlinLogging.logger {}
+
+    /**
+     * The delegate [FlowSource].
+     */
+    private var delegate: FlowSource? = null
+
+    /**
+     * A flag to indicate that the delegate was started.
+     */
+    private var hasDelegateStarted: Boolean = false
+
+    /**
+     * The exposed [FlowConnection].
+     */
+    private val _ctx = object : FlowConnection {
+        override var shouldSourceConverge: Boolean = false
+            set(value) {
+                field = value
+                _innerCtx?.shouldSourceConverge = value
+            }
+
+        override val capacity: Double
+            get() = _innerCtx?.capacity ?: 0.0
+
+        override val demand: Double
+            get() = _innerCtx?.demand ?: 0.0
+
+        override val rate: Double
+            get() = _innerCtx?.rate ?: 0.0
+
+        override fun pull() {
+            _innerCtx?.pull()
+        }
+
+        @JvmField var lastPull = Long.MAX_VALUE
+
+        override fun push(rate: Double) {
+            if (delegate == null) {
+                return
+            }
+
+            _innerCtx?.push(rate)
+            _demand = rate
+        }
+
+        override fun close() {
+            val delegate = delegate ?: return
+            val hasDelegateStarted = hasDelegateStarted
+
+            // Warning: resumption of the continuation might change the entire state of the forwarder. Make sure we
+            // reset beforehand the existing state and check whether it has been updated afterwards
+            reset()
+
+            if (hasDelegateStarted) {
+                val now = engine.clock.millis()
+                val delta = max(0, now - lastPull)
+                delegate.onStop(this, now, delta)
+            }
+        }
+    }
+
+    /**
+     * The [FlowConnection] in which the forwarder runs.
+     */
+    private var _innerCtx: FlowConnection? = null
+
+    override val isActive: Boolean
+        get() = delegate != null
+
+    override val capacity: Double
+        get() = _ctx.capacity
+
+    override val rate: Double
+        get() = _ctx.rate
+
+    override val demand: Double
+        get() = _ctx.demand
+
+    override val counters: FlowCounters
+        get() = _counters
+    private val _counters = FlowCountersImpl()
+
+    override fun startConsumer(source: FlowSource) {
+        check(delegate == null) { "Forwarder already active" }
+
+        delegate = source
+
+        // Pull to replace the source
+        pull()
+    }
+
+    override fun pull() {
+        _ctx.pull()
+    }
+
+    override fun cancel() {
+        _ctx.close()
+    }
+
+    override fun close() {
+        val ctx = _innerCtx
+
+        if (ctx != null) {
+            this._innerCtx = null
+            ctx.pull()
+        }
+    }
+
+    override fun onStart(conn: FlowConnection, now: Long) {
+        _innerCtx = conn
+
+        if (_ctx.shouldSourceConverge) {
+            conn.shouldSourceConverge = true
+        }
+    }
+
+    override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+        _innerCtx = null
+
+        val delegate = delegate
+        if (delegate != null) {
+            reset()
+
+            try {
+                delegate.onStop(this._ctx, now, delta)
+            } catch (cause: Throwable) {
+                logger.error(cause) { "Uncaught exception" }
+            }
+        }
+    }
+
+    override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+        val delegate = delegate
+
+        if (!hasDelegateStarted) {
+            start()
+        }
+
+        _ctx.lastPull = now
+        updateCounters(conn, delta)
+
+        return try {
+            delegate?.onPull(_ctx, now, delta) ?: Long.MAX_VALUE
+        } catch (cause: Throwable) {
+            logger.error(cause) { "Uncaught exception" }
+
+            reset()
+            Long.MAX_VALUE
+        }
+    }
+
+    override fun onConverge(conn: FlowConnection, now: Long, delta: Long) {
+        try {
+            delegate?.onConverge(this._ctx, now, delta)
+        } catch (cause: Throwable) {
+            logger.error(cause) { "Uncaught exception" }
+
+            _innerCtx = null
+            reset()
+        }
+    }
+
+    /**
+     * Start the delegate.
+     */
+    private fun start() {
+        val delegate = delegate ?: return
+
+        try {
+            delegate.onStart(_ctx, engine.clock.millis())
+            hasDelegateStarted = true
+        } catch (cause: Throwable) {
+            logger.error(cause) { "Uncaught exception" }
+            reset()
+        }
+    }
+
+    /**
+     * Reset the delegate.
+     */
+    private fun reset() {
+        if (isCoupled)
+            _innerCtx?.close()
+        else
+            _innerCtx?.push(0.0)
+
+        delegate = null
+        hasDelegateStarted = false
+    }
+
+    /**
+     * The requested flow rate.
+     */
+    private var _demand: Double = 0.0
+
+    /**
+     * Update the flow counters for the transformer.
+     */
+    private fun updateCounters(ctx: FlowConnection, delta: Long) {
+        if (delta <= 0) {
+            return
+        }
+
+        val counters = _counters
+        val deltaS = delta / 1000.0
+        val total = ctx.capacity * deltaS
+        val work = _demand * deltaS
+        val actualWork = ctx.rate * deltaS
+        counters.demand += work
+        counters.actual += actualWork
+        counters.remaining += (total - actualWork)
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowMapper.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowMapper.kt
new file mode 100644
index 00000000..6867bcef
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowMapper.kt
@@ -0,0 +1,75 @@
+/*
+ * 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.flow
+
+/**
+ * A [FlowConsumer] that maps the pushed flow through [transform].
+ *
+ * @param source The source of the flow.
+ * @param transform The method to transform the flow.
+ */
+public class FlowMapper(
+    private val source: FlowSource,
+    private val transform: (FlowConnection, Double) -> Double
+) : FlowSource {
+
+    /**
+     * The current active connection.
+     */
+    private var _conn: Connection? = null
+
+    override fun onStart(conn: FlowConnection, now: Long) {
+        check(_conn == null) { "Concurrent access" }
+        val delegate = Connection(conn, transform)
+        _conn = delegate
+        source.onStart(delegate, now)
+    }
+
+    override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+        val delegate = checkNotNull(_conn) { "Invariant violation" }
+        _conn = null
+        source.onStop(delegate, now, delta)
+    }
+
+    override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+        val delegate = checkNotNull(_conn) { "Invariant violation" }
+        return source.onPull(delegate, now, delta)
+    }
+
+    override fun onConverge(conn: FlowConnection, now: Long, delta: Long) {
+        val delegate = _conn ?: return
+        source.onConverge(delegate, now, delta)
+    }
+
+    /**
+     * The wrapper [FlowConnection] that is used to transform the flow.
+     */
+    private class Connection(
+        private val delegate: FlowConnection,
+        private val transform: (FlowConnection, Double) -> Double
+    ) : FlowConnection by delegate {
+        override fun push(rate: Double) {
+            delegate.push(transform(this, rate))
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSink.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSink.kt
new file mode 100644
index 00000000..b4eb6a38
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSink.kt
@@ -0,0 +1,70 @@
+/*
+ * 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.flow
+
+/**
+ * A [FlowSink] represents a sink with a fixed capacity.
+ *
+ * @param initialCapacity The initial capacity of the resource.
+ * @param engine The engine that is used for driving the flow simulation.
+ * @param parent The parent flow system.
+ */
+public class FlowSink(
+    private val engine: FlowEngine,
+    initialCapacity: Double,
+    private val parent: FlowConvergenceListener? = null
+) : AbstractFlowConsumer(engine, initialCapacity) {
+
+    override fun start(ctx: FlowConsumerContext) {
+        if (parent != null) {
+            ctx.shouldConsumerConverge = true
+        }
+        super.start(ctx)
+    }
+
+    override fun createLogic(): FlowConsumerLogic {
+        return object : FlowConsumerLogic {
+            private val parent = this@FlowSink.parent
+
+            override fun onPush(
+                ctx: FlowConsumerContext,
+                now: Long,
+                delta: Long,
+                rate: Double
+            ) {
+                updateCounters(ctx, delta)
+            }
+
+            override fun onFinish(ctx: FlowConsumerContext, now: Long, delta: Long, cause: Throwable?) {
+                updateCounters(ctx, delta)
+                cancel()
+            }
+
+            override fun onConverge(ctx: FlowConsumerContext, now: Long, delta: Long) {
+                parent?.onConverge(now, delta)
+            }
+        }
+    }
+
+    override fun toString(): String = "FlowSink[capacity=$capacity]"
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSource.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSource.kt
new file mode 100644
index 00000000..3a7e52aa
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSource.kt
@@ -0,0 +1,70 @@
+/*
+ * 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.flow
+
+/**
+ * A source of flow that is consumed by a [FlowConsumer].
+ *
+ * Implementations of this interface should be considered stateful and must be assumed not to be re-usable
+ * (concurrently) for multiple [FlowConsumer]s, unless explicitly said otherwise.
+ */
+public interface FlowSource {
+    /**
+     * This method is invoked when the source is started.
+     *
+     * @param conn The connection between the source and consumer.
+     * @param now The virtual timestamp in milliseconds at which the provider finished.
+     */
+    public fun onStart(conn: FlowConnection, now: Long) {}
+
+    /**
+     * This method is invoked when the source is finished.
+     *
+     * @param conn The connection between the source and consumer.
+     * @param now The virtual timestamp in milliseconds at which the source finished.
+     * @param delta The virtual duration between this call and the last call to [onPull] in milliseconds.
+     */
+    public fun onStop(conn: FlowConnection, now: Long, delta: Long) {}
+
+    /**
+     * This method is invoked when the source is pulled.
+     *
+     * @param conn The connection between the source and consumer.
+     * @param now The virtual timestamp in milliseconds at which the pull is occurring.
+     * @param delta The virtual duration between this call and the last call to [onPull] in milliseconds.
+     * @return The duration after which the resource consumer should be pulled again.
+     */
+    public fun onPull(conn: FlowConnection, now: Long, delta: Long): Long
+
+    /**
+     * This method is invoked when the flow graph has converged into a steady-state system.
+     *
+     * Make sure to enable [FlowConnection.shouldSourceConverge] if you need this callback. By default, this method
+     * will not be invoked.
+     *
+     * @param conn The connection between the source and consumer.
+     * @param now The virtual timestamp in milliseconds at which the system converged.
+     * @param delta The virtual duration between this call and the last call to [onConverge] in milliseconds.
+     */
+    public fun onConverge(conn: FlowConnection, now: Long, delta: Long) {}
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceDomain.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceDomain.kt
new file mode 100644
index 00000000..aa2713b6
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceDomain.kt
@@ -0,0 +1,19 @@
+package org.opendc.simulator.flow.interference
+
+import org.opendc.simulator.flow.FlowSource
+
+/**
+ * An interference domain represents a system of flow stages where [flow sources][FlowSource] may incur
+ * performance variability due to operating on the same resource and therefore causing interference.
+ */
+public interface InterferenceDomain {
+    /**
+     * Compute the performance score of a participant in this interference domain.
+     *
+     * @param key The participant to obtain the score of or `null` if the participant has no key.
+     * @param load The overall load on the interference domain.
+     * @return A score representing the performance score to be applied to the resource consumer, with 1
+     * meaning no influence, <1 means that performance degrades, and >1 means that performance improves.
+     */
+    public fun apply(key: InterferenceKey?, load: Double): Double
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceKey.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceKey.kt
new file mode 100644
index 00000000..d28ebde5
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceKey.kt
@@ -0,0 +1,28 @@
+/*
+ * 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.flow.interference
+
+/**
+ * A key that uniquely identifies a participant of an interference domain.
+ */
+public interface InterferenceKey
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Flags.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Flags.kt
new file mode 100644
index 00000000..81ed9f26
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Flags.kt
@@ -0,0 +1,43 @@
+/*
+ * 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.flow.internal
+
+/**
+ * States of the flow connection.
+ */
+internal const val ConnPending = 0 // Connection is pending and the consumer is waiting to consume the source
+internal const val ConnActive = 1 // Connection is active and the source is currently being consumed
+internal const val ConnClosed = 2 // Connection is closed and source cannot be consumed through this connection anymore
+internal const val ConnState = 0b11 // Mask for accessing the state of the flow connection
+
+/**
+ * Flags of the flow connection
+ */
+internal const val ConnPulled = 1 shl 2 // The source should be pulled
+internal const val ConnPushed = 1 shl 3 // The source has pushed a value
+internal const val ConnUpdateActive = 1 shl 4 // An update for the connection is active
+internal const val ConnUpdatePending = 1 shl 5 // An (immediate) update of the connection is pending
+internal const val ConnUpdateSkipped = 1 shl 6 // An update of the connection was not necessary
+internal const val ConnConvergePending = 1 shl 7 // Indication that a convergence is already pending
+internal const val ConnConvergeSource = 1 shl 8 // Enable convergence of the source
+internal const val ConnConvergeConsumer = 1 shl 9 // Enable convergence of the consumer
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt
new file mode 100644
index 00000000..9d36483e
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt
@@ -0,0 +1,419 @@
+/*
+ * 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.flow.internal
+
+import mu.KotlinLogging
+import org.opendc.simulator.flow.*
+import java.util.*
+import kotlin.math.max
+import kotlin.math.min
+
+/**
+ * Implementation of a [FlowConnection] managing the communication between flow sources and consumers.
+ */
+internal class FlowConsumerContextImpl(
+    private val engine: FlowEngineImpl,
+    private val source: FlowSource,
+    private val logic: FlowConsumerLogic
+) : FlowConsumerContext {
+    /**
+     * The logging instance of this connection.
+     */
+    private val logger = KotlinLogging.logger {}
+
+    /**
+     * The capacity of the connection.
+     */
+    override var capacity: Double
+        get() = _capacity
+        set(value) {
+            val oldValue = _capacity
+
+            // Only changes will be propagated
+            if (value != oldValue) {
+                _capacity = value
+                pull()
+            }
+        }
+    private var _capacity: Double = 0.0
+
+    /**
+     * The current processing rate of the connection.
+     */
+    override val rate: Double
+        get() = _rate
+    private var _rate = 0.0
+
+    /**
+     * The current flow processing demand.
+     */
+    override val demand: Double
+        get() = _demand
+
+    /**
+     * Flags to control the convergence of the consumer and source.
+     */
+    override var shouldSourceConverge: Boolean = false
+        set(value) {
+            field = value
+            _flags =
+                if (value)
+                    _flags or ConnConvergeSource
+                else
+                    _flags and ConnConvergeSource.inv()
+        }
+    override var shouldConsumerConverge: Boolean = false
+        set(value) {
+            field = value
+
+            _flags =
+                if (value)
+                    _flags or ConnConvergeConsumer
+                else
+                    _flags and ConnConvergeConsumer.inv()
+        }
+
+    /**
+     * The clock to track simulation time.
+     */
+    private val _clock = engine.clock
+
+    /**
+     * The current state of the connection.
+     */
+    private var _demand: Double = 0.0 // The current (pending) demand of the source
+    private var _deadline: Long = Long.MAX_VALUE // The deadline of the source's timer
+
+    /**
+     * The flags of the flow connection, indicating certain actions.
+     */
+    private var _flags: Int = 0
+
+    /**
+     * The timestamp of calls to the callbacks.
+     */
+    private var _lastPull: Long = Long.MIN_VALUE // Last call to `onPull`
+    private var _lastPush: Long = Long.MIN_VALUE // Last call to `onPush`
+    private var _lastSourceConvergence: Long = Long.MAX_VALUE // Last call to source `onConvergence`
+    private var _lastConsumerConvergence: Long = Long.MAX_VALUE // Last call to consumer `onConvergence`
+
+    /**
+     * The timers at which the context is scheduled to be interrupted.
+     */
+    private var _timer: FlowEngineImpl.Timer? = null
+    private val _pendingTimers: ArrayDeque<FlowEngineImpl.Timer> = ArrayDeque(5)
+
+    override fun start() {
+        check(_flags and ConnState == ConnPending) { "Consumer is already started" }
+        engine.batch {
+            val now = _clock.millis()
+            source.onStart(this, now)
+
+            // Mark the connection as active and pulled
+            val newFlags = (_flags and ConnState.inv()) or ConnActive or ConnPulled
+            scheduleImmediate(now, newFlags)
+        }
+    }
+
+    override fun close() {
+        var flags = _flags
+        if (flags and ConnState == ConnClosed) {
+            return
+        }
+
+        engine.batch {
+            // Mark the connection as closed and pulled (in order to converge)
+            flags = (flags and ConnState.inv()) or ConnClosed or ConnPulled
+            _flags = flags
+
+            if (flags and ConnUpdateActive == 0) {
+                val now = _clock.millis()
+                doStopSource(now)
+
+                // FIX: Make sure the context converges
+                scheduleImmediate(now, flags)
+            }
+        }
+    }
+
+    override fun pull() {
+        val flags = _flags
+        if (flags and ConnState != ConnActive) {
+            return
+        }
+
+        // Mark connection as pulled
+        scheduleImmediate(_clock.millis(), flags or ConnPulled)
+    }
+
+    override fun flush() {
+        val flags = _flags
+
+        // Do not attempt to flush the connection if the connection is closed or an update is already active
+        if (flags and ConnState != ConnActive || flags and ConnUpdateActive != 0) {
+            return
+        }
+
+        engine.scheduleSync(_clock.millis(), this)
+    }
+
+    override fun push(rate: Double) {
+        if (_demand == rate) {
+            return
+        }
+
+        _demand = rate
+
+        val flags = _flags
+
+        if (flags and ConnUpdateActive != 0) {
+            // If an update is active, it will already get picked up at the end of the update
+            _flags = flags or ConnPushed
+        } else {
+            // Invalidate only if no update is active
+            scheduleImmediate(_clock.millis(), flags or ConnPushed)
+        }
+    }
+
+    /**
+     * Update the state of the flow connection.
+     *
+     * @param now The current virtual timestamp.
+     * @param visited The queue of connections that have been visited during the cycle.
+     * @param timerQueue The queue of all pending timers.
+     * @param isImmediate A flag to indicate that this invocation is an immediate update or a delayed update.
+     */
+    fun doUpdate(
+        now: Long,
+        visited: ArrayDeque<FlowConsumerContextImpl>,
+        timerQueue: PriorityQueue<FlowEngineImpl.Timer>,
+        isImmediate: Boolean
+    ) {
+        var flags = _flags
+
+        // Precondition: The flow connection must be active
+        if (flags and ConnState != ConnActive) {
+            return
+        }
+
+        val deadline = _deadline
+        val reachedDeadline = deadline == now
+        var newDeadline = deadline
+        var hasUpdated = false
+
+        try {
+            // Pull the source if (1) `pull` is called or (2) the timer of the source has expired
+            newDeadline = if (flags and ConnPulled != 0 || reachedDeadline) {
+                val lastPull = _lastPull
+                val delta = max(0, now - lastPull)
+
+                // Update state before calling into the outside world, so it observes a consistent state
+                _lastPull = now
+                _flags = (flags and ConnPulled.inv()) or ConnUpdateActive
+                hasUpdated = true
+
+                val duration = source.onPull(this, now, delta)
+
+                // IMPORTANT: Re-fetch the flags after the callback might have changed those
+                flags = _flags
+
+                if (duration != Long.MAX_VALUE)
+                    now + duration
+                else
+                    duration
+            } else {
+                deadline
+            }
+
+            // Push to the consumer if the rate of the source has changed (after a call to `push`)
+            val newState = flags and ConnState
+            if (newState == ConnActive && flags and ConnPushed != 0) {
+                val lastPush = _lastPush
+                val delta = max(0, now - lastPush)
+
+                // Update state before calling into the outside world, so it observes a consistent state
+                _lastPush = now
+                _flags = (flags and ConnPushed.inv()) or ConnUpdateActive
+                hasUpdated = true
+
+                logic.onPush(this, now, delta, _demand)
+
+                // IMPORTANT: Re-fetch the flags after the callback might have changed those
+                flags = _flags
+            } else if (newState == ConnClosed) {
+                hasUpdated = true
+
+                // The source has called [FlowConnection.close], so clean up the connection
+                doStopSource(now)
+            }
+        } catch (cause: Throwable) {
+            // Mark the connection as closed
+            flags = (flags and ConnState.inv()) or ConnClosed
+
+            doFailSource(now, cause)
+        }
+
+        // Check whether the connection needs to be added to the visited queue. This is the case when:
+        //  (1) An update was performed (either a push or a pull)
+        //  (2) Either the source or consumer want to converge, and
+        //  (3) Convergence is not already pending (ConnConvergePending)
+        if (hasUpdated && flags and (ConnConvergeSource or ConnConvergeConsumer) != 0 && flags and ConnConvergePending == 0) {
+            visited.add(this)
+            flags = flags or ConnConvergePending
+        }
+
+        // Compute the new flow rate of the connection
+        // Note: _demand might be changed by [logic.onConsume], so we must re-fetch the value
+        _rate = min(_capacity, _demand)
+
+        // Indicate that no update is active anymore and flush the flags
+        _flags = flags and ConnUpdateActive.inv() and ConnUpdatePending.inv()
+
+        val pendingTimers = _pendingTimers
+
+        // Prune the head timer if this is a delayed update
+        val timer = if (!isImmediate) {
+            // Invariant: Any pending timer should only point to a future timestamp
+            // See also `scheduleDelayed`
+            val timer = pendingTimers.poll()
+            _timer = timer
+            timer
+        } else {
+            _timer
+        }
+
+        // Set the new deadline and schedule a delayed update for that deadline
+        _deadline = newDeadline
+
+        // Check whether we need to schedule a new timer for this connection. That is the case when:
+        // (1) The deadline is valid (not the maximum value)
+        // (2) The connection is active
+        // (3) The current active timer for the connection points to a later deadline
+        if (newDeadline == Long.MAX_VALUE || flags and ConnState != ConnActive || (timer != null && newDeadline >= timer.target)) {
+            // Ignore any deadline scheduled at the maximum value
+            // This indicates that the source does not want to register a timer
+            return
+        }
+
+        // Construct a timer for the new deadline and add it to the global queue of timers
+        val newTimer = FlowEngineImpl.Timer(this, newDeadline)
+        _timer = newTimer
+        timerQueue.add(newTimer)
+
+        // A timer already exists for this connection, so add it to the queue of pending timers
+        if (timer != null) {
+            pendingTimers.addFirst(timer)
+        }
+    }
+
+    /**
+     * This method is invoked when the system converges into a steady state.
+     */
+    fun onConverge(now: Long) {
+        try {
+            val flags = _flags
+
+            // The connection is converging now, so unset the convergence pending flag
+            _flags = flags and ConnConvergePending.inv()
+
+            // Call the source converge callback if it has enabled convergence and the connection is active
+            if (flags and ConnState == ConnActive && flags and ConnConvergeSource != 0) {
+                val delta = max(0, now - _lastSourceConvergence)
+                _lastSourceConvergence = now
+
+                source.onConverge(this, now, delta)
+            }
+
+            // Call the consumer callback if it has enabled convergence
+            if (flags and ConnConvergeConsumer != 0) {
+                val delta = max(0, now - _lastConsumerConvergence)
+                _lastConsumerConvergence = now
+
+                logic.onConverge(this, now, delta)
+            }
+        } catch (cause: Throwable) {
+            doFailSource(now, cause)
+        }
+    }
+
+    override fun toString(): String = "FlowConsumerContextImpl[capacity=$capacity,rate=$_rate]"
+
+    /**
+     * Stop the [FlowSource].
+     */
+    private fun doStopSource(now: Long) {
+        try {
+            source.onStop(this, now, max(0, now - _lastPull))
+            doFinishConsumer(now, null)
+        } catch (cause: Throwable) {
+            doFinishConsumer(now, cause)
+            return
+        } finally {
+            _deadline = Long.MAX_VALUE
+            _demand = 0.0
+        }
+    }
+
+    /**
+     * Fail the [FlowSource].
+     */
+    private fun doFailSource(now: Long, cause: Throwable) {
+        try {
+            source.onStop(this, now, max(0, now - _lastPull))
+        } catch (e: Throwable) {
+            e.addSuppressed(cause)
+            doFinishConsumer(now, e)
+        } finally {
+            _deadline = Long.MAX_VALUE
+            _demand = 0.0
+        }
+    }
+
+    /**
+     * Finish the consumer.
+     */
+    private fun doFinishConsumer(now: Long, cause: Throwable?) {
+        try {
+            logic.onFinish(this, now, max(0, now - _lastPush), cause)
+        } catch (e: Throwable) {
+            e.addSuppressed(cause)
+            logger.error(e) { "Uncaught exception" }
+        }
+    }
+
+    /**
+     * Schedule an immediate update for this connection.
+     */
+    private fun scheduleImmediate(now: Long, flags: Int) {
+        // In case an immediate update is already scheduled, no need to do anything
+        if (flags and ConnUpdatePending != 0) {
+            _flags = flags
+            return
+        }
+
+        // Mark the connection that there is an update pending
+        _flags = flags or ConnUpdatePending
+
+        engine.scheduleImmediate(now, this)
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowCountersImpl.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowCountersImpl.kt
new file mode 100644
index 00000000..d2fa5228
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowCountersImpl.kt
@@ -0,0 +1,46 @@
+/*
+ * 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.flow.internal
+
+import org.opendc.simulator.flow.FlowCounters
+
+/**
+ * Mutable implementation of the [FlowCounters] interface.
+ */
+internal class FlowCountersImpl : FlowCounters {
+    override var demand: Double = 0.0
+    override var actual: Double = 0.0
+    override var remaining: Double = 0.0
+    override var interference: Double = 0.0
+
+    override fun reset() {
+        demand = 0.0
+        actual = 0.0
+        remaining = 0.0
+        interference = 0.0
+    }
+
+    override fun toString(): String {
+        return "FlowCounters[demand=$demand,actual=$actual,remaining=$remaining,interference=$interference]"
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt
new file mode 100644
index 00000000..019b5f10
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt
@@ -0,0 +1,247 @@
+/*
+ * 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.flow.internal
+
+import kotlinx.coroutines.Delay
+import kotlinx.coroutines.DisposableHandle
+import kotlinx.coroutines.InternalCoroutinesApi
+import kotlinx.coroutines.Runnable
+import org.opendc.simulator.flow.*
+import java.time.Clock
+import java.util.*
+import kotlin.coroutines.ContinuationInterceptor
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * Internal implementation of the [FlowEngine] interface.
+ *
+ * @param context The coroutine context to use.
+ * @param clock The virtual simulation clock.
+ */
+internal class FlowEngineImpl(private val context: CoroutineContext, override val clock: Clock) : FlowEngine {
+    /**
+     * The [Delay] instance that provides scheduled execution of [Runnable]s.
+     */
+    @OptIn(InternalCoroutinesApi::class)
+    private val delay = requireNotNull(context[ContinuationInterceptor] as? Delay) { "Invalid CoroutineDispatcher: no delay implementation" }
+
+    /**
+     * The queue of connection updates that are scheduled for immediate execution.
+     */
+    private val queue = ArrayDeque<FlowConsumerContextImpl>()
+
+    /**
+     * A priority queue containing the connection updates to be scheduled in the future.
+     */
+    private val futureQueue = PriorityQueue<Timer>()
+
+    /**
+     * The stack of engine invocations to occur in the future.
+     */
+    private val futureInvocations = ArrayDeque<Invocation>()
+
+    /**
+     * The systems that have been visited during the engine cycle.
+     */
+    private val visited: ArrayDeque<FlowConsumerContextImpl> = ArrayDeque()
+
+    /**
+     * The index in the batch stack.
+     */
+    private var batchIndex = 0
+
+    /**
+     * Update the specified [ctx] synchronously.
+     */
+    fun scheduleSync(now: Long, ctx: FlowConsumerContextImpl) {
+        ctx.doUpdate(now, visited, futureQueue, isImmediate = true)
+
+        // In-case the engine is already running in the call-stack, return immediately. The changes will be picked
+        // up by the active engine.
+        if (batchIndex > 0) {
+            return
+        }
+
+        runEngine(now)
+    }
+
+    /**
+     * Enqueue the specified [ctx] to be updated immediately during the active engine cycle.
+     *
+     * This method should be used when the state of a flow context is invalidated/interrupted and needs to be
+     * re-computed. In case no engine is currently active, the engine will be started.
+     */
+    fun scheduleImmediate(now: Long, ctx: FlowConsumerContextImpl) {
+        queue.add(ctx)
+
+        // In-case the engine is already running in the call-stack, return immediately. The changes will be picked
+        // up by the active engine.
+        if (batchIndex > 0) {
+            return
+        }
+
+        runEngine(now)
+    }
+
+    override fun newContext(consumer: FlowSource, provider: FlowConsumerLogic): FlowConsumerContext = FlowConsumerContextImpl(this, consumer, provider)
+
+    override fun pushBatch() {
+        batchIndex++
+    }
+
+    override fun popBatch() {
+        try {
+            // Flush the work if the engine is not already running
+            if (batchIndex == 1 && queue.isNotEmpty()) {
+                doRunEngine(clock.millis())
+            }
+        } finally {
+            batchIndex--
+        }
+    }
+
+    /**
+     * Run the engine and mark as active while running.
+     */
+    private fun runEngine(now: Long) {
+        try {
+            batchIndex++
+            doRunEngine(now)
+        } finally {
+            batchIndex--
+        }
+    }
+
+    /**
+     * Run all the enqueued actions for the specified [timestamp][now].
+     */
+    private fun doRunEngine(now: Long) {
+        val queue = queue
+        val futureQueue = futureQueue
+        val futureInvocations = futureInvocations
+        val visited = visited
+
+        // Remove any entries in the `futureInvocations` queue from the past
+        while (true) {
+            val head = futureInvocations.peek()
+            if (head == null || head.timestamp > now) {
+                break
+            }
+            futureInvocations.poll()
+        }
+
+        // Execute all scheduled updates at current timestamp
+        while (true) {
+            val timer = futureQueue.peek() ?: break
+            val target = timer.target
+
+            if (target > now) {
+                break
+            }
+
+            assert(target >= now) { "Internal inconsistency: found update of the past" }
+
+            futureQueue.poll()
+            timer.ctx.doUpdate(now, visited, futureQueue, isImmediate = false)
+        }
+
+        // Repeat execution of all immediate updates until the system has converged to a steady-state
+        // We have to take into account that the onConverge callback can also trigger new actions.
+        do {
+            // Execute all immediate updates
+            while (true) {
+                val ctx = queue.poll() ?: break
+                ctx.doUpdate(now, visited, futureQueue, isImmediate = true)
+            }
+
+            while (true) {
+                val ctx = visited.poll() ?: break
+                ctx.onConverge(now)
+            }
+        } while (queue.isNotEmpty())
+
+        // Schedule an engine invocation for the next update to occur.
+        val headTimer = futureQueue.peek()
+        if (headTimer != null) {
+            trySchedule(now, futureInvocations, headTimer.target)
+        }
+    }
+
+    /**
+     * Try to schedule an engine invocation at the specified [target].
+     *
+     * @param now The current virtual timestamp.
+     * @param target The virtual timestamp at which the engine invocation should happen.
+     * @param scheduled The queue of scheduled invocations.
+     */
+    private fun trySchedule(now: Long, scheduled: ArrayDeque<Invocation>, target: Long) {
+        while (true) {
+            val invocation = scheduled.peekFirst()
+            if (invocation == null || invocation.timestamp > target) {
+                // Case 2: A new timer was registered ahead of the other timers.
+                // Solution: Schedule a new scheduler invocation
+                @OptIn(InternalCoroutinesApi::class)
+                val handle = delay.invokeOnTimeout(target - now, { runEngine(target) }, context)
+                scheduled.addFirst(Invocation(target, handle))
+                break
+            } else if (invocation.timestamp < target) {
+                // Case 2: A timer was cancelled and the head of the timer queue is now later than excepted
+                // Solution: Cancel the next scheduler invocation
+                scheduled.pollFirst()
+
+                invocation.cancel()
+            } else {
+                break
+            }
+        }
+    }
+
+    /**
+     * A future engine invocation.
+     *
+     * This class is used to keep track of the future engine invocations created using the [Delay] instance. In case
+     * the invocation is not needed anymore, it can be cancelled via [cancel].
+     */
+    private class Invocation(
+        @JvmField val timestamp: Long,
+        @JvmField val handle: DisposableHandle
+    ) {
+        /**
+         * Cancel the engine invocation.
+         */
+        fun cancel() = handle.dispose()
+    }
+
+    /**
+     * An update call for [ctx] that is scheduled for [target].
+     *
+     * This class represents an update in the future at [target] requested by [ctx].
+     */
+    class Timer(@JvmField val ctx: FlowConsumerContextImpl, @JvmField val target: Long) : Comparable<Timer> {
+        override fun compareTo(other: Timer): Int {
+            return target.compareTo(other.target)
+        }
+
+        override fun toString(): String = "Timer[ctx=$ctx,timestamp=$target]"
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt
new file mode 100644
index 00000000..04ba7f21
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt
@@ -0,0 +1,100 @@
+/*
+ * 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.flow.mux
+
+import org.opendc.simulator.flow.FlowConsumer
+import org.opendc.simulator.flow.FlowCounters
+import org.opendc.simulator.flow.FlowSource
+import org.opendc.simulator.flow.interference.InterferenceKey
+
+/**
+ * A [FlowMultiplexer] enables multiplexing multiple [FlowSource]s over possibly multiple [FlowConsumer]s.
+ */
+public interface FlowMultiplexer {
+    /**
+     * The inputs of the multiplexer that can be used to consume sources.
+     */
+    public val inputs: Set<FlowConsumer>
+
+    /**
+     * The outputs of the multiplexer over which the flows will be distributed.
+     */
+    public val outputs: Set<FlowSource>
+
+    /**
+     * The actual processing rate of the multiplexer.
+     */
+    public val rate: Double
+
+    /**
+     * The demanded processing rate of the input.
+     */
+    public val demand: Double
+
+    /**
+     * The capacity of the outputs.
+     */
+    public val capacity: Double
+
+    /**
+     * The flow counters to track the flow metrics of all multiplexer inputs.
+     */
+    public val counters: FlowCounters
+
+    /**
+     * Create a new input on this multiplexer.
+     *
+     * @param key The key of the interference member to which the input belongs.
+     */
+    public fun newInput(key: InterferenceKey? = null): FlowConsumer
+
+    /**
+     * Remove [input] from this multiplexer.
+     */
+    public fun removeInput(input: FlowConsumer)
+
+    /**
+     * Create a new output on this multiplexer.
+     */
+    public fun newOutput(): FlowSource
+
+    /**
+     * Remove [output] from this multiplexer.
+     */
+    public fun removeOutput(output: FlowSource)
+
+    /**
+     * Clear all inputs and outputs from the multiplexer.
+     */
+    public fun clear()
+
+    /**
+     * Clear the inputs of the multiplexer.
+     */
+    public fun clearInputs()
+
+    /**
+     * Clear the outputs of the multiplexer.
+     */
+    public fun clearOutputs()
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt
new file mode 100644
index 00000000..125d10fe
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt
@@ -0,0 +1,154 @@
+/*
+ * 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.flow.mux
+
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.interference.InterferenceKey
+import java.util.ArrayDeque
+
+/**
+ * A [FlowMultiplexer] implementation that allocates inputs to the outputs of the multiplexer exclusively. This means
+ * that a single input is directly connected to an output and that the multiplexer can only support as many
+ * inputs as outputs.
+ *
+ * @param engine The [FlowEngine] driving the simulation.
+ */
+public class ForwardingFlowMultiplexer(private val engine: FlowEngine) : FlowMultiplexer {
+    override val inputs: Set<FlowConsumer>
+        get() = _inputs
+    private val _inputs = mutableSetOf<Input>()
+
+    override val outputs: Set<FlowSource>
+        get() = _outputs
+    private val _outputs = mutableSetOf<Output>()
+    private val _availableOutputs = ArrayDeque<Output>()
+
+    override val counters: FlowCounters = object : FlowCounters {
+        override val demand: Double
+            get() = _outputs.sumOf { it.forwarder.counters.demand }
+        override val actual: Double
+            get() = _outputs.sumOf { it.forwarder.counters.actual }
+        override val remaining: Double
+            get() = _outputs.sumOf { it.forwarder.counters.remaining }
+        override val interference: Double
+            get() = _outputs.sumOf { it.forwarder.counters.interference }
+
+        override fun reset() {
+            for (output in _outputs) {
+                output.forwarder.counters.reset()
+            }
+        }
+
+        override fun toString(): String = "FlowCounters[demand=$demand,actual=$actual,remaining=$remaining]"
+    }
+
+    override val rate: Double
+        get() = _outputs.sumOf { it.forwarder.rate }
+
+    override val demand: Double
+        get() = _outputs.sumOf { it.forwarder.demand }
+
+    override val capacity: Double
+        get() = _outputs.sumOf { it.forwarder.capacity }
+
+    override fun newInput(key: InterferenceKey?): FlowConsumer {
+        val output = checkNotNull(_availableOutputs.poll()) { "No capacity to serve request" }
+        val input = Input(output)
+        _inputs += input
+        return input
+    }
+
+    override fun removeInput(input: FlowConsumer) {
+        if (!_inputs.remove(input)) {
+            return
+        }
+
+        val output = (input as Input).output
+        output.forwarder.cancel()
+        _availableOutputs += output
+    }
+
+    override fun newOutput(): FlowSource {
+        val forwarder = FlowForwarder(engine)
+        val output = Output(forwarder)
+
+        _outputs += output
+        return output
+    }
+
+    override fun removeOutput(output: FlowSource) {
+        if (!_outputs.remove(output)) {
+            return
+        }
+
+        val forwarder = (output as Output).forwarder
+        forwarder.close()
+    }
+
+    override fun clearInputs() {
+        for (input in _inputs) {
+            val output = input.output
+            output.forwarder.cancel()
+            _availableOutputs += output
+        }
+
+        _inputs.clear()
+    }
+
+    override fun clearOutputs() {
+        for (output in _outputs) {
+            output.forwarder.cancel()
+        }
+        _outputs.clear()
+        _availableOutputs.clear()
+    }
+
+    override fun clear() {
+        clearOutputs()
+        clearInputs()
+    }
+
+    /**
+     * An input on the multiplexer.
+     */
+    private inner class Input(@JvmField val output: Output) : FlowConsumer by output.forwarder {
+        override fun toString(): String = "ForwardingFlowMultiplexer.Input"
+    }
+
+    /**
+     * An output on the multiplexer.
+     */
+    private inner class Output(@JvmField val forwarder: FlowForwarder) : FlowSource by forwarder {
+        override fun onStart(conn: FlowConnection, now: Long) {
+            _availableOutputs += this
+            forwarder.onStart(conn, now)
+        }
+
+        override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+            forwarder.cancel()
+            forwarder.onStop(conn, now, delta)
+        }
+
+        override fun toString(): String = "ForwardingFlowMultiplexer.Output"
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt
new file mode 100644
index 00000000..5ff0fb8d
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt
@@ -0,0 +1,490 @@
+/*
+ * 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.flow.mux
+
+import org.opendc.simulator.flow.*
+import org.opendc.simulator.flow.interference.InterferenceDomain
+import org.opendc.simulator.flow.interference.InterferenceKey
+import org.opendc.simulator.flow.internal.FlowCountersImpl
+import kotlin.math.max
+import kotlin.math.min
+
+/**
+ * A [FlowMultiplexer] implementation that multiplexes flows over the available outputs using max-min fair sharing.
+ *
+ * @param engine The [FlowEngine] to drive the flow simulation.
+ * @param parent The parent flow system of the multiplexer.
+ * @param interferenceDomain The interference domain of the multiplexer.
+ */
+public class MaxMinFlowMultiplexer(
+    private val engine: FlowEngine,
+    private val parent: FlowConvergenceListener? = null,
+    private val interferenceDomain: InterferenceDomain? = null
+) : FlowMultiplexer {
+    /**
+     * The inputs of the multiplexer.
+     */
+    override val inputs: Set<FlowConsumer>
+        get() = _inputs
+    private val _inputs = mutableSetOf<Input>()
+    private val _activeInputs = mutableListOf<Input>()
+
+    /**
+     * The outputs of the multiplexer.
+     */
+    override val outputs: Set<FlowSource>
+        get() = _outputs
+    private val _outputs = mutableSetOf<Output>()
+    private val _activeOutputs = mutableListOf<Output>()
+
+    /**
+     * The flow counters of this multiplexer.
+     */
+    public override val counters: FlowCounters
+        get() = _counters
+    private val _counters = FlowCountersImpl()
+
+    /**
+     * The actual processing rate of the multiplexer.
+     */
+    public override val rate: Double
+        get() = _rate
+    private var _rate = 0.0
+
+    /**
+     * The demanded processing rate of the input.
+     */
+    public override val demand: Double
+        get() = _demand
+    private var _demand = 0.0
+
+    /**
+     * The capacity of the outputs.
+     */
+    public override val capacity: Double
+        get() = _capacity
+    private var _capacity = 0.0
+
+    /**
+     * Flag to indicate that the scheduler is active.
+     */
+    private var _schedulerActive = false
+    private var _lastSchedulerCycle = Long.MAX_VALUE
+
+    /**
+     * The last convergence timestamp and the input.
+     */
+    private var _lastConverge: Long = Long.MIN_VALUE
+    private var _lastConvergeInput: Input? = null
+
+    override fun newInput(key: InterferenceKey?): FlowConsumer {
+        val provider = Input(_capacity, key)
+        _inputs.add(provider)
+        return provider
+    }
+
+    override fun removeInput(input: FlowConsumer) {
+        if (!_inputs.remove(input)) {
+            return
+        }
+        // This cast should always succeed since only `Input` instances should be added to `_inputs`
+        (input as Input).close()
+    }
+
+    override fun newOutput(): FlowSource {
+        val output = Output()
+        _outputs.add(output)
+        return output
+    }
+
+    override fun removeOutput(output: FlowSource) {
+        if (!_outputs.remove(output)) {
+            return
+        }
+
+        // This cast should always succeed since only `Output` instances should be added to `_outputs`
+        (output as Output).cancel()
+    }
+
+    override fun clearInputs() {
+        for (input in _inputs) {
+            input.cancel()
+        }
+        _inputs.clear()
+    }
+
+    override fun clearOutputs() {
+        for (output in _outputs) {
+            output.cancel()
+        }
+        _outputs.clear()
+    }
+
+    override fun clear() {
+        clearOutputs()
+        clearInputs()
+    }
+
+    /**
+     * Converge the scheduler of the multiplexer.
+     */
+    private fun runScheduler(now: Long) {
+        if (_schedulerActive) {
+            return
+        }
+        val lastSchedulerCycle = _lastSchedulerCycle
+        val delta = max(0, now - lastSchedulerCycle)
+        _schedulerActive = true
+        _lastSchedulerCycle = now
+
+        try {
+            doSchedule(now, delta)
+        } finally {
+            _schedulerActive = false
+        }
+    }
+
+    /**
+     * Schedule the inputs over the outputs.
+     */
+    private fun doSchedule(now: Long, delta: Long) {
+        val activeInputs = _activeInputs
+        val activeOutputs = _activeOutputs
+
+        // Update the counters of the scheduler
+        updateCounters(delta)
+
+        // If there is no work yet, mark the inputs as idle.
+        if (activeInputs.isEmpty()) {
+            _demand = 0.0
+            _rate = 0.0
+            return
+        }
+
+        val capacity = _capacity
+        var availableCapacity = capacity
+
+        // Pull in the work of the outputs
+        val inputIterator = activeInputs.listIterator()
+        for (input in inputIterator) {
+            input.pull(now)
+
+            // Remove outputs that have finished
+            if (!input.isActive) {
+                input.actualRate = 0.0
+                inputIterator.remove()
+            }
+        }
+
+        var demand = 0.0
+
+        // Sort in-place the inputs based on their pushed flow.
+        // Profiling shows that it is faster than maintaining some kind of sorted set.
+        activeInputs.sort()
+
+        // Divide the available output capacity fairly over the inputs using max-min fair sharing
+        var remaining = activeInputs.size
+        for (i in activeInputs.indices) {
+            val input = activeInputs[i]
+            val availableShare = availableCapacity / remaining--
+            val grantedRate = min(input.allowedRate, availableShare)
+
+            // Ignore empty sources
+            if (grantedRate <= 0.0) {
+                input.actualRate = 0.0
+                continue
+            }
+
+            input.actualRate = grantedRate
+            demand += input.limit
+            availableCapacity -= grantedRate
+        }
+
+        val rate = capacity - availableCapacity
+
+        _demand = demand
+        _rate = rate
+
+        // Sort all consumers by their capacity
+        activeOutputs.sort()
+
+        // Divide the requests over the available capacity of the input resources fairly
+        for (i in activeOutputs.indices) {
+            val output = activeOutputs[i]
+            val inputCapacity = output.capacity
+            val fraction = inputCapacity / capacity
+            val grantedSpeed = rate * fraction
+
+            output.push(grantedSpeed)
+        }
+    }
+
+    /**
+     * Recompute the capacity of the multiplexer.
+     */
+    private fun updateCapacity() {
+        val newCapacity = _activeOutputs.sumOf(Output::capacity)
+
+        // No-op if the capacity is unchanged
+        if (_capacity == newCapacity) {
+            return
+        }
+
+        _capacity = newCapacity
+
+        for (input in _inputs) {
+            input.capacity = newCapacity
+        }
+    }
+
+    /**
+     * The previous capacity of the multiplexer.
+     */
+    private var _previousCapacity = 0.0
+
+    /**
+     * Update the counters of the scheduler.
+     */
+    private fun updateCounters(delta: Long) {
+        val previousCapacity = _previousCapacity
+        _previousCapacity = _capacity
+
+        if (delta <= 0) {
+            return
+        }
+
+        val deltaS = delta / 1000.0
+
+        _counters.demand += _demand * deltaS
+        _counters.actual += _rate * deltaS
+        _counters.remaining += (previousCapacity - _rate) * deltaS
+    }
+
+    /**
+     * An internal [FlowConsumer] implementation for multiplexer inputs.
+     */
+    private inner class Input(capacity: Double, val key: InterferenceKey?) :
+        AbstractFlowConsumer(engine, capacity),
+        FlowConsumerLogic,
+        Comparable<Input> {
+        /**
+         * The requested limit.
+         */
+        @JvmField var limit: Double = 0.0
+
+        /**
+         * The actual processing speed.
+         */
+        @JvmField var actualRate: Double = 0.0
+
+        /**
+         * The processing rate that is allowed by the model constraints.
+         */
+        val allowedRate: Double
+            get() = min(capacity, limit)
+
+        /**
+         * A flag to indicate that the input is closed.
+         */
+        private var _isClosed: Boolean = false
+
+        /**
+         * The timestamp at which we received the last command.
+         */
+        private var _lastPull: Long = Long.MIN_VALUE
+
+        /**
+         * The interference domain this input belongs to.
+         */
+        private val interferenceDomain = this@MaxMinFlowMultiplexer.interferenceDomain
+
+        /**
+         * Close the input.
+         *
+         * This method is invoked when the user removes an input from the switch.
+         */
+        fun close() {
+            _isClosed = true
+            cancel()
+        }
+
+        /* AbstractFlowConsumer */
+        override fun createLogic(): FlowConsumerLogic = this
+
+        override fun start(ctx: FlowConsumerContext) {
+            check(!_isClosed) { "Cannot re-use closed input" }
+
+            _activeInputs += this
+            if (parent != null) {
+                ctx.shouldConsumerConverge = true
+            }
+
+            super.start(ctx)
+        }
+
+        /* FlowConsumerLogic */
+        override fun onPush(
+            ctx: FlowConsumerContext,
+            now: Long,
+            delta: Long,
+            rate: Double
+        ) {
+            doUpdateCounters(delta)
+
+            actualRate = 0.0
+            limit = rate
+            _lastPull = now
+
+            runScheduler(now)
+        }
+
+        override fun onConverge(ctx: FlowConsumerContext, now: Long, delta: Long) {
+            val lastConverge = _lastConverge
+            val parent = parent
+
+            if (parent != null && (lastConverge < now || _lastConvergeInput == null)) {
+                _lastConverge = now
+                _lastConvergeInput = this
+
+                parent.onConverge(now, max(0, now - lastConverge))
+            }
+        }
+
+        override fun onFinish(ctx: FlowConsumerContext, now: Long, delta: Long, cause: Throwable?) {
+            doUpdateCounters(delta)
+
+            limit = 0.0
+            actualRate = 0.0
+            _lastPull = now
+
+            // Assign a new input responsible for handling the convergence events
+            if (_lastConvergeInput == this) {
+                _lastConvergeInput = null
+            }
+
+            // Re-run scheduler to distribute new load
+            runScheduler(now)
+        }
+
+        /* Comparable */
+        override fun compareTo(other: Input): Int = allowedRate.compareTo(other.allowedRate)
+
+        /**
+         * Pull the source if necessary.
+         */
+        fun pull(now: Long) {
+            val ctx = ctx
+            if (ctx != null && _lastPull < now) {
+                ctx.flush()
+            }
+        }
+
+        /**
+         * Helper method to update the flow counters of the multiplexer.
+         */
+        private fun doUpdateCounters(delta: Long) {
+            if (delta <= 0L) {
+                return
+            }
+
+            // Compute the performance penalty due to flow interference
+            val perfScore = if (interferenceDomain != null) {
+                val load = _rate / _capacity
+                interferenceDomain.apply(key, load)
+            } else {
+                1.0
+            }
+
+            val deltaS = delta / 1000.0
+            val demand = limit * deltaS
+            val actual = actualRate * deltaS
+            val remaining = (capacity - actualRate) * deltaS
+
+            updateCounters(demand, actual, remaining)
+
+            _counters.interference += actual * max(0.0, 1 - perfScore)
+        }
+    }
+
+    /**
+     * An internal [FlowSource] implementation for multiplexer outputs.
+     */
+    private inner class Output : FlowSource, Comparable<Output> {
+        /**
+         * The active [FlowConnection] of this source.
+         */
+        private var _conn: FlowConnection? = null
+
+        /**
+         * The capacity of this output.
+         */
+        @JvmField var capacity: Double = 0.0
+
+        /**
+         * Push the specified rate to the consumer.
+         */
+        fun push(rate: Double) {
+            _conn?.push(rate)
+        }
+
+        /**
+         * Cancel this output.
+         */
+        fun cancel() {
+            _conn?.close()
+        }
+
+        override fun onStart(conn: FlowConnection, now: Long) {
+            assert(_conn == null) { "Source running concurrently" }
+            _conn = conn
+            capacity = conn.capacity
+            _activeOutputs.add(this)
+
+            updateCapacity()
+        }
+
+        override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+            _conn = null
+            capacity = 0.0
+            _activeOutputs.remove(this)
+
+            updateCapacity()
+
+            runScheduler(now)
+        }
+
+        override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+            val capacity = capacity
+            if (capacity != conn.capacity) {
+                this.capacity = capacity
+                updateCapacity()
+            }
+
+            // Re-run scheduler to distribute new load
+            runScheduler(now)
+            return Long.MAX_VALUE
+        }
+
+        override fun compareTo(other: Output): Int = capacity.compareTo(other.capacity)
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FixedFlowSource.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FixedFlowSource.kt
new file mode 100644
index 00000000..d9779c6a
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FixedFlowSource.kt
@@ -0,0 +1,57 @@
+/*
+ * 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.flow.source
+
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
+import kotlin.math.roundToLong
+
+/**
+ * A [FlowSource] that contains a fixed [amount] and is pushed with a given [utilization].
+ */
+public class FixedFlowSource(private val amount: Double, private val utilization: Double) : FlowSource {
+
+    init {
+        require(amount >= 0.0) { "Amount must be positive" }
+        require(utilization > 0.0) { "Utilization must be positive" }
+    }
+
+    private var remainingAmount = amount
+
+    override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+        val consumed = conn.rate * delta / 1000.0
+        val limit = conn.capacity * utilization
+
+        remainingAmount -= consumed
+
+        val duration = (remainingAmount / limit * 1000).roundToLong()
+
+        return if (duration > 0) {
+            conn.push(limit)
+            duration
+        } else {
+            conn.close()
+            Long.MAX_VALUE
+        }
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceBarrier.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceBarrier.kt
new file mode 100644
index 00000000..b3191ad3
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceBarrier.kt
@@ -0,0 +1,52 @@
+/*
+ * 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.flow.source
+
+/**
+ * The [FlowSourceBarrier] is a barrier that allows multiple sources to wait for a select number of other sources to
+ * finish a pull, before proceeding its operation.
+ */
+public class FlowSourceBarrier(public val parties: Int) {
+    private var counter = 0
+
+    /**
+     * Enter the barrier and determine whether the caller is the last to reach the barrier.
+     *
+     * @return `true` if the caller is the last to reach the barrier, `false` otherwise.
+     */
+    public fun enter(): Boolean {
+        val last = ++counter == parties
+        if (last) {
+            counter = 0
+            return true
+        }
+        return false
+    }
+
+    /**
+     * Reset the barrier.
+     */
+    public fun reset() {
+        counter = 0
+    }
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceRateAdapter.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceRateAdapter.kt
new file mode 100644
index 00000000..6dd60d95
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceRateAdapter.kt
@@ -0,0 +1,77 @@
+/*
+ * 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.flow.source
+
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
+
+/**
+ * Helper class to expose an observable [rate] field describing the flow rate of the source.
+ */
+public class FlowSourceRateAdapter(
+    private val delegate: FlowSource,
+    private val callback: (Double) -> Unit = {}
+) : FlowSource by delegate {
+    /**
+     * The resource processing speed at this instant.
+     */
+    public var rate: Double = 0.0
+        private set(value) {
+            if (field != value) {
+                callback(value)
+                field = value
+            }
+        }
+
+    init {
+        callback(0.0)
+    }
+
+    override fun onStart(conn: FlowConnection, now: Long) {
+        conn.shouldSourceConverge = true
+
+        delegate.onStart(conn, now)
+    }
+
+    override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+        try {
+            delegate.onStop(conn, now, delta)
+        } finally {
+            rate = 0.0
+        }
+    }
+
+    override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+        return delegate.onPull(conn, now, delta)
+    }
+
+    override fun onConverge(conn: FlowConnection, now: Long, delta: Long) {
+        try {
+            delegate.onConverge(conn, now, delta)
+        } finally {
+            rate = conn.rate
+        }
+    }
+
+    override fun toString(): String = "FlowSourceRateAdapter[delegate=$delegate]"
+}
diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/TraceFlowSource.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/TraceFlowSource.kt
new file mode 100644
index 00000000..ae537845
--- /dev/null
+++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/TraceFlowSource.kt
@@ -0,0 +1,67 @@
+/*
+ * 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.flow.source
+
+import org.opendc.simulator.flow.FlowConnection
+import org.opendc.simulator.flow.FlowSource
+
+/**
+ * A [FlowSource] that replays a sequence of [Fragment], each indicating the flow rate for some period of time.
+ */
+public class TraceFlowSource(private val trace: Sequence<Fragment>) : FlowSource {
+    private var _iterator: Iterator<Fragment>? = null
+    private var _nextTarget = Long.MIN_VALUE
+
+    override fun onStart(conn: FlowConnection, now: Long) {
+        check(_iterator == null) { "Source already running" }
+        _iterator = trace.iterator()
+    }
+
+    override fun onStop(conn: FlowConnection, now: Long, delta: Long) {
+        _iterator = null
+    }
+
+    override fun onPull(conn: FlowConnection, now: Long, delta: Long): Long {
+        // Check whether the trace fragment was fully consumed, otherwise wait until we have done so
+        val nextTarget = _nextTarget
+        if (nextTarget > now) {
+            return now - nextTarget
+        }
+
+        val iterator = checkNotNull(_iterator)
+        return if (iterator.hasNext()) {
+            val fragment = iterator.next()
+            _nextTarget = now + fragment.duration
+            conn.push(fragment.usage)
+            fragment.duration
+        } else {
+            conn.close()
+            Long.MAX_VALUE
+        }
+    }
+
+    /**
+     * A fragment of the trace.
+     */
+    public data class Fragment(val duration: Long, val usage: Double)
+}