Merge branch '2.x-perf-interf-optimization' into '2.x'

Optimize the performance interference model

See merge request opendc/opendc-simulator!50

2 files changed, 30 insertions, 5 deletions

diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt
index 3086f4e6..b4626def 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/driver/SimpleVirtDriver.kt
@@ -50,6 +50,9 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import java.util.UUID
 import kotlin.math.ceil
@@ -87,6 +90,17 @@ class SimpleVirtDriver(
 
     override val events: Flow<HypervisorEvent> = eventFlow
 
+    init {
+        events.filter { it is HypervisorEvent.VmsUpdated }.onEach {
+            val imagesRunning = vms.map { it.server.image }.toSet()
+            vms.forEach {
+                val performanceModel =
+                    it.server.image.tags[IMAGE_PERF_INTERFERENCE_MODEL] as? PerformanceInterferenceModel?
+                performanceModel?.computeIntersectingItems(imagesRunning)
+            }
+        }.launchIn(coroutineScope)
+    }
+
     override suspend fun spawn(
         name: String,
         image: Image,
@@ -193,13 +207,12 @@ class SimpleVirtDriver(
 
             val totalRemainder = remainder.sum()
             val totalBurst = burst.sum()
-            val imagesRunning = vms.map { it.server.image }.toSet()
 
             for (vm in vms) {
                 // Apply performance interference model
                 val performanceModel =
                     vm.server.image.tags[IMAGE_PERF_INTERFERENCE_MODEL] as? PerformanceInterferenceModel?
-                val performanceScore = performanceModel?.apply(imagesRunning, serverLoad) ?: 1.0
+                val performanceScore = performanceModel?.apply(serverLoad) ?: 1.0
 
                 for ((i, req) in vm.requests.withIndex()) {
                     // Compute the fraction of compute time allocated to the VM
@@ -222,7 +235,15 @@ class SimpleVirtDriver(
                 }
             }
 
-            eventFlow.emit(HypervisorEvent.SliceFinished(this@SimpleVirtDriver, totalBurst, totalBurst - totalRemainder, vms.size, server))
+            eventFlow.emit(
+                HypervisorEvent.SliceFinished(
+                    this@SimpleVirtDriver,
+                    totalBurst,
+                    totalBurst - totalRemainder,
+                    vms.size,
+                    server
+                )
+            )
         }
         this.call = call
     }
diff --git a/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/workload/PerformanceInterferenceModel.kt b/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/workload/PerformanceInterferenceModel.kt
index e2032ff1..ea110934 100644
--- a/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/workload/PerformanceInterferenceModel.kt
+++ b/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/workload/PerformanceInterferenceModel.kt
@@ -16,12 +16,16 @@ const val IMAGE_PERF_INTERFERENCE_MODEL = "image:performance-interference"
 data class PerformanceInterferenceModel(
     val items: Set<PerformanceInterferenceModelItem>
 ) {
-    fun apply(colocatedWorkloads: Set<Resource>, currentServerLoad: Double): Double {
+    private var intersectingItems: List<PerformanceInterferenceModelItem> = emptyList()
+
+    fun computeIntersectingItems(colocatedWorkloads: Set<Resource>) {
         val colocatedWorkloadIds = colocatedWorkloads.map { it.name }
-        val intersectingItems = items.filter { item ->
+        intersectingItems = items.filter { item ->
             colocatedWorkloadIds.intersect(item.workloadNames).size > 1
         }
+    }
 
+    fun apply(currentServerLoad: Double): Double {
         if (intersectingItems.isEmpty()) {
             return 1.0
         }