diff options
29 files changed, 416 insertions, 356 deletions
diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt index 0fa91d52..c04573b5 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimHost.kt @@ -35,7 +35,6 @@ import org.opendc.compute.simulator.internal.Guest import org.opendc.compute.simulator.internal.GuestListener import org.opendc.simulator.compute.* import org.opendc.simulator.compute.kernel.SimHypervisor -import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.compute.model.MemoryUnit import org.opendc.simulator.compute.workload.SimWorkload @@ -57,7 +56,6 @@ public class SimHost( private val machine: SimBareMetalMachine, private val hypervisor: SimHypervisor, private val mapper: SimWorkloadMapper = SimMetaWorkloadMapper(), - private val interferenceDomain: VmInterferenceDomain? = null, private val optimize: Boolean = false ) : Host, AutoCloseable { /** @@ -119,8 +117,7 @@ public class SimHost( val guest = guests.computeIfAbsent(server) { key -> require(canFit(key)) { "Server does not fit" } - val interferenceKey = interferenceDomain?.getMember(key.name) - val machine = hypervisor.newMachine(key.flavor.toMachineModel(), interferenceKey) + val machine = hypervisor.newMachine(key.flavor.toMachineModel()) val newGuest = Guest( context, clock, diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt index ea3c6549..cc084526 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/internal/Guest.kt @@ -217,7 +217,7 @@ internal class Guest( */ private suspend fun runMachine(workload: SimWorkload) { delay(1) // TODO Introduce model for boot time - machine.runWorkload(workload, mapOf("driver" to host, "server" to server)) + machine.runWorkload(workload, mapOf("driver" to host, "server" to server) + server.meta) } /** diff --git a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt index 06500a06..879f15b2 100644 --- a/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt +++ b/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimHostTest.kt @@ -71,7 +71,7 @@ internal class SimHostTest { val duration = 5 * 60L val engine = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) - val hypervisor = SimFairShareHypervisor(engine, null, SplittableRandom(1)) + val hypervisor = SimFairShareHypervisor(engine, SplittableRandom(1), null) val host = SimHost( uid = UUID.randomUUID(), name = "test", @@ -155,7 +155,7 @@ internal class SimHostTest { val duration = 5 * 60L val engine = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) - val hypervisor = SimFairShareHypervisor(engine, null, SplittableRandom(1)) + val hypervisor = SimFairShareHypervisor(engine, SplittableRandom(1), null) val host = SimHost( uid = UUID.randomUUID(), name = "test", diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt index 92652329..ad132efe 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeServiceHelper.kt @@ -32,7 +32,7 @@ import org.opendc.compute.service.scheduler.ComputeScheduler import org.opendc.compute.simulator.SimHost import org.opendc.compute.workload.topology.HostSpec import org.opendc.simulator.compute.SimBareMetalMachine -import org.opendc.simulator.compute.kernel.interference.VmInterferenceModel +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.compute.workload.SimTraceWorkload import org.opendc.simulator.flow.FlowEngine import java.time.Clock @@ -48,7 +48,6 @@ import kotlin.math.max * @param clock [Clock] instance tracking simulation time. * @param scheduler [ComputeScheduler] implementation to use for the service. * @param failureModel A failure model to use for injecting failures. - * @param interferenceModel The model to use for performance interference. * @param schedulingQuantum The scheduling quantum of the scheduler. */ public class ComputeServiceHelper( @@ -57,7 +56,6 @@ public class ComputeServiceHelper( scheduler: ComputeScheduler, seed: Long, private val failureModel: FailureModel? = null, - private val interferenceModel: VmInterferenceModel? = null, schedulingQuantum: Duration = Duration.ofMinutes(5) ) : AutoCloseable { /** @@ -91,11 +89,13 @@ public class ComputeServiceHelper( * @param trace The trace to simulate. * @param servers A list to which the created servers is added. * @param submitImmediately A flag to indicate that the servers are scheduled immediately (so not at their start time). + * @param interference A flag to indicate that VM interference needs to be enabled. */ public suspend fun run( trace: List<VirtualMachine>, servers: MutableList<Server>? = null, - submitImmediately: Boolean = false + submitImmediately: Boolean = false, + interference: Boolean = false, ) { val injector = failureModel?.createInjector(context, clock, service, Random(random.nextLong())) val client = service.newClient() @@ -125,10 +125,16 @@ public class ComputeServiceHelper( delay(max(0, (start - offset) - now)) } - launch { - val workloadOffset = -offset + 300001 - val workload = SimTraceWorkload(entry.trace, workloadOffset) + val workloadOffset = -offset + 300001 + val workload = SimTraceWorkload(entry.trace, workloadOffset) + val meta = mutableMapOf<String, Any>("workload" to workload) + + val interferenceProfile = entry.interferenceProfile + if (interference && interferenceProfile != null) { + meta["interference-profile"] = interferenceProfile + } + launch { val server = client.newServer( entry.name, image, @@ -138,7 +144,7 @@ public class ComputeServiceHelper( entry.memCapacity, meta = if (entry.cpuCapacity > 0.0) mapOf("cpu-capacity" to entry.cpuCapacity) else emptyMap() ), - meta = mapOf("workload" to workload) + meta = meta ) servers?.add(server) @@ -169,7 +175,7 @@ public class ComputeServiceHelper( */ public fun registerHost(spec: HostSpec, optimize: Boolean = false): SimHost { val machine = SimBareMetalMachine(engine, spec.model, spec.powerDriver) - val hypervisor = spec.hypervisor.create(engine, random) + val hypervisor = spec.hypervisor.create(engine, random, interferenceDomain = VmInterferenceDomain()) val host = SimHost( spec.uid, @@ -179,7 +185,6 @@ public class ComputeServiceHelper( clock, machine, hypervisor, - interferenceDomain = interferenceModel?.newDomain(), optimize = optimize ) diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt index aa0b5eaf..78002c2f 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkload.kt @@ -22,7 +22,6 @@ package org.opendc.compute.workload -import org.opendc.simulator.compute.kernel.interference.VmInterferenceModel import java.util.* /** @@ -32,10 +31,5 @@ public interface ComputeWorkload { /** * Resolve the workload into a list of [VirtualMachine]s to simulate. */ - public fun resolve(loader: ComputeWorkloadLoader, random: Random): Resolved - - /** - * A concrete instance of a workload. - */ - public data class Resolved(val vms: List<VirtualMachine>, val interferenceModel: VmInterferenceModel?) + public fun resolve(loader: ComputeWorkloadLoader, random: Random): List<VirtualMachine> } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt index 7ed04994..387a3ec2 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/ComputeWorkloadLoader.kt @@ -48,7 +48,7 @@ public class ComputeWorkloadLoader(private val baseDir: File) { /** * The cache of workloads. */ - private val cache = ConcurrentHashMap<String, SoftReference<ComputeWorkload.Resolved>>() + private val cache = ConcurrentHashMap<String, SoftReference<List<VirtualMachine>>>() /** * Read the fragments into memory. @@ -87,7 +87,7 @@ public class ComputeWorkloadLoader(private val baseDir: File) { /** * Read the metadata into a workload. */ - private fun parseMeta(trace: Trace, fragments: Map<String, Builder>): List<VirtualMachine> { + private fun parseMeta(trace: Trace, fragments: Map<String, Builder>, interferenceModel: VmInterferenceModel): List<VirtualMachine> { val reader = checkNotNull(trace.getTable(TABLE_RESOURCES)).newReader() val idCol = reader.resolve(RESOURCE_ID) @@ -128,7 +128,8 @@ public class ComputeWorkloadLoader(private val baseDir: File) { totalLoad, submissionTime, endTime, - builder.build() + builder.build(), + interferenceModel.getProfile(id) ) ) } @@ -159,7 +160,6 @@ public class ComputeWorkloadLoader(private val baseDir: File) { val modelBuilder = VmInterferenceModel.builder() while (reader.nextRow()) { - @Suppress("UNCHECKED_CAST") val members = reader.getSet(membersCol, String::class.java)!! val target = reader.getDouble(targetCol) val score = reader.getDouble(scoreCol) @@ -177,7 +177,7 @@ public class ComputeWorkloadLoader(private val baseDir: File) { /** * Load the trace with the specified [name] and [format]. */ - public fun get(name: String, format: String): ComputeWorkload.Resolved { + public fun get(name: String, format: String): List<VirtualMachine> { val ref = cache.compute(name) { key, oldVal -> val inst = oldVal?.get() if (inst == null) { @@ -188,11 +188,10 @@ public class ComputeWorkloadLoader(private val baseDir: File) { val trace = Trace.open(path, format) val fragments = parseFragments(trace) - val vms = parseMeta(trace, fragments) val interferenceModel = parseInterferenceModel(trace) - val instance = ComputeWorkload.Resolved(vms, interferenceModel) + val vms = parseMeta(trace, fragments, interferenceModel) - SoftReference(instance) + SoftReference(vms) } else { oldVal } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt index 88e80719..8560b537 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/VirtualMachine.kt @@ -22,6 +22,7 @@ package org.opendc.compute.workload +import org.opendc.simulator.compute.kernel.interference.VmInterferenceProfile import org.opendc.simulator.compute.workload.SimTrace import java.time.Instant import java.util.* @@ -37,6 +38,7 @@ import java.util.* * @param startTime The start time of the VM. * @param stopTime The stop time of the VM. * @param trace The trace that belong to this VM. + * @param interferenceProfile The interference profile of this virtual machine. */ public data class VirtualMachine( val uid: UUID, @@ -48,4 +50,5 @@ public data class VirtualMachine( val startTime: Instant, val stopTime: Instant, val trace: SimTrace, + val interferenceProfile: VmInterferenceProfile? ) diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt index 1959c48d..9b2bec55 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/CompositeComputeWorkload.kt @@ -37,17 +37,17 @@ internal class CompositeComputeWorkload(val sources: Map<ComputeWorkload, Double */ private val logger = KotlinLogging.logger {} - override fun resolve(loader: ComputeWorkloadLoader, random: Random): ComputeWorkload.Resolved { + override fun resolve(loader: ComputeWorkloadLoader, random: Random): List<VirtualMachine> { val traces = sources.map { (source, fraction) -> fraction to source.resolve(loader, random) } - val totalLoad = traces.sumOf { (_, w) -> w.vms.sumOf { it.totalLoad } } + val totalLoad = traces.sumOf { (_, vms) -> vms.sumOf { it.totalLoad } } val res = mutableListOf<VirtualMachine>() - for ((fraction, w) in traces) { + for ((fraction, vms) in traces) { var currentLoad = 0.0 - for (entry in w.vms) { + for (entry in vms) { val entryLoad = entry.totalLoad if ((currentLoad + entryLoad) / totalLoad > fraction) { break @@ -58,9 +58,9 @@ internal class CompositeComputeWorkload(val sources: Map<ComputeWorkload, Double } } - val vmCount = traces.sumOf { (_, w) -> w.vms.size } + val vmCount = traces.sumOf { (_, vms) -> vms.size } logger.info { "Sampled $vmCount VMs into subset of ${res.size} VMs" } - return ComputeWorkload.Resolved(res, null) + return res } } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt index 84a77f0f..52f4c672 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/HpcSampledComputeWorkload.kt @@ -45,8 +45,8 @@ internal class HpcSampledComputeWorkload(val source: ComputeWorkload, val fracti */ private val pattern = Regex("^(ComputeNode|cn).*") - override fun resolve(loader: ComputeWorkloadLoader, random: Random): ComputeWorkload.Resolved { - val (vms, interferenceModel) = source.resolve(loader, random) + override fun resolve(loader: ComputeWorkloadLoader, random: Random): List<VirtualMachine> { + val vms = source.resolve(loader, random) val (hpc, nonHpc) = vms.partition { entry -> val name = entry.name @@ -130,7 +130,7 @@ internal class HpcSampledComputeWorkload(val source: ComputeWorkload, val fracti logger.debug { "Total sampled load: ${hpcLoad + nonHpcLoad}" } logger.info { "Sampled ${vms.size} VMs (fraction $fraction) into subset of ${res.size} VMs" } - return ComputeWorkload.Resolved(res, interferenceModel) + return res } /** diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt index bc13560c..ef6de729 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/LoadSampledComputeWorkload.kt @@ -37,8 +37,8 @@ internal class LoadSampledComputeWorkload(val source: ComputeWorkload, val fract */ private val logger = KotlinLogging.logger {} - override fun resolve(loader: ComputeWorkloadLoader, random: Random): ComputeWorkload.Resolved { - val (vms, interferenceModel) = source.resolve(loader, random) + override fun resolve(loader: ComputeWorkloadLoader, random: Random): List<VirtualMachine> { + val vms = source.resolve(loader, random) val res = mutableListOf<VirtualMachine>() val totalLoad = vms.sumOf { it.totalLoad } @@ -56,6 +56,6 @@ internal class LoadSampledComputeWorkload(val source: ComputeWorkload, val fract logger.info { "Sampled ${vms.size} VMs (fraction $fraction) into subset of ${res.size} VMs" } - return ComputeWorkload.Resolved(res, interferenceModel) + return res } } diff --git a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt index dc9abaef..c20cb8f3 100644 --- a/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt +++ b/opendc-compute/opendc-compute-workload/src/main/kotlin/org/opendc/compute/workload/internal/TraceComputeWorkload.kt @@ -24,13 +24,14 @@ package org.opendc.compute.workload.internal import org.opendc.compute.workload.ComputeWorkload import org.opendc.compute.workload.ComputeWorkloadLoader +import org.opendc.compute.workload.VirtualMachine import java.util.* /** * A [ComputeWorkload] from a trace. */ internal class TraceComputeWorkload(val name: String, val format: String) : ComputeWorkload { - override fun resolve(loader: ComputeWorkloadLoader, random: Random): ComputeWorkload.Resolved { + override fun resolve(loader: ComputeWorkloadLoader, random: Random): List<VirtualMachine> { return loader.get(name, format) } } diff --git a/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt b/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt index 074ffc3e..c09ce96a 100644 --- a/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt +++ b/opendc-experiments/opendc-experiments-capelin/src/jmh/kotlin/org/opendc/experiments/capelin/CapelinBenchmarks.kt @@ -55,7 +55,7 @@ class CapelinBenchmarks { fun setUp() { val loader = ComputeWorkloadLoader(File("src/test/resources/trace")) val source = trace("bitbrains-small") - vms = source.resolve(loader, Random(1L)).vms + vms = trace("bitbrains-small").resolve(loader, Random(1L)) topology = checkNotNull(object {}.javaClass.getResourceAsStream("/topology.txt")).use { clusterTopology(it) } } @@ -69,12 +69,12 @@ class CapelinBenchmarks { coroutineContext, clock, computeScheduler, - seed = 0L, + seed = 0L ) try { runner.apply(topology, isOptimized) - runner.run(vms) + runner.run(vms, interference = true) } finally { runner.close() } diff --git a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt index 2f417172..7be09ff5 100644 --- a/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt +++ b/opendc-experiments/opendc-experiments-capelin/src/main/kotlin/org/opendc/experiments/capelin/CapelinRunner.kt @@ -68,14 +68,13 @@ public class CapelinRunner( grid5000(Duration.ofSeconds((operationalPhenomena.failureFrequency * 60).roundToLong())) else null - val (vms, interferenceModel) = scenario.workload.source.resolve(workloadLoader, seeder) + val vms = scenario.workload.source.resolve(workloadLoader, seeder) val runner = ComputeServiceHelper( coroutineContext, clock, computeScheduler, seed, - failureModel, - interferenceModel?.takeIf { operationalPhenomena.hasInterference } + failureModel ) val topology = clusterTopology(File(envPath, "${scenario.topology.name}.txt")) @@ -105,7 +104,7 @@ public class CapelinRunner( runner.apply(topology, optimize = true) // Run the workload trace - runner.run(vms, servers) + runner.run(vms, servers, interference = operationalPhenomena.hasInterference) // Stop the metric collection exporter?.close() diff --git a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt index ff9faef7..af846dd6 100644 --- a/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt +++ b/opendc-experiments/opendc-experiments-capelin/src/test/kotlin/org/opendc/experiments/capelin/CapelinIntegrationTest.kt @@ -82,7 +82,7 @@ class CapelinIntegrationTest { @Test fun testLarge() = runBlockingSimulation { val seed = 0L - val (workload, _) = createTestWorkload(1.0, seed) + val workload = createTestWorkload(1.0, seed) val runner = ComputeServiceHelper( coroutineContext, clock, @@ -132,7 +132,7 @@ class CapelinIntegrationTest { @Test fun testSmall() = runBlockingSimulation { val seed = 1L - val (workload, _) = createTestWorkload(0.25, seed) + val workload = createTestWorkload(0.25, seed) val runner = ComputeServiceHelper( coroutineContext, clock, @@ -177,14 +177,13 @@ class CapelinIntegrationTest { @Test fun testInterference() = runBlockingSimulation { val seed = 0L - val (workload, interferenceModel) = createTestWorkload(1.0, seed) + val workload = createTestWorkload(1.0, seed) val simulator = ComputeServiceHelper( coroutineContext, clock, computeScheduler, - seed, - interferenceModel = interferenceModel + seed ) val topology = createTopology("single") val servers = mutableListOf<Server>() @@ -192,7 +191,7 @@ class CapelinIntegrationTest { try { simulator.apply(topology) - simulator.run(workload, servers) + simulator.run(workload, servers, interference = true) val serviceMetrics = simulator.service.getSchedulerStats() println( @@ -213,7 +212,7 @@ class CapelinIntegrationTest { { assertEquals(6028050, this@CapelinIntegrationTest.monitor.idleTime) { "Idle time incorrect" } }, { assertEquals(14712749, this@CapelinIntegrationTest.monitor.activeTime) { "Active time incorrect" } }, { assertEquals(12532907, this@CapelinIntegrationTest.monitor.stealTime) { "Steal time incorrect" } }, - { assertEquals(477068, this@CapelinIntegrationTest.monitor.lostTime) { "Lost time incorrect" } } + { assertEquals(485510, this@CapelinIntegrationTest.monitor.lostTime) { "Lost time incorrect" } } ) } @@ -231,7 +230,7 @@ class CapelinIntegrationTest { grid5000(Duration.ofDays(7)) ) val topology = createTopology("single") - val (workload, _) = createTestWorkload(0.25, seed) + val workload = createTestWorkload(0.25, seed) val servers = mutableListOf<Server>() val reader = ComputeMetricReader(this, clock, simulator.service, servers, monitor) @@ -266,7 +265,7 @@ class CapelinIntegrationTest { /** * Obtain the trace reader for the test. */ - private fun createTestWorkload(fraction: Double, seed: Long): ComputeWorkload.Resolved { + private fun createTestWorkload(fraction: Double, seed: Long): List<VirtualMachine> { val source = trace("bitbrains-small").sampleByLoad(fraction) return source.resolve(workloadLoader, Random(seed)) } diff --git a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt index c3332d66..e862e4d1 100644 --- a/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt +++ b/opendc-simulator/opendc-simulator-compute/src/jmh/kotlin/org/opendc/simulator/compute/SimMachineBenchmarks.kt @@ -87,7 +87,7 @@ class SimMachineBenchmarks { engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(engine, null, random) + val hypervisor = SimSpaceSharedHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } @@ -110,7 +110,7 @@ class SimMachineBenchmarks { engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimFairShareHypervisor(engine, null, random) + val hypervisor = SimFairShareHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } @@ -133,7 +133,7 @@ class SimMachineBenchmarks { engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimFairShareHypervisor(engine, null, random) + val hypervisor = SimFairShareHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt index 2cabeece..77088b74 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimAbstractHypervisor.kt @@ -25,7 +25,9 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.* import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor import org.opendc.simulator.compute.kernel.cpufreq.ScalingPolicy +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.compute.kernel.interference.VmInterferenceMember +import org.opendc.simulator.compute.kernel.interference.VmInterferenceProfile import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.compute.model.ProcessingUnit import org.opendc.simulator.compute.workload.SimWorkload @@ -38,13 +40,15 @@ import kotlin.math.roundToLong * Abstract implementation of the [SimHypervisor] interface. * * @param engine The [FlowEngine] to drive the simulation. - * @param scalingGovernor The scaling governor to use for scaling the CPU frequency of the underlying hardware. * @param random A randomness generator for the interference calculations. + * @param scalingGovernor The scaling governor to use for scaling the CPU frequency of the underlying hardware. + * @param interferenceDomain The interference domain to which the hypervisor belongs. */ public abstract class SimAbstractHypervisor( protected val engine: FlowEngine, + private val random: SplittableRandom, private val scalingGovernor: ScalingGovernor?, - private val random: SplittableRandom + private val interferenceDomain: VmInterferenceDomain ) : SimHypervisor, FlowConvergenceListener { /** * The machine on which the hypervisor runs. @@ -94,9 +98,9 @@ public abstract class SimAbstractHypervisor( private val governors = mutableListOf<ScalingGovernor.Logic>() /* SimHypervisor */ - override fun newMachine(model: MachineModel, interferenceKey: VmInterferenceMember?): SimVirtualMachine { + override fun newMachine(model: MachineModel): SimVirtualMachine { require(canFit(model)) { "Machine does not fit" } - val vm = VirtualMachine(model, interferenceKey) + val vm = VirtualMachine(model) _vms.add(vm) return vm } @@ -164,12 +168,8 @@ public abstract class SimAbstractHypervisor( * A virtual machine running on the hypervisor. * * @param model The machine model of the virtual machine. - * @param interferenceKey The interference key of this virtual machine. */ - private inner class VirtualMachine( - model: MachineModel, - private val interferenceKey: VmInterferenceMember? = null - ) : SimAbstractMachine(engine, model), SimVirtualMachine, AutoCloseable { + private inner class VirtualMachine(model: MachineModel) : SimAbstractMachine(engine, model), SimVirtualMachine, AutoCloseable { /** * A flag to indicate that the machine is closed. */ @@ -185,7 +185,7 @@ public abstract class SimAbstractHypervisor( */ override val counters: SimHypervisorCounters get() = _counters - @JvmField val _counters = VmCountersImpl(cpus, interferenceKey) + @JvmField val _counters = VmCountersImpl(cpus, null) /** * The CPU capacity of the hypervisor in MHz. @@ -208,21 +208,27 @@ public abstract class SimAbstractHypervisor( override fun startWorkload(workload: SimWorkload, meta: Map<String, Any>): SimMachineContext { check(!isClosed) { "Machine is closed" } + val profile = meta["interference-profile"] as? VmInterferenceProfile + val interferenceMember = if (profile != null) interferenceDomain.join(profile) else null + + val counters = _counters + counters.member = interferenceMember + return super.startWorkload( object : SimWorkload { override fun onStart(ctx: SimMachineContext) { - val interferenceKey = interferenceKey try { - interferenceKey?.activate() + interferenceMember?.activate() workload.onStart(ctx) } catch (cause: Throwable) { - interferenceKey?.deactivate() + interferenceMember?.deactivate() throw cause } } override fun onStop(ctx: SimMachineContext) { - interferenceKey?.deactivate() + interferenceMember?.deactivate() + counters.member = null workload.onStop(ctx) } }, @@ -339,7 +345,7 @@ public abstract class SimAbstractHypervisor( */ private inner class VmCountersImpl( private val cpus: List<VCpu>, - private val key: VmInterferenceMember? + @JvmField var member: VmInterferenceMember? ) : SimHypervisorCounters { private val d = cpus.size / cpus.sumOf { it.model.frequency } * 1000 @@ -388,9 +394,9 @@ public abstract class SimAbstractHypervisor( cpuTime[2] += ((demandDelta - actualDelta) * d).roundToLong() // Compute the performance penalty due to flow interference - val key = key - if (key != null) { - val penalty = 1 - key.apply(random, load) + val member = member + if (member != null) { + val penalty = 1 - member.apply(random, load) val interference = (actualDelta * d * penalty).roundToLong() if (interference > 0) { diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt index 4435a422..fbee46da 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisor.kt @@ -24,6 +24,7 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.SimMachine import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.compute.workload.SimWorkload import org.opendc.simulator.flow.FlowEngine @@ -35,15 +36,17 @@ import java.util.SplittableRandom * A [SimHypervisor] that distributes the computing requirements of multiple [SimWorkload]s on a single [SimMachine] * concurrently using weighted fair sharing. * - * @param engine The [FlowEngine] to manage the machine's resources. - * @param scalingGovernor The CPU frequency scaling governor to use for the hypervisor. + * @param engine The [FlowEngine] to drive the simulation. * @param random A randomness generator for the interference calculations. + * @param scalingGovernor The scaling governor to use for scaling the CPU frequency of the underlying hardware. + * @param interferenceDomain The interference domain to which the hypervisor belongs. */ public class SimFairShareHypervisor( engine: FlowEngine, + random: SplittableRandom, scalingGovernor: ScalingGovernor?, - random: SplittableRandom -) : SimAbstractHypervisor(engine, scalingGovernor, random) { + interferenceDomain: VmInterferenceDomain = VmInterferenceDomain() +) : SimAbstractHypervisor(engine, random, scalingGovernor, interferenceDomain) { /** * The multiplexer that distributes the computing capacity. */ diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt index c7008652..81dfc43d 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorProvider.kt @@ -23,6 +23,7 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.flow.FlowEngine import java.util.* @@ -36,5 +37,6 @@ public class SimFairShareHypervisorProvider : SimHypervisorProvider { engine: FlowEngine, random: SplittableRandom, scalingGovernor: ScalingGovernor?, - ): SimHypervisor = SimFairShareHypervisor(engine, scalingGovernor, random) + interferenceDomain: VmInterferenceDomain, + ): SimHypervisor = SimFairShareHypervisor(engine, random, scalingGovernor, interferenceDomain) } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt index f53d0c5d..d8e4e7cd 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisor.kt @@ -23,7 +23,6 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.SimMachine -import org.opendc.simulator.compute.kernel.interference.VmInterferenceMember import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.compute.workload.SimWorkload @@ -66,9 +65,8 @@ public interface SimHypervisor : SimWorkload { * Create a [SimMachine] instance on which users may run a [SimWorkload]. * * @param model The machine to create. - * @param interferenceKey The key of the machine in the interference model. */ - public fun newMachine(model: MachineModel, interferenceKey: VmInterferenceMember? = null): SimVirtualMachine + public fun newMachine(model: MachineModel): SimVirtualMachine /** * Remove the specified [machine] from the hypervisor. diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt index 020a2a60..2c86854e 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimHypervisorProvider.kt @@ -23,6 +23,7 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.flow.FlowEngine import java.util.SplittableRandom @@ -41,5 +42,10 @@ public interface SimHypervisorProvider { /** * Create a new [SimHypervisor] instance. */ - public fun create(engine: FlowEngine, random: SplittableRandom, scalingGovernor: ScalingGovernor? = null): SimHypervisor + public fun create( + engine: FlowEngine, + random: SplittableRandom, + scalingGovernor: ScalingGovernor? = null, + interferenceDomain: VmInterferenceDomain + ): SimHypervisor } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt index 51bf4ce5..c32dd027 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisor.kt @@ -23,6 +23,7 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.compute.model.MachineModel import org.opendc.simulator.flow.FlowEngine import org.opendc.simulator.flow.mux.FlowMultiplexer @@ -32,15 +33,17 @@ import java.util.SplittableRandom /** * A [SimHypervisor] that allocates its sub-resources exclusively for the virtual machine that it hosts. * - * @param engine The [FlowEngine] to manage the machine's resources. - * @param scalingGovernor The CPU frequency scaling governor to use for the hypervisor. + * @param engine The [FlowEngine] to drive the simulation. * @param random A randomness generator for the interference calculations. + * @param scalingGovernor The scaling governor to use for scaling the CPU frequency of the underlying hardware. + * @param interferenceDomain The interference domain to which the hypervisor belongs. */ public class SimSpaceSharedHypervisor( engine: FlowEngine, + random: SplittableRandom, scalingGovernor: ScalingGovernor?, - random: SplittableRandom -) : SimAbstractHypervisor(engine, scalingGovernor, random) { + interferenceDomain: VmInterferenceDomain = VmInterferenceDomain() +) : SimAbstractHypervisor(engine, random, scalingGovernor, interferenceDomain) { override val mux: FlowMultiplexer = ForwardingFlowMultiplexer(engine, this) override fun canFit(model: MachineModel): Boolean { diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt index 05c54528..cc303bbd 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorProvider.kt @@ -23,6 +23,7 @@ package org.opendc.simulator.compute.kernel import org.opendc.simulator.compute.kernel.cpufreq.ScalingGovernor +import org.opendc.simulator.compute.kernel.interference.VmInterferenceDomain import org.opendc.simulator.flow.FlowEngine import java.util.* @@ -36,5 +37,6 @@ public class SimSpaceSharedHypervisorProvider : SimHypervisorProvider { engine: FlowEngine, random: SplittableRandom, scalingGovernor: ScalingGovernor?, - ): SimHypervisor = SimSpaceSharedHypervisor(engine, scalingGovernor, random) + interferenceDomain: VmInterferenceDomain, + ): SimHypervisor = SimSpaceSharedHypervisor(engine, random, scalingGovernor, interferenceDomain) } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt index 3b355f1e..6861823b 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceDomain.kt @@ -1,5 +1,5 @@ /* - * Copyright (c) 2021 AtLarge Research + * Copyright (c) 2022 AtLarge Research * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to deal @@ -22,16 +22,110 @@ package org.opendc.simulator.compute.kernel.interference +import java.util.ArrayDeque +import java.util.ArrayList +import java.util.WeakHashMap + /** - * The interference domain of a hypervisor. + * A domain where virtual machines may incur performance variability due to operating on the same resource and + * therefore causing interference. */ -public interface VmInterferenceDomain { +public class VmInterferenceDomain { + /** + * A cache to maintain a mapping between the active profiles in this domain. + */ + private val cache = WeakHashMap<VmInterferenceProfile, VmInterferenceMember>() + + /** + * The set of members active in this domain. + */ + private val activeKeys = ArrayList<VmInterferenceMember>() + + /** + * Queue of participants that will be removed or added to the active groups. + */ + private val participants = ArrayDeque<VmInterferenceMember>() + + /** + * Join this interference domain with the specified [profile] and return the [VmInterferenceMember] associated with + * the profile. If the member does not exist, it will be created. + */ + public fun join(profile: VmInterferenceProfile): VmInterferenceMember { + return cache.computeIfAbsent(profile) { key -> key.newMember(this) } + } + + /** + * Mark the specified [member] as active in this interference domain. + */ + internal fun activate(member: VmInterferenceMember) { + val activeKeys = activeKeys + val pos = activeKeys.binarySearch(member) + if (pos < 0) { + activeKeys.add(-pos - 1, member) + } + + computeActiveGroups(activeKeys, member) + } + /** - * Return the [VmInterferenceMember] associated with the specified [id]. - * - * @param id The identifier of the virtual machine. - * @return A [VmInterferenceMember] representing the virtual machine as part of the interference domain. `null` if - * the virtual machine does not participate in the domain. + * Mark the specified [member] as inactive in this interference domain. */ - public fun getMember(id: String): VmInterferenceMember? + internal fun deactivate(member: VmInterferenceMember) { + val activeKeys = activeKeys + activeKeys.remove(member) + computeActiveGroups(activeKeys, member) + } + + /** + * (Re-)compute the active groups. + */ + private fun computeActiveGroups(activeKeys: ArrayList<VmInterferenceMember>, member: VmInterferenceMember) { + if (activeKeys.isEmpty()) { + return + } + + val groups = member.membership + val members = member.members + val participants = participants + + for (group in groups) { + val groupMembers = members[group] + + var i = 0 + var j = 0 + var intersection = 0 + + // Compute the intersection of the group members and the current active members + while (i < groupMembers.size && j < activeKeys.size) { + val l = groupMembers[i] + val rightEntry = activeKeys[j] + val r = rightEntry.id + + if (l < r) { + i++ + } else if (l > r) { + j++ + } else { + if (++intersection > 1) { + rightEntry.addGroup(group) + } else { + participants.add(rightEntry) + } + + i++ + j++ + } + } + + while (true) { + val participant = participants.poll() ?: break + + if (intersection <= 1) { + participant.removeGroup(group) + } else { + participant.addGroup(group) + } + } + } + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceMember.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceMember.kt index 04203c63..762bb568 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceMember.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceMember.kt @@ -1,5 +1,5 @@ /* - * Copyright (c) 2021 AtLarge Research + * Copyright (c) 2022 AtLarge Research * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to deal @@ -27,16 +27,43 @@ import java.util.* /** * A participant of an interference domain. */ -public interface VmInterferenceMember { +public class VmInterferenceMember( + private val domain: VmInterferenceDomain, + private val model: VmInterferenceModel, + @JvmField internal val id: Int, + @JvmField internal val membership: IntArray, + @JvmField internal val members: Array<IntArray>, + private val targets: DoubleArray, + private val scores: DoubleArray +) : Comparable<VmInterferenceMember> { + /** + * The active groups to which the key belongs. + */ + private var groups: IntArray = IntArray(2) + private var groupsSize: Int = 0 + + /** + * The number of users of the interference key. + */ + private var refCount: Int = 0 + /** * Mark this member as active in this interference domain. */ - public fun activate() + public fun activate() { + if (refCount++ <= 0) { + domain.activate(this) + } + } /** * Mark this member as inactive in this interference domain. */ - public fun deactivate() + public fun deactivate() { + if (--refCount <= 0) { + domain.deactivate(this) + } + } /** * Compute the performance score of the member in this interference domain. @@ -46,5 +73,91 @@ public interface VmInterferenceMember { * @return A score representing the performance score to be applied to the member, with 1 * meaning no influence, <1 means that performance degrades, and >1 means that performance improves. */ - public fun apply(random: SplittableRandom, load: Double): Double + public fun apply(random: SplittableRandom, load: Double): Double { + val groupsSize = groupsSize + + if (groupsSize == 0) { + return 1.0 + } + + val groups = groups + val targets = targets + + var low = 0 + var high = groupsSize - 1 + var group = -1 + + // Perform binary search over the groups based on target load + while (low <= high) { + val mid = low + high ushr 1 + val midGroup = groups[mid] + val target = targets[midGroup] + + if (target < load) { + low = mid + 1 + group = midGroup + } else if (target > load) { + high = mid - 1 + } else { + group = midGroup + break + } + } + + return if (group >= 0 && random.nextInt(members[group].size) == 0) { + scores[group] + } else { + 1.0 + } + } + + /** + * Add an active group to this member. + */ + internal fun addGroup(group: Int) { + var groups = groups + val groupsSize = groupsSize + val pos = groups.binarySearch(group, toIndex = groupsSize) + + if (pos >= 0) { + return + } + + val idx = -pos - 1 + + if (groups.size == groupsSize) { + val newSize = groupsSize + (groupsSize shr 1) + groups = groups.copyOf(newSize) + this.groups = groups + } + + groups.copyInto(groups, idx + 1, idx, groupsSize) + groups[idx] = group + this.groupsSize += 1 + } + + /** + * Remove an active group from this member. + */ + internal fun removeGroup(group: Int) { + val groups = groups + val groupsSize = groupsSize + val pos = groups.binarySearch(group, toIndex = groupsSize) + + if (pos < 0) { + return + } + + groups.copyInto(groups, pos, pos + 1, groupsSize) + this.groupsSize -= 1 + } + + override fun compareTo(other: VmInterferenceMember): Int { + val cmp = model.hashCode().compareTo(other.model.hashCode()) + if (cmp != 0) { + return cmp + } + + return id.compareTo(other.id) + } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt index 3ea869d4..018c6e3d 100644 --- a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceModel.kt @@ -27,7 +27,7 @@ import java.util.* /** * An interference model that models the resource interference between virtual machines on a host. * - * @param targets The target load of each group. + * @param members The target load of each group. * @param scores The performance score of each group. * @param members The members belonging to each group. * @param membership The identifier of each key. @@ -42,9 +42,16 @@ public class VmInterferenceModel private constructor( private val size: Int ) { /** - * Construct a new [VmInterferenceDomain]. + * Return the [VmInterferenceProfile] associated with the specified [id]. + * + * @param id The identifier of the virtual machine. + * @return A [VmInterferenceProfile] representing the virtual machine as part of interference model or `null` if + * there is no profile for the virtual machine. */ - public fun newDomain(): VmInterferenceDomain = InterferenceDomainImpl(idMapping, members, membership, targets, scores) + public fun getProfile(id: String): VmInterferenceProfile? { + val intId = idMapping[id] ?: return null + return VmInterferenceProfile(this, intId, membership[intId], members, targets, scores) + } public companion object { /** @@ -112,8 +119,8 @@ public class VmInterferenceModel private constructor( val scores = _scores val members = _members - val indices = Array(size) { it } - indices.sortWith( + val indices = IntArray(size) { it } + indices.sortedWith( Comparator { l, r -> var cmp = targets[l].compareTo(targets[r]) // Order by target load if (cmp != 0) { @@ -179,224 +186,4 @@ public class VmInterferenceModel private constructor( const val INITIAL_CAPACITY = 256 } } - - /** - * Internal implementation of [VmInterferenceDomain]. - */ - private class InterferenceDomainImpl( - private val idMapping: Map<String, Int>, - private val members: Array<IntArray>, - private val membership: Array<IntArray>, - private val targets: DoubleArray, - private val scores: DoubleArray, - ) : VmInterferenceDomain { - /** - * Keys registered with this domain. - */ - private val keys = HashMap<Int, InterferenceMemberImpl>() - - /** - * The set of keys active in this domain. - */ - private val activeKeys = ArrayList<InterferenceMemberImpl>() - - /** - * Queue of participants that will be removed or added to the active groups. - */ - private val participants = ArrayDeque<InterferenceMemberImpl>() - - override fun getMember(id: String): VmInterferenceMember? { - val intId = idMapping[id] ?: return null - return keys.computeIfAbsent(intId) { InterferenceMemberImpl(it, this, membership[it], members, targets, scores) } - } - - override fun toString(): String = "VmInterferenceDomain" - - fun join(key: InterferenceMemberImpl) { - val activeKeys = activeKeys - val pos = activeKeys.binarySearch(key) - if (pos < 0) { - activeKeys.add(-pos - 1, key) - } - - computeActiveGroups(activeKeys, key) - } - - fun leave(key: InterferenceMemberImpl) { - val activeKeys = activeKeys - activeKeys.remove(key) - computeActiveGroups(activeKeys, key) - } - - /** - * (Re-)compute the active groups. - */ - private fun computeActiveGroups(activeKeys: ArrayList<InterferenceMemberImpl>, key: InterferenceMemberImpl) { - if (activeKeys.isEmpty()) { - return - } - - val groups = key.membership - val members = members - val participants = participants - - for (group in groups) { - val groupMembers = members[group] - - var i = 0 - var j = 0 - var intersection = 0 - - // Compute the intersection of the group members and the current active members - while (i < groupMembers.size && j < activeKeys.size) { - val l = groupMembers[i] - val rightEntry = activeKeys[j] - val r = rightEntry.id - - if (l < r) { - i++ - } else if (l > r) { - j++ - } else { - if (++intersection > 1) { - rightEntry.addGroup(group) - } else { - participants.add(rightEntry) - } - - i++ - j++ - } - } - - while (true) { - val participant = participants.poll() ?: break - - if (intersection <= 1) { - participant.removeGroup(group) - } else { - participant.addGroup(group) - } - } - } - } - } - - /** - * An interference key. - * - * @param id The identifier of the member. - */ - private class InterferenceMemberImpl( - @JvmField val id: Int, - private val domain: InterferenceDomainImpl, - @JvmField val membership: IntArray, - private val members: Array<IntArray>, - private val targets: DoubleArray, - private val scores: DoubleArray - ) : VmInterferenceMember, Comparable<InterferenceMemberImpl> { - /** - * The active groups to which the key belongs. - */ - private var groups: IntArray = IntArray(2) - private var groupsSize: Int = 0 - - /** - * The number of users of the interference key. - */ - private var refCount: Int = 0 - - override fun activate() { - if (refCount++ <= 0) { - domain.join(this) - } - } - - override fun deactivate() { - if (--refCount <= 0) { - domain.leave(this) - } - } - - override fun apply(random: SplittableRandom, load: Double): Double { - val groupsSize = groupsSize - - if (groupsSize == 0) { - return 1.0 - } - - val groups = groups - val targets = targets - - var low = 0 - var high = groupsSize - 1 - var group = -1 - - // Perform binary search over the groups based on target load - while (low <= high) { - val mid = low + high ushr 1 - val midGroup = groups[mid] - val target = targets[midGroup] - - if (target < load) { - low = mid + 1 - group = midGroup - } else if (target > load) { - high = mid - 1 - } else { - group = midGroup - break - } - } - - return if (group >= 0 && random.nextInt(members[group].size) == 0) { - scores[group] - } else { - 1.0 - } - } - - /** - * Add an active group to this member. - */ - fun addGroup(group: Int) { - var groups = groups - val groupsSize = groupsSize - val pos = groups.binarySearch(group, toIndex = groupsSize) - - if (pos >= 0) { - return - } - - val idx = -pos - 1 - - if (groups.size == groupsSize) { - val newSize = groupsSize + (groupsSize shr 1) - groups = groups.copyOf(newSize) - this.groups = groups - } - - groups.copyInto(groups, idx + 1, idx, groupsSize) - groups[idx] = group - this.groupsSize += 1 - } - - /** - * Remove an active group from this member. - */ - fun removeGroup(group: Int) { - val groups = groups - val groupsSize = groupsSize - val pos = groups.binarySearch(group, toIndex = groupsSize) - - if (pos < 0) { - return - } - - groups.copyInto(groups, pos, pos + 1, groupsSize) - this.groupsSize -= 1 - } - - override fun compareTo(other: InterferenceMemberImpl): Int = id.compareTo(other.id) - } } diff --git a/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceProfile.kt b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceProfile.kt new file mode 100644 index 00000000..004dbd07 --- /dev/null +++ b/opendc-simulator/opendc-simulator-compute/src/main/kotlin/org/opendc/simulator/compute/kernel/interference/VmInterferenceProfile.kt @@ -0,0 +1,51 @@ +/* + * Copyright (c) 2022 AtLarge Research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package org.opendc.simulator.compute.kernel.interference + +/** + * A profile of a particular virtual machine describing its interference pattern with other virtual machines. + * + * @param model The model to which this profile belongs. + * @property id The identifier of the profile inside the model. + * @property membership The membership of the profile in the groups. + * @param members The members in the model. + * @param targets The targets in the model. + * @param scores The scores in the model. + */ +public class VmInterferenceProfile internal constructor( + private val model: VmInterferenceModel, + private val id: Int, + private val membership: IntArray, + private val members: Array<IntArray>, + private val targets: DoubleArray, + private val scores: DoubleArray +) { + /** + * Create a new [VmInterferenceMember] based on this profile for the specified [domain]. + */ + internal fun newMember(domain: VmInterferenceDomain): VmInterferenceMember { + return VmInterferenceMember(domain, model, id, membership, members, targets, scores) + } + + override fun toString(): String = "VmInterferenceProfile[id=$id]" +} diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt index 23d832e8..d401f8b5 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimFairShareHypervisorTest.kt @@ -79,7 +79,7 @@ internal class SimFairShareHypervisorTest { val platform = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0))) val random = SplittableRandom(1) - val hypervisor = SimFairShareHypervisor(platform, PerformanceScalingGovernor(), random) + val hypervisor = SimFairShareHypervisor(platform, random, PerformanceScalingGovernor()) launch { machine.runWorkload(hypervisor) @@ -131,7 +131,7 @@ internal class SimFairShareHypervisorTest { platform, model, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimFairShareHypervisor(platform, null, random) + val hypervisor = SimFairShareHypervisor(platform, random, null) launch { machine.runWorkload(hypervisor) @@ -171,7 +171,7 @@ internal class SimFairShareHypervisorTest { val platform = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(platform, model, SimplePowerDriver(ConstantPowerModel(0.0))) val random = SplittableRandom(1) - val hypervisor = SimFairShareHypervisor(platform, null, random) + val hypervisor = SimFairShareHypervisor(platform, random, null) assertDoesNotThrow { launch { @@ -195,14 +195,13 @@ internal class SimFairShareHypervisorTest { .addGroup(targetLoad = 0.0, score = 0.6, members = setOf("a", "c")) .addGroup(targetLoad = 0.1, score = 0.8, members = setOf("a", "n")) .build() - val interferenceDomain = interferenceModel.newDomain() - val platform = FlowEngine(coroutineContext, clock) + val engine = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine( - platform, model, SimplePowerDriver(ConstantPowerModel(0.0)) + engine, model, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimFairShareHypervisor(platform, null, random) + val hypervisor = SimFairShareHypervisor(engine, random, null) val duration = 5 * 60L val workloadA = @@ -230,12 +229,12 @@ internal class SimFairShareHypervisorTest { coroutineScope { launch { - val vm = hypervisor.newMachine(model, interferenceDomain.getMember("a")) - vm.runWorkload(workloadA) + val vm = hypervisor.newMachine(model) + vm.runWorkload(workloadA, meta = mapOf("interference-model" to interferenceModel.getProfile("a")!!)) hypervisor.removeMachine(vm) } - val vm = hypervisor.newMachine(model, interferenceDomain.getMember("b")) - vm.runWorkload(workloadB) + val vm = hypervisor.newMachine(model) + vm.runWorkload(workloadB, meta = mapOf("interference-model" to interferenceModel.getProfile("b")!!)) hypervisor.removeMachine(vm) } diff --git a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt index 9471f548..9b31acf4 100644 --- a/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt +++ b/opendc-simulator/opendc-simulator-compute/src/test/kotlin/org/opendc/simulator/compute/kernel/SimSpaceSharedHypervisorTest.kt @@ -77,7 +77,7 @@ internal class SimSpaceSharedHypervisorTest { val engine = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(engine, null, random) + val hypervisor = SimSpaceSharedHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } val vm = hypervisor.newMachine(machineModel) @@ -100,7 +100,7 @@ internal class SimSpaceSharedHypervisorTest { val engine = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(engine, null, random) + val hypervisor = SimSpaceSharedHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } yield() @@ -125,7 +125,7 @@ internal class SimSpaceSharedHypervisorTest { engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(engine, null, random) + val hypervisor = SimSpaceSharedHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } yield() @@ -147,7 +147,7 @@ internal class SimSpaceSharedHypervisorTest { engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(engine, null, random) + val hypervisor = SimSpaceSharedHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } yield() @@ -175,7 +175,7 @@ internal class SimSpaceSharedHypervisorTest { val engine = FlowEngine(coroutineContext, clock) val machine = SimBareMetalMachine(engine, machineModel, SimplePowerDriver(ConstantPowerModel(0.0))) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(engine, null, random) + val hypervisor = SimSpaceSharedHypervisor(engine, random, null) launch { machine.runWorkload(hypervisor) } yield() @@ -200,7 +200,7 @@ internal class SimSpaceSharedHypervisorTest { interpreter, machineModel, SimplePowerDriver(ConstantPowerModel(0.0)) ) val random = SplittableRandom(1) - val hypervisor = SimSpaceSharedHypervisor(interpreter, null, random) + val hypervisor = SimSpaceSharedHypervisor(interpreter, random, null) launch { machine.runWorkload(hypervisor) } yield() diff --git a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt index d5dbed1c..b7e550ef 100644 --- a/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt +++ b/opendc-web/opendc-web-runner/src/main/kotlin/org/opendc/web/runner/OpenDCRunner.kt @@ -208,7 +208,7 @@ public class OpenDCRunner( val phenomena = scenario.phenomena val computeScheduler = createComputeScheduler(scenario.schedulerName, seeder) val workload = trace(workloadName).sampleByLoad(workloadFraction) - val (vms, interferenceModel) = workload.resolve(workloadLoader, seeder) + val vms = workload.resolve(workloadLoader, seeder) val failureModel = if (phenomena.failures) @@ -221,8 +221,7 @@ public class OpenDCRunner( clock, computeScheduler, seed = 0L, - failureModel, - interferenceModel.takeIf { phenomena.interference } + failureModel ) val servers = mutableListOf<Server>() val reader = ComputeMetricReader(this, clock, simulator.service, servers, monitor) @@ -231,7 +230,7 @@ public class OpenDCRunner( // Instantiate the topology onto the simulator simulator.apply(topology) // Run workload trace - simulator.run(vms, servers) + simulator.run(vms, servers, interference = phenomena.interference) val serviceMetrics = simulator.service.getSchedulerStats() logger.debug { |