From ad88144923d76dfc421f0b22a0b4e670b3f6366e Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Sun, 21 Aug 2022 14:27:41 +0200 Subject: perf(sim/flow): Add support for multi-flow stages This change adds support for creating nodes in a flow graph that support multiple inputs and outputs directly, instead of our current approach where we need to re-implement the `FlowConsumerContext` interface in order to support multiple inputs or outputs. --- .../org/opendc/simulator/flow/FlowBenchmarks.kt | 2 +- .../org/opendc/simulator/flow2/FlowBenchmarks.kt | 108 ++++++ .../org/opendc/simulator/flow2/FlowEngine.java | 260 ++++++++++++++ .../java/org/opendc/simulator/flow2/FlowGraph.java | 63 ++++ .../opendc/simulator/flow2/FlowGraphInternal.java | 93 +++++ .../java/org/opendc/simulator/flow2/FlowStage.java | 303 ++++++++++++++++ .../org/opendc/simulator/flow2/FlowStageLogic.java | 38 ++ .../org/opendc/simulator/flow2/FlowStageQueue.java | 109 ++++++ .../org/opendc/simulator/flow2/FlowTimerQueue.java | 208 +++++++++++ .../java/org/opendc/simulator/flow2/InHandler.java | 54 +++ .../org/opendc/simulator/flow2/InHandlers.java | 53 +++ .../java/org/opendc/simulator/flow2/InPort.java | 214 ++++++++++++ .../java/org/opendc/simulator/flow2/Inlet.java | 38 ++ .../opendc/simulator/flow2/InvocationStack.java | 95 +++++ .../org/opendc/simulator/flow2/OutHandler.java | 47 +++ .../org/opendc/simulator/flow2/OutHandlers.java | 53 +++ .../java/org/opendc/simulator/flow2/OutPort.java | 224 ++++++++++++ .../java/org/opendc/simulator/flow2/Outlet.java | 38 ++ .../simulator/flow2/mux/FlowMultiplexer.java | 70 ++++ .../simulator/flow2/mux/MaxMinFlowMultiplexer.java | 268 ++++++++++++++ .../org/opendc/simulator/flow2/sink/FlowSink.java | 36 ++ .../simulator/flow2/sink/SimpleFlowSink.java | 123 +++++++ .../simulator/flow2/source/EmptyFlowSource.java | 65 ++++ .../opendc/simulator/flow2/source/FlowSource.java | 36 ++ .../simulator/flow2/source/RuntimeFlowSource.java | 129 +++++++ .../simulator/flow2/source/SimpleFlowSource.java | 132 +++++++ .../simulator/flow2/source/TraceFlowSource.java | 152 ++++++++ .../org/opendc/simulator/flow2/FlowEngineTest.kt | 197 +++++++++++ .../opendc/simulator/flow2/FlowTimerQueueTest.kt | 385 +++++++++++++++++++++ .../opendc/simulator/flow2/InvocationStackTest.kt | 71 ++++ .../flow2/mux/MaxMinFlowMultiplexerTest.kt | 54 +++ .../opendc/simulator/flow2/sink/FlowSinkTest.kt | 124 +++++++ 32 files changed, 3841 insertions(+), 1 deletion(-) create mode 100644 opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowEngine.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraph.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraphInternal.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStage.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageLogic.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageQueue.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowTimerQueue.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandler.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandlers.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InPort.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Inlet.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InvocationStack.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandler.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandlers.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutPort.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Outlet.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/FlowSink.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/SimpleFlowSink.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/EmptyFlowSource.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/FlowSource.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowEngineTest.kt create mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowTimerQueueTest.kt create mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/InvocationStackTest.kt create mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexerTest.kt create mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/sink/FlowSinkTest.kt (limited to 'opendc-simulator/opendc-simulator-flow/src') diff --git a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt index 58f84d82..9e0a4a5e 100644 --- a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt +++ b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt @@ -47,7 +47,7 @@ class FlowBenchmarks { @Setup fun setUp() { val random = ThreadLocalRandom.current() - val entries = List(10000) { TraceFlowSource.Fragment(1000, random.nextDouble(0.0, 4500.0)) } + val entries = List(1000000) { TraceFlowSource.Fragment(1000, random.nextDouble(0.0, 4500.0)) } trace = entries.asSequence() } diff --git a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt new file mode 100644 index 00000000..1b0e2e9e --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt @@ -0,0 +1,108 @@ +/* + * 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.flow2 + +import kotlinx.coroutines.launch +import org.opendc.simulator.flow2.mux.MaxMinFlowMultiplexer +import org.opendc.simulator.flow2.sink.SimpleFlowSink +import org.opendc.simulator.flow2.source.TraceFlowSource +import org.opendc.simulator.kotlin.runSimulation +import org.openjdk.jmh.annotations.Benchmark +import org.openjdk.jmh.annotations.Fork +import org.openjdk.jmh.annotations.Measurement +import org.openjdk.jmh.annotations.Scope +import org.openjdk.jmh.annotations.Setup +import org.openjdk.jmh.annotations.State +import org.openjdk.jmh.annotations.Warmup +import java.util.concurrent.ThreadLocalRandom +import java.util.concurrent.TimeUnit + +@State(Scope.Thread) +@Fork(1) +@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 3, timeUnit = TimeUnit.SECONDS) +class FlowBenchmarks { + private lateinit var trace: TraceFlowSource.Trace + + @Setup + fun setUp() { + val random = ThreadLocalRandom.current() + val traceSize = 10_000_000 + trace = TraceFlowSource.Trace( + LongArray(traceSize) { (it + 1) * 1000L }, + FloatArray(traceSize) { random.nextFloat(0.0f, 4500.0f) }, + traceSize + ) + } + + @Benchmark + fun benchmarkSink() { + return runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + val sink = SimpleFlowSink(graph, 4200.0f) + val source = TraceFlowSource(graph, trace) + graph.connect(source.output, sink.input) + } + } + + @Benchmark + fun benchmarkMuxMaxMinSingleSource() { + return runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + val switch = MaxMinFlowMultiplexer(graph) + + val sinkA = SimpleFlowSink(graph, 3000.0f) + val sinkB = SimpleFlowSink(graph, 3000.0f) + + graph.connect(switch.newOutput(), sinkA.input) + graph.connect(switch.newOutput(), sinkB.input) + + val source = TraceFlowSource(graph, trace) + graph.connect(source.output, switch.newInput()) + } + } + + @Benchmark + fun benchmarkMuxMaxMinTripleSource() { + return runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + val switch = MaxMinFlowMultiplexer(graph) + + val sinkA = SimpleFlowSink(graph, 3000.0f) + val sinkB = SimpleFlowSink(graph, 3000.0f) + + graph.connect(switch.newOutput(), sinkA.input) + graph.connect(switch.newOutput(), sinkB.input) + + repeat(3) { + launch { + val source = TraceFlowSource(graph, trace) + graph.connect(source.output, switch.newInput()) + } + } + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowEngine.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowEngine.java new file mode 100644 index 00000000..0ebb0da9 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowEngine.java @@ -0,0 +1,260 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.time.Clock; +import java.util.ArrayList; +import java.util.List; +import kotlin.coroutines.ContinuationInterceptor; +import kotlin.coroutines.CoroutineContext; +import kotlinx.coroutines.Delay; + +/** + * A {@link FlowEngine} simulates a generic flow network. + *

+ * 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 final class FlowEngine implements Runnable { + /** + * The queue of {@link FlowStage} updates that are scheduled for immediate execution. + */ + private final FlowStageQueue queue = new FlowStageQueue(256); + + /** + * A priority queue containing the {@link FlowStage} updates to be scheduled in the future. + */ + private final FlowTimerQueue timerQueue = new FlowTimerQueue(256); + + /** + * The stack of engine invocations to occur in the future. + */ + private final InvocationStack futureInvocations = new InvocationStack(256); + + /** + * A flag to indicate that the engine is active. + */ + private boolean active; + + private final CoroutineContext coroutineContext; + private final Clock clock; + private final Delay delay; + + /** + * Create a new {@link FlowEngine} instance using the specified {@link CoroutineContext} and {@link Clock}. + */ + public static FlowEngine create(CoroutineContext coroutineContext, Clock clock) { + return new FlowEngine(coroutineContext, clock); + } + + FlowEngine(CoroutineContext coroutineContext, Clock clock) { + this.coroutineContext = coroutineContext; + this.clock = clock; + + CoroutineContext.Key key = ContinuationInterceptor.Key; + this.delay = (Delay) coroutineContext.get(key); + } + + /** + * Obtain the (virtual) {@link Clock} driving the simulation. + */ + public Clock getClock() { + return clock; + } + + /** + * Return a new {@link FlowGraph} that can be used to build a flow network. + */ + public FlowGraph newGraph() { + return new RootGraph(this); + } + + /** + * Enqueue the specified {@link FlowStage} 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. + */ + void scheduleImmediate(long now, FlowStage ctx) { + scheduleImmediateInContext(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 (active) { + return; + } + + trySchedule(futureInvocations, now, now); + } + + /** + * Enqueue the specified {@link FlowStage} 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. + *

+ * This method should only be invoked while inside an engine cycle. + */ + void scheduleImmediateInContext(FlowStage ctx) { + queue.add(ctx); + } + + /** + * Enqueue the specified {@link FlowStage} to be updated at its updated deadline. + */ + void scheduleDelayed(FlowStage ctx) { + scheduleDelayedInContext(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 (active) { + return; + } + + long deadline = timerQueue.peekDeadline(); + if (deadline != Long.MAX_VALUE) { + trySchedule(futureInvocations, clock.millis(), deadline); + } + } + + /** + * Enqueue the specified {@link FlowStage} to be updated at its updated deadline. + *

+ * This method should only be invoked while inside an engine cycle. + */ + void scheduleDelayedInContext(FlowStage ctx) { + FlowTimerQueue timerQueue = this.timerQueue; + timerQueue.enqueue(ctx); + } + + /** + * Run all the enqueued actions for the specified timestamp (now). + */ + private void doRunEngine(long now) { + final FlowStageQueue queue = this.queue; + final FlowTimerQueue timerQueue = this.timerQueue; + + try { + // Mark the engine as active to prevent concurrent calls to this method + active = true; + + // Execute all scheduled updates at current timestamp + while (true) { + final FlowStage ctx = timerQueue.poll(now); + if (ctx == null) { + break; + } + + ctx.onUpdate(now); + } + + // Execute all immediate updates + while (true) { + final FlowStage ctx = queue.poll(); + if (ctx == null) { + break; + } + + ctx.onUpdate(now); + } + } finally { + active = false; + } + + // Schedule an engine invocation for the next update to occur. + long headDeadline = timerQueue.peekDeadline(); + if (headDeadline != Long.MAX_VALUE && headDeadline >= now) { + trySchedule(futureInvocations, now, headDeadline); + } + } + + @Override + public void run() { + doRunEngine(futureInvocations.poll()); + } + + /** + * Try to schedule an engine invocation at the specified [target]. + * + * @param scheduled The queue of scheduled invocations. + * @param now The current virtual timestamp. + * @param target The virtual timestamp at which the engine invocation should happen. + */ + private void trySchedule(InvocationStack scheduled, long now, long target) { + // Only schedule a new scheduler invocation in case the target is earlier than all other pending + // scheduler invocations + if (scheduled.tryAdd(target)) { + delay.invokeOnTimeout(target - now, this, coroutineContext); + } + } + + /** + * Internal implementation of a root {@link FlowGraph}. + */ + private static final class RootGraph implements FlowGraphInternal { + private final FlowEngine engine; + private final List stages = new ArrayList<>(); + + public RootGraph(FlowEngine engine) { + this.engine = engine; + } + + @Override + public FlowEngine getEngine() { + return engine; + } + + @Override + public FlowStage newStage(FlowStageLogic logic) { + final FlowEngine engine = this.engine; + final FlowStage stage = new FlowStage(this, logic); + stages.add(stage); + long now = engine.getClock().millis(); + stage.invalidate(now); + return stage; + } + + @Override + public void connect(Outlet outlet, Inlet inlet) { + FlowGraphInternal.connect(this, outlet, inlet); + } + + @Override + public void disconnect(Outlet outlet) { + FlowGraphInternal.disconnect(this, outlet); + } + + @Override + public void disconnect(Inlet inlet) { + FlowGraphInternal.disconnect(this, inlet); + } + + /** + * Internal method to remove the specified {@link FlowStage} from the graph. + */ + @Override + public void detach(FlowStage stage) { + stages.remove(stage); + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraph.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraph.java new file mode 100644 index 00000000..f45be6cd --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraph.java @@ -0,0 +1,63 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * A representation of a flow network. A flow network is a directed graph where each edge has a capacity and receives an + * amount of flow that cannot exceed the edge's capacity. + */ +public interface FlowGraph { + /** + * Return the {@link FlowEngine} driving the simulation of the graph. + */ + FlowEngine getEngine(); + + /** + * Create a new {@link FlowStage} representing a node in the flow network. + * + * @param logic The logic for handling the events of the stage. + */ + FlowStage newStage(FlowStageLogic logic); + + /** + * Add an edge between the specified outlet port and inlet port in this graph. + * + * @param outlet The outlet of the source from which the flow originates. + * @param inlet The inlet of the sink that should receive the flow. + */ + void connect(Outlet outlet, Inlet inlet); + + /** + * Disconnect the specified {@link Outlet} (if connected). + * + * @param outlet The outlet to disconnect. + */ + void disconnect(Outlet outlet); + + /** + * Disconnect the specified {@link Inlet} (if connected). + * + * @param inlet The inlet to disconnect. + */ + void disconnect(Inlet inlet); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraphInternal.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraphInternal.java new file mode 100644 index 00000000..0f608b60 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowGraphInternal.java @@ -0,0 +1,93 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * Interface implemented by {@link FlowGraph} implementations. + */ +interface FlowGraphInternal extends FlowGraph { + /** + * Internal method to remove the specified {@link FlowStage} from the graph. + */ + void detach(FlowStage stage); + + /** + * Helper method to connect an outlet to an inlet. + */ + static void connect(FlowGraph graph, Outlet outlet, Inlet inlet) { + if (!(outlet instanceof OutPort) || !(inlet instanceof InPort)) { + throw new IllegalArgumentException("Invalid outlet or inlet passed to graph"); + } + + InPort inPort = (InPort) inlet; + OutPort outPort = (OutPort) outlet; + + if (!graph.equals(outPort.getGraph()) || !graph.equals(inPort.getGraph())) { + throw new IllegalArgumentException("Outlet or inlet does not belong to graph"); + } else if (outPort.input != null || inPort.output != null) { + throw new IllegalStateException("Inlet or outlet already connected"); + } + + outPort.input = inPort; + inPort.output = outPort; + + inPort.connect(); + outPort.connect(); + } + + /** + * Helper method to disconnect an outlet. + */ + static void disconnect(FlowGraph graph, Outlet outlet) { + if (!(outlet instanceof OutPort)) { + throw new IllegalArgumentException("Invalid outlet passed to graph"); + } + + OutPort outPort = (OutPort) outlet; + + if (!graph.equals(outPort.getGraph())) { + throw new IllegalArgumentException("Outlet or inlet does not belong to graph"); + } + + outPort.cancel(null); + outPort.complete(); + } + + /** + * Helper method to disconnect an inlet. + */ + static void disconnect(FlowGraph graph, Inlet inlet) { + if (!(inlet instanceof InPort)) { + throw new IllegalArgumentException("Invalid outlet passed to graph"); + } + + InPort inPort = (InPort) inlet; + + if (!graph.equals(inPort.getGraph())) { + throw new IllegalArgumentException("Outlet or inlet does not belong to graph"); + } + + inPort.finish(null); + inPort.cancel(null); + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStage.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStage.java new file mode 100644 index 00000000..4d098043 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStage.java @@ -0,0 +1,303 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.time.Clock; +import java.util.HashMap; +import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link FlowStage} represents a node in a {@link FlowGraph}. + */ +public final class FlowStage { + private static final Logger LOGGER = LoggerFactory.getLogger(FlowStage.class); + + /** + * States of the flow stage. + */ + private static final int STAGE_PENDING = 0; // Stage is pending to be started + + private static final int STAGE_ACTIVE = 1; // Stage is actively running + private static final int STAGE_CLOSED = 2; // Stage is closed + private static final int STAGE_STATE = 0b11; // Mask for accessing the state of the flow stage + + /** + * Flags of the flow connection + */ + private static final int STAGE_INVALIDATE = 1 << 2; // The stage is invalidated + + private static final int STAGE_CLOSE = 1 << 3; // The stage should be closed + private static final int STAGE_UPDATE_ACTIVE = 1 << 4; // An update for the connection is active + private static final int STAGE_UPDATE_PENDING = 1 << 5; // An (immediate) update of the connection is pending + + /** + * The flags representing the state and pending actions for the stage. + */ + private int flags = STAGE_PENDING; + + /** + * The deadline of the stage after which an update should run. + */ + long deadline = Long.MAX_VALUE; + + /** + * The index of the timer in the {@link FlowTimerQueue}. + */ + int timerIndex = -1; + + final Clock clock; + private final FlowStageLogic logic; + final FlowGraphInternal parentGraph; + private final FlowEngine engine; + + private final Map inlets = new HashMap<>(); + private final Map outlets = new HashMap<>(); + private int nextInlet = 0; + private int nextOutlet = 0; + + /** + * Construct a new {@link FlowStage} instance. + * + * @param parentGraph The {@link FlowGraph} this stage belongs to. + * @param logic The logic of the stage. + */ + FlowStage(FlowGraphInternal parentGraph, FlowStageLogic logic) { + this.parentGraph = parentGraph; + this.logic = logic; + this.engine = parentGraph.getEngine(); + this.clock = engine.getClock(); + } + + /** + * Return the {@link FlowGraph} to which this stage belongs. + */ + public FlowGraph getGraph() { + return parentGraph; + } + + /** + * Return the {@link Inlet} (an in-going edge) with the specified name for this {@link FlowStage}. + * If an inlet with that name does not exist, a new one is allocated for the stage. + * + * @param name The name of the inlet. + * @return The {@link InPort} representing an {@link Inlet} with the specified name. + */ + public InPort getInlet(String name) { + return inlets.computeIfAbsent(name, (key) -> new InPort(this, key, nextInlet++)); + } + + /** + * Return the {@link Outlet} (an out-going edge) with the specified name for this {@link FlowStage}. + * If an outlet with that name does not exist, a new one is allocated for the stage. + * + * @param name The name of the outlet. + * @return The {@link OutPort} representing an {@link Outlet} with the specified name. + */ + public OutPort getOutlet(String name) { + return outlets.computeIfAbsent(name, (key) -> new OutPort(this, key, nextOutlet++)); + } + + /** + * Return the current deadline of the {@link FlowStage}'s timer (in milliseconds after epoch). + */ + public long getDeadline() { + return deadline; + } + + /** + * Set the deadline of the {@link FlowStage}'s timer. + * + * @param deadline The new deadline (in milliseconds after epoch) when the stage should be interrupted. + */ + public void setDeadline(long deadline) { + this.deadline = deadline; + + if ((flags & STAGE_UPDATE_ACTIVE) == 0) { + // Update the timer queue with the new deadline + engine.scheduleDelayed(this); + } + } + + /** + * Invalidate the {@link FlowStage} forcing the stage to update. + */ + public void invalidate() { + int flags = this.flags; + + if ((flags & STAGE_UPDATE_ACTIVE) == 0) { + scheduleImmediate(clock.millis(), flags | STAGE_INVALIDATE); + } + } + + /** + * Close the {@link FlowStage} and disconnect all inlets and outlets. + */ + public void close() { + int flags = this.flags; + + if ((flags & STAGE_STATE) == STAGE_CLOSED) { + return; + } + + // Toggle the close bit. In case no update is active, schedule a new update. + if ((flags & STAGE_UPDATE_ACTIVE) != 0) { + this.flags = flags | STAGE_CLOSE; + } else { + scheduleImmediate(clock.millis(), flags | STAGE_CLOSE); + } + } + + /** + * Update the state of the flow stage. + * + * @param now The current virtual timestamp. + */ + void onUpdate(long now) { + int flags = this.flags; + int state = flags & STAGE_STATE; + + if (state == STAGE_ACTIVE) { + doUpdate(now, flags); + } else if (state == STAGE_PENDING) { + doStart(now, flags); + } + } + + /** + * Invalidate the {@link FlowStage} forcing the stage to update. + * + *

+ * This method is similar to {@link #invalidate()}, but allows the user to manually pass the current timestamp to + * prevent having to re-query the clock. This method should not be called during an update. + */ + void invalidate(long now) { + scheduleImmediate(now, flags | STAGE_INVALIDATE); + } + + /** + * Schedule an immediate update for this stage. + */ + private void scheduleImmediate(long now, int flags) { + // In case an immediate update is already scheduled, no need to do anything + if ((flags & STAGE_UPDATE_PENDING) != 0) { + this.flags = flags; + return; + } + + // Mark the stage that there is an update pending + this.flags = flags | STAGE_UPDATE_PENDING; + + engine.scheduleImmediate(now, this); + } + + /** + * Start the stage. + */ + private void doStart(long now, int flags) { + // Update state before calling into the outside world, so it observes a consistent state + flags = flags | STAGE_ACTIVE | STAGE_UPDATE_ACTIVE; + + doUpdate(now, flags); + } + + /** + * Update the state of the stage. + */ + private void doUpdate(long now, int flags) { + long deadline = this.deadline; + long newDeadline = deadline; + + // Update the stage if: + // (1) the timer of the stage has expired. + // (2) one of the input ports is pushed, + // (3) one of the output ports is pulled, + if ((flags & STAGE_INVALIDATE) != 0 || deadline == now) { + // Update state before calling into the outside world, so it observes a consistent state + this.flags = (flags & ~STAGE_INVALIDATE) | STAGE_UPDATE_ACTIVE; + + try { + newDeadline = logic.onUpdate(this, now); + + // IMPORTANT: Re-fetch the flags after the callback might have changed those + flags = this.flags; + } catch (Exception e) { + doFail(e); + } + } + + // Check whether the stage is marked as closing. + if ((flags & STAGE_CLOSE) != 0) { + doClose(flags, null); + + // IMPORTANT: Re-fetch the flags after the callback might have changed those + flags = this.flags; + } + + // Indicate that no update is active anymore and flush the flags + this.flags = flags & ~(STAGE_UPDATE_ACTIVE | STAGE_UPDATE_PENDING); + this.deadline = newDeadline; + + // Update the timer queue with the new deadline + engine.scheduleDelayedInContext(this); + } + + /** + * This method is invoked when an uncaught exception is caught by the engine. When this happens, the + * {@link FlowStageLogic} "fails" and disconnects all its inputs and outputs. + */ + void doFail(Throwable cause) { + LOGGER.warn("Uncaught exception (closing stage)", cause); + + doClose(flags, cause); + } + + /** + * This method is invoked when the {@link FlowStageLogic} exits successfully or due to failure. + */ + private void doClose(int flags, Throwable cause) { + // Mark the stage as closed + this.flags = flags & ~(STAGE_STATE | STAGE_INVALIDATE | STAGE_CLOSE) | STAGE_CLOSED; + + // Remove stage from parent graph + parentGraph.detach(this); + + // Remove stage from the timer queue + setDeadline(Long.MAX_VALUE); + + // Cancel all input ports + for (InPort port : inlets.values()) { + if (port != null) { + port.cancel(cause); + } + } + + // Cancel all output ports + for (OutPort port : outlets.values()) { + if (port != null) { + port.fail(cause); + } + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageLogic.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageLogic.java new file mode 100644 index 00000000..70986a35 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageLogic.java @@ -0,0 +1,38 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * The {@link FlowStageLogic} interface is responsible for describing the behaviour of a {@link FlowStage} via + * out-going flows based on its potential inputs. + */ +public interface FlowStageLogic { + /** + * This method is invoked when the one of the stage's inlets or outlets is invalidated. + * + * @param ctx The context in which the stage runs. + * @param now The virtual timestamp in milliseconds after epoch at which the update is occurring. + * @return The next deadline for the stage. + */ + long onUpdate(FlowStage ctx, long now); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageQueue.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageQueue.java new file mode 100644 index 00000000..56ec7702 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowStageQueue.java @@ -0,0 +1,109 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.util.ArrayDeque; +import java.util.Arrays; + +/** + * A specialized {@link ArrayDeque} implementation that contains the {@link FlowStageLogic}s + * that have been updated during the engine cycle and should converge. + *

+ * By using a specialized class, we reduce the overhead caused by type-erasure. + */ +final class FlowStageQueue { + /** + * The array of elements in the queue. + */ + private FlowStage[] elements; + + private int head = 0; + private int tail = 0; + + public FlowStageQueue(int initialCapacity) { + elements = new FlowStage[initialCapacity]; + } + + /** + * Add the specified context to the queue. + */ + void add(FlowStage ctx) { + final FlowStage[] es = elements; + int tail = this.tail; + + es[tail] = ctx; + + tail = inc(tail, es.length); + this.tail = tail; + + if (head == tail) { + doubleCapacity(); + } + } + + /** + * Remove a {@link FlowStage} from the queue or null if the queue is empty. + */ + FlowStage poll() { + final FlowStage[] es = elements; + int head = this.head; + FlowStage ctx = es[head]; + + if (ctx != null) { + es[head] = null; + this.head = inc(head, es.length); + } + + return ctx; + } + + /** + * Doubles the capacity of this deque + */ + private void doubleCapacity() { + int oldCapacity = elements.length; + int newCapacity = oldCapacity + (oldCapacity >> 1); + if (newCapacity < 0) { + throw new IllegalStateException("Sorry, deque too big"); + } + + final FlowStage[] es = elements = Arrays.copyOf(elements, newCapacity); + + // Exceptionally, here tail == head needs to be disambiguated + if (tail < head || (tail == head && es[head] != null)) { + // wrap around; slide first leg forward to end of array + int newSpace = newCapacity - oldCapacity; + System.arraycopy(es, head, es, head + newSpace, oldCapacity - head); + for (int i = head, to = (head += newSpace); i < to; i++) es[i] = null; + } + } + + /** + * Circularly increments i, mod modulus. + * Precondition and postcondition: 0 <= i < modulus. + */ + private static int inc(int i, int modulus) { + if (++i >= modulus) i = 0; + return i; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowTimerQueue.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowTimerQueue.java new file mode 100644 index 00000000..4b746202 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/FlowTimerQueue.java @@ -0,0 +1,208 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.util.Arrays; + +/** + * A specialized priority queue for timers of {@link FlowStageLogic}s. + *

+ * By using a specialized priority queue, we reduce the overhead caused by the default priority queue implementation + * being generic. + */ +final class FlowTimerQueue { + /** + * Array representation of binary heap of {@link FlowStage} instances. + */ + private FlowStage[] queue; + + /** + * The number of elements in the priority queue. + */ + private int size = 0; + + /** + * Construct a {@link FlowTimerQueue} with the specified initial capacity. + * + * @param initialCapacity The initial capacity of the queue. + */ + public FlowTimerQueue(int initialCapacity) { + this.queue = new FlowStage[initialCapacity]; + } + + /** + * Enqueue a timer for the specified context or update the existing timer. + */ + void enqueue(FlowStage ctx) { + FlowStage[] es = queue; + int k = ctx.timerIndex; + + if (ctx.deadline != Long.MAX_VALUE) { + if (k >= 0) { + update(es, ctx, k); + } else { + add(es, ctx); + } + } else if (k >= 0) { + delete(es, k); + } + } + + /** + * Retrieve the head of the queue if its deadline does not exceed now. + * + * @param now The timestamp that the deadline of the head of the queue should not exceed. + * @return The head of the queue if its deadline does not exceed now, otherwise null. + */ + FlowStage poll(long now) { + int size = this.size; + if (size == 0) { + return null; + } + + final FlowStage[] es = queue; + final FlowStage head = es[0]; + + if (now < head.deadline) { + return null; + } + + int n = size - 1; + this.size = n; + final FlowStage next = es[n]; + es[n] = null; // Clear the last element of the queue + + if (n > 0) { + siftDown(0, next, es, n); + } + + head.timerIndex = -1; + return head; + } + + /** + * Find the earliest deadline in the queue. + */ + long peekDeadline() { + if (size > 0) { + return queue[0].deadline; + } + + return Long.MAX_VALUE; + } + + /** + * Add a new entry to the queue. + */ + private void add(FlowStage[] es, FlowStage ctx) { + int i = size; + + if (i >= es.length) { + // Re-fetch the resized array + es = grow(); + } + + siftUp(i, ctx, es); + + size = i + 1; + } + + /** + * Update the deadline of an existing entry in the queue. + */ + private void update(FlowStage[] es, FlowStage ctx, int k) { + if (k > 0) { + int parent = (k - 1) >>> 1; + if (es[parent].deadline > ctx.deadline) { + siftUp(k, ctx, es); + return; + } + } + + siftDown(k, ctx, es, size); + } + + /** + * Deadline an entry from the queue. + */ + private void delete(FlowStage[] es, int k) { + int s = --size; + if (s == k) { + es[k] = null; // Element is last in the queue + } else { + FlowStage moved = es[s]; + es[s] = null; + + siftDown(k, moved, es, s); + + if (es[k] == moved) { + siftUp(k, moved, es); + } + } + } + + /** + * Increases the capacity of the array. + */ + private FlowStage[] grow() { + FlowStage[] queue = this.queue; + int oldCapacity = queue.length; + int newCapacity = oldCapacity + (oldCapacity >> 1); + + queue = Arrays.copyOf(queue, newCapacity); + this.queue = queue; + return queue; + } + + private static void siftUp(int k, FlowStage key, FlowStage[] es) { + while (k > 0) { + int parent = (k - 1) >>> 1; + FlowStage e = es[parent]; + if (key.deadline >= e.deadline) break; + es[k] = e; + e.timerIndex = k; + k = parent; + } + es[k] = key; + key.timerIndex = k; + } + + private static void siftDown(int k, FlowStage key, FlowStage[] es, int n) { + int half = n >>> 1; // loop while a non-leaf + while (k < half) { + int child = (k << 1) + 1; // assume left child is least + FlowStage c = es[child]; + int right = child + 1; + if (right < n && c.deadline > es[right].deadline) c = es[child = right]; + + if (key.deadline <= c.deadline) break; + + es[k] = c; + c.timerIndex = k; + k = child; + } + + es[k] = key; + key.timerIndex = k; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandler.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandler.java new file mode 100644 index 00000000..839b01db --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandler.java @@ -0,0 +1,54 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * Collection of callbacks for the input port (a {@link InPort}) of a {@link FlowStageLogic}. + */ +public interface InHandler { + /** + * Return the actual flow rate over the input port. + * + * @param port The input port to which the flow was pushed. + * @return The actual flow rate over the port. + */ + default float getRate(InPort port) { + return Math.min(port.getDemand(), port.getCapacity()); + } + + /** + * This method is invoked when another {@link FlowStageLogic} changes the rate of flow to the specified inlet. + * + * @param port The input port to which the flow was pushed. + * @param demand The rate of flow the output attempted to push to the port. + */ + void onPush(InPort port, float demand); + + /** + * This method is invoked when the input port is finished. + * + * @param port The input port that has finished. + * @param cause The cause of the input port being finished or null if the port completed successfully. + */ + void onUpstreamFinish(InPort port, Throwable cause); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandlers.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandlers.java new file mode 100644 index 00000000..9d5b4bef --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InHandlers.java @@ -0,0 +1,53 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * A collection of common {@link InHandler} implementations. + */ +public class InHandlers { + /** + * Prevent construction of this class. + */ + private InHandlers() {} + + /** + * Return an {@link InHandler} that does nothing. + */ + public static InHandler noop() { + return NoopInHandler.INSTANCE; + } + + /** + * No-op implementation of {@link InHandler}. + */ + private static final class NoopInHandler implements InHandler { + public static final InHandler INSTANCE = new NoopInHandler(); + + @Override + public void onPush(InPort port, float demand) {} + + @Override + public void onUpstreamFinish(InPort port, Throwable cause) {} + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InPort.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InPort.java new file mode 100644 index 00000000..fba12aaf --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InPort.java @@ -0,0 +1,214 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.time.Clock; +import java.util.Objects; + +/** + * A port that consumes a flow. + *

+ * Input ports are represented as in-going edges in the flow graph. + */ +public final class InPort implements Inlet { + private final int id; + + private float capacity; + private float demand; + + private boolean mask; + + OutPort output; + private InHandler handler = InHandlers.noop(); + private final Clock clock; + private final String name; + private final FlowStage stage; + + InPort(FlowStage stage, String name, int id) { + this.name = name; + this.id = id; + this.stage = stage; + this.clock = stage.clock; + } + + @Override + public FlowGraph getGraph() { + return stage.parentGraph; + } + + @Override + public String getName() { + return name; + } + + /** + * Return the identifier of the {@link InPort} with respect to its stage. + */ + public int getId() { + return id; + } + + /** + * Return the current capacity of the input port. + */ + public float getCapacity() { + return capacity; + } + + /** + * Return the current demand of flow of the input port. + */ + public float getDemand() { + return demand; + } + + /** + * Return the current rate of flow of the input port. + */ + public float getRate() { + return handler.getRate(this); + } + + /** + * Pull the flow with the specified capacity from the input port. + * + * @param capacity The maximum throughput that the stage can receive from the input port. + */ + public void pull(float capacity) { + this.capacity = capacity; + + OutPort output = this.output; + if (output != null) { + output.pull(capacity); + } + } + + /** + * Return the current {@link InHandler} of the input port. + */ + public InHandler getHandler() { + return handler; + } + + /** + * Set the {@link InHandler} of the input port. + */ + public void setHandler(InHandler handler) { + this.handler = handler; + } + + /** + * Return the mask of this port. + *

+ * Stages ignore events originating from masked ports. + */ + public boolean getMask() { + return mask; + } + + /** + * (Un)mask the port. + */ + public void setMask(boolean mask) { + this.mask = mask; + } + + /** + * Disconnect the input port from its (potentially) connected outlet. + *

+ * The inlet can still be used and re-connected to another outlet. + * + * @param cause The cause for disconnecting the port or null when no more flow is needed. + */ + public void cancel(Throwable cause) { + demand = 0.f; + + OutPort output = this.output; + if (output != null) { + this.output = null; + output.input = null; + output.cancel(cause); + } + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + InPort port = (InPort) o; + return stage.equals(port.stage) && name.equals(port.name); + } + + @Override + public int hashCode() { + return Objects.hash(stage.parentGraph, name); + } + + /** + * This method is invoked when the inlet is connected to an outlet. + */ + void connect() { + OutPort output = this.output; + output.pull(capacity); + } + + /** + * Push a flow from an outlet to this inlet. + * + * @param demand The rate of flow to push. + */ + void push(float demand) { + // No-op when the rate is unchanged + if (this.demand == demand) { + return; + } + + try { + handler.onPush(this, demand); + this.demand = demand; + + if (!mask) { + stage.invalidate(clock.millis()); + } + } catch (Exception e) { + stage.doFail(e); + } + } + + /** + * This method is invoked by the connected {@link OutPort} when it finishes. + */ + void finish(Throwable cause) { + try { + long now = clock.millis(); + handler.onUpstreamFinish(this, cause); + this.demand = 0.f; + + if (!mask) { + stage.invalidate(now); + } + } catch (Exception e) { + stage.doFail(e); + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Inlet.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Inlet.java new file mode 100644 index 00000000..4a9ea6a5 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Inlet.java @@ -0,0 +1,38 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * An in-going edge in a {@link FlowGraph}. + */ +public interface Inlet { + /** + * Return the {@link FlowGraph} to which the inlet is exposed. + */ + FlowGraph getGraph(); + + /** + * Return the name of the inlet. + */ + String getName(); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InvocationStack.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InvocationStack.java new file mode 100644 index 00000000..a5b5114b --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/InvocationStack.java @@ -0,0 +1,95 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.util.Arrays; + +/** + * A specialized monotonic stack implementation for tracking the scheduled engine invocations. + *

+ * By using a specialized class, we reduce the overhead caused by type-erasure. + */ +final class InvocationStack { + /** + * The array of elements in the stack. + */ + private long[] elements; + + private int head = -1; + + public InvocationStack(int initialCapacity) { + elements = new long[initialCapacity]; + Arrays.fill(elements, Long.MIN_VALUE); + } + + /** + * Try to add the specified invocation to the monotonic stack. + * + * @param invocation The timestamp of the invocation. + * @return true if the invocation was added, false otherwise. + */ + boolean tryAdd(long invocation) { + final long[] es = elements; + int head = this.head; + + if (head < 0 || es[head] > invocation) { + es[head + 1] = invocation; + this.head = head + 1; + + if (head + 2 == es.length) { + doubleCapacity(); + } + + return true; + } + + return false; + } + + /** + * Remove the head invocation from the stack or return {@link Long#MAX_VALUE} if the stack is empty. + */ + long poll() { + final long[] es = elements; + int head = this.head--; + + if (head >= 0) { + return es[head]; + } + + return Long.MAX_VALUE; + } + + /** + * Doubles the capacity of this deque + */ + private void doubleCapacity() { + int oldCapacity = elements.length; + int newCapacity = oldCapacity + (oldCapacity >> 1); + if (newCapacity < 0) { + throw new IllegalStateException("Sorry, deque too big"); + } + + elements = Arrays.copyOf(elements, newCapacity); + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandler.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandler.java new file mode 100644 index 00000000..723c6d6b --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandler.java @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * Collection of callbacks for the output port (a {@link OutPort}) of a {@link FlowStageLogic}. + */ +public interface OutHandler { + /** + * This method is invoked when another {@link FlowStageLogic} changes the capacity of the outlet. + * + * @param port The output port of which the capacity was changed. + * @param capacity The new capacity of the outlet. + */ + void onPull(OutPort port, float capacity); + + /** + * This method is invoked when the output port no longer accepts any flow. + *

+ * After this callback no other callbacks will be called for this port. + * + * @param port The outlet that no longer accepts any flow. + * @param cause The cause of the output port no longer accepting any flow or null if the port closed + * successfully. + */ + void onDownstreamFinish(OutPort port, Throwable cause); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandlers.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandlers.java new file mode 100644 index 00000000..8fbfda0d --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutHandlers.java @@ -0,0 +1,53 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * A collection of common {@link OutHandler} implementations. + */ +public class OutHandlers { + /** + * Prevent construction of this class. + */ + private OutHandlers() {} + + /** + * Return an {@link OutHandler} that does nothing. + */ + public static OutHandler noop() { + return NoopOutHandler.INSTANCE; + } + + /** + * No-op implementation of {@link OutHandler}. + */ + private static final class NoopOutHandler implements OutHandler { + public static final OutHandler INSTANCE = new NoopOutHandler(); + + @Override + public void onPull(OutPort port, float capacity) {} + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) {} + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutPort.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutPort.java new file mode 100644 index 00000000..332296a0 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/OutPort.java @@ -0,0 +1,224 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +import java.time.Clock; +import java.util.Objects; + +/** + * A port that outputs a flow. + *

+ * Output ports are represented as out-going edges in the flow graph. + */ +public final class OutPort implements Outlet { + private final int id; + + private float capacity; + private float demand; + + private boolean mask; + + InPort input; + private OutHandler handler = OutHandlers.noop(); + private final String name; + private final FlowStage stage; + private final Clock clock; + + OutPort(FlowStage stage, String name, int id) { + this.name = name; + this.id = id; + this.stage = stage; + this.clock = stage.clock; + } + + @Override + public FlowGraph getGraph() { + return stage.parentGraph; + } + + @Override + public String getName() { + return name; + } + + /** + * Return the identifier of the {@link OutPort} with respect to its stage. + */ + public int getId() { + return id; + } + + /** + * Return the capacity of the output port. + */ + public float getCapacity() { + return capacity; + } + + /** + * Return the current demand of flow of the output port. + */ + public float getDemand() { + return demand; + } + + /** + * Return the current rate of flow of the input port. + */ + public float getRate() { + InPort input = this.input; + if (input != null) { + return input.getRate(); + } + + return 0.f; + } + + /** + * Return the current {@link OutHandler} of the output port. + */ + public OutHandler getHandler() { + return handler; + } + + /** + * Set the {@link OutHandler} of the output port. + */ + public void setHandler(OutHandler handler) { + this.handler = handler; + } + + /** + * Return the mask of this port. + *

+ * Stages ignore events originating from masked ports. + */ + public boolean getMask() { + return mask; + } + + /** + * (Un)mask the port. + */ + public void setMask(boolean mask) { + this.mask = mask; + } + + /** + * Push the given flow rate over output port. + * + * @param rate The rate of the flow to push. + */ + public void push(float rate) { + demand = rate; + InPort input = this.input; + + if (input != null) { + input.push(rate); + } + } + + /** + * Signal to the downstream port that the output has completed successfully and disconnect the port from its input. + *

+ * The output port can still be used and re-connected to another input. + */ + public void complete() { + fail(null); + } + + /** + * Signal a failure to the downstream port and disconnect the port from its input. + *

+ * The output can still be used and re-connected to another input. + */ + public void fail(Throwable cause) { + capacity = 0.f; + + InPort input = this.input; + if (input != null) { + this.input = null; + input.output = null; + input.finish(cause); + } + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + OutPort port = (OutPort) o; + return stage.equals(port.stage) && name.equals(port.name); + } + + @Override + public int hashCode() { + return Objects.hash(stage.parentGraph, name); + } + + /** + * This method is invoked when the outlet is connected to an inlet. + */ + void connect() { + input.push(demand); + } + + /** + * Pull from this outlet with a specified capacity. + * + * @param capacity The capacity of the inlet. + */ + void pull(float capacity) { + // No-op when outlet is not active or the rate is unchanged + if (this.capacity == capacity) { + return; + } + + try { + handler.onPull(this, capacity); + this.capacity = capacity; + + if (!mask) { + stage.invalidate(clock.millis()); + } + } catch (Exception e) { + stage.doFail(e); + } + } + + /** + * This method is invoked by the connected {@link InPort} when downstream cancels the connection. + */ + void cancel(Throwable cause) { + try { + handler.onDownstreamFinish(this, cause); + this.capacity = 0.f; + + if (!mask) { + stage.invalidate(clock.millis()); + } + } catch (Exception e) { + stage.doFail(e); + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Outlet.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Outlet.java new file mode 100644 index 00000000..32e19a3b --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/Outlet.java @@ -0,0 +1,38 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2; + +/** + * An out-going edge in a {@link FlowGraph}. + */ +public interface Outlet { + /** + * Return the {@link FlowGraph} to which the outlet is exposed. + */ + FlowGraph getGraph(); + + /** + * Return the name of the outlet. + */ + String getName(); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java new file mode 100644 index 00000000..1a99d0cf --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java @@ -0,0 +1,70 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.mux; + +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.Inlet; +import org.opendc.simulator.flow2.Outlet; + +/** + * A {@link FlowStageLogic} that multiplexes multiple inputs over (possibly) multiple outputs. + */ +public interface FlowMultiplexer { + /** + * Return the number of active inputs on this multiplexer. + */ + int getInputCount(); + + /** + * Allocate a new input on this multiplexer with the specified capacity.. + * + * @return The identifier of the input for this stage. + */ + Inlet newInput(); + + /** + * Release the input at the specified slot. + * + * @param inlet The inlet to release. + */ + void releaseInput(Inlet inlet); + + /** + * Return the number of active outputs on this multiplexer. + */ + int getOutputCount(); + + /** + * Allocate a new output on this multiplexer. + * + * @return The outlet for this stage. + */ + Outlet newOutput(); + + /** + * Release the output at the specified slot. + * + * @param outlet The outlet to release. + */ + void releaseOutput(Outlet outlet); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java new file mode 100644 index 00000000..ca0639f5 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java @@ -0,0 +1,268 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.mux; + +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.InHandler; +import org.opendc.simulator.flow2.InPort; +import org.opendc.simulator.flow2.Inlet; +import org.opendc.simulator.flow2.OutHandler; +import org.opendc.simulator.flow2.OutPort; +import org.opendc.simulator.flow2.Outlet; + +import java.util.Arrays; +import java.util.BitSet; + +/** + * A {@link FlowMultiplexer} implementation that distributes the available capacity of the outputs over the inputs + * using max-min fair sharing. + *

+ * The max-min fair sharing algorithm of this multiplexer ensures that each input receives a fair share of the combined + * output capacity, but allows individual inputs to use more capacity if there is still capacity left. + */ +public final class MaxMinFlowMultiplexer implements FlowMultiplexer, FlowStageLogic { + private final FlowStage stage; + private final BitSet activeInputs; + private final BitSet activeOutputs; + + private float capacity = 0.f; + private float demand = 0.f; + private float rate = 0.f; + + private InPort[] inlets; + private long[] inputs; + private float[] rates; + private OutPort[] outlets; + + private final MultiplexerInHandler inHandler = new MultiplexerInHandler(); + private final MultiplexerOutHandler outHandler = new MultiplexerOutHandler(); + + /** + * Construct a {@link MaxMinFlowMultiplexer} instance. + * + * @param graph The {@link FlowGraph} to add the multiplexer to. + */ + public MaxMinFlowMultiplexer(FlowGraph graph) { + this.stage = graph.newStage(this); + this.activeInputs = new BitSet(); + this.activeOutputs = new BitSet(); + + this.inlets = new InPort[4]; + this.inputs = new long[4]; + this.rates = new float[4]; + this.outlets = new OutPort[4]; + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + float capacity = this.capacity; + float demand = this.demand; + float rate = demand; + + if (demand > capacity) { + rate = redistributeCapacity(inlets, inputs, rates, capacity); + } + + if (this.rate != rate) { + // Only update the outputs if the output rate has changed + this.rate = rate; + + changeRate(activeOutputs, outlets, capacity, rate); + } + + return Long.MAX_VALUE; + } + + @Override + public int getInputCount() { + return activeInputs.length(); + } + + @Override + public Inlet newInput() { + final BitSet activeInputs = this.activeInputs; + int slot = activeInputs.nextClearBit(0); + + InPort port = stage.getInlet("in" + slot); + port.setHandler(inHandler); + port.pull(this.capacity); + + InPort[] inlets = this.inlets; + if (slot >= inlets.length) { + int newLength = inlets.length + (inlets.length >> 1); + inlets = Arrays.copyOf(inlets, newLength); + inputs = Arrays.copyOf(inputs, newLength); + rates = Arrays.copyOf(rates, newLength); + this.inlets = inlets; + } + inlets[slot] = port; + + activeInputs.set(slot); + return port; + } + + @Override + public void releaseInput(Inlet inlet) { + InPort port = (InPort) inlet; + + activeInputs.clear(port.getId()); + port.cancel(null); + } + + @Override + public int getOutputCount() { + return activeOutputs.length(); + } + + @Override + public Outlet newOutput() { + final BitSet activeOutputs = this.activeOutputs; + int slot = activeOutputs.nextClearBit(0); + + OutPort port = stage.getOutlet("out" + slot); + port.setHandler(outHandler); + + OutPort[] outlets = this.outlets; + if (slot >= outlets.length) { + int newLength = outlets.length + (outlets.length >> 1); + outlets = Arrays.copyOf(outlets, newLength); + this.outlets = outlets; + } + outlets[slot] = port; + + activeOutputs.set(slot); + return port; + } + + @Override + public void releaseOutput(Outlet outlet) { + OutPort port = (OutPort) outlet; + activeInputs.clear(port.getId()); + port.complete(); + } + + /** + * Helper function to redistribute the specified capacity across the inlets. + */ + private static float redistributeCapacity(InPort[] inlets, long[] inputs, float[] rates, float capacity) { + // If the demand is higher than the capacity, we need use max-min fair sharing to distribute the + // constrained capacity across the inputs. + for (int i = 0; i < inputs.length; i++) { + InPort inlet = inlets[i]; + if (inlet == null) { + break; + } + + inputs[i] = ((long) Float.floatToRawIntBits(inlet.getDemand()) << 32) | (i & 0xFFFFFFFFL); + } + Arrays.sort(inputs); + + float availableCapacity = capacity; + int inputSize = inputs.length; + + // Divide the available output capacity fairly over the inputs using max-min fair sharing + for (int i = 0; i < inputs.length; i++) { + long v = inputs[i]; + int slot = (int) v; + float d = Float.intBitsToFloat((int) (v >> 32)); + + if (d == 0.0) { + continue; + } + + float availableShare = availableCapacity / (inputSize - i); + float r = Math.min(d, availableShare); + + rates[slot] = r; + availableCapacity -= r; + } + + return capacity - availableCapacity; + } + + /** + * Helper method to change the rate of the outlets. + */ + private static void changeRate(BitSet activeOutputs, OutPort[] outlets, float capacity, float rate) { + // Divide the requests over the available capacity of the input resources fairly + for (int i = activeOutputs.nextSetBit(0); i != -1; i = activeOutputs.nextSetBit(i + 1)) { + OutPort outlet = outlets[i]; + float fraction = outlet.getCapacity() / capacity; + outlet.push(rate * fraction); + } + } + + /** + * A {@link InHandler} implementation for the multiplexer inputs. + */ + private class MultiplexerInHandler implements InHandler { + @Override + public float getRate(InPort port) { + return rates[port.getId()]; + } + + @Override + public void onPush(InPort port, float demand) { + MaxMinFlowMultiplexer.this.demand += -port.getDemand() + demand; + rates[port.getId()] = demand; + } + + @Override + public void onUpstreamFinish(InPort port, Throwable cause) { + MaxMinFlowMultiplexer.this.demand -= port.getDemand(); + releaseInput(port); + rates[port.getId()] = 0.f; + } + } + + /** + * A {@link OutHandler} implementation for the multiplexer outputs. + */ + private class MultiplexerOutHandler implements OutHandler { + @Override + public void onPull(OutPort port, float capacity) { + float newCapacity = MaxMinFlowMultiplexer.this.capacity - port.getCapacity() + capacity; + MaxMinFlowMultiplexer.this.capacity = newCapacity; + changeInletCapacity(newCapacity); + } + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) { + float newCapacity = MaxMinFlowMultiplexer.this.capacity - port.getCapacity(); + MaxMinFlowMultiplexer.this.capacity = newCapacity; + releaseOutput(port); + changeInletCapacity(newCapacity); + } + + private void changeInletCapacity(float capacity) { + BitSet activeInputs = MaxMinFlowMultiplexer.this.activeInputs; + InPort[] inlets = MaxMinFlowMultiplexer.this.inlets; + + for (int i = activeInputs.nextSetBit(0); i != -1; i = activeInputs.nextSetBit(i + 1)) { + inlets[i].pull(capacity); + } + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/FlowSink.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/FlowSink.java new file mode 100644 index 00000000..69c94708 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/FlowSink.java @@ -0,0 +1,36 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.sink; + +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.Inlet; + +/** + * A {@link FlowStage} with a single input. + */ +public interface FlowSink { + /** + * Return the input of this {@link FlowSink}. + */ + Inlet getInput(); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/SimpleFlowSink.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/SimpleFlowSink.java new file mode 100644 index 00000000..fdfe5ee8 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/sink/SimpleFlowSink.java @@ -0,0 +1,123 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.sink; + +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.InHandler; +import org.opendc.simulator.flow2.InPort; +import org.opendc.simulator.flow2.Inlet; + +/** + * A sink with a fixed capacity. + */ +public final class SimpleFlowSink implements FlowSink, FlowStageLogic { + private final FlowStage stage; + private final InPort input; + private final Handler handler; + + /** + * Construct a new {@link SimpleFlowSink} with the specified initial capacity. + * + * @param graph The graph to add the sink to. + * @param initialCapacity The initial capacity of the sink. + */ + public SimpleFlowSink(FlowGraph graph, float initialCapacity) { + this.stage = graph.newStage(this); + this.handler = new Handler(); + this.input = stage.getInlet("in"); + this.input.pull(initialCapacity); + this.input.setMask(true); + this.input.setHandler(handler); + } + + /** + * Return the {@link Inlet} of this sink. + */ + @Override + public Inlet getInput() { + return input; + } + + /** + * Return the capacity of the sink. + */ + public float getCapacity() { + return input.getCapacity(); + } + + /** + * Update the capacity of the sink. + * + * @param capacity The new capacity to update the sink to. + */ + public void setCapacity(float capacity) { + input.pull(capacity); + stage.invalidate(); + } + + /** + * Return the flow rate of the sink. + */ + public float getRate() { + return input.getRate(); + } + + /** + * Remove this node from the graph. + */ + public void close() { + stage.close(); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + InPort input = this.input; + handler.rate = Math.min(input.getDemand(), input.getCapacity()); + return Long.MAX_VALUE; + } + + /** + * The {@link InHandler} implementation for the sink. + */ + private static final class Handler implements InHandler { + float rate; + + @Override + public float getRate(InPort port) { + return rate; + } + + @Override + public void onPush(InPort port, float demand) { + float capacity = port.getCapacity(); + rate = Math.min(demand, capacity); + } + + @Override + public void onUpstreamFinish(InPort port, Throwable cause) { + rate = 0.f; + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/EmptyFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/EmptyFlowSource.java new file mode 100644 index 00000000..2dcc66e4 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/EmptyFlowSource.java @@ -0,0 +1,65 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.source; + +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.OutPort; +import org.opendc.simulator.flow2.Outlet; + +/** + * An empty {@link FlowSource}. + */ +public final class EmptyFlowSource implements FlowSource, FlowStageLogic { + private final FlowStage stage; + private final OutPort output; + + /** + * Construct a new {@link EmptyFlowSource}. + */ + public EmptyFlowSource(FlowGraph graph) { + this.stage = graph.newStage(this); + this.output = stage.getOutlet("out"); + } + + /** + * Return the {@link Outlet} of the source. + */ + @Override + public Outlet getOutput() { + return output; + } + + /** + * Remove this node from the graph. + */ + public void close() { + stage.close(); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + return Long.MAX_VALUE; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/FlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/FlowSource.java new file mode 100644 index 00000000..f9432c33 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/FlowSource.java @@ -0,0 +1,36 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.source; + +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.Outlet; + +/** + * A {@link FlowStage} with a single output. + */ +public interface FlowSource { + /** + * Return the output of this {@link FlowSource}. + */ + Outlet getOutput(); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java new file mode 100644 index 00000000..a237c81e --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java @@ -0,0 +1,129 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.source; + +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.OutHandler; +import org.opendc.simulator.flow2.OutPort; +import org.opendc.simulator.flow2.Outlet; + +import java.util.function.Consumer; + +/** + * A {@link FlowSource} that ensures a flow is emitted for a specified amount of time at some utilization. + */ +public class RuntimeFlowSource implements FlowSource, FlowStageLogic { + private final float utilization; + + private final FlowStage stage; + private final OutPort output; + private final Consumer completionHandler; + + private long duration; + private long lastPull; + + /** + * Construct a {@link RuntimeFlowSource} instance. + * + * @param graph The {@link FlowGraph} to which this source belongs. + * @param duration The duration of the source. + * @param utilization The utilization of the capacity of the outlet. + * @param completionHandler A callback invoked when the source completes. + */ + public RuntimeFlowSource( + FlowGraph graph, long duration, float utilization, Consumer completionHandler) { + if (duration <= 0) { + throw new IllegalArgumentException("Duration must be positive and non-zero"); + } + + if (utilization <= 0.0) { + throw new IllegalArgumentException("Utilization must be positive and non-zero"); + } + + this.stage = graph.newStage(this); + this.output = stage.getOutlet("out"); + this.output.setHandler(new OutHandler() { + @Override + public void onPull(OutPort port, float capacity) {} + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) { + // Source cannot complete without re-connecting to another sink, so mark the source as completed + completionHandler.accept(RuntimeFlowSource.this); + } + }); + this.duration = duration; + this.utilization = utilization; + this.completionHandler = completionHandler; + this.lastPull = graph.getEngine().getClock().millis(); + } + + /** + * Construct a new {@link RuntimeFlowSource}. + * + * @param graph The {@link FlowGraph} to which this source belongs. + * @param duration The duration of the source. + * @param utilization The utilization of the capacity of the outlet. + */ + public RuntimeFlowSource(FlowGraph graph, long duration, float utilization) { + this(graph, duration, utilization, RuntimeFlowSource::close); + } + + /** + * Return the {@link Outlet} of the source. + */ + @Override + public Outlet getOutput() { + return output; + } + + /** + * Remove this node from the graph. + */ + public void close() { + stage.close(); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + long lastPull = this.lastPull; + this.lastPull = now; + + long delta = Math.max(0, now - lastPull); + + OutPort output = this.output; + float limit = output.getCapacity() * utilization; + long duration = this.duration - delta; + + if (duration <= 0) { + completionHandler.accept(this); + return Long.MAX_VALUE; + } + + this.duration = duration; + output.push(limit); + return now + duration; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java new file mode 100644 index 00000000..764a20a8 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.source; + +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.OutHandler; +import org.opendc.simulator.flow2.OutPort; +import org.opendc.simulator.flow2.Outlet; + +import java.util.function.Consumer; + +/** + * A flow source that contains a fixed amount and is pushed with a given utilization. + */ +public final class SimpleFlowSource implements FlowSource, FlowStageLogic { + private final float utilization; + private float remainingAmount; + private long lastPull; + + private final FlowStage stage; + private final OutPort output; + private final Consumer completionHandler; + + /** + * Construct a new {@link SimpleFlowSource}. + * + * @param graph The {@link FlowGraph} to which this source belongs. + * @param amount The amount to transfer via the outlet. + * @param utilization The utilization of the capacity of the outlet. + * @param completionHandler A callback invoked when the source completes. + */ + public SimpleFlowSource( + FlowGraph graph, float amount, float utilization, Consumer completionHandler) { + if (amount < 0.0) { + throw new IllegalArgumentException("Amount must be non-negative"); + } + + if (utilization <= 0.0) { + throw new IllegalArgumentException("Utilization must be positive and non-zero"); + } + + this.stage = graph.newStage(this); + this.output = stage.getOutlet("out"); + this.output.setHandler(new OutHandler() { + @Override + public void onPull(OutPort port, float capacity) {} + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) { + // Source cannot complete without re-connecting to another sink, so mark the source as completed + completionHandler.accept(SimpleFlowSource.this); + } + }); + this.completionHandler = completionHandler; + this.utilization = utilization; + this.remainingAmount = amount; + this.lastPull = graph.getEngine().getClock().millis(); + } + + /** + * Construct a new {@link SimpleFlowSource}. + * + * @param graph The {@link FlowGraph} to which this source belongs. + * @param amount The amount to transfer via the outlet. + * @param utilization The utilization of the capacity of the outlet. + */ + public SimpleFlowSource(FlowGraph graph, float amount, float utilization) { + this(graph, amount, utilization, SimpleFlowSource::close); + } + + /** + * Return the {@link Outlet} of the source. + */ + @Override + public Outlet getOutput() { + return output; + } + + /** + * Remove this node from the graph. + */ + public void close() { + stage.close(); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + long lastPull = this.lastPull; + this.lastPull = now; + + long delta = Math.max(0, now - lastPull); + + OutPort output = this.output; + float consumed = output.getRate() * delta / 1000.f; + float limit = output.getCapacity() * utilization; + + float remainingAmount = this.remainingAmount - consumed; + this.remainingAmount = remainingAmount; + + long duration = (long) Math.ceil(remainingAmount / limit * 1000); + + if (duration <= 0) { + completionHandler.accept(this); + return Long.MAX_VALUE; + } + + output.push(limit); + return now + duration; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java new file mode 100644 index 00000000..96d43aef --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java @@ -0,0 +1,152 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.source; + +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.OutHandler; +import org.opendc.simulator.flow2.OutPort; +import org.opendc.simulator.flow2.Outlet; + +import java.util.function.Consumer; + +/** + * A flow source that replays a sequence of fragments, each indicating the flow rate for some period of time. + */ +public final class TraceFlowSource implements FlowSource, FlowStageLogic { + private final OutPort output; + private final long[] deadlines; + private final float[] usages; + private final int size; + private int index; + + private final FlowStage stage; + private final Consumer completionHandler; + + /** + * Construct a {@link TraceFlowSource}. + * + * @param graph The {@link FlowGraph} to which the source belongs. + * @param trace The {@link Trace} to replay. + * @param completionHandler The completion handler to invoke when the source finishes. + */ + public TraceFlowSource(FlowGraph graph, Trace trace, Consumer completionHandler) { + this.stage = graph.newStage(this); + this.output = stage.getOutlet("out"); + this.output.setHandler(new OutHandler() { + @Override + public void onPull(OutPort port, float capacity) {} + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) { + // Source cannot complete without re-connecting to another sink, so mark the source as completed + completionHandler.accept(TraceFlowSource.this); + } + }); + this.deadlines = trace.deadlines; + this.usages = trace.usages; + this.size = trace.size; + this.completionHandler = completionHandler; + } + + /** + * Construct a {@link TraceFlowSource}. + * + * @param graph The {@link FlowGraph} to which the source belongs. + * @param trace The {@link Trace} to replay. + */ + public TraceFlowSource(FlowGraph graph, Trace trace) { + this(graph, trace, TraceFlowSource::close); + } + + @Override + public Outlet getOutput() { + return output; + } + + /** + * Remove this node from the graph. + */ + public void close() { + stage.close(); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + int size = this.size; + int index = this.index; + long[] deadlines = this.deadlines; + long deadline; + + do { + deadline = deadlines[index]; + } while (deadline <= now && ++index < size); + + if (index >= size) { + output.push(0.0f); + completionHandler.accept(this); + return Long.MAX_VALUE; + } + + this.index = index; + float usage = usages[index]; + output.push(usage); + + return deadline; + } + + /** + * A trace describes the workload over time. + */ + public static final class Trace { + private final long[] deadlines; + private final float[] usages; + private final int size; + + /** + * Construct a {@link Trace}. + * + * @param deadlines The deadlines of the trace fragments. + * @param usages The usages of the trace fragments. + * @param size The size of the trace. + */ + public Trace(long[] deadlines, float[] usages, int size) { + this.deadlines = deadlines; + this.usages = usages; + this.size = size; + } + + public long[] getDeadlines() { + return deadlines; + } + + public float[] getUsages() { + return usages; + } + + public int getSize() { + return size; + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowEngineTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowEngineTest.kt new file mode 100644 index 00000000..839835ce --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowEngineTest.kt @@ -0,0 +1,197 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2 + +import io.mockk.mockk +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotEquals +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import org.opendc.simulator.flow2.mux.MaxMinFlowMultiplexer +import org.opendc.simulator.flow2.sink.SimpleFlowSink +import org.opendc.simulator.flow2.source.SimpleFlowSource +import org.opendc.simulator.kotlin.runSimulation + +/** + * Smoke tests for the Flow API. + */ +class FlowEngineTest { + @Test + fun testSmoke() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val multiplexer = MaxMinFlowMultiplexer(graph) + val sink = SimpleFlowSink(graph, 2.0f) + + graph.connect(multiplexer.newOutput(), sink.input) + + val sourceA = SimpleFlowSource(graph, 2000.0f, 0.8f) + val sourceB = SimpleFlowSource(graph, 2000.0f, 0.8f) + + graph.connect(sourceA.output, multiplexer.newInput()) + graph.connect(sourceB.output, multiplexer.newInput()) + } + + @Test + fun testConnectInvalidInlet() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val inlet = mockk() + val source = SimpleFlowSource(graph, 2000.0f, 0.8f) + assertThrows { graph.connect(source.output, inlet) } + } + + @Test + fun testConnectInvalidOutlet() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val outlet = mockk() + val sink = SimpleFlowSink(graph, 2.0f) + assertThrows { graph.connect(outlet, sink.input) } + } + + @Test + fun testConnectInletBelongsToDifferentGraph() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graphA = engine.newGraph() + val graphB = engine.newGraph() + + val sink = SimpleFlowSink(graphB, 2.0f) + val source = SimpleFlowSource(graphA, 2000.0f, 0.8f) + + assertThrows { graphA.connect(source.output, sink.input) } + } + + @Test + fun testConnectOutletBelongsToDifferentGraph() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graphA = engine.newGraph() + val graphB = engine.newGraph() + + val sink = SimpleFlowSink(graphA, 2.0f) + val source = SimpleFlowSource(graphB, 2000.0f, 0.8f) + + assertThrows { graphA.connect(source.output, sink.input) } + } + + @Test + fun testConnectInletAlreadyConnected() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sink = SimpleFlowSink(graph, 2.0f) + val sourceA = SimpleFlowSource(graph, 2000.0f, 0.8f) + val sourceB = SimpleFlowSource(graph, 2000.0f, 0.8f) + + graph.connect(sourceA.output, sink.input) + assertThrows { graph.connect(sourceB.output, sink.input) } + } + + @Test + fun testConnectOutletAlreadyConnected() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sinkA = SimpleFlowSink(graph, 2.0f) + val sinkB = SimpleFlowSink(graph, 2.0f) + val source = SimpleFlowSource(graph, 2000.0f, 0.8f) + + graph.connect(source.output, sinkA.input) + assertThrows { graph.connect(source.output, sinkB.input) } + } + + @Test + fun testDisconnectInletInvalid() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val inlet = mockk() + assertThrows { graph.disconnect(inlet) } + } + + @Test + fun testDisconnectOutletInvalid() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val outlet = mockk() + assertThrows { graph.disconnect(outlet) } + } + + @Test + fun testDisconnectInletInvalidGraph() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graphA = engine.newGraph() + val graphB = engine.newGraph() + + val sink = SimpleFlowSink(graphA, 2.0f) + + assertThrows { graphB.disconnect(sink.input) } + } + + @Test + fun testDisconnectOutletInvalidGraph() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graphA = engine.newGraph() + val graphB = engine.newGraph() + + val source = SimpleFlowSource(graphA, 2000.0f, 0.8f) + + assertThrows { graphB.disconnect(source.output) } + } + + @Test + fun testInletEquality() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sinkA = SimpleFlowSink(graph, 2.0f) + val sinkB = SimpleFlowSink(graph, 2.0f) + + val multiplexer = MaxMinFlowMultiplexer(graph) + + assertEquals(sinkA.input, sinkA.input) + assertNotEquals(sinkA.input, sinkB.input) + + assertNotEquals(multiplexer.newInput(), multiplexer.newInput()) + } + + @Test + fun testOutletEquality() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sourceA = SimpleFlowSource(graph, 2000.0f, 0.8f) + val sourceB = SimpleFlowSource(graph, 2000.0f, 0.8f) + + val multiplexer = MaxMinFlowMultiplexer(graph) + + assertEquals(sourceA.output, sourceA.output) + assertNotEquals(sourceA.output, sourceB.output) + + assertNotEquals(multiplexer.newOutput(), multiplexer.newOutput()) + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowTimerQueueTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowTimerQueueTest.kt new file mode 100644 index 00000000..1824959c --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/FlowTimerQueueTest.kt @@ -0,0 +1,385 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2 + +import io.mockk.mockk +import org.junit.jupiter.api.Assertions.assertAll +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test + +/** + * Test suite for the [FlowTimerQueue] class. + */ +class FlowTimerQueueTest { + private lateinit var queue: FlowTimerQueue + + @BeforeEach + fun setUp() { + queue = FlowTimerQueue(3) + } + + /** + * Test whether a call to [FlowTimerQueue.poll] returns `null` for an empty queue. + */ + @Test + fun testPollEmpty() { + assertAll( + { assertEquals(Long.MAX_VALUE, queue.peekDeadline()) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test whether a call to [FlowTimerQueue.poll] returns the proper value for a queue with a single entry. + */ + @Test + fun testSingleEntry() { + val entry = mockk() + entry.deadline = 100 + entry.timerIndex = -1 + + queue.enqueue(entry) + + assertAll( + { assertEquals(100, queue.peekDeadline()) }, + { assertNull(queue.poll(10L)) }, + { assertEquals(entry, queue.poll(200L)) }, + { assertNull(queue.poll(200L)) } + ) + } + + /** + * Test whether [FlowTimerQueue.poll] returns values in the queue in the proper order. + */ + @Test + fun testMultipleEntries() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 10 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + assertAll( + { assertEquals(10, queue.peekDeadline()) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that the queue is properly resized when the number of entries exceed the capacity. + */ + @Test + fun testResize() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + val entryD = mockk() + entryD.deadline = 31 + entryD.timerIndex = -1 + + queue.enqueue(entryD) + + assertAll( + { assertEquals(20, queue.peekDeadline()) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryD, queue.poll(100L)) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test to verify that we can change the deadline of the last element in the queue. + */ + @Test + fun testChangeDeadlineTail() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + entryA.deadline = 10 + queue.enqueue(entryA) + + assertAll( + { assertEquals(10, queue.peekDeadline()) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that we can change the deadline of the head entry in the queue. + */ + @Test + fun testChangeDeadlineMiddle() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + entryC.deadline = 10 + queue.enqueue(entryC) + + assertAll( + { assertEquals(10, queue.peekDeadline()) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that we can change the deadline of the head entry in the queue. + */ + @Test + fun testChangeDeadlineHead() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + entryB.deadline = 30 + queue.enqueue(entryB) + + assertAll( + { assertEquals(30, queue.peekDeadline()) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that an unchanged deadline results in a no-op. + */ + @Test + fun testChangeDeadlineNop() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + // Should be a no-op + queue.enqueue(entryA) + + assertAll( + { assertEquals(20, queue.peekDeadline()) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that we can remove an entry from the end of the queue. + */ + @Test + fun testRemoveEntryTail() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + entryC.deadline = Long.MAX_VALUE + queue.enqueue(entryC) + + assertAll( + { assertEquals(20, queue.peekDeadline()) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that we can remove an entry from the head of the queue. + */ + @Test + fun testRemoveEntryHead() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + entryB.deadline = Long.MAX_VALUE + queue.enqueue(entryB) + + assertAll( + { assertEquals(58, queue.peekDeadline()) }, + { assertEquals(entryC, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } + + /** + * Test that we can remove an entry from the middle of a queue. + */ + @Test + fun testRemoveEntryMiddle() { + val entryA = mockk() + entryA.deadline = 100 + entryA.timerIndex = -1 + + queue.enqueue(entryA) + + val entryB = mockk() + entryB.deadline = 20 + entryB.timerIndex = -1 + + queue.enqueue(entryB) + + val entryC = mockk() + entryC.deadline = 58 + entryC.timerIndex = -1 + + queue.enqueue(entryC) + + entryC.deadline = Long.MAX_VALUE + queue.enqueue(entryC) + + assertAll( + { assertEquals(20, queue.peekDeadline()) }, + { assertEquals(entryB, queue.poll(100L)) }, + { assertEquals(entryA, queue.poll(100L)) }, + { assertNull(queue.poll(100L)) } + ) + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/InvocationStackTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/InvocationStackTest.kt new file mode 100644 index 00000000..2250fe87 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/InvocationStackTest.kt @@ -0,0 +1,71 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2 + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +/** + * Test suite for the [InvocationStack] class. + */ +class InvocationStackTest { + private val stack = InvocationStack(2) + + @Test + fun testPollEmpty() { + assertEquals(Long.MAX_VALUE, stack.poll()) + } + + @Test + fun testAddSingle() { + assertTrue(stack.tryAdd(10)) + assertEquals(10, stack.poll()) + } + + @Test + fun testAddLater() { + assertTrue(stack.tryAdd(10)) + assertFalse(stack.tryAdd(15)) + assertEquals(10, stack.poll()) + } + + @Test + fun testAddEarlier() { + assertTrue(stack.tryAdd(10)) + assertTrue(stack.tryAdd(5)) + assertEquals(5, stack.poll()) + assertEquals(10, stack.poll()) + } + + @Test + fun testCapacityExceeded() { + assertTrue(stack.tryAdd(10)) + assertTrue(stack.tryAdd(5)) + assertTrue(stack.tryAdd(2)) + assertEquals(2, stack.poll()) + assertEquals(5, stack.poll()) + assertEquals(10, stack.poll()) + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexerTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexerTest.kt new file mode 100644 index 00000000..ba339ee3 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexerTest.kt @@ -0,0 +1,54 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.mux + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.opendc.simulator.flow2.FlowEngine +import org.opendc.simulator.flow2.sink.SimpleFlowSink +import org.opendc.simulator.flow2.source.SimpleFlowSource +import org.opendc.simulator.kotlin.runSimulation + +/** + * Test suite for the [MaxMinFlowMultiplexer] class. + */ +class MaxMinFlowMultiplexerTest { + @Test + fun testSmoke() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + val switch = MaxMinFlowMultiplexer(graph) + + val sinks = List(2) { SimpleFlowSink(graph, 2000.0f) } + for (source in sinks) { + graph.connect(switch.newOutput(), source.input) + } + + val source = SimpleFlowSource(graph, 2000.0f, 1.0f) + graph.connect(source.output, switch.newInput()) + + advanceUntilIdle() + + assertEquals(500, clock.millis()) + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/sink/FlowSinkTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/sink/FlowSinkTest.kt new file mode 100644 index 00000000..a75efba3 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/sink/FlowSinkTest.kt @@ -0,0 +1,124 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.sink + +import kotlinx.coroutines.delay +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.opendc.simulator.flow2.FlowEngine +import org.opendc.simulator.flow2.source.SimpleFlowSource +import org.opendc.simulator.flow2.source.TraceFlowSource +import org.opendc.simulator.kotlin.runSimulation +import java.util.concurrent.ThreadLocalRandom + +/** + * Test suite for the [SimpleFlowSink] class. + */ +class FlowSinkTest { + @Test + fun testSmoke() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sink = SimpleFlowSink(graph, 1.0f) + val source = SimpleFlowSource(graph, 2.0f, 1.0f) + + graph.connect(source.output, sink.input) + advanceUntilIdle() + + assertEquals(2000, clock.millis()) + } + + @Test + fun testAdjustCapacity() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sink = SimpleFlowSink(graph, 1.0f) + val source = SimpleFlowSource(graph, 2.0f, 1.0f) + + graph.connect(source.output, sink.input) + + delay(1000) + sink.capacity = 0.5f + + advanceUntilIdle() + + assertEquals(3000, clock.millis()) + } + + @Test + fun testUtilization() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sink = SimpleFlowSink(graph, 1.0f) + val source = SimpleFlowSource(graph, 2.0f, 0.5f) + + graph.connect(source.output, sink.input) + advanceUntilIdle() + + assertEquals(4000, clock.millis()) + } + + @Test + fun testFragments() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val sink = SimpleFlowSink(graph, 1.0f) + val trace = TraceFlowSource.Trace( + longArrayOf(1000, 2000, 3000, 4000), + floatArrayOf(1.0f, 0.5f, 2.0f, 1.0f), + 4 + ) + val source = TraceFlowSource( + graph, + trace + ) + + graph.connect(source.output, sink.input) + advanceUntilIdle() + + assertEquals(4000, clock.millis()) + } + + @Test + fun benchmarkSink() { + val random = ThreadLocalRandom.current() + val traceSize = 10000000 + val trace = TraceFlowSource.Trace( + LongArray(traceSize) { it * 1000L }, + FloatArray(traceSize) { random.nextDouble(0.0, 4500.0).toFloat() }, + traceSize + ) + + return runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + val sink = SimpleFlowSink(graph, 4200.0f) + val source = TraceFlowSource(graph, trace) + graph.connect(source.output, sink.input) + } + } +} -- cgit v1.2.3 From e4f4e1c4ebd02278dc1c24ee481357989af6abe5 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Thu, 1 Sep 2022 12:02:06 +0200 Subject: feat(sim/flow): Support flow transformations This change adds a new component, FlowTransform, which is able to transform the flow from one component to another, based on some inversible transformation. Such as component is useful when converting between different units of flow between different components. --- .../org/opendc/simulator/flow2/FlowBenchmarks.kt | 16 +++ .../opendc/simulator/flow2/util/FlowTransform.java | 41 +++++++ .../simulator/flow2/util/FlowTransformer.java | 124 +++++++++++++++++++++ .../simulator/flow2/util/FlowTransforms.java | 57 ++++++++++ 4 files changed, 238 insertions(+) create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransform.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransformer.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransforms.java (limited to 'opendc-simulator/opendc-simulator-flow/src') diff --git a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt index 1b0e2e9e..fb112082 100644 --- a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt +++ b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow2/FlowBenchmarks.kt @@ -26,6 +26,8 @@ import kotlinx.coroutines.launch import org.opendc.simulator.flow2.mux.MaxMinFlowMultiplexer import org.opendc.simulator.flow2.sink.SimpleFlowSink import org.opendc.simulator.flow2.source.TraceFlowSource +import org.opendc.simulator.flow2.util.FlowTransformer +import org.opendc.simulator.flow2.util.FlowTransforms import org.opendc.simulator.kotlin.runSimulation import org.openjdk.jmh.annotations.Benchmark import org.openjdk.jmh.annotations.Fork @@ -66,6 +68,20 @@ class FlowBenchmarks { } } + @Benchmark + fun benchmarkForward() { + return runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + val sink = SimpleFlowSink(graph, 4200.0f) + val source = TraceFlowSource(graph, trace) + val forwarder = FlowTransformer(graph, FlowTransforms.noop()) + + graph.connect(source.output, forwarder.input) + graph.connect(forwarder.output, sink.input) + } + } + @Benchmark fun benchmarkMuxMaxMinSingleSource() { return runSimulation { diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransform.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransform.java new file mode 100644 index 00000000..51ea7df3 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransform.java @@ -0,0 +1,41 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.util; + +import org.opendc.simulator.flow2.FlowGraph; + +/** + * A {@link FlowTransform} describes a transformation between two components in a {@link FlowGraph} that might operate + * at different units of flow. + */ +public interface FlowTransform { + /** + * Apply the transform to the specified flow rate. + */ + float apply(float value); + + /** + * Apply the inverse of the transformation to the specified flow rate. + */ + float applyInverse(float value); +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransformer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransformer.java new file mode 100644 index 00000000..852240d8 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransformer.java @@ -0,0 +1,124 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.util; + +import org.opendc.simulator.flow2.*; +import org.opendc.simulator.flow2.sink.FlowSink; +import org.opendc.simulator.flow2.source.FlowSource; + +/** + * Helper class to transform flow from outlet to inlet. + */ +public final class FlowTransformer implements FlowStageLogic, FlowSource, FlowSink { + private final FlowStage stage; + private final InPort input; + private final OutPort output; + + /** + * Construct a new {@link FlowTransformer}. + */ + public FlowTransformer(FlowGraph graph, FlowTransform transform) { + this.stage = graph.newStage(this); + this.input = stage.getInlet("in"); + this.output = stage.getOutlet("out"); + + this.input.setHandler(new ForwardInHandler(output, transform)); + this.input.setMask(true); + this.output.setHandler(new ForwardOutHandler(input, transform)); + this.output.setMask(true); + } + + /** + * Return the {@link Outlet} of the transformer. + */ + @Override + public Outlet getOutput() { + return output; + } + + /** + * Return the {@link Inlet} of the transformer. + */ + @Override + public Inlet getInput() { + return input; + } + + /** + * Close the transformer. + */ + void close() { + stage.close(); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + return Long.MAX_VALUE; + } + + private static class ForwardInHandler implements InHandler { + private final OutPort output; + private final FlowTransform transform; + + ForwardInHandler(OutPort output, FlowTransform transform) { + this.output = output; + this.transform = transform; + } + + @Override + public float getRate(InPort port) { + return transform.applyInverse(output.getRate()); + } + + @Override + public void onPush(InPort port, float demand) { + float rate = transform.apply(demand); + output.push(rate); + } + + @Override + public void onUpstreamFinish(InPort port, Throwable cause) { + output.fail(cause); + } + } + + private static class ForwardOutHandler implements OutHandler { + private final InPort input; + private final FlowTransform transform; + + ForwardOutHandler(InPort input, FlowTransform transform) { + this.input = input; + this.transform = transform; + } + + @Override + public void onPull(OutPort port, float capacity) { + input.pull(transform.applyInverse(capacity)); + } + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) { + input.cancel(cause); + } + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransforms.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransforms.java new file mode 100644 index 00000000..428dbfca --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/util/FlowTransforms.java @@ -0,0 +1,57 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.util; + +/** + * A collection of common {@link FlowTransform} implementations. + */ +public class FlowTransforms { + /** + * Prevent construction of this class. + */ + private FlowTransforms() {} + + /** + * Return a {@link FlowTransform} that forwards the flow rate unmodified. + */ + public static FlowTransform noop() { + return NoopFlowTransform.INSTANCE; + } + + /** + * No-op implementation of a {@link FlowTransform}. + */ + private static final class NoopFlowTransform implements FlowTransform { + static final NoopFlowTransform INSTANCE = new NoopFlowTransform(); + + @Override + public float apply(float value) { + return value; + } + + @Override + public float applyInverse(float value) { + return value; + } + } +} -- cgit v1.2.3 From 66a5266c2e6f060a11fffef1e9eed9e716056853 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Thu, 1 Sep 2022 17:10:52 +0200 Subject: feat(sim/flow): Add forwarding flow multiplexer This change implements a new `FlowMultiplexer` that forwards the inputs directly to one of the pre-allocated outputs. --- .../flow2/mux/ForwardingFlowMultiplexer.java | 233 +++++++++++++++++++++ .../flow2/mux/ForwardingFlowMultiplexerTest.kt | 66 ++++++ 2 files changed, 299 insertions(+) create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java create mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexerTest.kt (limited to 'opendc-simulator/opendc-simulator-flow/src') diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java new file mode 100644 index 00000000..9ff6a4c8 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java @@ -0,0 +1,233 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.mux; + +import java.util.Arrays; +import java.util.BitSet; +import org.opendc.simulator.flow2.FlowGraph; +import org.opendc.simulator.flow2.FlowStage; +import org.opendc.simulator.flow2.FlowStageLogic; +import org.opendc.simulator.flow2.InHandler; +import org.opendc.simulator.flow2.InPort; +import org.opendc.simulator.flow2.Inlet; +import org.opendc.simulator.flow2.OutHandler; +import org.opendc.simulator.flow2.OutPort; +import org.opendc.simulator.flow2.Outlet; + +/** + * A {@link 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. + */ +public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowStageLogic { + public final IdleInHandler IDLE_IN_HANDLER = new IdleInHandler(); + public final IdleOutHandler IDLE_OUT_HANDLER = new IdleOutHandler(); + + private final FlowStage stage; + + private InPort[] inlets; + private OutPort[] outlets; + private final BitSet activeInputs; + private final BitSet activeOutputs; + private final BitSet availableOutputs; + + public ForwardingFlowMultiplexer(FlowGraph graph) { + this.stage = graph.newStage(this); + + this.inlets = new InPort[4]; + this.activeInputs = new BitSet(); + this.outlets = new OutPort[4]; + this.activeOutputs = new BitSet(); + this.availableOutputs = new BitSet(); + } + + @Override + public int getInputCount() { + return activeInputs.length(); + } + + @Override + public Inlet newInput() { + final BitSet activeInputs = this.activeInputs; + int slot = activeInputs.nextClearBit(0); + + InPort inPort = stage.getInlet("in" + slot); + inPort.setMask(true); + + InPort[] inlets = this.inlets; + if (slot >= inlets.length) { + int newLength = inlets.length + (inlets.length >> 1); + inlets = Arrays.copyOf(inlets, newLength); + this.inlets = inlets; + } + + final BitSet availableOutputs = this.availableOutputs; + int outSlot = availableOutputs.nextSetBit(0); + + if (outSlot < 0) { + throw new IllegalStateException("No capacity available for a new input"); + } + + inlets[slot] = inPort; + activeInputs.set(slot); + + OutPort outPort = outlets[outSlot]; + availableOutputs.clear(outSlot); + + inPort.setHandler(new ForwardingInHandler(outPort)); + outPort.setHandler(new ForwardingOutHandler(inPort)); + + inPort.pull(outPort.getCapacity()); + + return inPort; + } + + @Override + public void releaseInput(Inlet inlet) { + InPort port = (InPort) inlet; + int slot = port.getId(); + + final BitSet activeInputs = this.activeInputs; + + if (!activeInputs.get(slot)) { + return; + } + + port.cancel(null); + activeInputs.clear(slot); + + ForwardingInHandler inHandler = (ForwardingInHandler) port.getHandler(); + availableOutputs.set(inHandler.output.getId()); + + port.setHandler(IDLE_IN_HANDLER); + } + + @Override + public int getOutputCount() { + return activeOutputs.length(); + } + + @Override + public Outlet newOutput() { + final BitSet activeOutputs = this.activeOutputs; + int slot = activeOutputs.nextClearBit(0); + + OutPort port = stage.getOutlet("out" + slot); + OutPort[] outlets = this.outlets; + if (slot >= outlets.length) { + int newLength = outlets.length + (outlets.length >> 1); + outlets = Arrays.copyOf(outlets, newLength); + this.outlets = outlets; + } + outlets[slot] = port; + + activeOutputs.set(slot); + availableOutputs.set(slot); + return port; + } + + @Override + public void releaseOutput(Outlet outlet) { + OutPort port = (OutPort) outlet; + int slot = port.getId(); + activeInputs.clear(slot); + availableOutputs.clear(slot); + port.complete(); + + port.setHandler(IDLE_OUT_HANDLER); + } + + @Override + public long onUpdate(FlowStage ctx, long now) { + return Long.MAX_VALUE; + } + + class ForwardingInHandler implements InHandler { + final OutPort output; + + ForwardingInHandler(OutPort output) { + this.output = output; + } + + @Override + public float getRate(InPort port) { + return output.getRate(); + } + + @Override + public void onPush(InPort port, float rate) { + output.push(rate); + } + + @Override + public void onUpstreamFinish(InPort port, Throwable cause) { + final OutPort output = this.output; + output.push(0.f); + + releaseInput(port); + } + } + + private class ForwardingOutHandler implements OutHandler { + private final InPort input; + + ForwardingOutHandler(InPort input) { + this.input = input; + } + + @Override + public void onPull(OutPort port, float capacity) { + input.pull(capacity); + } + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) { + input.cancel(cause); + + releaseOutput(port); + } + } + + private static class IdleInHandler implements InHandler { + @Override + public float getRate(InPort port) { + return 0.f; + } + + @Override + public void onPush(InPort port, float rate) { + port.cancel(new IllegalStateException("Inlet is not allocated")); + } + + @Override + public void onUpstreamFinish(InPort port, Throwable cause) {} + } + + private static class IdleOutHandler implements OutHandler { + @Override + public void onPull(OutPort port, float capacity) {} + + @Override + public void onDownstreamFinish(OutPort port, Throwable cause) {} + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexerTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexerTest.kt new file mode 100644 index 00000000..a2ed2195 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexerTest.kt @@ -0,0 +1,66 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.mux + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertAll +import org.opendc.simulator.flow2.FlowEngine +import org.opendc.simulator.flow2.sink.SimpleFlowSink +import org.opendc.simulator.flow2.source.TraceFlowSource +import org.opendc.simulator.kotlin.runSimulation + +/** + * Test suite for the [ForwardingFlowMultiplexer] class. + */ +class ForwardingFlowMultiplexerTest { + /** + * Test a trace workload. + */ + @Test + fun testTrace() = runSimulation { + val engine = FlowEngine.create(coroutineContext, clock) + val graph = engine.newGraph() + + val switch = ForwardingFlowMultiplexer(graph) + val sink = SimpleFlowSink(graph, 3200.0f) + graph.connect(switch.newOutput(), sink.input) + + val workload = + TraceFlowSource( + graph, + TraceFlowSource.Trace( + longArrayOf(1000, 2000, 3000, 4000), + floatArrayOf(28.0f, 3500.0f, 0.0f, 183.0f), + 4 + ) + ) + graph.connect(workload.output, switch.newInput()) + + advanceUntilIdle() + + assertAll( + { assertEquals(4000, clock.millis()) { "Took enough time" } } + ) + } +} -- cgit v1.2.3 From 5abcbaa672d029fb390156a83c29d8d47a215f4f Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 12 Oct 2022 15:02:10 +0200 Subject: feat(sim/flow): Expose metrics on FlowMultiplexer --- .../simulator/flow2/mux/FlowMultiplexer.java | 15 ++++++++++ .../flow2/mux/ForwardingFlowMultiplexer.java | 32 ++++++++++++++++++++++ .../simulator/flow2/mux/MaxMinFlowMultiplexer.java | 15 ++++++++++ 3 files changed, 62 insertions(+) (limited to 'opendc-simulator/opendc-simulator-flow/src') diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java index 1a99d0cf..2a23b039 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java @@ -67,4 +67,19 @@ public interface FlowMultiplexer { * @param outlet The outlet to release. */ void releaseOutput(Outlet outlet); + + /** + * Return the total input capacity of the {@link FlowMultiplexer}. + */ + float getCapacity(); + + /** + * Return the total input demand for the {@link FlowMultiplexer}. + */ + float getDemand(); + + /** + * Return the total input rate for the {@link FlowMultiplexer}. + */ + float getRate(); } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java index 9ff6a4c8..6394c3fd 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java @@ -51,6 +51,9 @@ public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowSta private final BitSet activeOutputs; private final BitSet availableOutputs; + private float capacity = 0.f; + private float demand = 0.f; + public ForwardingFlowMultiplexer(FlowGraph graph) { this.stage = graph.newStage(this); @@ -61,6 +64,27 @@ public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowSta this.availableOutputs = new BitSet(); } + @Override + public float getCapacity() { + return capacity; + } + + @Override + public float getDemand() { + return demand; + } + + @Override + public float getRate() { + final BitSet activeOutputs = this.activeOutputs; + final OutPort[] outlets = this.outlets; + float rate = 0.f; + for (int i = activeOutputs.nextSetBit(0); i != -1; i = activeOutputs.nextSetBit(i + 1)) { + rate += outlets[i].getRate(); + } + return rate; + } + @Override public int getInputCount() { return activeInputs.length(); @@ -176,11 +200,15 @@ public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowSta @Override public void onPush(InPort port, float rate) { + ForwardingFlowMultiplexer.this.demand += -port.getDemand() + rate; + output.push(rate); } @Override public void onUpstreamFinish(InPort port, Throwable cause) { + ForwardingFlowMultiplexer.this.demand -= port.getDemand(); + final OutPort output = this.output; output.push(0.f); @@ -197,11 +225,15 @@ public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowSta @Override public void onPull(OutPort port, float capacity) { + ForwardingFlowMultiplexer.this.capacity += -port.getCapacity() + capacity; + input.pull(capacity); } @Override public void onDownstreamFinish(OutPort port, Throwable cause) { + ForwardingFlowMultiplexer.this.capacity -= port.getCapacity(); + input.cancel(cause); releaseOutput(port); diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java index ca0639f5..77922066 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java @@ -75,6 +75,21 @@ public final class MaxMinFlowMultiplexer implements FlowMultiplexer, FlowStageLo this.outlets = new OutPort[4]; } + @Override + public float getCapacity() { + return capacity; + } + + @Override + public float getDemand() { + return demand; + } + + @Override + public float getRate() { + return rate; + } + @Override public long onUpdate(FlowStage ctx, long now) { float capacity = this.capacity; -- cgit v1.2.3 From 44215bd668c5fa3efe2f57fc577824478b00af57 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Thu, 1 Sep 2022 14:38:34 +0200 Subject: refactor(sim/compute): Re-implement using flow2 This change re-implements the OpenDC compute simulator framework using the new flow2 framework for modelling multi-edge flow networks. The re-implementation is written in Java and focusses on performance and clean API surface. --- .../simulator/flow2/mux/FlowMultiplexer.java | 10 +++++ .../flow2/mux/FlowMultiplexerFactory.java | 51 ++++++++++++++++++++++ .../flow2/mux/ForwardingFlowMultiplexer.java | 15 +++++++ .../simulator/flow2/mux/MaxMinFlowMultiplexer.java | 20 +++++++-- .../simulator/flow2/source/RuntimeFlowSource.java | 5 +-- .../simulator/flow2/source/SimpleFlowSource.java | 5 +-- .../simulator/flow2/source/TraceFlowSource.java | 3 +- 7 files changed, 98 insertions(+), 11 deletions(-) create mode 100644 opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexerFactory.java (limited to 'opendc-simulator/opendc-simulator-flow/src') diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java index 2a23b039..dec98955 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexer.java @@ -30,6 +30,16 @@ import org.opendc.simulator.flow2.Outlet; * A {@link FlowStageLogic} that multiplexes multiple inputs over (possibly) multiple outputs. */ public interface FlowMultiplexer { + /** + * Return maximum number of inputs supported by the multiplexer. + */ + int getMaxInputs(); + + /** + * Return maximum number of outputs supported by the multiplexer. + */ + int getMaxOutputs(); + /** * Return the number of active inputs on this multiplexer. */ diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexerFactory.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexerFactory.java new file mode 100644 index 00000000..0b5b9141 --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/FlowMultiplexerFactory.java @@ -0,0 +1,51 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.flow2.mux; + +import org.opendc.simulator.flow2.FlowGraph; + +/** + * Factory interface for a {@link FlowMultiplexer} implementation. + */ +public interface FlowMultiplexerFactory { + /** + * Construct a new {@link FlowMultiplexer} belonging to the specified {@link FlowGraph}. + * + * @param graph The graph to which the multiplexer belongs. + */ + FlowMultiplexer newMultiplexer(FlowGraph graph); + + /** + * Return a {@link FlowMultiplexerFactory} for {@link ForwardingFlowMultiplexer} instances. + */ + static FlowMultiplexerFactory forwardingMultiplexer() { + return ForwardingFlowMultiplexer.FACTORY; + } + + /** + * Return a {@link FlowMultiplexerFactory} for {@link MaxMinFlowMultiplexer} instances. + */ + static FlowMultiplexerFactory maxMinMultiplexer() { + return MaxMinFlowMultiplexer.FACTORY; + } +} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java index 6394c3fd..abe3510b 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/ForwardingFlowMultiplexer.java @@ -40,6 +40,11 @@ import org.opendc.simulator.flow2.Outlet; * inputs as outputs. */ public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowStageLogic { + /** + * Factory implementation for this implementation. + */ + static FlowMultiplexerFactory FACTORY = ForwardingFlowMultiplexer::new; + public final IdleInHandler IDLE_IN_HANDLER = new IdleInHandler(); public final IdleOutHandler IDLE_OUT_HANDLER = new IdleOutHandler(); @@ -85,6 +90,16 @@ public final class ForwardingFlowMultiplexer implements FlowMultiplexer, FlowSta return rate; } + @Override + public int getMaxInputs() { + return getOutputCount(); + } + + @Override + public int getMaxOutputs() { + return Integer.MAX_VALUE; + } + @Override public int getInputCount() { return activeInputs.length(); diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java index 77922066..ac5c4f5c 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/mux/MaxMinFlowMultiplexer.java @@ -22,6 +22,8 @@ package org.opendc.simulator.flow2.mux; +import java.util.Arrays; +import java.util.BitSet; import org.opendc.simulator.flow2.FlowGraph; import org.opendc.simulator.flow2.FlowStage; import org.opendc.simulator.flow2.FlowStageLogic; @@ -32,9 +34,6 @@ import org.opendc.simulator.flow2.OutHandler; import org.opendc.simulator.flow2.OutPort; import org.opendc.simulator.flow2.Outlet; -import java.util.Arrays; -import java.util.BitSet; - /** * A {@link FlowMultiplexer} implementation that distributes the available capacity of the outputs over the inputs * using max-min fair sharing. @@ -43,6 +42,11 @@ import java.util.BitSet; * output capacity, but allows individual inputs to use more capacity if there is still capacity left. */ public final class MaxMinFlowMultiplexer implements FlowMultiplexer, FlowStageLogic { + /** + * Factory implementation for this implementation. + */ + static FlowMultiplexerFactory FACTORY = MaxMinFlowMultiplexer::new; + private final FlowStage stage; private final BitSet activeInputs; private final BitSet activeOutputs; @@ -90,6 +94,16 @@ public final class MaxMinFlowMultiplexer implements FlowMultiplexer, FlowStageLo return rate; } + @Override + public int getMaxInputs() { + return Integer.MAX_VALUE; + } + + @Override + public int getMaxOutputs() { + return Integer.MAX_VALUE; + } + @Override public long onUpdate(FlowStage ctx, long now) { float capacity = this.capacity; diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java index a237c81e..c09987cd 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/RuntimeFlowSource.java @@ -22,6 +22,7 @@ package org.opendc.simulator.flow2.source; +import java.util.function.Consumer; import org.opendc.simulator.flow2.FlowGraph; import org.opendc.simulator.flow2.FlowStage; import org.opendc.simulator.flow2.FlowStageLogic; @@ -29,8 +30,6 @@ import org.opendc.simulator.flow2.OutHandler; import org.opendc.simulator.flow2.OutPort; import org.opendc.simulator.flow2.Outlet; -import java.util.function.Consumer; - /** * A {@link FlowSource} that ensures a flow is emitted for a specified amount of time at some utilization. */ @@ -53,7 +52,7 @@ public class RuntimeFlowSource implements FlowSource, FlowStageLogic { * @param completionHandler A callback invoked when the source completes. */ public RuntimeFlowSource( - FlowGraph graph, long duration, float utilization, Consumer completionHandler) { + FlowGraph graph, long duration, float utilization, Consumer completionHandler) { if (duration <= 0) { throw new IllegalArgumentException("Duration must be positive and non-zero"); } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java index 764a20a8..a0e9cb9d 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/SimpleFlowSource.java @@ -22,6 +22,7 @@ package org.opendc.simulator.flow2.source; +import java.util.function.Consumer; import org.opendc.simulator.flow2.FlowGraph; import org.opendc.simulator.flow2.FlowStage; import org.opendc.simulator.flow2.FlowStageLogic; @@ -29,8 +30,6 @@ import org.opendc.simulator.flow2.OutHandler; import org.opendc.simulator.flow2.OutPort; import org.opendc.simulator.flow2.Outlet; -import java.util.function.Consumer; - /** * A flow source that contains a fixed amount and is pushed with a given utilization. */ @@ -52,7 +51,7 @@ public final class SimpleFlowSource implements FlowSource, FlowStageLogic { * @param completionHandler A callback invoked when the source completes. */ public SimpleFlowSource( - FlowGraph graph, float amount, float utilization, Consumer completionHandler) { + FlowGraph graph, float amount, float utilization, Consumer completionHandler) { if (amount < 0.0) { throw new IllegalArgumentException("Amount must be non-negative"); } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java index 96d43aef..e8abc2d7 100644 --- a/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java +++ b/opendc-simulator/opendc-simulator-flow/src/main/java/org/opendc/simulator/flow2/source/TraceFlowSource.java @@ -22,6 +22,7 @@ package org.opendc.simulator.flow2.source; +import java.util.function.Consumer; import org.opendc.simulator.flow2.FlowGraph; import org.opendc.simulator.flow2.FlowStage; import org.opendc.simulator.flow2.FlowStageLogic; @@ -29,8 +30,6 @@ import org.opendc.simulator.flow2.OutHandler; import org.opendc.simulator.flow2.OutPort; import org.opendc.simulator.flow2.Outlet; -import java.util.function.Consumer; - /** * A flow source that replays a sequence of fragments, each indicating the flow rate for some period of time. */ -- cgit v1.2.3 From 2a457d9e5480407d76440a2277817cb735c86ae1 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Tue, 4 Oct 2022 16:39:28 +0200 Subject: refactor(sim/flow): Remove old flow simulator This change removes the old version of the flow simulator from the OpenDC repository. The old version has been replaced by the new flow2 framework which is able to simulate flows more efficiently. --- .../org/opendc/simulator/flow/FlowBenchmarks.kt | 137 ---- .../org/opendc/simulator/flow/FlowConnection.kt | 72 -- .../org/opendc/simulator/flow/FlowConsumer.kt | 131 ---- .../opendc/simulator/flow/FlowConsumerContext.kt | 62 -- .../org/opendc/simulator/flow/FlowConsumerLogic.kt | 57 -- .../simulator/flow/FlowConvergenceListener.kt | 35 - .../org/opendc/simulator/flow/FlowCounters.kt | 48 -- .../kotlin/org/opendc/simulator/flow/FlowEngine.kt | 95 --- .../org/opendc/simulator/flow/FlowForwarder.kt | 264 ------- .../kotlin/org/opendc/simulator/flow/FlowMapper.kt | 75 -- .../kotlin/org/opendc/simulator/flow/FlowSink.kt | 160 ---- .../kotlin/org/opendc/simulator/flow/FlowSource.kt | 67 -- .../opendc/simulator/flow/internal/Constants.kt | 28 - .../org/opendc/simulator/flow/internal/Flags.kt | 44 -- .../flow/internal/FlowConsumerContextImpl.kt | 436 ----------- .../opendc/simulator/flow/internal/FlowDeque.kt | 119 --- .../simulator/flow/internal/FlowEngineImpl.kt | 218 ------ .../simulator/flow/internal/FlowTimerQueue.kt | 200 ----- .../simulator/flow/internal/MutableFlowCounters.kt | 53 -- .../opendc/simulator/flow/mux/FlowMultiplexer.kt | 124 ---- .../simulator/flow/mux/FlowMultiplexerFactory.kt | 60 -- .../flow/mux/ForwardingFlowMultiplexer.kt | 177 ----- .../simulator/flow/mux/MaxMinFlowMultiplexer.kt | 811 --------------------- .../simulator/flow/source/FixedFlowSource.kt | 66 -- .../simulator/flow/source/FlowSourceBarrier.kt | 52 -- .../simulator/flow/source/FlowSourceRateAdapter.kt | 77 -- .../simulator/flow/source/TraceFlowSource.kt | 67 -- .../simulator/flow/FlowConsumerContextTest.kt | 107 --- .../org/opendc/simulator/flow/FlowForwarderTest.kt | 331 --------- .../org/opendc/simulator/flow/FlowSinkTest.kt | 245 ------- .../flow/mux/ForwardingFlowMultiplexerTest.kt | 158 ---- .../flow/mux/MaxMinFlowMultiplexerTest.kt | 150 ---- .../simulator/flow/source/FixedFlowSourceTest.kt | 57 -- 33 files changed, 4783 deletions(-) delete mode 100644 opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConnection.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumer.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerContext.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerLogic.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConvergenceListener.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowCounters.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowEngine.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowMapper.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSink.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSource.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Constants.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Flags.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowDeque.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowTimerQueue.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/MutableFlowCounters.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexerFactory.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FixedFlowSource.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceBarrier.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceRateAdapter.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/TraceFlowSource.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt delete mode 100644 opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt (limited to 'opendc-simulator/opendc-simulator-flow/src') diff --git a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt b/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt deleted file mode 100644 index 9e0a4a5e..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/jmh/kotlin/org/opendc/simulator/flow/FlowBenchmarks.kt +++ /dev/null @@ -1,137 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow - -import kotlinx.coroutines.launch -import org.opendc.simulator.flow.mux.ForwardingFlowMultiplexer -import org.opendc.simulator.flow.mux.MaxMinFlowMultiplexer -import org.opendc.simulator.flow.source.TraceFlowSource -import org.opendc.simulator.kotlin.runSimulation -import org.openjdk.jmh.annotations.Benchmark -import org.openjdk.jmh.annotations.Fork -import org.openjdk.jmh.annotations.Measurement -import org.openjdk.jmh.annotations.Scope -import org.openjdk.jmh.annotations.Setup -import org.openjdk.jmh.annotations.State -import org.openjdk.jmh.annotations.Warmup -import java.util.concurrent.ThreadLocalRandom -import java.util.concurrent.TimeUnit - -@State(Scope.Thread) -@Fork(1) -@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS) -@Measurement(iterations = 5, time = 3, timeUnit = TimeUnit.SECONDS) -class FlowBenchmarks { - private lateinit var trace: Sequence - - @Setup - fun setUp() { - val random = ThreadLocalRandom.current() - val entries = List(1000000) { TraceFlowSource.Fragment(1000, random.nextDouble(0.0, 4500.0)) } - trace = entries.asSequence() - } - - @Benchmark - fun benchmarkSink() { - return runSimulation { - val engine = FlowEngine(coroutineContext, clock) - val provider = FlowSink(engine, 4200.0) - return@runSimulation provider.consume(TraceFlowSource(trace)) - } - } - - @Benchmark - fun benchmarkForward() { - return runSimulation { - val engine = FlowEngine(coroutineContext, clock) - val provider = FlowSink(engine, 4200.0) - val forwarder = FlowForwarder(engine) - provider.startConsumer(forwarder) - return@runSimulation forwarder.consume(TraceFlowSource(trace)) - } - } - - @Benchmark - fun benchmarkMuxMaxMinSingleSource() { - return runSimulation { - val engine = FlowEngine(coroutineContext, clock) - val switch = MaxMinFlowMultiplexer(engine) - - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - - val provider = switch.newInput() - return@runSimulation provider.consume(TraceFlowSource(trace)) - } - } - - @Benchmark - fun benchmarkMuxMaxMinTripleSource() { - return runSimulation { - val engine = FlowEngine(coroutineContext, clock) - val switch = MaxMinFlowMultiplexer(engine) - - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - - repeat(3) { - launch { - val provider = switch.newInput() - provider.consume(TraceFlowSource(trace)) - } - } - } - } - - @Benchmark - fun benchmarkMuxExclusiveSingleSource() { - return runSimulation { - val engine = FlowEngine(coroutineContext, clock) - val switch = ForwardingFlowMultiplexer(engine) - - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - - val provider = switch.newInput() - return@runSimulation provider.consume(TraceFlowSource(trace)) - } - } - - @Benchmark - fun benchmarkMuxExclusiveTripleSource() { - return runSimulation { - val engine = FlowEngine(coroutineContext, clock) - val switch = ForwardingFlowMultiplexer(engine) - - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - FlowSink(engine, 3000.0).startConsumer(switch.newOutput()) - - repeat(2) { - launch { - val provider = switch.newInput() - provider.consume(TraceFlowSource(trace)) - } - } - } - } -} 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 deleted file mode 100644 index 8ff0bc76..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConnection.kt +++ /dev/null @@ -1,72 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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() - - /** - * Pull the source. - * - * @param now The timestamp at which the connection is pulled. - */ - public fun pull(now: Long) - - /** - * 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 deleted file mode 100644 index a49826f4..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumer.kt +++ /dev/null @@ -1,131 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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) { - try { - source.onStop(conn, now) - - if (!cont.isCompleted) { - cont.resume(Unit) - } - } catch (cause: Throwable) { - cont.resumeWithException(cause) - throw cause - } - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - return try { - source.onPull(conn, now) - } catch (cause: Throwable) { - cont.resumeWithException(cause) - throw cause - } - } - - override fun onConverge(conn: FlowConnection, now: Long) { - try { - source.onConverge(conn, now) - } 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 deleted file mode 100644 index 98922ab3..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerContext.kt +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 deadline of the source. - */ - public val deadline: Long - - /** - * 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 - - /** - * A flag to control whether the timers for the [FlowSource] should be enabled. - */ - public var enableTimers: Boolean - - /** - * Start the flow over the connection. - */ - public fun start() - - /** - * Synchronously pull the source of the connection. - * - * @param now The timestamp at which the connection is pulled. - */ - public fun pullSync(now: Long) -} 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 deleted file mode 100644 index 1d3adb10..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConsumerLogic.kt +++ /dev/null @@ -1,57 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 rate The requested processing rate of the source. - */ - public fun onPush(ctx: FlowConsumerContext, now: 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. - */ - public fun onConverge(ctx: FlowConsumerContext, now: 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 cause The cause of the failure or `null` if the source completed. - */ - public fun onFinish(ctx: FlowConsumerContext, now: 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 deleted file mode 100644 index 62cb10d1..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowConvergenceListener.kt +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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. - */ - public fun onConverge(now: 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 deleted file mode 100644 index d8ad7978..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowCounters.kt +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 - - /** - * 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 deleted file mode 100644 index 65224827..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowEngine.kt +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 deleted file mode 100644 index 5202c252..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt +++ /dev/null @@ -1,264 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow - -import mu.KotlinLogging -import org.opendc.simulator.flow.internal.D_MS_TO_S -import org.opendc.simulator.flow.internal.MutableFlowCounters - -/** - * The logging instance of this connection. - */ -private val logger = KotlinLogging.logger {} - -/** - * A class that acts as a [FlowSource] and [FlowConsumer] at the same time. - * - * @param engine The [FlowEngine] the forwarder runs in. - * @param listener The convergence lister to use. - * @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 listener: FlowConvergenceListener? = null, - private val isCoupled: Boolean = false -) : FlowSource, FlowConsumer, AutoCloseable { - /** - * 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() - } - - override fun pull(now: Long) { - _innerCtx?.pull(now) - } - - 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() - delegate.onStop(this, now) - } - } - } - - /** - * 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 = MutableFlowCounters() - - 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 (listener != null || _ctx.shouldSourceConverge) { - conn.shouldSourceConverge = true - } - } - - override fun onStop(conn: FlowConnection, now: Long) { - _innerCtx = null - - val delegate = delegate - if (delegate != null) { - reset() - - try { - delegate.onStop(this._ctx, now) - } catch (cause: Throwable) { - logger.error(cause) { "Uncaught exception" } - } - } - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - val delegate = delegate - - if (!hasDelegateStarted) { - start() - } - - updateCounters(conn, now) - - return try { - delegate?.onPull(_ctx, now) ?: Long.MAX_VALUE - } catch (cause: Throwable) { - logger.error(cause) { "Uncaught exception" } - - reset() - Long.MAX_VALUE - } - } - - override fun onConverge(conn: FlowConnection, now: Long) { - try { - delegate?.onConverge(this._ctx, now) - listener?.onConverge(now) - } catch (cause: Throwable) { - logger.error(cause) { "Uncaught exception" } - - _innerCtx = null - reset() - } - } - - /** - * Start the delegate. - */ - private fun start() { - val delegate = delegate ?: return - - try { - val now = engine.clock.millis() - delegate.onStart(_ctx, now) - hasDelegateStarted = true - _lastUpdate = now - } 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 - private var _lastUpdate = 0L - - /** - * Update the flow counters for the transformer. - */ - private fun updateCounters(ctx: FlowConnection, now: Long) { - val lastUpdate = _lastUpdate - _lastUpdate = now - val delta = now - lastUpdate - if (delta <= 0) { - return - } - - val counters = _counters - val deltaS = delta * D_MS_TO_S - val total = ctx.capacity * deltaS - val work = _demand * deltaS - val actualWork = ctx.rate * deltaS - - counters.increment(work, actualWork, (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 deleted file mode 100644 index af702701..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowMapper.kt +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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) { - val delegate = checkNotNull(_conn) { "Invariant violation" } - _conn = null - source.onStop(delegate, now) - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - val delegate = checkNotNull(_conn) { "Invariant violation" } - return source.onPull(delegate, now) - } - - override fun onConverge(conn: FlowConnection, now: Long) { - val delegate = _conn ?: return - source.onConverge(delegate, now) - } - - /** - * 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 deleted file mode 100644 index ee8cd739..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSink.kt +++ /dev/null @@ -1,160 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow - -import org.opendc.simulator.flow.internal.D_MS_TO_S -import org.opendc.simulator.flow.internal.MutableFlowCounters - -/** - * 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 -) : 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 = MutableFlowCounters() - - /** - * The current active [FlowConsumerLogic] of this sink. - */ - private var _ctx: FlowConsumerContext? = null - - override fun startConsumer(source: FlowSource) { - check(_ctx == null) { "Consumer is in invalid state" } - - val ctx = engine.newContext(source, Logic(parent, _counters)) - _ctx = ctx - - ctx.capacity = capacity - if (parent != null) { - ctx.shouldConsumerConverge = true - } - - ctx.start() - } - - override fun pull() { - _ctx?.pull() - } - - override fun cancel() { - _ctx?.close() - } - - override fun toString(): String = "FlowSink[capacity=$capacity]" - - /** - * [FlowConsumerLogic] of a sink. - */ - private inner class Logic(private val parent: FlowConvergenceListener?, private val counters: MutableFlowCounters) : FlowConsumerLogic { - - override fun onPush( - ctx: FlowConsumerContext, - now: Long, - rate: Double - ) { - updateCounters(ctx, now, rate, ctx.capacity) - } - - override fun onFinish(ctx: FlowConsumerContext, now: Long, cause: Throwable?) { - updateCounters(ctx, now, 0.0, 0.0) - - _ctx = null - } - - override fun onConverge(ctx: FlowConsumerContext, now: Long) { - parent?.onConverge(now) - } - - /** - * The previous demand and capacity for the consumer. - */ - private val _previous = DoubleArray(2) - private var _previousUpdate = Long.MAX_VALUE - - /** - * Update the counters of the flow consumer. - */ - private fun updateCounters(ctx: FlowConnection, now: Long, nextDemand: Double, nextCapacity: Double) { - val previousUpdate = _previousUpdate - _previousUpdate = now - val delta = now - previousUpdate - - val counters = counters - val previous = _previous - val demand = previous[0] - val capacity = previous[1] - - previous[0] = nextDemand - previous[1] = nextCapacity - - if (delta <= 0) { - return - } - - val deltaS = delta * D_MS_TO_S - val total = demand * deltaS - val work = capacity * deltaS - val actualWork = ctx.rate * deltaS - - counters.increment(work, actualWork, (total - actualWork)) - } - } -} 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 deleted file mode 100644 index a48ac18e..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowSource.kt +++ /dev/null @@ -1,67 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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. - */ - public fun onStop(conn: FlowConnection, now: 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. - * @return The duration after which the resource consumer should be pulled again. - */ - public fun onPull(conn: FlowConnection, now: 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. - */ - public fun onConverge(conn: FlowConnection, now: Long) {} -} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Constants.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Constants.kt deleted file mode 100644 index 450195ec..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Constants.kt +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.internal - -/** - * Constant for converting milliseconds into seconds. - */ -internal const val D_MS_TO_S = 1 / 1000.0 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 deleted file mode 100644 index 97d56fff..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/Flags.kt +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 ConnClose = 1 shl 4 // The connection should be closed -internal const val ConnUpdateActive = 1 shl 5 // An update for the connection is active -internal const val ConnUpdatePending = 1 shl 6 // An (immediate) update of the connection is pending -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 -internal const val ConnDisableTimers = 1 shl 10 // Disable timers for the source 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 deleted file mode 100644 index fba3af5f..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowConsumerContextImpl.kt +++ /dev/null @@ -1,436 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.internal - -import mu.KotlinLogging -import org.opendc.simulator.flow.FlowConsumerContext -import org.opendc.simulator.flow.FlowConsumerLogic -import org.opendc.simulator.flow.FlowSource -import org.opendc.simulator.flow.batch -import java.util.ArrayDeque -import kotlin.math.min - -/** - * The logging instance of this connection. - */ -private val logger = KotlinLogging.logger {} - -/** - * 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 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 - private var _demand: Double = 0.0 // The current (pending) demand of the source - - /** - * The deadline of the source. - */ - override val deadline: Long - get() = _deadline - private var _deadline: Long = Long.MAX_VALUE // The deadline of the source's timer - - /** - * Flags to control the convergence of the consumer and source. - */ - override var shouldSourceConverge: Boolean - get() = _flags and ConnConvergeSource == ConnConvergeSource - set(value) { - _flags = - if (value) { - _flags or ConnConvergeSource - } else { - _flags and ConnConvergeSource.inv() - } - } - override var shouldConsumerConverge: Boolean - get() = _flags and ConnConvergeConsumer == ConnConvergeConsumer - set(value) { - _flags = - if (value) { - _flags or ConnConvergeConsumer - } else { - _flags and ConnConvergeConsumer.inv() - } - } - - /** - * Flag to control the timers on the [FlowSource] - */ - override var enableTimers: Boolean - get() = _flags and ConnDisableTimers != ConnDisableTimers - set(value) { - _flags = - if (!value) { - _flags or ConnDisableTimers - } else { - _flags and ConnDisableTimers.inv() - } - } - - /** - * The clock to track simulation time. - */ - private val _clock = engine.clock - - /** - * The flags of the flow connection, indicating certain actions. - */ - private var _flags: Int = 0 - - /** - * The timers at which the context is scheduled to be interrupted. - */ - private var _timer: Long = Long.MAX_VALUE - private val _pendingTimers: ArrayDeque = 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() { - val flags = _flags - if (flags and ConnState == ConnClosed) { - return - } - - // Toggle the close bit. In case no update is active, schedule a new update. - if (flags and ConnUpdateActive == 0) { - val now = _clock.millis() - scheduleImmediate(now, flags or ConnClose) - } else { - _flags = flags or ConnClose - } - } - - override fun pull(now: Long) { - val flags = _flags - if (flags and ConnState != ConnActive) { - return - } - - // Mark connection as pulled - scheduleImmediate(now, flags or ConnPulled) - } - - override fun pull() { - pull(_clock.millis()) - } - - override fun pullSync(now: Long) { - 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 - } - - if (flags and (ConnPulled or ConnPushed) != 0 || _deadline == now) { - engine.scheduleSync(now, 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: FlowDeque, - timerQueue: FlowTimerQueue, - 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: Long - 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) { - // Update state before calling into the outside world, so it observes a consistent state - _flags = (flags and ConnPulled.inv()) or ConnUpdateActive - hasUpdated = true - - val duration = source.onPull(this, now) - - // 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 - } - - // Make the new deadline available for the consumer if it has changed - if (newDeadline != deadline) { - _deadline = newDeadline - } - - // Push to the consumer if the rate of the source has changed (after a call to `push`) - if (flags and ConnPushed != 0) { - // Update state before calling into the outside world, so it observes a consistent state - _flags = (flags and ConnPushed.inv()) or ConnUpdateActive - hasUpdated = true - - logic.onPush(this, now, _demand) - - // IMPORTANT: Re-fetch the flags after the callback might have changed those - flags = _flags - } - - // Check whether the source or consumer have tried to close the connection - if (flags and ConnClose != 0) { - hasUpdated = true - - // The source has called [FlowConnection.close], so clean up the connection - doStopSource(now) - - // IMPORTANT: Re-fetch the flags after the callback might have changed those - // We now also mark the connection as closed - flags = (_flags and ConnState.inv()) or ConnClosed - - _demand = 0.0 - newDeadline = Long.MAX_VALUE - } - } catch (cause: Throwable) { - hasUpdated = true - - // Clean up the connection - doFailSource(now, cause) - - // Mark the connection as closed - flags = (flags and ConnState.inv()) or ConnClosed - - _demand = 0.0 - newDeadline = Long.MAX_VALUE - } - - // 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 - val timer = pendingTimers.poll() ?: Long.MAX_VALUE - _timer = timer - timer - } else { - _timer - } - - // 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) Timers are not disabled for the source - // (4) The current active timer for the connection points to a later deadline - if (newDeadline == Long.MAX_VALUE || - flags and ConnState != ConnActive || - flags and ConnDisableTimers != 0 || - (timer != Long.MAX_VALUE && newDeadline >= timer) - ) { - // 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 - _timer = newDeadline - timerQueue.add(this, newDeadline) - - // Slow-path: a timer already exists for this connection, so add it to the queue of pending timers - if (timer != Long.MAX_VALUE) { - 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 - if (flags and ConnConvergeSource != 0) { - source.onConverge(this, now) - } - - // Call the consumer callback if it has enabled convergence - if (flags and ConnConvergeConsumer != 0) { - logic.onConverge(this, now) - } - } catch (cause: Throwable) { - // Invoke the finish callbacks - doFailSource(now, cause) - - // Mark the connection as closed - _flags = (_flags and ConnState.inv()) or ConnClosed - _demand = 0.0 - _deadline = Long.MAX_VALUE - } - } - - override fun toString(): String = "FlowConsumerContextImpl[capacity=$capacity,rate=$_rate]" - - /** - * Stop the [FlowSource]. - */ - private fun doStopSource(now: Long) { - try { - source.onStop(this, now) - doFinishConsumer(now, null) - } catch (cause: Throwable) { - doFinishConsumer(now, cause) - } - } - - /** - * Fail the [FlowSource]. - */ - private fun doFailSource(now: Long, cause: Throwable) { - try { - source.onStop(this, now) - } catch (e: Throwable) { - e.addSuppressed(cause) - doFinishConsumer(now, e) - } - } - - /** - * Finish the consumer. - */ - private fun doFinishConsumer(now: Long, cause: Throwable?) { - try { - logic.onFinish(this, now, 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/FlowDeque.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowDeque.kt deleted file mode 100644 index 403a9aec..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowDeque.kt +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.internal - -import java.util.ArrayDeque - -/** - * A specialized [ArrayDeque] that tracks the [FlowConsumerContextImpl] instances that have updated in an interpreter - * cycle. - * - * By using a specialized class, we reduce the overhead caused by type-erasure. - */ -internal class FlowDeque(initialCapacity: Int = 256) { - /** - * The array of elements in the queue. - */ - private var _elements: Array = arrayOfNulls(initialCapacity) - private var _head = 0 - private var _tail = 0 - - /** - * Determine whether this queue is not empty. - */ - fun isNotEmpty(): Boolean { - return _head != _tail - } - - /** - * Add the specified [ctx] to the queue. - */ - fun add(ctx: FlowConsumerContextImpl) { - val es = _elements - var tail = _tail - - es[tail] = ctx - - tail = inc(tail, es.size) - _tail = tail - - if (_head == tail) { - doubleCapacity() - } - } - - /** - * Remove a [FlowConsumerContextImpl] from the queue or `null` if the queue is empty. - */ - fun poll(): FlowConsumerContextImpl? { - val es = _elements - val head = _head - val ctx = es[head] - - if (ctx != null) { - es[head] = null - _head = inc(head, es.size) - } - - return ctx - } - - /** - * Clear the queue. - */ - fun clear() { - _elements.fill(null) - _head = 0 - _tail = 0 - } - - private fun inc(i: Int, modulus: Int): Int { - var x = i - if (++x >= modulus) { - x = 0 - } - return x - } - - /** - * Doubles the capacity of this deque - */ - private fun doubleCapacity() { - assert(_head == _tail) - val p = _head - val n = _elements.size - val r = n - p // number of elements to the right of p - - val newCapacity = n shl 1 - check(newCapacity >= 0) { "Sorry, deque too big" } - - val a = arrayOfNulls(newCapacity) - - _elements.copyInto(a, 0, p, n) - _elements.copyInto(a, r, 0, p) - - _elements = a - _head = 0 - _tail = n - } -} 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 deleted file mode 100644 index 6fd1ef31..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowEngineImpl.kt +++ /dev/null @@ -1,218 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.internal - -import kotlinx.coroutines.Delay -import kotlinx.coroutines.DisposableHandle -import kotlinx.coroutines.InternalCoroutinesApi -import kotlinx.coroutines.Runnable -import org.opendc.simulator.flow.FlowConsumerContext -import org.opendc.simulator.flow.FlowConsumerLogic -import org.opendc.simulator.flow.FlowEngine -import org.opendc.simulator.flow.FlowSource -import java.time.Clock -import java.util.ArrayDeque -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, clock: Clock) : FlowEngine, Runnable { - /** - * 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 = FlowDeque() - - /** - * A priority queue containing the connection updates to be scheduled in the future. - */ - private val futureQueue = FlowTimerQueue() - - /** - * The stack of engine invocations to occur in the future. - */ - private val futureInvocations = ArrayDeque() - - /** - * The systems that have been visited during the engine cycle. - */ - private val visited = FlowDeque() - - /** - * The index in the batch stack. - */ - private var batchIndex = 0 - - /** - * The virtual [Clock] of this engine. - */ - override val clock: Clock - get() = _clock - private val _clock: Clock = clock - - /** - * 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 - } - - doRunEngine(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 - } - - doRunEngine(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-- - } - } - - /* Runnable */ - override fun run() { - val invocation = futureInvocations.poll() // Clear invocation from future invocation queue - doRunEngine(invocation.timestamp) - } - - /** - * 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 - - try { - // Increment batch index so synchronous calls will not launch concurrent engine invocations - batchIndex++ - - // Execute all scheduled updates at current timestamp - while (true) { - val ctx = futureQueue.poll(now) ?: break - 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()) - } finally { - // Decrement batch index to indicate no engine is active at the moment - batchIndex-- - } - - // Schedule an engine invocation for the next update to occur. - val headDeadline = futureQueue.peekDeadline() - if (headDeadline != Long.MAX_VALUE) { - trySchedule(now, futureInvocations, headDeadline) - } - } - - /** - * 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, target: Long) { - val head = scheduled.peek() - - // Only schedule a new scheduler invocation in case the target is earlier than all other pending - // scheduler invocations - if (head == null || target < head.timestamp) { - @OptIn(InternalCoroutinesApi::class) - val handle = delay.invokeOnTimeout(target - now, this, context) - scheduled.addFirst(Invocation(target, handle)) - } - } - - /** - * 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() - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowTimerQueue.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowTimerQueue.kt deleted file mode 100644 index 47061a91..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/FlowTimerQueue.kt +++ /dev/null @@ -1,200 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.internal - -/** - * Specialized priority queue for flow timers. - * - * By using a specialized priority queue, we reduce the overhead caused by the default priority queue implementation - * being generic. - */ -internal class FlowTimerQueue(initialCapacity: Int = 256) { - /** - * The binary heap of deadlines. - */ - private var _deadlines = LongArray(initialCapacity) { Long.MIN_VALUE } - - /** - * The binary heap of [FlowConsumerContextImpl]s. - */ - private var _pending = arrayOfNulls(initialCapacity) - - /** - * The number of elements in the priority queue. - */ - private var size = 0 - - /** - * Register a timer for [ctx] with [deadline]. - */ - fun add(ctx: FlowConsumerContextImpl, deadline: Long) { - val i = size - var deadlines = _deadlines - if (i >= deadlines.size) { - grow() - // Re-fetch the resized array - deadlines = _deadlines - } - - siftUp(deadlines, _pending, i, ctx, deadline) - - size = i + 1 - } - - /** - * Update all pending [FlowConsumerContextImpl]s at the timestamp [now]. - */ - fun poll(now: Long): FlowConsumerContextImpl? { - if (size == 0) { - return null - } - - val deadlines = _deadlines - val deadline = deadlines[0] - - if (now < deadline) { - return null - } - - val pending = _pending - val res = pending[0] - val s = --size - - val nextDeadline = deadlines[s] - val next = pending[s]!! - - // Clear the last element of the queue - pending[s] = null - deadlines[s] = Long.MIN_VALUE - - if (s != 0) { - siftDown(deadlines, pending, next, nextDeadline) - } - - return res - } - - /** - * Find the earliest deadline in the queue. - */ - fun peekDeadline(): Long { - return if (size == 0) Long.MAX_VALUE else _deadlines[0] - } - - /** - * Increases the capacity of the array. - */ - private fun grow() { - val oldCapacity = _deadlines.size - // Double size if small; else grow by 50% - val newCapacity = oldCapacity + if (oldCapacity < 64) oldCapacity + 2 else oldCapacity shr 1 - - _deadlines = _deadlines.copyOf(newCapacity) - _pending = _pending.copyOf(newCapacity) - } - - /** - * Insert item [ctx] at position [pos], maintaining heap invariant by promoting [ctx] up the tree until it is - * greater than or equal to its parent, or is the root. - * - * @param deadlines The heap of deadlines. - * @param pending The heap of contexts. - * @param pos The position to fill. - * @param ctx The [FlowConsumerContextImpl] to insert. - * @param deadline The deadline of the context. - */ - private fun siftUp( - deadlines: LongArray, - pending: Array, - pos: Int, - ctx: FlowConsumerContextImpl, - deadline: Long - ) { - var k = pos - - while (k > 0) { - val parent = (k - 1) ushr 1 - val parentDeadline = deadlines[parent] - - if (deadline >= parentDeadline) { - break - } - - deadlines[k] = parentDeadline - pending[k] = pending[parent] - - k = parent - } - - deadlines[k] = deadline - pending[k] = ctx - } - - /** - * Inserts [ctx] at the top, maintaining heap invariant by demoting [ctx] down the tree repeatedly until it - * is less than or equal to its children or is a leaf. - * - * @param deadlines The heap of deadlines. - * @param pending The heap of contexts. - * @param ctx The [FlowConsumerContextImpl] to insert. - * @param deadline The deadline of the context. - */ - private fun siftDown( - deadlines: LongArray, - pending: Array, - ctx: FlowConsumerContextImpl, - deadline: Long - ) { - var k = 0 - val size = size - val half = size ushr 1 - - while (k < half) { - var child = (k shl 1) + 1 - - var childDeadline = deadlines[child] - val right = child + 1 - - if (right < size) { - val rightDeadline = deadlines[right] - - if (childDeadline > rightDeadline) { - child = right - childDeadline = rightDeadline - } - } - - if (deadline <= childDeadline) { - break - } - - deadlines[k] = childDeadline - pending[k] = pending[child] - - k = child - } - - deadlines[k] = deadline - pending[k] = ctx - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/MutableFlowCounters.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/MutableFlowCounters.kt deleted file mode 100644 index c320a362..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/MutableFlowCounters.kt +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.internal - -import org.opendc.simulator.flow.FlowCounters - -/** - * Mutable implementation of the [FlowCounters] interface. - */ -public class MutableFlowCounters : FlowCounters { - override val demand: Double - get() = _counters[0] - override val actual: Double - get() = _counters[1] - override val remaining: Double - get() = _counters[2] - private val _counters = DoubleArray(3) - - override fun reset() { - _counters.fill(0.0) - } - - public fun increment(demand: Double, actual: Double, remaining: Double) { - val counters = _counters - counters[0] += demand - counters[1] += actual - counters[2] += remaining - } - - override fun toString(): String { - return "FlowCounters[demand=$demand,actual=$actual,remaining=$remaining]" - } -} 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 deleted file mode 100644 index 8752c559..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexer.kt +++ /dev/null @@ -1,124 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.mux - -import org.opendc.simulator.flow.FlowConsumer -import org.opendc.simulator.flow.FlowCounters -import org.opendc.simulator.flow.FlowSource - -/** - * A [FlowMultiplexer] enables multiplexing multiple [FlowSource]s over possibly multiple [FlowConsumer]s. - */ -public interface FlowMultiplexer { - /** - * The maximum number of inputs supported by the multiplexer. - */ - public val maxInputs: Int - - /** - * The maximum number of outputs supported by the multiplexer. - */ - public val maxOutputs: Int - - /** - * The inputs of the multiplexer that can be used to consume sources. - */ - public val inputs: Set - - /** - * The outputs of the multiplexer over which the flows will be distributed. - */ - public val outputs: Set - - /** - * 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 with a coupled capacity. - */ - public fun newInput(): FlowConsumer - - /** - * Create a new input on this multiplexer with the specified [capacity]. - * - * @param capacity The capacity of the input. - */ - public fun newInput(capacity: Double): 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() - - /** - * Flush the counters of the multiplexer. - */ - public fun flushCounters() - - /** - * Flush the counters of the specified [input]. - */ - public fun flushCounters(input: FlowConsumer) -} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexerFactory.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexerFactory.kt deleted file mode 100644 index a863e3ad..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexerFactory.kt +++ /dev/null @@ -1,60 +0,0 @@ -/* - * Copyright (c) 2022 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.mux - -import org.opendc.simulator.flow.FlowConvergenceListener -import org.opendc.simulator.flow.FlowEngine - -/** - * Factory interface for a [FlowMultiplexer] implementation. - */ -public fun interface FlowMultiplexerFactory { - /** - * Construct a new [FlowMultiplexer] using the specified [engine] and [listener]. - */ - public fun newMultiplexer(engine: FlowEngine, listener: FlowConvergenceListener?): FlowMultiplexer - - public companion object { - /** - * A [FlowMultiplexerFactory] constructing a [MaxMinFlowMultiplexer]. - */ - private val MAX_MIN_FACTORY = FlowMultiplexerFactory { engine, listener -> MaxMinFlowMultiplexer(engine, listener) } - - /** - * A [FlowMultiplexerFactory] constructing a [ForwardingFlowMultiplexer]. - */ - private val FORWARDING_FACTORY = FlowMultiplexerFactory { engine, listener -> ForwardingFlowMultiplexer(engine, listener) } - - /** - * Return a [FlowMultiplexerFactory] that returns [MaxMinFlowMultiplexer] instances. - */ - @JvmStatic - public fun maxMinMultiplexer(): FlowMultiplexerFactory = MAX_MIN_FACTORY - - /** - * Return a [ForwardingFlowMultiplexer] that returns [ForwardingFlowMultiplexer] instances. - */ - @JvmStatic - public fun forwardingMultiplexer(): FlowMultiplexerFactory = FORWARDING_FACTORY - } -} 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 deleted file mode 100644 index 53f94a94..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexer.kt +++ /dev/null @@ -1,177 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.mux - -import org.opendc.simulator.flow.FlowConnection -import org.opendc.simulator.flow.FlowConsumer -import org.opendc.simulator.flow.FlowConvergenceListener -import org.opendc.simulator.flow.FlowCounters -import org.opendc.simulator.flow.FlowEngine -import org.opendc.simulator.flow.FlowForwarder -import org.opendc.simulator.flow.FlowSource -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. - * @param listener The convergence listener of the multiplexer. - */ -public class ForwardingFlowMultiplexer( - private val engine: FlowEngine, - private val listener: FlowConvergenceListener? = null -) : FlowMultiplexer, FlowConvergenceListener { - - override val maxInputs: Int - get() = _outputs.size - - override val maxOutputs: Int = Int.MAX_VALUE - - override val inputs: Set - get() = _inputs - private val _inputs = mutableSetOf() - - override val outputs: Set - get() = _outputs - private val _outputs = mutableSetOf() - private val _availableOutputs = ArrayDeque() - - 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 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(): FlowConsumer { - val output = checkNotNull(_availableOutputs.poll()) { "No capacity to serve request" } - val input = Input(output) - _inputs += input - return input - } - - override fun newInput(capacity: Double): FlowConsumer = newInput() - - 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, this) - 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() - } - - override fun flushCounters() {} - - override fun flushCounters(input: FlowConsumer) {} - - override fun onConverge(now: Long) { - listener?.onConverge(now) - } - - /** - * 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) { - forwarder.cancel() - forwarder.onStop(conn, now) - } - - 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 deleted file mode 100644 index d9c6f893..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexer.kt +++ /dev/null @@ -1,811 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.mux - -import org.opendc.simulator.flow.FlowConnection -import org.opendc.simulator.flow.FlowConsumer -import org.opendc.simulator.flow.FlowConsumerContext -import org.opendc.simulator.flow.FlowConsumerLogic -import org.opendc.simulator.flow.FlowConvergenceListener -import org.opendc.simulator.flow.FlowCounters -import org.opendc.simulator.flow.FlowEngine -import org.opendc.simulator.flow.FlowSource -import org.opendc.simulator.flow.internal.D_MS_TO_S -import org.opendc.simulator.flow.internal.MutableFlowCounters -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. - */ -public class MaxMinFlowMultiplexer( - private val engine: FlowEngine, - parent: FlowConvergenceListener? = null -) : FlowMultiplexer { - - override val maxInputs: Int = Int.MAX_VALUE - - override val maxOutputs: Int = Int.MAX_VALUE - - /** - * The inputs of the multiplexer. - */ - override val inputs: Set - get() = _inputs - private val _inputs = mutableSetOf() - - /** - * The outputs of the multiplexer. - */ - override val outputs: Set - get() = _outputs - private val _outputs = mutableSetOf() - - /** - * The flow counters of this multiplexer. - */ - public override val counters: FlowCounters - get() = scheduler.counters - - /** - * The actual processing rate of the multiplexer. - */ - public override val rate: Double - get() = scheduler.rate - - /** - * The demanded processing rate of the input. - */ - public override val demand: Double - get() = scheduler.demand - - /** - * The capacity of the outputs. - */ - public override val capacity: Double - get() = scheduler.capacity - - /** - * The [Scheduler] instance of this multiplexer. - */ - private val scheduler = Scheduler(engine, parent) - - override fun newInput(): FlowConsumer { - return newInput(isCoupled = true, Double.POSITIVE_INFINITY) - } - - override fun newInput(capacity: Double): FlowConsumer { - return newInput(isCoupled = false, capacity) - } - - private fun newInput(isCoupled: Boolean, initialCapacity: Double): FlowConsumer { - val provider = Input(engine, scheduler, isCoupled, initialCapacity) - _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(scheduler) - _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() - } - - override fun flushCounters() { - scheduler.updateCounters(engine.clock.millis()) - } - - override fun flushCounters(input: FlowConsumer) { - (input as Input).doUpdateCounters(engine.clock.millis()) - } - - /** - * Helper class containing the scheduler state. - */ - private class Scheduler(engine: FlowEngine, private val parent: FlowConvergenceListener?) { - /** - * The flow counters of this scheduler. - */ - @JvmField val counters = MutableFlowCounters() - - /** - * The flow rate of the multiplexer. - */ - @JvmField var rate = 0.0 - - /** - * The demand for the multiplexer. - */ - @JvmField var demand = 0.0 - - /** - * The capacity of the multiplexer. - */ - @JvmField var capacity = 0.0 - - /** - * An [Output] that is used to activate the scheduler. - */ - @JvmField var activationOutput: Output? = null - - /** - * The active inputs registered with the scheduler. - */ - private val _activeInputs = mutableListOf() - - /** - * An array containing the active inputs, which is used to reduce the overhead of an [ArrayList]. - */ - private var _inputArray = emptyArray() - - /** - * The active outputs registered with the scheduler. - */ - private val _activeOutputs = mutableListOf() - - /** - * Flag to indicate that the scheduler is active. - */ - private var _schedulerActive = false - - /** - * The last convergence timestamp and the input. - */ - private var _lastConverge: Long = Long.MIN_VALUE - private var _lastConvergeInput: Input? = null - - /** - * The simulation clock. - */ - private val _clock = engine.clock - - /** - * Register the specified [input] to this scheduler. - */ - fun registerInput(input: Input) { - _activeInputs.add(input) - _inputArray = _activeInputs.toTypedArray() - - val hasActivationOutput = activationOutput != null - - // Disable timers and convergence of the source if one of the output manages it - input.shouldConsumerConverge = !hasActivationOutput - input.enableTimers = !hasActivationOutput - - if (input.isCoupled) { - input.capacity = capacity - } - - trigger(_clock.millis()) - } - - /** - * De-register the specified [input] from this scheduler. - */ - fun deregisterInput(input: Input, now: Long) { - // Assign a new input responsible for handling the convergence events - if (_lastConvergeInput == input) { - _lastConvergeInput = null - } - - _activeInputs.remove(input) - - // Re-run scheduler to distribute new load - trigger(now) - } - - /** - * This method is invoked when one of the inputs converges. - */ - fun convergeInput(input: Input, now: Long) { - val lastConverge = _lastConverge - val lastConvergeInput = _lastConvergeInput - val parent = parent - - if (parent != null && (now > lastConverge || lastConvergeInput == null || lastConvergeInput == input)) { - _lastConverge = now - _lastConvergeInput = input - - parent.onConverge(now) - } - } - - /** - * Register the specified [output] to this scheduler. - */ - fun registerOutput(output: Output) { - _activeOutputs.add(output) - - updateCapacity() - updateActivationOutput() - } - - /** - * De-register the specified [output] from this scheduler. - */ - fun deregisterOutput(output: Output, now: Long) { - _activeOutputs.remove(output) - updateCapacity() - - trigger(now) - } - - /** - * This method is invoked when one of the outputs converges. - */ - fun convergeOutput(output: Output, now: Long) { - val parent = parent - - if (parent != null) { - _lastConverge = now - parent.onConverge(now) - } - - if (!output.isActive) { - output.isActivationOutput = false - updateActivationOutput() - } - } - - /** - * Trigger the scheduler of the multiplexer. - * - * @param now The current virtual timestamp of the simulation. - */ - fun trigger(now: Long) { - if (_schedulerActive) { - // No need to trigger the scheduler in case it is already active - return - } - - val activationOutput = activationOutput - - // We can run the scheduler in two ways: - // (1) We can pull one of the multiplexer's outputs. This allows us to cascade multiple pushes by the input - // into a single scheduling cycle, but is slower in case of a few changes at the same timestamp. - // (2) We run the scheduler directly from this method call. This is the fastest approach when there are only - // a few inputs and little changes at the same timestamp. - // We always pick for option (1) unless there are no outputs available. - if (activationOutput != null) { - activationOutput.pull(now) - return - } else { - runScheduler(now) - } - } - - /** - * Synchronously run the scheduler of the multiplexer. - */ - fun runScheduler(now: Long): Long { - return try { - _schedulerActive = true - doRunScheduler(now) - } finally { - _schedulerActive = false - } - } - - /** - * Recompute the capacity of the multiplexer. - */ - fun updateCapacity() { - val newCapacity = _activeOutputs.sumOf(Output::capacity) - - // No-op if the capacity is unchanged - if (capacity == newCapacity) { - return - } - - capacity = newCapacity - - for (input in _activeInputs) { - if (input.isCoupled) { - input.capacity = newCapacity - } - } - - // Sort outputs by their capacity - _activeOutputs.sort() - } - - /** - * Updates the output that is used for scheduler activation. - */ - private fun updateActivationOutput() { - val output = _activeOutputs.firstOrNull() - activationOutput = output - - if (output != null) { - output.isActivationOutput = true - } - - val hasActivationOutput = output != null - - for (input in _activeInputs) { - input.shouldConsumerConverge = !hasActivationOutput - input.enableTimers = !hasActivationOutput - } - } - - /** - * Schedule the inputs over the outputs. - * - * @return The deadline after which a new scheduling cycle should start. - */ - private fun doRunScheduler(now: Long): Long { - val activeInputs = _activeInputs - val activeOutputs = _activeOutputs - var inputArray = _inputArray - var inputSize = _inputArray.size - - // Update the counters of the scheduler - updateCounters(now) - - // If there is no work yet, mark the inputs as idle. - if (inputSize == 0) { - demand = 0.0 - rate = 0.0 - return Long.MAX_VALUE - } - - val capacity = capacity - var availableCapacity = capacity - var deadline = Long.MAX_VALUE - var demand = 0.0 - var shouldRebuild = false - - // Pull in the work of the inputs - for (i in 0 until inputSize) { - val input = inputArray[i] - - input.pullSync(now) - - // Remove inputs that have finished - if (!input.isActive) { - input.actualRate = 0.0 - shouldRebuild = true - } else { - demand += input.limit - deadline = min(deadline, input.deadline) - } - } - - // Slow-path: Rebuild the input array based on the (apparently) updated `activeInputs` - if (shouldRebuild) { - inputArray = activeInputs.toTypedArray() - inputSize = inputArray.size - _inputArray = inputArray - } - - val rate = if (demand > capacity) { - // If the demand is higher than the capacity, we need use max-min fair sharing to distribute the - // constrained capacity across the inputs. - - // Sort in-place the inputs based on their pushed flow. - // Profiling shows that it is faster than maintaining some kind of sorted set. - inputArray.sort() - - // Divide the available output capacity fairly over the inputs using max-min fair sharing - for (i in 0 until inputSize) { - val input = inputArray[i] - val availableShare = availableCapacity / (inputSize - i) - val grantedRate = min(input.allowedRate, availableShare) - - availableCapacity -= grantedRate - input.actualRate = grantedRate - } - - capacity - availableCapacity - } else { - demand - } - - this.demand = demand - if (this.rate != rate) { - // Only update the outputs if the output rate has changed - this.rate = rate - - // 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) - } - } - - return deadline - } - - /** - * The previous capacity of the multiplexer. - */ - private var _previousCapacity = 0.0 - private var _previousUpdate = Long.MIN_VALUE - - /** - * Update the counters of the scheduler. - */ - fun updateCounters(now: Long) { - val previousCapacity = _previousCapacity - _previousCapacity = capacity - - val previousUpdate = _previousUpdate - _previousUpdate = now - - val delta = now - previousUpdate - if (delta <= 0) { - return - } - - val deltaS = delta * D_MS_TO_S - val demand = demand - val rate = rate - - counters.increment( - demand = demand * deltaS, - actual = rate * deltaS, - remaining = (previousCapacity - rate) * deltaS - ) - } - } - - /** - * An internal [FlowConsumer] implementation for multiplexer inputs. - */ - private class Input( - private val engine: FlowEngine, - private val scheduler: Scheduler, - @JvmField val isCoupled: Boolean, - initialCapacity: Double - ) : FlowConsumer, FlowConsumerLogic, Comparable { - /** - * A flag to indicate that the consumer is active. - */ - override val isActive: Boolean - get() = _ctx != null - - /** - * The demand of the consumer. - */ - override val demand: Double - get() = limit - - /** - * The processing rate of the consumer. - */ - override val rate: Double - get() = actualRate - - /** - * The capacity of the input. - */ - override var capacity: Double - get() = _capacity - set(value) { - allowedRate = min(limit, value) - _capacity = value - _ctx?.capacity = value - } - private var _capacity = initialCapacity - - /** - * The flow counters to track the flow metrics of the consumer. - */ - override val counters: FlowCounters - get() = _counters - private val _counters = MutableFlowCounters() - - /** - * A flag to enable timers for the input. - */ - var enableTimers: Boolean = true - set(value) { - field = value - _ctx?.enableTimers = value - } - - /** - * A flag to control whether the input should converge. - */ - var shouldConsumerConverge: Boolean = true - set(value) { - field = value - _ctx?.shouldConsumerConverge = value - } - - /** - * 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. - */ - @JvmField var allowedRate: Double = 0.0 - - /** - * The deadline of the input. - */ - val deadline: Long - get() = _ctx?.deadline ?: Long.MAX_VALUE - - /** - * The [FlowConsumerContext] that is currently running. - */ - private var _ctx: FlowConsumerContext? = null - - /** - * A flag to indicate that the input is closed. - */ - private var _isClosed: Boolean = false - - /** - * Close the input. - * - * This method is invoked when the user removes an input from the switch. - */ - fun close() { - _isClosed = true - cancel() - } - - /** - * Pull the source if necessary. - */ - fun pullSync(now: Long) { - _ctx?.pullSync(now) - } - - /* FlowConsumer */ - override fun startConsumer(source: FlowSource) { - check(!_isClosed) { "Cannot re-use closed input" } - check(_ctx == null) { "Consumer is in invalid state" } - - val ctx = engine.newContext(source, this) - _ctx = ctx - - ctx.capacity = capacity - scheduler.registerInput(this) - - ctx.start() - } - - override fun pull() { - _ctx?.pull() - } - - override fun cancel() { - _ctx?.close() - } - - /* FlowConsumerLogic */ - override fun onPush( - ctx: FlowConsumerContext, - now: Long, - rate: Double - ) { - doUpdateCounters(now) - - val allowed = min(rate, capacity) - limit = rate - actualRate = allowed - allowedRate = allowed - - scheduler.trigger(now) - } - - override fun onFinish(ctx: FlowConsumerContext, now: Long, cause: Throwable?) { - doUpdateCounters(now) - - limit = 0.0 - actualRate = 0.0 - allowedRate = 0.0 - - scheduler.deregisterInput(this, now) - - _ctx = null - } - - override fun onConverge(ctx: FlowConsumerContext, now: Long) { - scheduler.convergeInput(this, now) - } - - /* Comparable */ - override fun compareTo(other: Input): Int = allowedRate.compareTo(other.allowedRate) - - /** - * The timestamp that the counters where last updated. - */ - private var _lastUpdate = Long.MIN_VALUE - - /** - * Helper method to update the flow counters of the multiplexer. - */ - fun doUpdateCounters(now: Long) { - val lastUpdate = _lastUpdate - _lastUpdate = now - - val delta = (now - lastUpdate).coerceAtLeast(0) - if (delta <= 0L) { - return - } - - val actualRate = actualRate - - val deltaS = delta * D_MS_TO_S - val demand = limit * deltaS - val actual = actualRate * deltaS - val remaining = (_capacity - actualRate) * deltaS - - _counters.increment(demand, actual, remaining) - scheduler.counters.increment(0.0, 0.0, 0.0) - } - } - - /** - * An internal [FlowSource] implementation for multiplexer outputs. - */ - private class Output(private val scheduler: Scheduler) : FlowSource, Comparable { - /** - * The active [FlowConnection] of this source. - */ - private var _conn: FlowConnection? = null - - /** - * The capacity of this output. - */ - @JvmField var capacity: Double = 0.0 - - /** - * A flag to indicate that this output is the activation output. - */ - var isActivationOutput: Boolean - get() = _isActivationOutput - set(value) { - _isActivationOutput = value - _conn?.shouldSourceConverge = value - } - private var _isActivationOutput: Boolean = false - - /** - * A flag to indicate that the output is active. - */ - @JvmField var isActive = false - - /** - * Push the specified rate to the consumer. - */ - fun push(rate: Double) { - _conn?.push(rate) - } - - /** - * Cancel this output. - */ - fun cancel() { - _conn?.close() - } - - /** - * Pull this output. - */ - fun pull(now: Long) { - _conn?.pull(now) - } - - override fun onStart(conn: FlowConnection, now: Long) { - assert(_conn == null) { "Source running concurrently" } - _conn = conn - capacity = conn.capacity - isActive = true - - scheduler.registerOutput(this) - } - - override fun onStop(conn: FlowConnection, now: Long) { - _conn = null - capacity = 0.0 - isActive = false - - scheduler.deregisterOutput(this, now) - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - val capacity = capacity - if (capacity != conn.capacity) { - this.capacity = capacity - scheduler.updateCapacity() - } - - return if (_isActivationOutput) { - // If this output is the activation output, synchronously run the scheduler and return the new deadline - val deadline = scheduler.runScheduler(now) - if (deadline == Long.MAX_VALUE) { - deadline - } else { - deadline - now - } - } else { - // Output is not the activation output, so trigger activation output and do not install timer for this - // output (by returning `Long.MAX_VALUE`) - scheduler.trigger(now) - - Long.MAX_VALUE - } - } - - override fun onConverge(conn: FlowConnection, now: Long) { - if (_isActivationOutput) { - scheduler.convergeOutput(this, now) - } - } - - 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 deleted file mode 100644 index 6cfcc82c..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FixedFlowSource.kt +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 - private var lastPull: Long = 0L - - override fun onStart(conn: FlowConnection, now: Long) { - lastPull = now - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - val lastPull = lastPull - this.lastPull = now - val delta = (now - lastPull).coerceAtLeast(0) - - 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 deleted file mode 100644 index b3191ad3..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceBarrier.kt +++ /dev/null @@ -1,52 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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 deleted file mode 100644 index 80127fb5..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/FlowSourceRateAdapter.kt +++ /dev/null @@ -1,77 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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) { - try { - delegate.onStop(conn, now) - } finally { - rate = 0.0 - } - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - return delegate.onPull(conn, now) - } - - override fun onConverge(conn: FlowConnection, now: Long) { - try { - delegate.onConverge(conn, now) - } 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 deleted file mode 100644 index c9a52128..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/source/TraceFlowSource.kt +++ /dev/null @@ -1,67 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.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) : FlowSource { - private var _iterator: Iterator? = 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) { - _iterator = null - } - - override fun onPull(conn: FlowConnection, now: 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) -} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt deleted file mode 100644 index f89133dd..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowConsumerContextTest.kt +++ /dev/null @@ -1,107 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow - -import io.mockk.spyk -import io.mockk.verify -import net.bytebuddy.matcher.ElementMatchers.any -import org.junit.jupiter.api.Test -import org.junit.jupiter.api.assertThrows -import org.opendc.simulator.flow.internal.FlowConsumerContextImpl -import org.opendc.simulator.flow.internal.FlowEngineImpl -import org.opendc.simulator.kotlin.runSimulation - -/** - * A test suite for the [FlowConsumerContextImpl] class. - */ -class FlowConsumerContextTest { - @Test - fun testFlushWithoutCommand() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val consumer = object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (now == 0L) { - conn.push(1.0) - 1000 - } else { - conn.close() - Long.MAX_VALUE - } - } - } - - val logic = object : FlowConsumerLogic {} - val context = FlowConsumerContextImpl(engine, consumer, logic) - - engine.scheduleSync(engine.clock.millis(), context) - } - - @Test - fun testDoubleStart() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val consumer = object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (now == 0L) { - conn.push(0.0) - 1000 - } else { - conn.close() - Long.MAX_VALUE - } - } - } - - val logic = object : FlowConsumerLogic {} - val context = FlowConsumerContextImpl(engine, consumer, logic) - - context.start() - - assertThrows { - context.start() - } - } - - @Test - fun testIdempotentCapacityChange() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val consumer = spyk(object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (now == 0L) { - conn.push(1.0) - 1000 - } else { - conn.close() - Long.MAX_VALUE - } - } - }) - - val logic = object : FlowConsumerLogic {} - val context = FlowConsumerContextImpl(engine, consumer, logic) - context.capacity = 4200.0 - context.start() - context.capacity = 4200.0 - - verify(exactly = 1) { consumer.onPull(any(), any()) } - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt deleted file mode 100644 index f75e5037..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowForwarderTest.kt +++ /dev/null @@ -1,331 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow - -import io.mockk.spyk -import io.mockk.verify -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.delay -import kotlinx.coroutines.launch -import kotlinx.coroutines.yield -import net.bytebuddy.matcher.ElementMatchers.any -import org.junit.jupiter.api.Assertions.assertAll -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Assertions.assertFalse -import org.junit.jupiter.api.Assertions.assertTrue -import org.junit.jupiter.api.Disabled -import org.junit.jupiter.api.Test -import org.junit.jupiter.api.assertThrows -import org.opendc.simulator.flow.internal.FlowEngineImpl -import org.opendc.simulator.flow.source.FixedFlowSource -import org.opendc.simulator.kotlin.runSimulation - -/** - * A test suite for the [FlowForwarder] class. - */ -internal class FlowForwarderTest { - @Test - fun testCancelImmediately() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 2000.0) - - launch { source.consume(forwarder) } - - forwarder.consume(object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - conn.close() - return Long.MAX_VALUE - } - }) - - forwarder.close() - source.cancel() - } - - @Test - fun testCancel() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 2000.0) - - launch { source.consume(forwarder) } - - forwarder.consume(object : FlowSource { - var isFirst = true - - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (isFirst) { - isFirst = false - conn.push(1.0) - 10 * 1000 - } else { - conn.close() - Long.MAX_VALUE - } - } - }) - - forwarder.close() - source.cancel() - } - - @Test - fun testState() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val consumer = object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - conn.close() - return Long.MAX_VALUE - } - } - - assertFalse(forwarder.isActive) - - forwarder.startConsumer(consumer) - assertTrue(forwarder.isActive) - - assertThrows { forwarder.startConsumer(consumer) } - - forwarder.cancel() - assertFalse(forwarder.isActive) - - forwarder.close() - assertFalse(forwarder.isActive) - } - - @Test - fun testCancelPendingDelegate() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - - val consumer = spyk(object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - conn.close() - return Long.MAX_VALUE - } - }) - - forwarder.startConsumer(consumer) - forwarder.cancel() - - verify(exactly = 0) { consumer.onStop(any(), any()) } - } - - @Test - fun testCancelStartedDelegate() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 2000.0) - - val consumer = spyk(FixedFlowSource(2000.0, 1.0)) - - source.startConsumer(forwarder) - yield() - forwarder.startConsumer(consumer) - yield() - forwarder.cancel() - - verify(exactly = 1) { consumer.onStart(any(), any()) } - verify(exactly = 1) { consumer.onStop(any(), any()) } - } - - @Test - fun testCancelPropagation() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 2000.0) - - val consumer = spyk(FixedFlowSource(2000.0, 1.0)) - - source.startConsumer(forwarder) - yield() - forwarder.startConsumer(consumer) - yield() - source.cancel() - - verify(exactly = 1) { consumer.onStart(any(), any()) } - verify(exactly = 1) { consumer.onStop(any(), any()) } - } - - @Test - fun testExitPropagation() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine, isCoupled = true) - val source = FlowSink(engine, 2000.0) - - val consumer = object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - conn.close() - return Long.MAX_VALUE - } - } - - source.startConsumer(forwarder) - forwarder.consume(consumer) - yield() - - assertFalse(forwarder.isActive) - } - - @Test - @Disabled // Due to Kotlin bug: https://github.com/mockk/mockk/issues/368 - fun testAdjustCapacity() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val sink = FlowSink(engine, 1.0) - - val source = spyk(FixedFlowSource(2.0, 1.0)) - sink.startConsumer(forwarder) - - coroutineScope { - launch { forwarder.consume(source) } - delay(1000) - sink.capacity = 0.5 - } - - assertEquals(3000, clock.millis()) - verify(exactly = 1) { source.onPull(any(), any()) } - } - - @Test - fun testCounters() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 1.0) - - val consumer = FixedFlowSource(2.0, 1.0) - source.startConsumer(forwarder) - - forwarder.consume(consumer) - - yield() - - assertAll( - { assertEquals(2.0, source.counters.actual) }, - { assertEquals(source.counters.actual, forwarder.counters.actual) { "Actual work" } }, - { assertEquals(source.counters.demand, forwarder.counters.demand) { "Work demand" } }, - { assertEquals(source.counters.remaining, forwarder.counters.remaining) { "Overcommitted work" } }, - { assertEquals(2000, clock.millis()) } - ) - } - - @Test - fun testCoupledExit() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine, isCoupled = true) - val source = FlowSink(engine, 2000.0) - - launch { source.consume(forwarder) } - - forwarder.consume(FixedFlowSource(2000.0, 1.0)) - - yield() - - assertFalse(source.isActive) - } - - @Test - fun testPullFailureCoupled() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine, isCoupled = true) - val source = FlowSink(engine, 2000.0) - - launch { source.consume(forwarder) } - - try { - forwarder.consume(object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - throw IllegalStateException("Test") - } - }) - } catch (cause: Throwable) { - // Ignore - } - - yield() - - assertFalse(source.isActive) - } - - @Test - fun testStartFailure() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 2000.0) - - launch { source.consume(forwarder) } - - try { - forwarder.consume(object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - return Long.MAX_VALUE - } - - override fun onStart(conn: FlowConnection, now: Long) { - throw IllegalStateException("Test") - } - }) - } catch (cause: Throwable) { - // Ignore - } - - yield() - - assertTrue(source.isActive) - source.cancel() - } - - @Test - fun testConvergeFailure() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val forwarder = FlowForwarder(engine) - val source = FlowSink(engine, 2000.0) - - launch { source.consume(forwarder) } - - try { - forwarder.consume(object : FlowSource { - override fun onStart(conn: FlowConnection, now: Long) { - conn.shouldSourceConverge = true - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - return Long.MAX_VALUE - } - - override fun onConverge(conn: FlowConnection, now: Long) { - throw IllegalStateException("Test") - } - }) - } catch (cause: Throwable) { - // Ignore - } - - yield() - - assertTrue(source.isActive) - source.cancel() - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt deleted file mode 100644 index 746d752d..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/FlowSinkTest.kt +++ /dev/null @@ -1,245 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow - -import io.mockk.spyk -import io.mockk.verify -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.delay -import kotlinx.coroutines.launch -import kotlinx.coroutines.yield -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Test -import org.junit.jupiter.api.assertThrows -import org.opendc.simulator.flow.internal.FlowEngineImpl -import org.opendc.simulator.flow.source.FixedFlowSource -import org.opendc.simulator.flow.source.FlowSourceRateAdapter -import org.opendc.simulator.kotlin.runSimulation - -/** - * A test suite for the [FlowSink] class. - */ -internal class FlowSinkTest { - @Test - fun testSpeed() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = FixedFlowSource(4200.0, 1.0) - - val res = mutableListOf() - val adapter = FlowSourceRateAdapter(consumer, res::add) - - provider.consume(adapter) - - assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" } - } - - @Test - fun testAdjustCapacity() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val provider = FlowSink(engine, 1.0) - - val consumer = spyk(FixedFlowSource(2.0, 1.0)) - - coroutineScope { - launch { provider.consume(consumer) } - delay(1000) - provider.capacity = 0.5 - } - assertEquals(3000, clock.millis()) - verify(exactly = 3) { consumer.onPull(any(), any()) } - } - - @Test - fun testSpeedLimit() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = FixedFlowSource(capacity, 2.0) - - val res = mutableListOf() - val adapter = FlowSourceRateAdapter(consumer, res::add) - - provider.consume(adapter) - - assertEquals(listOf(0.0, capacity, 0.0), res) { "Speed is reported correctly" } - } - - /** - * Test to see whether no infinite recursion occurs when interrupting during [FlowSource.onStart] or - * [FlowSource.onPull]. - */ - @Test - fun testIntermediateInterrupt() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long { - conn.close() - return Long.MAX_VALUE - } - - override fun onStart(conn: FlowConnection, now: Long) { - conn.pull() - } - } - - provider.consume(consumer) - } - - @Test - fun testInterrupt() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - lateinit var resCtx: FlowConnection - - val consumer = object : FlowSource { - var isFirst = true - - override fun onStart(conn: FlowConnection, now: Long) { - resCtx = conn - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (isFirst) { - isFirst = false - conn.push(1.0) - 4000 - } else { - conn.close() - Long.MAX_VALUE - } - } - } - - launch { - yield() - resCtx.pull() - } - provider.consume(consumer) - - assertEquals(0, clock.millis()) - } - - @Test - fun testFailure() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = object : FlowSource { - override fun onStart(conn: FlowConnection, now: Long) { - throw IllegalStateException("Hi") - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - return Long.MAX_VALUE - } - } - - assertThrows { - provider.consume(consumer) - } - } - - @Test - fun testExceptionPropagationOnNext() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = object : FlowSource { - var isFirst = true - - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (isFirst) { - isFirst = false - conn.push(1.0) - 1000 - } else { - throw IllegalStateException() - } - } - } - - assertThrows { - provider.consume(consumer) - } - } - - @Test - fun testConcurrentConsumption() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = FixedFlowSource(capacity, 1.0) - - assertThrows { - coroutineScope { - launch { provider.consume(consumer) } - provider.consume(consumer) - } - } - } - - @Test - fun testCancelDuringConsumption() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = FixedFlowSource(capacity, 1.0) - - launch { provider.consume(consumer) } - delay(500) - provider.cancel() - - yield() - - assertEquals(500, clock.millis()) - } - - @Test - fun testInfiniteSleep() { - assertThrows { - runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - val capacity = 4200.0 - val provider = FlowSink(engine, capacity) - - val consumer = object : FlowSource { - override fun onPull(conn: FlowConnection, now: Long): Long = Long.MAX_VALUE - } - - provider.consume(consumer) - } - } - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt deleted file mode 100644 index 2409e174..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/ForwardingFlowMultiplexerTest.kt +++ /dev/null @@ -1,158 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.mux - -import kotlinx.coroutines.yield -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Test -import org.junit.jupiter.api.assertAll -import org.junit.jupiter.api.assertThrows -import org.opendc.simulator.flow.FlowConnection -import org.opendc.simulator.flow.FlowForwarder -import org.opendc.simulator.flow.FlowSink -import org.opendc.simulator.flow.FlowSource -import org.opendc.simulator.flow.consume -import org.opendc.simulator.flow.internal.FlowEngineImpl -import org.opendc.simulator.flow.source.FixedFlowSource -import org.opendc.simulator.flow.source.FlowSourceRateAdapter -import org.opendc.simulator.flow.source.TraceFlowSource -import org.opendc.simulator.kotlin.runSimulation - -/** - * Test suite for the [ForwardingFlowMultiplexer] class. - */ -internal class ForwardingFlowMultiplexerTest { - /** - * Test a trace workload. - */ - @Test - fun testTrace() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - - val speed = mutableListOf() - - val duration = 5 * 60L - val workload = - TraceFlowSource( - sequenceOf( - TraceFlowSource.Fragment(duration * 1000, 28.0), - TraceFlowSource.Fragment(duration * 1000, 3500.0), - TraceFlowSource.Fragment(duration * 1000, 0.0), - TraceFlowSource.Fragment(duration * 1000, 183.0) - ) - ) - - val switch = ForwardingFlowMultiplexer(engine) - val source = FlowSink(engine, 3200.0) - val forwarder = FlowForwarder(engine) - val adapter = FlowSourceRateAdapter(forwarder, speed::add) - source.startConsumer(adapter) - forwarder.startConsumer(switch.newOutput()) - - val provider = switch.newInput() - provider.consume(workload) - yield() - - assertAll( - { assertEquals(listOf(0.0, 28.0, 3200.0, 0.0, 183.0, 0.0), speed) { "Correct speed" } }, - { assertEquals(5 * 60L * 4000, clock.millis()) { "Took enough time" } } - ) - } - - /** - * Test runtime workload on hypervisor. - */ - @Test - fun testRuntimeWorkload() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - - val duration = 5 * 60L * 1000 - val workload = FixedFlowSource(duration * 3.2, 1.0) - - val switch = ForwardingFlowMultiplexer(engine) - val source = FlowSink(engine, 3200.0) - - source.startConsumer(switch.newOutput()) - - val provider = switch.newInput() - provider.consume(workload) - yield() - - assertEquals(duration, clock.millis()) { "Took enough time" } - } - - /** - * Test two workloads running sequentially. - */ - @Test - fun testTwoWorkloads() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - - val duration = 5 * 60L * 1000 - val workload = object : FlowSource { - var isFirst = true - - override fun onStart(conn: FlowConnection, now: Long) { - isFirst = true - } - - override fun onPull(conn: FlowConnection, now: Long): Long { - return if (isFirst) { - isFirst = false - conn.push(1.0) - duration - } else { - conn.close() - Long.MAX_VALUE - } - } - } - - val switch = ForwardingFlowMultiplexer(engine) - val source = FlowSink(engine, 3200.0) - - source.startConsumer(switch.newOutput()) - - val provider = switch.newInput() - provider.consume(workload) - yield() - provider.consume(workload) - assertEquals(duration * 2, clock.millis()) { "Took enough time" } - } - - /** - * Test concurrent workloads on the machine. - */ - @Test - fun testConcurrentWorkloadFails() = runSimulation { - val engine = FlowEngineImpl(coroutineContext, clock) - - val switch = ForwardingFlowMultiplexer(engine) - val source = FlowSink(engine, 3200.0) - - source.startConsumer(switch.newOutput()) - - switch.newInput() - assertThrows { switch.newInput() } - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt deleted file mode 100644 index a6bf8ad8..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/mux/MaxMinFlowMultiplexerTest.kt +++ /dev/null @@ -1,150 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.mux - -import kotlinx.coroutines.coroutineScope -import kotlinx.coroutines.launch -import kotlinx.coroutines.yield -import org.junit.jupiter.api.Assertions.assertAll -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Test -import org.opendc.simulator.flow.FlowSink -import org.opendc.simulator.flow.consume -import org.opendc.simulator.flow.internal.FlowEngineImpl -import org.opendc.simulator.flow.source.FixedFlowSource -import org.opendc.simulator.flow.source.TraceFlowSource -import org.opendc.simulator.kotlin.runSimulation - -/** - * Test suite for the [FlowMultiplexer] implementations - */ -internal class MaxMinFlowMultiplexerTest { - @Test - fun testSmoke() = runSimulation { - val scheduler = FlowEngineImpl(coroutineContext, clock) - val switch = MaxMinFlowMultiplexer(scheduler) - - val sources = List(2) { FlowSink(scheduler, 2000.0) } - sources.forEach { it.startConsumer(switch.newOutput()) } - - val provider = switch.newInput() - val consumer = FixedFlowSource(2000.0, 1.0) - - try { - provider.consume(consumer) - yield() - } finally { - switch.clear() - } - } - - /** - * Test overcommitting of resources via the hypervisor with a single VM. - */ - @Test - fun testOvercommittedSingle() = runSimulation { - val scheduler = FlowEngineImpl(coroutineContext, clock) - - val duration = 5 * 60L - val workload = - TraceFlowSource( - sequenceOf( - TraceFlowSource.Fragment(duration * 1000, 28.0), - TraceFlowSource.Fragment(duration * 1000, 3500.0), - TraceFlowSource.Fragment(duration * 1000, 0.0), - TraceFlowSource.Fragment(duration * 1000, 183.0) - ) - ) - - val switch = MaxMinFlowMultiplexer(scheduler) - val sink = FlowSink(scheduler, 3200.0) - val provider = switch.newInput() - - try { - sink.startConsumer(switch.newOutput()) - provider.consume(workload) - yield() - } finally { - switch.clear() - } - - assertAll( - { assertEquals(1113300.0, switch.counters.demand, "Requested work does not match") }, - { assertEquals(1023300.0, switch.counters.actual, "Actual work does not match") }, - { assertEquals(2816700.0, switch.counters.remaining, "Remaining capacity does not match") }, - { assertEquals(1200000, clock.millis()) } - ) - } - - /** - * Test overcommitting of resources via the hypervisor with two VMs. - */ - @Test - fun testOvercommittedDual() = runSimulation { - val scheduler = FlowEngineImpl(coroutineContext, clock) - - val duration = 5 * 60L - val workloadA = - TraceFlowSource( - sequenceOf( - TraceFlowSource.Fragment(duration * 1000, 28.0), - TraceFlowSource.Fragment(duration * 1000, 3500.0), - TraceFlowSource.Fragment(duration * 1000, 0.0), - TraceFlowSource.Fragment(duration * 1000, 183.0) - ) - ) - val workloadB = - TraceFlowSource( - sequenceOf( - TraceFlowSource.Fragment(duration * 1000, 28.0), - TraceFlowSource.Fragment(duration * 1000, 3100.0), - TraceFlowSource.Fragment(duration * 1000, 0.0), - TraceFlowSource.Fragment(duration * 1000, 73.0) - ) - ) - - val switch = MaxMinFlowMultiplexer(scheduler) - val sink = FlowSink(scheduler, 3200.0) - val providerA = switch.newInput() - val providerB = switch.newInput() - - try { - sink.startConsumer(switch.newOutput()) - - coroutineScope { - launch { providerA.consume(workloadA) } - providerB.consume(workloadB) - } - - yield() - } finally { - switch.clear() - } - assertAll( - { assertEquals(2073600.0, switch.counters.demand, "Requested work does not match") }, - { assertEquals(1053600.0, switch.counters.actual, "Granted work does not match") }, - { assertEquals(2786400.0, switch.counters.remaining, "Remaining capacity does not match") }, - { assertEquals(1200000, clock.millis()) } - ) - } -} diff --git a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt b/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt deleted file mode 100644 index 552579ff..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/test/kotlin/org/opendc/simulator/flow/source/FixedFlowSourceTest.kt +++ /dev/null @@ -1,57 +0,0 @@ -/* - * Copyright (c) 2021 AtLarge Research - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in all - * copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE - * SOFTWARE. - */ - -package org.opendc.simulator.flow.source - -import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Test -import org.opendc.simulator.flow.FlowSink -import org.opendc.simulator.flow.consume -import org.opendc.simulator.flow.internal.FlowEngineImpl -import org.opendc.simulator.kotlin.runSimulation - -/** - * A test suite for the [FixedFlowSource] class. - */ -internal class FixedFlowSourceTest { - @Test - fun testSmoke() = runSimulation { - val scheduler = FlowEngineImpl(coroutineContext, clock) - val provider = FlowSink(scheduler, 1.0) - - val consumer = FixedFlowSource(1.0, 1.0) - - provider.consume(consumer) - assertEquals(1000, clock.millis()) - } - - @Test - fun testUtilization() = runSimulation { - val scheduler = FlowEngineImpl(coroutineContext, clock) - val provider = FlowSink(scheduler, 1.0) - - val consumer = FixedFlowSource(1.0, 0.5) - - provider.consume(consumer) - assertEquals(2000, clock.millis()) - } -} -- cgit v1.2.3