diff options
Diffstat (limited to 'opendc-simulator/opendc-simulator-resources/src/main')
17 files changed, 526 insertions, 265 deletions
diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt index c7fa6a17..6ae04f27 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceAggregator.kt @@ -22,30 +22,10 @@ package org.opendc.simulator.resources -import java.time.Clock - /** * Abstract implementation of [SimResourceAggregator]. */ -public abstract class SimAbstractResourceAggregator(private val clock: Clock) : SimResourceAggregator { - /** - * The available resource provider contexts. - */ - protected val inputContexts: Set<SimResourceContext> - get() = _inputContexts - private val _inputContexts = mutableSetOf<SimResourceContext>() - - /** - * The output context. - */ - protected val outputContext: SimResourceContext - get() = context - - /** - * The commands to submit to the underlying input resources. - */ - protected val commands: MutableMap<SimResourceContext, SimResourceCommand> = mutableMapOf() - +public abstract class SimAbstractResourceAggregator(private val scheduler: SimResourceScheduler) : SimResourceAggregator { /** * This method is invoked when the resource consumer consumes resources. */ @@ -54,37 +34,29 @@ public abstract class SimAbstractResourceAggregator(private val clock: Clock) : /** * This method is invoked when the resource consumer enters an idle state. */ - protected open fun doIdle(deadline: Long) { - for (input in inputContexts) { - commands[input] = SimResourceCommand.Idle(deadline) - } - } + protected abstract fun doIdle(deadline: Long) /** * This method is invoked when the resource consumer finishes processing. */ - protected open fun doFinish(cause: Throwable?) { - for (input in inputContexts) { - commands[input] = SimResourceCommand.Exit - } - } + protected abstract fun doFinish(cause: Throwable?) /** * This method is invoked when an input context is started. */ - protected open fun onContextStarted(ctx: SimResourceContext) { - _inputContexts.add(ctx) - } + protected abstract fun onInputStarted(input: Input) - protected open fun onContextFinished(ctx: SimResourceContext) { - assert(_inputContexts.remove(ctx)) { "Lost context" } - } + /** + * This method is invoked when an input is stopped. + */ + protected abstract fun onInputFinished(input: Input) override fun addInput(input: SimResourceProvider) { check(output.state != SimResourceState.Stopped) { "Aggregator has been stopped" } val consumer = Consumer() _inputs.add(input) + _inputConsumers.add(consumer) input.startConsumer(consumer) } @@ -99,15 +71,18 @@ public abstract class SimAbstractResourceAggregator(private val clock: Clock) : override val inputs: Set<SimResourceProvider> get() = _inputs private val _inputs = mutableSetOf<SimResourceProvider>() + private val _inputConsumers = mutableListOf<Consumer>() - private val context = object : SimAbstractResourceContext(inputContexts.sumByDouble { it.capacity }, clock, _output) { + protected val outputContext: SimResourceContext + get() = context + private val context = object : SimAbstractResourceContext(0.0, scheduler, _output) { override val remainingWork: Double get() { val now = clock.millis() return if (_remainingWorkFlush < now) { _remainingWorkFlush = now - _inputContexts.sumByDouble { it.remainingWork }.also { _remainingWork = it } + _inputConsumers.sumByDouble { it._ctx?.remainingWork ?: 0.0 }.also { _remainingWork = it } } else { _remainingWork } @@ -115,94 +90,89 @@ public abstract class SimAbstractResourceAggregator(private val clock: Clock) : private var _remainingWork: Double = 0.0 private var _remainingWorkFlush: Long = Long.MIN_VALUE - override fun interrupt() { - super.interrupt() - - interruptAll() - } - override fun onConsume(work: Double, limit: Double, deadline: Long) = doConsume(work, limit, deadline) override fun onIdle(deadline: Long) = doIdle(deadline) - override fun onFinish(cause: Throwable?) { - doFinish(cause) - - super.onFinish(cause) - - interruptAll() + override fun onFinish() { + doFinish(null) } } /** - * A flag to indicate that an interrupt is active. - */ - private var isInterrupting: Boolean = false - - /** - * Schedule the work over the input resources. + * An input for the resource aggregator. */ - private fun doSchedule() { - context.flush(isIntermediate = true) - interruptAll() + public interface Input { + /** + * The [SimResourceContext] associated with the input. + */ + public val ctx: SimResourceContext + + /** + * Push the specified [SimResourceCommand] to the input. + */ + public fun push(command: SimResourceCommand) } /** - * Interrupt all inputs. + * An internal [SimResourceConsumer] implementation for aggregator inputs. */ - private fun interruptAll() { - // Prevent users from interrupting the resource while they are constructing their next command, as this will - // only lead to infinite recursion. - if (isInterrupting) { - return - } - - try { - isInterrupting = true - - val iterator = _inputs.iterator() - while (iterator.hasNext()) { - val input = iterator.next() - input.interrupt() - - if (input.state != SimResourceState.Active) { - iterator.remove() - } - } - } finally { - isInterrupting = false + private inner class Consumer : Input, SimResourceConsumer { + /** + * The resource context associated with the input. + */ + override val ctx: SimResourceContext + get() = _ctx!! + var _ctx: SimResourceContext? = null + + /** + * The resource command to run next. + */ + private var command: SimResourceCommand? = null + + private fun updateCapacity() { + // Adjust capacity of output resource + context.capacity = _inputConsumers.sumByDouble { it._ctx?.capacity ?: 0.0 } } - } - /** - * An internal [SimResourceConsumer] implementation for aggregator inputs. - */ - private inner class Consumer : SimResourceConsumer { - override fun onStart(ctx: SimResourceContext) { - onContextStarted(ctx) - onCapacityChanged(ctx, false) - - // Make sure we initialize the output if we have not done so yet - if (context.state == SimResourceState.Pending) { - context.start() - } + /* Input */ + override fun push(command: SimResourceCommand) { + this.command = command + _ctx?.interrupt() } + /* SimResourceConsumer */ override fun onNext(ctx: SimResourceContext): SimResourceCommand { - doSchedule() - - return commands[ctx] ?: SimResourceCommand.Idle() + var next = command + + return if (next != null) { + this.command = null + next + } else { + context.flush(isIntermediate = true) + next = command + this.command = null + next ?: SimResourceCommand.Idle() + } } - override fun onCapacityChanged(ctx: SimResourceContext, isThrottled: Boolean) { - // Adjust capacity of output resource - context.capacity = inputContexts.sumByDouble { it.capacity } - } + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { + when (event) { + SimResourceEvent.Start -> { + _ctx = ctx + updateCapacity() - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { - onContextFinished(ctx) + // Make sure we initialize the output if we have not done so yet + if (context.state == SimResourceState.Pending) { + context.start() + } - super.onFinish(ctx, cause) + onInputStarted(this) + } + SimResourceEvent.Capacity -> updateCapacity() + SimResourceEvent.Exit -> onInputFinished(this) + else -> {} + } } } } diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt index 05ed0714..c03bfad5 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimAbstractResourceContext.kt @@ -31,9 +31,16 @@ import kotlin.math.min */ public abstract class SimAbstractResourceContext( initialCapacity: Double, - override val clock: Clock, + private val scheduler: SimResourceScheduler, private val consumer: SimResourceConsumer -) : SimResourceContext { +) : SimResourceContext, SimResourceFlushable { + + /** + * The clock of the context. + */ + public override val clock: Clock + get() = scheduler.clock + /** * The capacity of the resource. */ @@ -75,7 +82,7 @@ public abstract class SimAbstractResourceContext( /** * The current processing speed of the resource. */ - public var speed: Double = 0.0 + final override var speed: Double = 0.0 private set /** @@ -92,9 +99,7 @@ public abstract class SimAbstractResourceContext( /** * This method is invoked when the resource consumer has finished. */ - public open fun onFinish(cause: Throwable?) { - consumer.onFinish(this, cause) - } + public abstract fun onFinish() /** * Get the remaining work to process after a resource consumption. @@ -126,10 +131,10 @@ public abstract class SimAbstractResourceContext( latestFlush = now try { - consumer.onStart(this) + consumer.onEvent(this, SimResourceEvent.Start) activeCommand = interpret(consumer.onNext(this), now) } catch (cause: Throwable) { - doStop(cause) + doFail(cause) } finally { isProcessing = false } @@ -144,22 +149,13 @@ public abstract class SimAbstractResourceContext( latestFlush = clock.millis() flush(isIntermediate = true) - doStop(null) - } catch (cause: Throwable) { - doStop(cause) + doStop() } finally { isProcessing = false } } - /** - * Flush the current active resource consumption. - * - * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be - * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer - * will be asked to deliver a new command and is essentially interrupted. - */ - public fun flush(isIntermediate: Boolean = false) { + override fun flush(isIntermediate: Boolean) { // Flush is no-op when the consumer is finished or not yet started if (state != SimResourceState.Active) { return @@ -214,7 +210,7 @@ public abstract class SimAbstractResourceContext( // Flush remaining work cache _remainingWorkFlush = Long.MIN_VALUE } catch (cause: Throwable) { - doStop(cause) + doFail(cause) } finally { latestFlush = now isProcessing = false @@ -228,7 +224,7 @@ public abstract class SimAbstractResourceContext( return } - flush() + scheduler.schedule(this, isIntermediate = false) } override fun toString(): String = "SimAbstractResourceContext[capacity=$capacity]" @@ -236,7 +232,7 @@ public abstract class SimAbstractResourceContext( /** * A flag to indicate that the resource is currently processing a command. */ - protected var isProcessing: Boolean = false + private var isProcessing: Boolean = false /** * The current command that is being processed. @@ -251,13 +247,18 @@ public abstract class SimAbstractResourceContext( /** * Finish the consumer and resource provider. */ - private fun doStop(cause: Throwable?) { + private fun doStop() { val state = state this.state = SimResourceState.Stopped if (state == SimResourceState.Active) { activeCommand = null - onFinish(cause) + try { + consumer.onEvent(this, SimResourceEvent.Exit) + onFinish() + } catch (cause: Throwable) { + doFail(cause) + } } } @@ -272,9 +273,9 @@ public abstract class SimAbstractResourceContext( require(deadline >= now) { "Deadline already passed" } speed = 0.0 - consumer.onConfirm(this, 0.0) onIdle(deadline) + consumer.onEvent(this, SimResourceEvent.Run) } is SimResourceCommand.Consume -> { val work = command.work @@ -284,14 +285,13 @@ public abstract class SimAbstractResourceContext( require(deadline >= now) { "Deadline already passed" } speed = min(capacity, limit) - consumer.onConfirm(this, speed) - onConsume(work, limit, deadline) + consumer.onEvent(this, SimResourceEvent.Run) } is SimResourceCommand.Exit -> { speed = 0.0 - doStop(null) + doStop() // No need to set the next active command return null @@ -319,6 +319,23 @@ public abstract class SimAbstractResourceContext( } /** + * Fail the resource consumer. + */ + private fun doFail(cause: Throwable) { + state = SimResourceState.Stopped + activeCommand = null + + try { + consumer.onFailure(this, cause) + } catch (e: Throwable) { + e.addSuppressed(cause) + e.printStackTrace() + } + + onFinish() + } + + /** * Indicate that the capacity of the resource has changed. */ private fun onCapacityChange() { @@ -328,7 +345,8 @@ public abstract class SimAbstractResourceContext( } val isThrottled = speed > capacity - consumer.onCapacityChanged(this, isThrottled) + + consumer.onEvent(this, SimResourceEvent.Capacity) // Optimization: only flush changes if the new capacity cannot satisfy the active resource command. // Alternatively, if the consumer already interrupts the resource, the fast-path will be taken in flush(). diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt index 08bc064e..5665abd1 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceAggregatorMaxMin.kt @@ -22,42 +22,50 @@ package org.opendc.simulator.resources -import java.time.Clock - /** * A [SimResourceAggregator] that distributes the load equally across the input resources. */ -public class SimResourceAggregatorMaxMin(clock: Clock) : SimAbstractResourceAggregator(clock) { - private val consumers = mutableListOf<SimResourceContext>() +public class SimResourceAggregatorMaxMin(scheduler: SimResourceScheduler) : SimAbstractResourceAggregator(scheduler) { + private val consumers = mutableListOf<Input>() override fun doConsume(work: Double, limit: Double, deadline: Long) { // Sort all consumers by their capacity - consumers.sortWith(compareBy { it.capacity }) + consumers.sortWith(compareBy { it.ctx.capacity }) // Divide the requests over the available capacity of the input resources fairly for (input in consumers) { - val inputCapacity = input.capacity + val inputCapacity = input.ctx.capacity val fraction = inputCapacity / outputContext.capacity val grantedSpeed = limit * fraction val grantedWork = fraction * work - commands[input] = - if (grantedWork > 0.0 && grantedSpeed > 0.0) - SimResourceCommand.Consume(grantedWork, grantedSpeed, deadline) - else - SimResourceCommand.Idle(deadline) + val command = if (grantedWork > 0.0 && grantedSpeed > 0.0) + SimResourceCommand.Consume(grantedWork, grantedSpeed, deadline) + else + SimResourceCommand.Idle(deadline) + input.push(command) } } - override fun onContextStarted(ctx: SimResourceContext) { - super.onContextStarted(ctx) + override fun doIdle(deadline: Long) { + for (input in consumers) { + input.push(SimResourceCommand.Idle(deadline)) + } + } - consumers.add(ctx) + override fun doFinish(cause: Throwable?) { + val iterator = consumers.iterator() + for (input in iterator) { + iterator.remove() + input.push(SimResourceCommand.Exit) + } } - override fun onContextFinished(ctx: SimResourceContext) { - super.onContextFinished(ctx) + override fun onInputStarted(input: Input) { + consumers.add(input) + } - consumers.remove(ctx) + override fun onInputFinished(input: Input) { + consumers.remove(input) } } diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceConsumer.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceConsumer.kt index 38672b13..4d937514 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceConsumer.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceConsumer.kt @@ -30,13 +30,6 @@ package org.opendc.simulator.resources */ public interface SimResourceConsumer { /** - * This method is invoked when the consumer is started for some resource. - * - * @param ctx The execution context in which the consumer runs. - */ - public fun onStart(ctx: SimResourceContext) {} - - /** * This method is invoked when a resource asks for the next [command][SimResourceCommand] to process, either because * the resource finished processing, reached its deadline or was interrupted. * @@ -46,34 +39,18 @@ public interface SimResourceConsumer { public fun onNext(ctx: SimResourceContext): SimResourceCommand /** - * This method is invoked when the resource provider confirms that the consumer is running at the given speed. + * This method is invoked when an event has occurred. * * @param ctx The execution context in which the consumer runs. - * @param speed The speed at which the consumer runs. + * @param event The event that has occurred. */ - public fun onConfirm(ctx: SimResourceContext, speed: Double) {} + public fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) {} /** - * This is method is invoked when the capacity of the resource changes. - * - * After being informed of such an event, the consumer might decide to adjust its consumption by interrupting the - * resource via [SimResourceContext.interrupt]. Alternatively, the consumer may decide to ignore the event, possibly - * causing the active resource command to finish at a later moment than initially planned. + * This method is invoked when a resource consumer throws an exception. * * @param ctx The execution context in which the consumer runs. - * @param isThrottled A flag to indicate that the active resource command will be throttled as a result of the - * capacity change. - */ - public fun onCapacityChanged(ctx: SimResourceContext, isThrottled: Boolean) {} - - /** - * This method is invoked when the consumer has finished, either because it exited via [SimResourceCommand.Exit], - * the resource finished itself, or a failure occurred at the resource. - * - * Note that throwing an exception in [onStart] or [onNext] is undefined behavior and up to the resource provider. - * - * @param ctx The execution context in which the consumer ran. - * @param cause The cause of the finish in case the resource finished exceptionally. + * @param cause The cause of the failure. */ - public fun onFinish(ctx: SimResourceContext, cause: Throwable? = null) {} + public fun onFailure(ctx: SimResourceContext, cause: Throwable) {} } diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceContext.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceContext.kt index 11dbb09f..7c76c634 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceContext.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceContext.kt @@ -40,6 +40,11 @@ public interface SimResourceContext { public val capacity: Double /** + * The resource processing speed at this instant. + */ + public val speed: Double + + /** * The amount of work still remaining at this instant. */ public val remainingWork: Double diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt index dfdd2c2e..a76cb1e3 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceDistributorMaxMin.kt @@ -22,7 +22,6 @@ package org.opendc.simulator.resources -import java.time.Clock import kotlin.math.max import kotlin.math.min @@ -31,7 +30,7 @@ import kotlin.math.min */ public class SimResourceDistributorMaxMin( override val input: SimResourceProvider, - private val clock: Clock, + private val scheduler: SimResourceScheduler, private val listener: Listener? = null ) : SimResourceDistributor { override val outputs: Set<SimResourceProvider> @@ -90,26 +89,28 @@ public class SimResourceDistributorMaxMin( val remainingWork: Double get() = ctx.remainingWork - override fun onStart(ctx: SimResourceContext) { - this.ctx = ctx - } - override fun onNext(ctx: SimResourceContext): SimResourceCommand { return doNext(ctx.capacity) } - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { - super.onFinish(ctx, cause) - - val iterator = _outputs.iterator() - while (iterator.hasNext()) { - val output = iterator.next() + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { + when (event) { + SimResourceEvent.Start -> { + this.ctx = ctx + } + SimResourceEvent.Exit -> { + val iterator = _outputs.iterator() + while (iterator.hasNext()) { + val output = iterator.next() - // Remove the output from the outputs to prevent ConcurrentModificationException when removing it - // during the call to output.close() - iterator.remove() + // Remove the output from the outputs to prevent ConcurrentModificationException when removing it + // during the call to output.close() + iterator.remove() - output.close() + output.close() + } + } + else -> {} } } } @@ -218,7 +219,7 @@ public class SimResourceDistributorMaxMin( } } - assert(deadline >= clock.millis()) { "Deadline already passed" } + assert(deadline >= scheduler.clock.millis()) { "Deadline already passed" } this.totalRequestedSpeed = totalRequestedSpeed this.totalRequestedWork = totalRequestedWork @@ -335,7 +336,7 @@ public class SimResourceDistributorMaxMin( private inner class OutputContext( private val provider: OutputProvider, consumer: SimResourceConsumer - ) : SimAbstractResourceContext(provider.capacity, clock, consumer), Comparable<OutputContext> { + ) : SimAbstractResourceContext(provider.capacity, scheduler, consumer), Comparable<OutputContext> { /** * The current command that is processed by the vCPU. */ @@ -370,13 +371,11 @@ public class SimResourceDistributorMaxMin( activeCommand = SimResourceCommand.Consume(work, limit, deadline) } - override fun onFinish(cause: Throwable?) { + override fun onFinish() { reportOvercommit() activeCommand = SimResourceCommand.Exit provider.cancel() - - super.onFinish(cause) } override fun getRemainingWork(work: Double, speed: Double, duration: Long): Double { @@ -402,6 +401,8 @@ public class SimResourceDistributorMaxMin( } } + private var isProcessing: Boolean = false + override fun interrupt() { // Prevent users from interrupting the CPU while it is constructing its next command, this will only lead // to infinite recursion. @@ -409,10 +410,16 @@ public class SimResourceDistributorMaxMin( return } - super.interrupt() + try { + isProcessing = false - // Force the scheduler to re-schedule - schedule() + super.interrupt() + + // Force the scheduler to re-schedule + schedule() + } finally { + isProcessing = true + } } override fun compareTo(other: OutputContext): Int = allowedSpeed.compareTo(other.allowedSpeed) diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceEvent.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceEvent.kt new file mode 100644 index 00000000..959427f1 --- /dev/null +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceEvent.kt @@ -0,0 +1,48 @@ +/* + * 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.resources + +/** + * A resource event that is communicated to the resource consumer. + */ +public enum class SimResourceEvent { + /** + * This event is emitted to the consumer when it has started. + */ + Start, + + /** + * This event is emitted to the consumer when it has exited. + */ + Exit, + + /** + * This event is emitted to the consumer when it has started a new resource consumption or idle cycle. + */ + Run, + + /** + * This event is emitted to the consumer when the capacity of the resource has changed. + */ + Capacity, +} diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlushable.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlushable.kt new file mode 100644 index 00000000..f6a1a42e --- /dev/null +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceFlushable.kt @@ -0,0 +1,37 @@ +/* + * 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.resources + +/** + * An interface used by the [SimResourceScheduler] to flush the progress of resource consumer. + */ +public interface SimResourceFlushable { + /** + * Flush the current active resource consumption. + * + * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be + * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer + * will be asked to deliver a new command and is essentially interrupted. + */ + public fun flush(isIntermediate: Boolean) +} diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt index 52b13c5c..2f567a5e 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceProvider.kt @@ -23,6 +23,8 @@ package org.opendc.simulator.resources import kotlinx.coroutines.suspendCancellableCoroutine +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException /** * A [SimResourceProvider] provides some resource of type [R]. @@ -65,15 +67,27 @@ public interface SimResourceProvider : AutoCloseable { public suspend fun SimResourceProvider.consume(consumer: SimResourceConsumer) { return suspendCancellableCoroutine { cont -> startConsumer(object : SimResourceConsumer by consumer { - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { - assert(!cont.isCompleted) { "Coroutine already completed" } + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { + consumer.onEvent(ctx, event) - consumer.onFinish(ctx, cause) + if (event == SimResourceEvent.Exit && !cont.isCompleted) { + cont.resume(Unit) + } + } - cont.resumeWith(if (cause != null) Result.failure(cause) else Result.success(Unit)) + override fun onFailure(ctx: SimResourceContext, cause: Throwable) { + try { + consumer.onFailure(ctx, cause) + cont.resumeWithException(cause) + } catch (e: Throwable) { + e.addSuppressed(cause) + cont.resumeWithException(e) + } } override fun toString(): String = "SimSuspendingResourceConsumer" }) + + cont.invokeOnCancellation { cancel() } } } diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceScheduler.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceScheduler.kt new file mode 100644 index 00000000..a228c47b --- /dev/null +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceScheduler.kt @@ -0,0 +1,69 @@ +/* + * 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.resources + +import java.time.Clock + +/** + * A resource scheduler is responsible for scheduling the communication and synchronization between multiple resource + * providers and consumers. + * + * By centralizing the scheduling logic, updates of resources within a single system can be scheduled and tracked more + * efficiently, reducing the overall work needed per update. + */ +public interface SimResourceScheduler { + /** + * The [Clock] associated with this scheduler. + */ + public val clock: Clock + + /** + * Schedule a direct interrupt for the resource context represented by [flushable]. + * + * @param flushable The resource context that needs to be flushed. + * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be + * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer + * will be asked to deliver a new command and is essentially interrupted. + */ + public fun schedule(flushable: SimResourceFlushable, isIntermediate: Boolean = false) + + /** + * Schedule an interrupt in the future for the resource context represented by [flushable]. + * + * This method will override earlier calls to this method for the same [flushable]. + * + * @param flushable The resource context that needs to be flushed. + * @param timestamp The timestamp when the interrupt should happen. + * @param isIntermediate A flag to indicate that the intermediate progress of the resource consumer should be + * flushed, but without interrupting the resource consumer to submit a new command. If false, the resource consumer + * will be asked to deliver a new command and is essentially interrupted. + */ + public fun schedule(flushable: SimResourceFlushable, timestamp: Long, isIntermediate: Boolean = false) + + /** + * 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 fun batch(block: () -> Unit) +} diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSchedulerTrampoline.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSchedulerTrampoline.kt new file mode 100644 index 00000000..cdbb4a6c --- /dev/null +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSchedulerTrampoline.kt @@ -0,0 +1,95 @@ +/* + * Copyright (c) 2021 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.resources + +import org.opendc.utils.TimerScheduler +import java.time.Clock +import java.util.ArrayDeque +import kotlin.coroutines.CoroutineContext + +/** + * A [SimResourceScheduler] queues all interrupts that occur during execution to be executed after. + * + * @param clock The virtual simulation clock. + */ +public class SimResourceSchedulerTrampoline(context: CoroutineContext, override val clock: Clock) : SimResourceScheduler { + /** + * The [TimerScheduler] to actually schedule the interrupts. + */ + private val timers = TimerScheduler<Any>(context, clock) + + /** + * A flag to indicate that an interrupt is currently running already. + */ + private var isRunning: Boolean = false + + /** + * The queue of resources to be flushed. + */ + private val queue = ArrayDeque<Pair<SimResourceFlushable, Boolean>>() + + override fun schedule(flushable: SimResourceFlushable, isIntermediate: Boolean) { + queue.add(flushable to isIntermediate) + + if (isRunning) { + return + } + + flush() + } + + override fun schedule(flushable: SimResourceFlushable, timestamp: Long, isIntermediate: Boolean) { + timers.startSingleTimerTo(flushable, timestamp) { + schedule(flushable, isIntermediate) + } + } + + override fun batch(block: () -> Unit) { + val wasAlreadyRunning = isRunning + try { + isRunning = true + block() + } finally { + if (!wasAlreadyRunning) { + isRunning = false + } + } + } + + /** + * Flush the scheduled queue. + */ + private fun flush() { + val visited = mutableSetOf<SimResourceFlushable>() + try { + isRunning = true + while (queue.isNotEmpty()) { + val (flushable, isIntermediate) = queue.poll() + flushable.flush(isIntermediate) + visited.add(flushable) + } + } finally { + isRunning = false + } + } +} diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt index 025b0406..3277b889 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSource.kt @@ -22,8 +22,6 @@ package org.opendc.simulator.resources -import org.opendc.utils.TimerScheduler -import java.time.Clock import kotlin.math.ceil import kotlin.math.min @@ -31,13 +29,11 @@ import kotlin.math.min * A [SimResourceSource] represents a source for some resource of type [R] that provides bounded processing capacity. * * @param initialCapacity The initial capacity of the resource. - * @param clock The virtual clock to track simulation time. * @param scheduler The scheduler to schedule the interrupts. */ public class SimResourceSource( initialCapacity: Double, - private val clock: Clock, - private val scheduler: TimerScheduler<Any> + private val scheduler: SimResourceScheduler ) : SimResourceProvider { /** * The current processing speed of the resource. @@ -96,25 +92,27 @@ public class SimResourceSource( /** * Internal implementation of [SimResourceContext] for this class. */ - private inner class Context(consumer: SimResourceConsumer) : SimAbstractResourceContext(capacity, clock, consumer) { + private inner class Context(consumer: SimResourceConsumer) : SimAbstractResourceContext(capacity, scheduler, consumer) { override fun onIdle(deadline: Long) { // Do not resume if deadline is "infinite" if (deadline != Long.MAX_VALUE) { - scheduler.startSingleTimerTo(this, deadline) { flush() } + scheduler.schedule(this, deadline) } } override fun onConsume(work: Double, limit: Double, deadline: Long) { val until = min(deadline, clock.millis() + getDuration(work, speed)) - - scheduler.startSingleTimerTo(this, until, ::flush) + scheduler.schedule(this, until) } - override fun onFinish(cause: Throwable?) { - scheduler.cancel(this) + override fun onFinish() { cancel() - super.onFinish(cause) + ctx = null + + if (this@SimResourceSource.state != SimResourceState.Stopped) { + this@SimResourceSource.state = SimResourceState.Pending + } } override fun toString(): String = "SimResourceSource.Context[capacity=$capacity]" diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt index 45e4c220..1a9dd0bc 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchExclusive.kt @@ -66,10 +66,13 @@ public class SimResourceSwitchExclusive : SimResourceSwitch { availableResources += forwarder input.startConsumer(object : SimResourceConsumer by forwarder { - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { - // De-register the input after it has finished - _inputs -= input - forwarder.onFinish(ctx, cause) + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { + if (event == SimResourceEvent.Exit) { + // De-register the input after it has finished + _inputs -= input + } + + forwarder.onEvent(ctx, event) } }) } diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt index c796c251..5dc1e68d 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceSwitchMaxMin.kt @@ -23,14 +23,13 @@ package org.opendc.simulator.resources import kotlinx.coroutines.* -import java.time.Clock /** * A [SimResourceSwitch] implementation that switches resource consumptions over the available resources using max-min * fair sharing. */ public class SimResourceSwitchMaxMin( - clock: Clock, + scheduler: SimResourceScheduler, private val listener: Listener? = null ) : SimResourceSwitch { private val _outputs = mutableSetOf<SimResourceProvider>() @@ -49,13 +48,13 @@ public class SimResourceSwitchMaxMin( /** * The aggregator to aggregate the resources. */ - private val aggregator = SimResourceAggregatorMaxMin(clock) + private val aggregator = SimResourceAggregatorMaxMin(scheduler) /** * The distributor to distribute the aggregated resources. */ private val distributor = SimResourceDistributorMaxMin( - aggregator.output, clock, + aggregator.output, scheduler, object : SimResourceDistributorMaxMin.Listener { override fun onSliceFinish( switch: SimResourceDistributor, diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt index de455021..32f3f573 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/SimResourceTransformer.kt @@ -75,7 +75,7 @@ public class SimResourceTransformer( if (delegate != null && ctx != null) { this.delegate = null - delegate.onFinish(ctx) + delegate.onEvent(ctx, SimResourceEvent.Exit) } } @@ -90,10 +90,6 @@ public class SimResourceTransformer( } } - override fun onStart(ctx: SimResourceContext) { - this.ctx = ctx - } - override fun onNext(ctx: SimResourceContext): SimResourceCommand { val delegate = delegate @@ -110,7 +106,7 @@ public class SimResourceTransformer( // reset beforehand the existing state and check whether it has been updated afterwards reset() - delegate.onFinish(ctx) + delegate.onEvent(ctx, SimResourceEvent.Exit) if (isCoupled || state == SimResourceState.Stopped) SimResourceCommand.Exit @@ -124,21 +120,31 @@ public class SimResourceTransformer( } } - override fun onConfirm(ctx: SimResourceContext, speed: Double) { - delegate?.onConfirm(ctx, speed) - } - - override fun onCapacityChanged(ctx: SimResourceContext, isThrottled: Boolean) { - delegate?.onCapacityChanged(ctx, isThrottled) + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { + when (event) { + SimResourceEvent.Start -> { + this.ctx = ctx + } + SimResourceEvent.Exit -> { + this.ctx = null + + val delegate = delegate + if (delegate != null) { + reset() + delegate.onEvent(ctx, SimResourceEvent.Exit) + } + } + else -> delegate?.onEvent(ctx, event) + } } - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { + override fun onFailure(ctx: SimResourceContext, cause: Throwable) { this.ctx = null val delegate = delegate if (delegate != null) { reset() - delegate.onFinish(ctx, cause) + delegate.onFailure(ctx, cause) } } @@ -147,7 +153,7 @@ public class SimResourceTransformer( */ private fun start() { val delegate = delegate ?: return - delegate.onStart(checkNotNull(ctx)) + delegate.onEvent(checkNotNull(ctx), SimResourceEvent.Start) hasDelegateStarted = true } diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimSpeedConsumerAdapter.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimSpeedConsumerAdapter.kt index 114c7312..4f4ebb14 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimSpeedConsumerAdapter.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimSpeedConsumerAdapter.kt @@ -25,6 +25,7 @@ package org.opendc.simulator.resources.consumer import org.opendc.simulator.resources.SimResourceCommand import org.opendc.simulator.resources.SimResourceConsumer import org.opendc.simulator.resources.SimResourceContext +import org.opendc.simulator.resources.SimResourceEvent import kotlin.math.min /** @@ -53,28 +54,29 @@ public class SimSpeedConsumerAdapter( return delegate.onNext(ctx) } - override fun onConfirm(ctx: SimResourceContext, speed: Double) { - delegate.onConfirm(ctx, speed) - - this.speed = speed - } - - override fun onCapacityChanged(ctx: SimResourceContext, isThrottled: Boolean) { + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { val oldSpeed = speed - delegate.onCapacityChanged(ctx, isThrottled) + delegate.onEvent(ctx, event) - // Check if the consumer interrupted the consumer and updated the resource consumption. If not, we might - // need to update the current speed. - if (oldSpeed == speed) { - speed = min(ctx.capacity, speed) + when (event) { + SimResourceEvent.Run -> speed = ctx.speed + SimResourceEvent.Capacity -> { + // Check if the consumer interrupted the consumer and updated the resource consumption. If not, we might + // need to update the current speed. + if (oldSpeed == speed) { + speed = min(ctx.capacity, speed) + } + } + SimResourceEvent.Exit -> speed = 0.0 + else -> {} } } - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { - super.onFinish(ctx, cause) - + override fun onFailure(ctx: SimResourceContext, cause: Throwable) { speed = 0.0 + + delegate.onFailure(ctx, cause) } override fun toString(): String = "SimSpeedConsumerAdapter[delegate=$delegate]" diff --git a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimTraceConsumer.kt b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimTraceConsumer.kt index a52d1d5d..2e94e1c1 100644 --- a/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimTraceConsumer.kt +++ b/opendc-simulator/opendc-simulator-resources/src/main/kotlin/org/opendc/simulator/resources/consumer/SimTraceConsumer.kt @@ -25,6 +25,7 @@ package org.opendc.simulator.resources.consumer import org.opendc.simulator.resources.SimResourceCommand import org.opendc.simulator.resources.SimResourceConsumer import org.opendc.simulator.resources.SimResourceContext +import org.opendc.simulator.resources.SimResourceEvent /** * A [SimResourceConsumer] that replays a workload trace consisting of multiple fragments, each indicating the resource @@ -33,11 +34,6 @@ import org.opendc.simulator.resources.SimResourceContext public class SimTraceConsumer(private val trace: Sequence<Fragment>) : SimResourceConsumer { private var iterator: Iterator<Fragment>? = null - override fun onStart(ctx: SimResourceContext) { - check(iterator == null) { "Consumer already running" } - iterator = trace.iterator() - } - override fun onNext(ctx: SimResourceContext): SimResourceCommand { val iterator = checkNotNull(iterator) return if (iterator.hasNext()) { @@ -57,8 +53,17 @@ public class SimTraceConsumer(private val trace: Sequence<Fragment>) : SimResour } } - override fun onFinish(ctx: SimResourceContext, cause: Throwable?) { - iterator = null + override fun onEvent(ctx: SimResourceContext, event: SimResourceEvent) { + when (event) { + SimResourceEvent.Start -> { + check(iterator == null) { "Consumer already running" } + iterator = trace.iterator() + } + SimResourceEvent.Exit -> { + iterator = null + } + else -> {} + } } /** |