refactor(sim/compute): Make interference domain independent of profile

This change updates the virtual machine performance interference model
so that the interference domain can be constructed independently of the
interference profile. As a consequence, the construction of the topology
now does not depend anymore on the interference profile.

8 files changed, 40 insertions, 38 deletions

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)
     }
 }