bug: Fix issues related to early termination

3 files changed, 37 insertions, 17 deletions

diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/image/VmImage.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/image/VmImage.kt
index e3227540..36bbfa45 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/image/VmImage.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/core/image/VmImage.kt
@@ -23,19 +23,17 @@ class VmImage(
         val clock = simulationContext.clock
         val job = coroutineContext[Job]!!
 
-        for (fragments in flopsHistory.chunked(128)) {
-            for (fragment in fragments) {
-                job.ensureActive()
+        for (fragment in flopsHistory) {
+            job.ensureActive()
 
-                if (fragment.flops == 0L) {
-                    delay(fragment.duration)
-                } else {
-                    val cores = min(fragment.cores, ctx.server.flavor.cpuCount)
-                    val burst = LongArray(cores) { fragment.flops / cores }
-                    val usage = DoubleArray(cores) { fragment.usage / cores }
+            if (fragment.flops == 0L) {
+                delay(fragment.duration)
+            } else {
+                val cores = min(fragment.cores, ctx.server.flavor.cpuCount)
+                val burst = LongArray(cores) { fragment.flops / cores }
+                val usage = DoubleArray(cores) { fragment.usage / cores }
 
-                    ctx.run(burst, usage, clock.millis() + fragment.duration)
-                }
+                ctx.run(burst, usage, clock.millis() + fragment.duration)
             }
         }
     }
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 8a32bc43..53fa463b 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
@@ -243,7 +243,7 @@ class SimpleVirtDriver(
             }
 
             // XXX We set the minimum duration to 5 minutes here to prevent the rounding issues that are occurring with the FLOPs.
-            duration = max(300.0, ceil(duration))
+            duration = 300.0
 
             val totalAllocatedUsage = maxUsage - availableUsage
             var totalAllocatedBurst = 0L
@@ -335,7 +335,7 @@ class SimpleVirtDriver(
             eventFlow.emit(
                 HypervisorEvent.SliceFinished(
                     this@SimpleVirtDriver,
-                    totalRequestedSubBurst,
+                    totalRequestedBurst,
                     min(totalRequestedSubBurst, totalGrantedBurst), // We can run more than requested due to timing
                     totalOvercommissionedBurst,
                     totalInterferedBurst, // Might be smaller than zero due to FP rounding errors,
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt
index 09036d0d..520f6dc5 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/SimpleVirtProvisioningService.kt
@@ -26,6 +26,7 @@ import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.withContext
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
+import kotlin.math.max
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class SimpleVirtProvisioningService(
@@ -57,6 +58,10 @@ class SimpleVirtProvisioningService(
     public var queuedVms = 0L
     public var runningVms = 0L
     public var finishedVms = 0L
+    public var unscheduledVms = 0L
+
+    private var maxCores = 0
+    private var maxMemory = 0L
 
     /**
      * The allocation logic to use.
@@ -124,15 +129,24 @@ class SimpleVirtProvisioningService(
     }
 
     private suspend fun schedule() {
-        val log = simulationContext.log
+        val clock = simulationContext.clock
         val imagesToBeScheduled = incomingImages.toSet()
 
         for (imageInstance in imagesToBeScheduled) {
             val requiredMemory = (imageInstance.image as VmImage).requiredMemory
-            val selectedHv = allocationLogic.select(availableHypervisors, imageInstance) ?: break
+            val selectedHv = allocationLogic.select(availableHypervisors, imageInstance)
+
+            if (selectedHv == null) {
+                if (requiredMemory > maxMemory || imageInstance.flavor.cpuCount > maxCores) {
+                    unscheduledVms++
+                    println("[${clock.millis()}] CANNOT SPAWN ${imageInstance.image}")
+                }
+
+                break
+            }
 
             try {
-                log.info("Spawning ${imageInstance.image} on ${selectedHv.server}")
+                println("[${clock.millis()}] SPAWN ${imageInstance.image} on ${selectedHv.server.uid} ${selectedHv.server.name} ${selectedHv.server.flavor}")
                 incomingImages -= imageInstance
 
                 // Speculatively update the hypervisor view information to prevent other images in the queue from
@@ -157,6 +171,7 @@ class SimpleVirtProvisioningService(
                         when (event) {
                             is ServerEvent.StateChanged -> {
                                 if (event.server.state == ServerState.SHUTOFF) {
+                                    println("[${clock.millis()}] FINISH ${event.server.uid} ${event.server.name} ${event.server.flavor}")
                                     runningVms--
                                     finishedVms++
 
@@ -178,6 +193,8 @@ class SimpleVirtProvisioningService(
                 selectedHv.numberOfActiveServers--
                 selectedHv.provisionedCores -= imageInstance.flavor.cpuCount
                 selectedHv.availableMemory += requiredMemory
+            } catch (e: Throwable) {
+                e.printStackTrace()
             }
         }
     }
@@ -196,13 +213,18 @@ class SimpleVirtProvisioningService(
                         server.flavor.memorySize,
                         0
                     )
+                    maxCores = max(maxCores, server.flavor.cpuCount)
+                    maxMemory = max(maxMemory, server.flavor.memorySize)
                     hypervisors[server] = hv
                 }
             }
             ServerState.SHUTOFF, ServerState.ERROR -> {
                 val hv = hypervisors[server] ?: return
                 availableHypervisors -= hv
-                requestCycle()
+
+                if (incomingImages.isNotEmpty()) {
+                    requestCycle()
+                }
             }
             else -> throw IllegalStateException()
         }