From bb3b8e207a08edff81b8c2fe30b476c94bfea086 Mon Sep 17 00:00:00 2001 From: Fabian Mastenbroek Date: Wed, 17 Mar 2021 16:23:48 +0100 Subject: simulator: Make hypervisors generic for the resource type This change moves the hypervisor implementations to the opendc-simulator-resources module and makes them generic to the resource type that is being used (e.g., CPU, disk or networking). --- .../simulator/compute/SimAbstractHypervisor.kt | 164 +++++++ .../opendc/simulator/compute/SimAbstractMachine.kt | 116 +++++ .../simulator/compute/SimBareMetalMachine.kt | 80 +--- .../simulator/compute/SimFairShareHypervisor.kt | 507 +-------------------- .../org/opendc/simulator/compute/SimHypervisor.kt | 5 + .../opendc/simulator/compute/SimMachineContext.kt | 2 +- .../simulator/compute/SimSpaceSharedHypervisor.kt | 263 +---------- .../compute/SimSpaceSharedHypervisorProvider.kt | 2 +- .../simulator/compute/workload/SimTraceWorkload.kt | 5 +- .../compute/workload/SimWorkloadBarrier.kt | 45 -- 10 files changed, 328 insertions(+), 861 deletions(-) create mode 100644 simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt create mode 100644 simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt delete mode 100644 simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadBarrier.kt (limited to 'simulator/opendc-simulator/opendc-simulator-compute/src/main') diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt new file mode 100644 index 00000000..a99b082a --- /dev/null +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractHypervisor.kt @@ -0,0 +1,164 @@ +/* + * 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.compute + +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.launch +import org.opendc.simulator.compute.interference.PerformanceInterferenceModel +import org.opendc.simulator.compute.model.SimMemoryUnit +import org.opendc.simulator.compute.model.SimProcessingUnit +import org.opendc.simulator.compute.workload.SimWorkload +import org.opendc.simulator.resources.* +import java.time.Clock +import kotlin.coroutines.CoroutineContext + +/** + * Abstract implementation of the [SimHypervisor] interface. + */ +public abstract class SimAbstractHypervisor : SimHypervisor { + /** + * The machine on which the hypervisor runs. + */ + private lateinit var context: SimMachineContext + + /** + * The resource switch to use. + */ + private lateinit var switch: SimResourceSwitch + + /** + * The virtual machines running on this hypervisor. + */ + private val _vms = mutableSetOf() + override val vms: Set + get() = _vms + + /** + * Construct the [SimResourceSwitch] implementation that performs the actual scheduling of the CPUs. + */ + public abstract fun createSwitch(ctx: SimMachineContext): SimResourceSwitch + + /** + * Check whether the specified machine model fits on this hypervisor. + */ + public abstract fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean + + override fun canFit(model: SimMachineModel): Boolean { + return canFit(model, switch) + } + + override fun createMachine( + model: SimMachineModel, + performanceInterferenceModel: PerformanceInterferenceModel? + ): SimMachine { + require(canFit(model)) { "Machine does not fit" } + val vm = VirtualMachine(model, performanceInterferenceModel) + _vms.add(vm) + return vm + } + + /** + * A virtual machine running on the hypervisor. + * + * @property model The machine model of the virtual machine. + * @property performanceInterferenceModel The performance interference model to utilize. + */ + private inner class VirtualMachine( + override val model: SimMachineModel, + val performanceInterferenceModel: PerformanceInterferenceModel? = null, + ) : SimMachine { + /** + * A [StateFlow] representing the CPU usage of the simulated machine. + */ + override val usage: MutableStateFlow = MutableStateFlow(0.0) + + /** + * A flag to indicate that the machine is terminated. + */ + private var isTerminated = false + + /** + * The vCPUs of the machine. + */ + private val cpus: Map> = model.cpus.associateWith { switch.addOutput(it) } + + /** + * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished. + */ + override suspend fun run(workload: SimWorkload, meta: Map) { + coroutineScope { + require(!isTerminated) { "Machine is terminated" } + + val ctx = object : SimMachineContext { + override val cpus: List + get() = model.cpus + + override val memory: List + get() = model.memory + + override val clock: Clock + get() = this@SimAbstractHypervisor.context.clock + + override val meta: Map = meta + mapOf("coroutine-context" to context.meta["coroutine-context"] as CoroutineContext) + + override fun interrupt(resource: SimResource) { + requireNotNull(this@VirtualMachine.cpus[resource]).interrupt() + } + } + + workload.onStart(ctx) + + for ((cpu, provider) in cpus) { + launch { + provider.consume(workload.getConsumer(ctx, cpu)) + } + } + } + } + + /** + * Terminate this VM instance. + */ + override fun close() { + if (!isTerminated) { + cpus.forEach { (_, provider) -> provider.close() } + _vms.remove(this) + } + + isTerminated = true + } + } + + override fun onStart(ctx: SimMachineContext) { + context = ctx + switch = createSwitch(ctx) + } + + override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer { + val forwarder = SimResourceForwarder(cpu) + switch.addInput(forwarder) + return forwarder + } +} diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt new file mode 100644 index 00000000..1bdbb7e8 --- /dev/null +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimAbstractMachine.kt @@ -0,0 +1,116 @@ +/* + * 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.compute + +import kotlinx.coroutines.* +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach +import org.opendc.simulator.compute.model.SimMemoryUnit +import org.opendc.simulator.compute.model.SimProcessingUnit +import org.opendc.simulator.compute.workload.SimWorkload +import org.opendc.simulator.resources.SimResource +import org.opendc.simulator.resources.SimResourceProvider +import org.opendc.simulator.resources.SimResourceSource +import java.time.Clock +import kotlin.coroutines.CoroutineContext + +/** + * Abstract implementation of the [SimMachine] interface. + * + * @param context The [CoroutineContext] in which the machine runs. + */ +public abstract class SimAbstractMachine(private val clock: Clock) : SimMachine { + private val _usage = MutableStateFlow(0.0) + override val usage: StateFlow + get() = _usage + + /** + * A flag to indicate that the machine is terminated. + */ + private var isTerminated = false + + /** + * The [CoroutineContext] to run in. + */ + protected abstract val context: CoroutineContext + + /** + * The resources allocated for this machine. + */ + protected abstract val resources: Map> + + /** + * The execution context in which the workload runs. + */ + private inner class Context( + val sources: Map>, + override val meta: Map + ) : SimMachineContext { + override val clock: Clock + get() = this@SimAbstractMachine.clock + + override val cpus: List = model.cpus + + override val memory: List = model.memory + + override fun interrupt(resource: SimResource) { + checkNotNull(sources[resource]) { "Invalid resource" }.interrupt() + } + } + + /** + * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished. + */ + override suspend fun run(workload: SimWorkload, meta: Map): Unit = withContext(context) { + val resources = resources + require(!isTerminated) { "Machine is terminated" } + val ctx = Context(resources, meta + mapOf("coroutine-context" to context)) + val totalCapacity = model.cpus.sumByDouble { it.frequency } + + workload.onStart(ctx) + + for ((cpu, source) in resources) { + val consumer = workload.getConsumer(ctx, cpu) + val job = source.speed + .onEach { + _usage.value = resources.values.sumByDouble { it.speed.value } / totalCapacity + } + .launchIn(this) + + launch { + source.consume(consumer) + job.cancel() + } + } + } + + override fun close() { + if (!isTerminated) { + resources.forEach { (_, provider) -> provider.close() } + } else { + isTerminated = true + } + } +} diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt index b1d1c0b7..79982ea8 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimBareMetalMachine.kt @@ -23,17 +23,10 @@ package org.opendc.simulator.compute import kotlinx.coroutines.* -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.flow.launchIn -import kotlinx.coroutines.flow.onEach -import org.opendc.simulator.compute.model.SimMemoryUnit import org.opendc.simulator.compute.model.SimProcessingUnit -import org.opendc.simulator.compute.workload.SimWorkload import org.opendc.simulator.resources.* import org.opendc.utils.TimerScheduler import java.time.Clock -import java.util.* import kotlin.coroutines.* /** @@ -42,83 +35,34 @@ import kotlin.coroutines.* * A [SimBareMetalMachine] is a stateful object and you should be careful when operating this object concurrently. For * example. the class expects only a single concurrent call to [run]. * - * @param coroutineScope The [CoroutineScope] to run the simulated workload in. + * @param context The [CoroutineContext] to run the simulated workload in. * @param clock The virtual clock to track the simulation time. * @param model The machine model to simulate. */ @OptIn(ExperimentalCoroutinesApi::class, InternalCoroutinesApi::class) public class SimBareMetalMachine( - private val coroutineScope: CoroutineScope, + context: CoroutineContext, private val clock: Clock, override val model: SimMachineModel -) : SimMachine { - private val _usage = MutableStateFlow(0.0) - override val usage: StateFlow - get() = _usage - +) : SimAbstractMachine(clock) { /** - * A flag to indicate that the machine is terminated. + * The [Job] associated with this machine. */ - private var isTerminated = false + private val job = Job() - /** - * The [TimerScheduler] to use for scheduling the interrupts. - */ - private val scheduler = TimerScheduler(coroutineScope, clock) + override val context: CoroutineContext = context + job /** - * The execution context in which the workload runs. - */ - private inner class Context(val map: Map>, - override val meta: Map) : SimMachineContext { - override val clock: Clock - get() = this@SimBareMetalMachine.clock - - override val cpus: List = model.cpus - - override val memory: List = model.memory - - override fun interrupt(resource: SimResource) { - val context = map[resource] - checkNotNull(context) { "Invalid resource" } - context.interrupt() - } - } - - /** - * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished. + * The [TimerScheduler] to use for scheduling the interrupts. */ - override suspend fun run(workload: SimWorkload, meta: Map): Unit = coroutineScope { - require(!isTerminated) { "Machine is terminated" } - val map = mutableMapOf>() - val ctx = Context(map, meta) - val sources = model.cpus.map { SimResourceSource(it, clock, scheduler) } - val totalCapacity = model.cpus.sumByDouble { it.frequency } + private val scheduler = TimerScheduler(this.context, clock) - workload.onStart(ctx) - - for (source in sources) { - val consumer = workload.getConsumer(ctx, source.resource) - val job = source.speed - .onEach { - _usage.value = sources.sumByDouble { it.speed.value } / totalCapacity - } - .launchIn(this) - - launch { - source.consume(object : SimResourceConsumer by consumer { - override fun onStart(ctx: SimResourceContext): SimResourceCommand { - map[ctx.resource] = ctx - return consumer.onStart(ctx) - } - }) - job.cancel() - } - } - } + override val resources: Map> = + model.cpus.associateWith { SimResourceSource(it, clock, scheduler) } override fun close() { - isTerminated = true + super.close() scheduler.close() + job.cancel() } } diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt index 12b3b428..bb97192d 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimFairShareHypervisor.kt @@ -22,22 +22,10 @@ package org.opendc.simulator.compute -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.suspendCancellableCoroutine -import org.opendc.simulator.compute.interference.PerformanceInterferenceModel -import org.opendc.simulator.compute.model.SimMemoryUnit import org.opendc.simulator.compute.model.SimProcessingUnit import org.opendc.simulator.compute.workload.SimWorkload -import org.opendc.simulator.compute.workload.SimWorkloadBarrier import org.opendc.simulator.resources.* -import java.time.Clock -import kotlin.coroutines.Continuation -import kotlin.coroutines.resume -import kotlin.coroutines.resumeWithException -import kotlin.math.ceil -import kotlin.math.max -import kotlin.math.min +import kotlin.coroutines.CoroutineContext /** * A [SimHypervisor] that distributes the computing requirements of multiple [SimWorkload] on a single @@ -45,482 +33,27 @@ import kotlin.math.min * * @param listener The hypervisor listener to use. */ -public class SimFairShareHypervisor(private val listener: SimHypervisor.Listener? = null) : SimHypervisor, SimResourceConsumer { - - override fun onStart(ctx: SimMachineContext) { - this.ctx = ctx - this.commands = Array(ctx.cpus.size) { SimResourceCommand.Idle() } - this.pCpus = ctx.cpus.indices.sortedBy { ctx.cpus[it].frequency }.toIntArray() - this.maxUsage = ctx.cpus.sumByDouble { it.frequency } - this.barrier = SimWorkloadBarrier(ctx.cpus.size) - } - - override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer { - return this - } - - override fun onNext(ctx: SimResourceContext, remainingWork: Double): SimResourceCommand { - val cpu = ctx.resource.id - totalRemainingWork += remainingWork - val isLast = barrier.enter() - - // Flush the progress of the guest after the barrier has been reached. - if (isLast && isDirty) { - isDirty = false - flushGuests() - } - - return if (isDirty) { - // Wait for the scheduler determine the work after the barrier has been reached by all CPUs. - SimResourceCommand.Idle() - } else { - // Indicate that the scheduler needs to run next call. - if (isLast) { - isDirty = true - } - - commands[cpu] - } - } - - override fun onStart(ctx: SimResourceContext): SimResourceCommand { - return commands[ctx.resource.id] - } - - override fun canFit(model: SimMachineModel): Boolean = true - - override fun createMachine( - model: SimMachineModel, - performanceInterferenceModel: PerformanceInterferenceModel? - ): SimMachine = SimVm(model, performanceInterferenceModel) - - /** - * The execution context in which the hypervisor runs. - */ - private lateinit var ctx: SimMachineContext - - /** - * The commands to submit to the underlying host. - */ - private lateinit var commands: Array - - /** - * The active vCPUs. - */ - private val vcpus: MutableList = mutableListOf() - - /** - * The indices of the physical CPU ordered by their speed. - */ - private lateinit var pCpus: IntArray - - /** - * The maximum amount of work to be performed per second. - */ - private var maxUsage: Double = 0.0 - - /** - * The current load on the hypervisor. - */ - private var load: Double = 0.0 - - /** - * The total amount of remaining work (of all pCPUs). - */ - private var totalRemainingWork: Double = 0.0 - - /** - * The total speed requested by the vCPUs. - */ - private var totalRequestedSpeed = 0.0 - - /** - * The total amount of work requested by the vCPUs. - */ - private var totalRequestedWork = 0.0 - - /** - * The total allocated speed for the vCPUs. - */ - private var totalAllocatedSpeed = 0.0 - - /** - * The total allocated work requested for the vCPUs. - */ - private var totalAllocatedWork = 0.0 - - /** - * The amount of work that could not be performed due to over-committing resources. - */ - private var totalOvercommittedWork = 0.0 - - /** - * The amount of work that was lost due to interference. - */ - private var totalInterferedWork = 0.0 - - /** - * A flag to indicate that the scheduler has submitted work that has not yet been completed. - */ - private var isDirty: Boolean = false - - /** - * The scheduler barrier. - */ - private lateinit var barrier: SimWorkloadBarrier - - /** - * Indicate that the workloads should be re-scheduled. - */ - private fun shouldSchedule() { - isDirty = true - ctx.interruptAll() - } - - /** - * Schedule the work over the physical CPUs. - */ - private fun doSchedule() { - // If there is no work yet, mark all pCPUs as idle. - if (vcpus.isEmpty()) { - commands.fill(SimResourceCommand.Idle()) - ctx.interruptAll() - } - - var duration: Double = Double.MAX_VALUE - var deadline: Long = Long.MAX_VALUE - var availableSpeed = maxUsage - var totalRequestedSpeed = 0.0 - var totalRequestedWork = 0.0 - - // Sort the vCPUs based on their requested usage - // Profiling shows that it is faster to sort every slice instead of maintaining some kind of sorted set - vcpus.sort() - - // Divide the available host capacity fairly across the vCPUs using max-min fair sharing - val vcpuIterator = vcpus.listIterator() - var remaining = vcpus.size - while (vcpuIterator.hasNext()) { - val vcpu = vcpuIterator.next() - val availableShare = availableSpeed / remaining-- - - when (val command = vcpu.activeCommand) { - is SimResourceCommand.Idle -> { - // Take into account the minimum deadline of this slice before we possible continue - deadline = min(deadline, command.deadline) - - vcpu.actualSpeed = 0.0 - } - is SimResourceCommand.Consume -> { - val grantedSpeed = min(vcpu.allowedSpeed, availableShare) - - // Take into account the minimum deadline of this slice before we possible continue - deadline = min(deadline, command.deadline) - - // Ignore idle computation - if (grantedSpeed <= 0.0 || command.work <= 0.0) { - vcpu.actualSpeed = 0.0 - continue - } - - totalRequestedSpeed += command.limit - totalRequestedWork += command.work - - vcpu.actualSpeed = grantedSpeed - availableSpeed -= grantedSpeed - - // The duration that we want to run is that of the shortest request from a vCPU - duration = min(duration, command.work / grantedSpeed) - } - SimResourceCommand.Exit -> { - // Apparently the vCPU has exited, so remove it from the scheduling queue. - vcpuIterator.remove() +public class SimFairShareHypervisor(private val listener: SimHypervisor.Listener? = null) : SimAbstractHypervisor() { + + override fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean = true + + override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch { + return SimResourceSwitchMaxMin( + ctx.clock, + ctx.meta["coroutine-context"] as CoroutineContext, + object : SimResourceSwitchMaxMin.Listener { + override fun onSliceFinish( + switch: SimResourceSwitchMaxMin, + requestedWork: Long, + grantedWork: Long, + overcommittedWork: Long, + interferedWork: Long, + cpuUsage: Double, + cpuDemand: Double + ) { + listener?.onSliceFinish(this@SimFairShareHypervisor, requestedWork, grantedWork, overcommittedWork, interferedWork, cpuUsage, cpuDemand) } } - } - - // Round the duration to milliseconds - duration = ceil(duration * 1000) / 1000 - - assert(deadline >= ctx.clock.millis()) { "Deadline already passed" } - - val totalAllocatedSpeed = maxUsage - availableSpeed - var totalAllocatedWork = 0.0 - availableSpeed = totalAllocatedSpeed - load = totalAllocatedSpeed / maxUsage - - // Divide the requests over the available capacity of the pCPUs fairly - for (i in pCpus) { - val maxCpuUsage = ctx.cpus[i].frequency - val fraction = maxCpuUsage / maxUsage - val grantedSpeed = min(maxCpuUsage, totalAllocatedSpeed * fraction) - val grantedWork = duration * grantedSpeed - - commands[i] = - if (grantedWork > 0.0 && grantedSpeed > 0.0) - SimResourceCommand.Consume(grantedWork, grantedSpeed, deadline) - else - SimResourceCommand.Idle(deadline) - - totalAllocatedWork += grantedWork - availableSpeed -= grantedSpeed - } - - this.totalRequestedSpeed = totalRequestedSpeed - this.totalRequestedWork = totalRequestedWork - this.totalAllocatedSpeed = totalAllocatedSpeed - this.totalAllocatedWork = totalAllocatedWork - - ctx.interruptAll() - } - - /** - * Flush the progress of the vCPUs. - */ - private fun flushGuests() { - // Flush all the vCPUs work - for (vcpu in vcpus) { - vcpu.flush(isIntermediate = true) - } - - // Report metrics - listener?.onSliceFinish( - this, - totalRequestedWork.toLong(), - (totalAllocatedWork - totalRemainingWork).toLong(), - totalOvercommittedWork.toLong(), - totalInterferedWork.toLong(), - totalRequestedSpeed, - totalAllocatedSpeed ) - totalRemainingWork = 0.0 - totalInterferedWork = 0.0 - totalOvercommittedWork = 0.0 - - // Force all pCPUs to re-schedule their work. - doSchedule() - } - - /** - * Interrupt all host CPUs. - */ - private fun SimMachineContext.interruptAll() { - for (cpu in ctx.cpus) { - interrupt(cpu) - } - } - - /** - * A virtual machine running on the hypervisor. - * - * @property model The machine model of the virtual machine. - * @property performanceInterferenceModel The performance interference model to utilize. - */ - private inner class SimVm( - override val model: SimMachineModel, - val performanceInterferenceModel: PerformanceInterferenceModel? = null, - ) : SimMachine { - /** - * A [StateFlow] representing the CPU usage of the simulated machine. - */ - override val usage: MutableStateFlow = MutableStateFlow(0.0) - - /** - * A flag to indicate that the machine is terminated. - */ - private var isTerminated = false - - /** - * The current active workload. - */ - private var cont: Continuation? = null - - /** - * The active CPUs of this virtual machine. - */ - private var cpus: List = emptyList() - - /** - * The execution context in which the workload runs. - */ - inner class Context(override val meta: Map) : SimMachineContext { - override val cpus: List - get() = model.cpus - - override val memory: List - get() = model.memory - - override val clock: Clock - get() = this@SimFairShareHypervisor.ctx.clock - - override fun interrupt(resource: SimResource) { - TODO() - } - } - - lateinit var ctx: SimMachineContext - - /** - * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished. - */ - override suspend fun run(workload: SimWorkload, meta: Map) { - require(!isTerminated) { "Machine is terminated" } - require(cont == null) { "Run should not be called concurrently" } - - ctx = Context(meta) - workload.onStart(ctx) - - return suspendCancellableCoroutine { cont -> - this.cont = cont - this.cpus = model.cpus.map { VCpu(this, it, workload.getConsumer(ctx, it), ctx.clock) } - - for (cpu in cpus) { - // Register vCPU to scheduler - vcpus.add(cpu) - - cpu.start() - } - - // Re-schedule the work over the pCPUs - shouldSchedule() - } - } - - /** - * Terminate this VM instance. - */ - override fun close() { - isTerminated = true - } - - /** - * Update the usage of the VM. - */ - fun updateUsage() { - usage.value = cpus.sumByDouble { it.actualSpeed } / cpus.sumByDouble { it.resource.frequency } - } - - /** - * This method is invoked when one of the CPUs has exited. - */ - fun onCpuExit() { - // Check whether all other CPUs have finished - if (cpus.all { it.hasExited }) { - val cont = cont - this.cont = null - cont?.resume(Unit) - } - } - - /** - * This method is invoked when one of the CPUs failed. - */ - fun onCpuFailure(e: Throwable) { - // In case the flush fails with an exception, immediately propagate to caller, cancelling all other - // tasks. - val cont = cont - this.cont = null - cont?.resumeWithException(e) - } - } - - /** - * A CPU of the virtual machine. - */ - private inner class VCpu( - val vm: SimVm, - resource: SimProcessingUnit, - consumer: SimResourceConsumer, - clock: Clock - ) : SimAbstractResourceContext(resource, clock, consumer), Comparable { - /** - * The current command that is processed by the vCPU. - */ - var activeCommand: SimResourceCommand = SimResourceCommand.Idle() - - /** - * The processing speed that is allowed by the model constraints. - */ - var allowedSpeed: Double = 0.0 - - /** - * The actual processing speed. - */ - var actualSpeed: Double = 0.0 - set(value) { - field = value - vm.updateUsage() - } - - /** - * A flag to indicate that the CPU has exited. - */ - var hasExited: Boolean = false - - override fun onIdle(deadline: Long) { - allowedSpeed = 0.0 - activeCommand = SimResourceCommand.Idle(deadline) - } - - override fun onConsume(work: Double, limit: Double, deadline: Long) { - allowedSpeed = getSpeed(limit) - activeCommand = SimResourceCommand.Consume(work, limit, deadline) - } - - override fun onFinish() { - hasExited = true - activeCommand = SimResourceCommand.Exit - vm.onCpuExit() - } - - override fun onFailure(cause: Throwable) { - hasExited = true - activeCommand = SimResourceCommand.Exit - vm.onCpuFailure(cause) - } - - override fun getRemainingWork(work: Double, speed: Double, duration: Long, isInterrupted: Boolean): Double { - // Apply performance interference model - val performanceScore = vm.performanceInterferenceModel?.apply(load) ?: 1.0 - - // Compute the remaining amount of work - val remainingWork = if (work > 0.0) { - // Compute the fraction of compute time allocated to the VM - val fraction = actualSpeed / totalAllocatedSpeed - - // Compute the work that was actually granted to the VM. - val processingAvailable = max(0.0, totalAllocatedWork - totalRemainingWork) * fraction - val processed = processingAvailable * performanceScore - - val interferedWork = processingAvailable - processed - - totalInterferedWork += interferedWork - - max(0.0, work - processed) - } else { - 0.0 - } - - if (!isInterrupted) { - totalOvercommittedWork += remainingWork - } - - return remainingWork - } - - override fun interrupt() { - // Prevent users from interrupting the CPU while it is constructing its next command, this will only lead - // to infinite recursion. - if (isProcessing) { - return - } - - super.interrupt() - - // Force the scheduler to re-schedule - shouldSchedule() - } - - override fun compareTo(other: VCpu): Int = allowedSpeed.compareTo(other.allowedSpeed) } } diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisor.kt index d8f00bef..4a233fec 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisor.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimHypervisor.kt @@ -30,6 +30,11 @@ import org.opendc.simulator.compute.workload.SimWorkload * to a [SimBareMetalMachine]. */ public interface SimHypervisor : SimWorkload { + /** + * The machines running on the hypervisor. + */ + public val vms: Set + /** * Determine whether the specified machine characterized by [model] can fit on this hypervisor at this moment. */ diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt index 5c67b990..cff70826 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimMachineContext.kt @@ -37,7 +37,7 @@ public interface SimMachineContext { * The virtual clock tracking simulation time. */ public val clock: Clock - + /** * The metadata associated with the context. */ diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt index 751873a5..2001a230 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisor.kt @@ -22,270 +22,19 @@ package org.opendc.simulator.compute -import kotlinx.coroutines.flow.MutableStateFlow -import kotlinx.coroutines.flow.StateFlow -import kotlinx.coroutines.suspendCancellableCoroutine -import org.opendc.simulator.compute.interference.PerformanceInterferenceModel -import org.opendc.simulator.compute.model.SimMemoryUnit import org.opendc.simulator.compute.model.SimProcessingUnit -import org.opendc.simulator.compute.workload.SimWorkload import org.opendc.simulator.resources.* -import java.time.Clock -import java.util.ArrayDeque -import kotlin.coroutines.Continuation -import kotlin.coroutines.resume -import kotlin.coroutines.resumeWithException +import kotlin.coroutines.CoroutineContext /** * A [SimHypervisor] that allocates its sub-resources exclusively for the virtual machine that it hosts. - * - * @param listener The hypervisor listener to use. */ -public class SimSpaceSharedHypervisor(private val listener: SimHypervisor.Listener? = null) : SimHypervisor, SimResourceConsumer { - /** - * The execution context in which the hypervisor runs. - */ - private lateinit var ctx: SimMachineContext - - /** - * The mapping from pCPU to vCPU. - */ - private lateinit var vcpus: Array - - /** - * The available physical CPUs to schedule on. - */ - private val availableCpus = ArrayDeque() - - override fun canFit(model: SimMachineModel): Boolean = availableCpus.size >= model.cpus.size - - override fun createMachine( - model: SimMachineModel, - performanceInterferenceModel: PerformanceInterferenceModel? - ): SimMachine { - require(canFit(model)) { "Cannot fit machine" } - return SimVm(model, performanceInterferenceModel) - } - - override fun onStart(ctx: SimMachineContext) { - this.ctx = ctx - this.vcpus = arrayOfNulls(ctx.cpus.size) - this.availableCpus.addAll(ctx.cpus.indices) - } - - override fun getConsumer(ctx: SimMachineContext, cpu: SimProcessingUnit): SimResourceConsumer { - return this +public class SimSpaceSharedHypervisor : SimAbstractHypervisor() { + override fun canFit(model: SimMachineModel, switch: SimResourceSwitch): Boolean { + return switch.inputs.size - switch.outputs.size >= model.cpus.size } - override fun onStart(ctx: SimResourceContext): SimResourceCommand { - return onNext(ctx, 0.0) - } - - override fun onNext(ctx: SimResourceContext, remainingWork: Double): SimResourceCommand { - val vcpu = vcpus[ctx.resource.id] ?: return SimResourceCommand.Idle() - - if (vcpu.isStarted) { - vcpu.remainingWork = remainingWork - vcpu.flush() - } else { - vcpu.isStarted = true - vcpu.start() - } - - if (vcpu.hasExited && vcpu != vcpus[ctx.resource.id]) { - return onNext(ctx, remainingWork) - } - - return vcpu.activeCommand - } - - /** - * A virtual machine running on the hypervisor. - * - * @property model The machine model of the virtual machine. - * @property performanceInterferenceModel The performance interference model to utilize. - */ - private inner class SimVm( - override val model: SimMachineModel, - val performanceInterferenceModel: PerformanceInterferenceModel? = null, - ) : SimMachine { - /** - * A flag to indicate that the machine is terminated. - */ - private var isTerminated = false - - /** - * A [StateFlow] representing the CPU usage of the simulated machine. - */ - override val usage: MutableStateFlow = MutableStateFlow(0.0) - - /** - * The current active workload. - */ - private var cont: Continuation? = null - - /** - * The physical CPUs that have been allocated. - */ - private val pCPUs = model.cpus.map { availableCpus.poll() }.toIntArray() - - /** - * The active CPUs of this virtual machine. - */ - private var cpus: List = emptyList() - - /** - * The execution context in which the workload runs. - */ - inner class Context(override val meta: Map) : SimMachineContext { - override val cpus: List - get() = model.cpus - - override val memory: List - get() = model.memory - - override val clock: Clock - get() = this@SimSpaceSharedHypervisor.ctx.clock - - override fun interrupt(resource: SimResource) { - TODO() - } - } - - lateinit var ctx: SimMachineContext - - /** - * Run the specified [SimWorkload] on this machine and suspend execution util the workload has finished. - */ - override suspend fun run(workload: SimWorkload, meta: Map) { - require(!isTerminated) { "Machine is terminated" } - require(cont == null) { "Run should not be called concurrently" } - - ctx = Context(meta) - workload.onStart(ctx) - - return suspendCancellableCoroutine { cont -> - this.cont = cont - try { - this.cpus = model.cpus.map { model -> VCpu(this, model, workload.getConsumer(ctx, model), ctx.clock) } - - for ((index, pCPU) in pCPUs.withIndex()) { - vcpus[pCPU] = cpus[index] - this@SimSpaceSharedHypervisor.ctx.interrupt(this@SimSpaceSharedHypervisor.ctx.cpus[pCPU]) - } - } catch (e: Throwable) { - cont.resumeWithException(e) - } - } - } - - override fun close() { - isTerminated = true - for (pCPU in pCPUs) { - vcpus[pCPU] = null - availableCpus.add(pCPU) - } - - val cont = cont - this.cont = null - cont?.resume(Unit) - } - - /** - * Update the usage of the VM. - */ - fun updateUsage() { - usage.value = cpus.sumByDouble { it.speed } / cpus.sumByDouble { it.resource.frequency } - } - - /** - * This method is invoked when one of the CPUs has exited. - */ - fun onCpuExit() { - // Check whether all other CPUs have finished - if (cpus.all { it.hasExited }) { - val cont = cont - this.cont = null - cont?.resume(Unit) - } - } - - /** - * This method is invoked when one of the CPUs failed. - */ - fun onCpuFailure(e: Throwable) { - // In case the flush fails with an exception, immediately propagate to caller, cancelling all other - // tasks. - val cont = cont - this.cont = null - cont?.resumeWithException(e) - } - } - - /** - * A CPU of the virtual machine. - */ - private inner class VCpu( - val vm: SimVm, - resource: SimProcessingUnit, - consumer: SimResourceConsumer, - clock: Clock - ) : SimAbstractResourceContext(resource, clock, consumer) { - /** - * Indicates that the vCPU was started. - */ - var isStarted: Boolean = false - - /** - * The current command that is processed by the vCPU. - */ - var activeCommand: SimResourceCommand = SimResourceCommand.Idle() - - /** - * The processing speed of the vCPU. - */ - var speed: Double = 0.0 - set(value) { - field = value - vm.updateUsage() - } - - /** - * The amount of work remaining from the previous consumption. - */ - var remainingWork: Double = 0.0 - - /** - * A flag to indicate that the CPU has exited. - */ - var hasExited: Boolean = false - - override fun onIdle(deadline: Long) { - speed = 0.0 - activeCommand = SimResourceCommand.Idle(deadline) - } - - override fun onConsume(work: Double, limit: Double, deadline: Long) { - speed = getSpeed(limit) - activeCommand = SimResourceCommand.Consume(work, speed, deadline) - } - - override fun onFinish() { - speed = 0.0 - hasExited = true - activeCommand = SimResourceCommand.Idle() - vm.onCpuExit() - } - - override fun onFailure(cause: Throwable) { - speed = 0.0 - hasExited = true - activeCommand = SimResourceCommand.Idle() - vm.onCpuFailure(cause) - } - - override fun getRemainingWork(work: Double, speed: Double, duration: Long, isInterrupted: Boolean): Double { - return remainingWork - } + override fun createSwitch(ctx: SimMachineContext): SimResourceSwitch { + return SimResourceSwitchExclusive(ctx.meta["coroutine-context"] as CoroutineContext) } } diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt index 3d49e544..e2044d05 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/SimSpaceSharedHypervisorProvider.kt @@ -28,5 +28,5 @@ package org.opendc.simulator.compute public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider { override val id: String = "space-shared" - override fun create(listener: SimHypervisor.Listener?): SimHypervisor = SimSpaceSharedHypervisor(listener) + override fun create(listener: SimHypervisor.Listener?): SimHypervisor = SimSpaceSharedHypervisor() } diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt index edef3843..31f58a0f 100644 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt +++ b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimTraceWorkload.kt @@ -27,6 +27,7 @@ import org.opendc.simulator.compute.model.SimProcessingUnit import org.opendc.simulator.resources.SimResourceCommand import org.opendc.simulator.resources.SimResourceConsumer import org.opendc.simulator.resources.SimResourceContext +import org.opendc.simulator.resources.consumer.SimConsumerBarrier /** * A [SimWorkload] that replays a workload trace consisting of multiple fragments, each indicating the resource @@ -36,10 +37,10 @@ public class SimTraceWorkload(public val trace: Sequence) : SimWorkloa private var offset = 0L private val iterator = trace.iterator() private var fragment: Fragment? = null - private lateinit var barrier: SimWorkloadBarrier + private lateinit var barrier: SimConsumerBarrier override fun onStart(ctx: SimMachineContext) { - barrier = SimWorkloadBarrier(ctx.cpus.size) + barrier = SimConsumerBarrier(ctx.cpus.size) fragment = nextFragment() offset = ctx.clock.millis() } diff --git a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadBarrier.kt b/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadBarrier.kt deleted file mode 100644 index 45a299be..00000000 --- a/simulator/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/workload/SimWorkloadBarrier.kt +++ /dev/null @@ -1,45 +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.compute.workload - -/** - * The [SimWorkloadBarrier] is a barrier that allows workloads to wait for a select number of CPUs to complete, before - * proceeding its operation. - */ -public class SimWorkloadBarrier(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 - } -} -- cgit v1.2.3