diff options
Diffstat (limited to 'opendc-simulator/opendc-simulator-flow/src')
10 files changed, 100 insertions, 98 deletions
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 index a717ae6e..d8ad7978 100644 --- 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 @@ -42,11 +42,6 @@ public interface FlowCounters { public val remaining: Double /** - * The accumulated flow lost due to interference between sources. - */ - public val interference: Double - - /** * Reset the flow counters. */ public fun reset() diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/FlowForwarder.kt index 0ad18f6a..6fa2971a 100644 --- 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 @@ -258,6 +258,6 @@ public class FlowForwarder( val work = _demand * deltaS val actualWork = ctx.rate * deltaS - counters.increment(work, actualWork, (total - actualWork), 0.0) + counters.increment(work, actualWork, (total - actualWork)) } } 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 index d0324ce8..ee8cd739 100644 --- 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 @@ -154,7 +154,7 @@ public class FlowSink( val work = capacity * deltaS val actualWork = ctx.rate * deltaS - counters.increment(work, actualWork, (total - actualWork), 0.0) + counters.increment(work, actualWork, (total - actualWork)) } } } diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceDomain.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceDomain.kt deleted file mode 100644 index aa2713b6..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceDomain.kt +++ /dev/null @@ -1,19 +0,0 @@ -package org.opendc.simulator.flow.interference - -import org.opendc.simulator.flow.FlowSource - -/** - * An interference domain represents a system of flow stages where [flow sources][FlowSource] may incur - * performance variability due to operating on the same resource and therefore causing interference. - */ -public interface InterferenceDomain { - /** - * Compute the performance score of a participant in this interference domain. - * - * @param key The participant to obtain the score of or `null` if the participant has no key. - * @param load The overall load on the interference domain. - * @return A score representing the performance score to be applied to the resource consumer, with 1 - * meaning no influence, <1 means that performance degrades, and >1 means that performance improves. - */ - public fun apply(key: InterferenceKey?, load: Double): Double -} diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceKey.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceKey.kt deleted file mode 100644 index d28ebde5..00000000 --- a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/interference/InterferenceKey.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.interference - -/** - * A key that uniquely identifies a participant of an interference domain. - */ -public interface InterferenceKey diff --git a/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/MutableFlowCounters.kt b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/internal/MutableFlowCounters.kt index d990dc61..c320a362 100644 --- 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 @@ -34,23 +34,20 @@ public class MutableFlowCounters : FlowCounters { get() = _counters[1] override val remaining: Double get() = _counters[2] - override val interference: Double - get() = _counters[3] - private val _counters = DoubleArray(4) + private val _counters = DoubleArray(3) override fun reset() { _counters.fill(0.0) } - public fun increment(demand: Double, actual: Double, remaining: Double, interference: Double) { + public fun increment(demand: Double, actual: Double, remaining: Double) { val counters = _counters counters[0] += demand counters[1] += actual counters[2] += remaining - counters[3] += interference } override fun toString(): String { - return "FlowCounters[demand=$demand,actual=$actual,remaining=$remaining,interference=$interference]" + 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 index 5f198944..8752c559 100644 --- 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 @@ -25,13 +25,22 @@ package org.opendc.simulator.flow.mux import org.opendc.simulator.flow.FlowConsumer import org.opendc.simulator.flow.FlowCounters import org.opendc.simulator.flow.FlowSource -import org.opendc.simulator.flow.interference.InterferenceKey /** * A [FlowMultiplexer] enables multiplexing multiple [FlowSource]s over possibly multiple [FlowConsumer]s. */ public interface FlowMultiplexer { /** + * The 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<FlowConsumer> @@ -63,18 +72,15 @@ public interface FlowMultiplexer { /** * Create a new input on this multiplexer with a coupled capacity. - * - * @param key The key of the interference member to which the input belongs. */ - public fun newInput(key: InterferenceKey? = null): FlowConsumer + public fun newInput(): FlowConsumer /** * Create a new input on this multiplexer with the specified [capacity]. * * @param capacity The capacity of the input. - * @param key The key of the interference member to which the input belongs. */ - public fun newInput(capacity: Double, key: InterferenceKey? = null): FlowConsumer + public fun newInput(capacity: Double): FlowConsumer /** * Remove [input] from this multiplexer. 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 new file mode 100644 index 00000000..a863e3ad --- /dev/null +++ b/opendc-simulator/opendc-simulator-flow/src/main/kotlin/org/opendc/simulator/flow/mux/FlowMultiplexerFactory.kt @@ -0,0 +1,60 @@ +/* + * 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 index 1d7d22ef..c50e9bbc 100644 --- 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 @@ -23,7 +23,6 @@ package org.opendc.simulator.flow.mux import org.opendc.simulator.flow.* -import org.opendc.simulator.flow.interference.InterferenceKey import java.util.ArrayDeque /** @@ -38,6 +37,12 @@ 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<FlowConsumer> get() = _inputs private val _inputs = mutableSetOf<Input>() @@ -54,8 +59,6 @@ public class ForwardingFlowMultiplexer( get() = _outputs.sumOf { it.forwarder.counters.actual } override val remaining: Double get() = _outputs.sumOf { it.forwarder.counters.remaining } - override val interference: Double - get() = _outputs.sumOf { it.forwarder.counters.interference } override fun reset() { for (output in _outputs) { @@ -75,14 +78,14 @@ public class ForwardingFlowMultiplexer( override val capacity: Double get() = _outputs.sumOf { it.forwarder.capacity } - override fun newInput(key: InterferenceKey?): FlowConsumer { + 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, key: InterferenceKey?): FlowConsumer = newInput(key) + override fun newInput(capacity: Double): FlowConsumer = newInput() override fun removeInput(input: FlowConsumer) { if (!_inputs.remove(input)) { 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 index cc831862..f2a4c1a4 100644 --- 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 @@ -23,11 +23,8 @@ package org.opendc.simulator.flow.mux import org.opendc.simulator.flow.* -import org.opendc.simulator.flow.interference.InterferenceDomain -import org.opendc.simulator.flow.interference.InterferenceKey import org.opendc.simulator.flow.internal.D_MS_TO_S import org.opendc.simulator.flow.internal.MutableFlowCounters -import kotlin.math.max import kotlin.math.min /** @@ -35,13 +32,16 @@ import kotlin.math.min * * @param engine The [FlowEngine] to drive the flow simulation. * @param parent The parent flow system of the multiplexer. - * @param interferenceDomain The interference domain of the multiplexer. */ public class MaxMinFlowMultiplexer( private val engine: FlowEngine, - parent: FlowConvergenceListener? = null, - private val interferenceDomain: InterferenceDomain? = null + parent: FlowConvergenceListener? = null ) : FlowMultiplexer { + + override val maxInputs: Int = Int.MAX_VALUE + + override val maxOutputs: Int = Int.MAX_VALUE + /** * The inputs of the multiplexer. */ @@ -85,16 +85,16 @@ public class MaxMinFlowMultiplexer( */ private val scheduler = Scheduler(engine, parent) - override fun newInput(key: InterferenceKey?): FlowConsumer { - return newInput(isCoupled = true, Double.POSITIVE_INFINITY, key) + override fun newInput(): FlowConsumer { + return newInput(isCoupled = true, Double.POSITIVE_INFINITY) } - override fun newInput(capacity: Double, key: InterferenceKey?): FlowConsumer { - return newInput(isCoupled = false, capacity, key) + override fun newInput(capacity: Double): FlowConsumer { + return newInput(isCoupled = false, capacity) } - private fun newInput(isCoupled: Boolean, initialCapacity: Double, key: InterferenceKey?): FlowConsumer { - val provider = Input(engine, scheduler, interferenceDomain, key, isCoupled, initialCapacity) + private fun newInput(isCoupled: Boolean, initialCapacity: Double): FlowConsumer { + val provider = Input(engine, scheduler, isCoupled, initialCapacity) _inputs.add(provider) return provider } @@ -499,8 +499,7 @@ public class MaxMinFlowMultiplexer( counters.increment( demand = demand * deltaS, actual = rate * deltaS, - remaining = (previousCapacity - rate) * deltaS, - interference = 0.0 + remaining = (previousCapacity - rate) * deltaS ) } } @@ -511,8 +510,6 @@ public class MaxMinFlowMultiplexer( private class Input( private val engine: FlowEngine, private val scheduler: Scheduler, - private val interferenceDomain: InterferenceDomain?, - @JvmField val key: InterferenceKey?, @JvmField val isCoupled: Boolean, initialCapacity: Double, ) : FlowConsumer, FlowConsumerLogic, Comparable<Input> { @@ -693,24 +690,15 @@ public class MaxMinFlowMultiplexer( return } - // Compute the performance penalty due to flow interference - val perfScore = if (interferenceDomain != null) { - val load = scheduler.rate / scheduler.capacity - interferenceDomain.apply(key, load) - } else { - 1.0 - } - val actualRate = actualRate val deltaS = delta * D_MS_TO_S val demand = limit * deltaS val actual = actualRate * deltaS val remaining = (_capacity - actualRate) * deltaS - val interference = actual * max(0.0, 1 - perfScore) - _counters.increment(demand, actual, remaining, interference) - scheduler.counters.increment(0.0, 0.0, 0.0, interference) + _counters.increment(demand, actual, remaining) + scheduler.counters.increment(0.0, 0.0, 0.0) } } |