bug: Record start and power off events in SC20 experiments

2 files changed, 42 insertions, 5 deletions

diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt
index 441b4498..120c4f81 100644
--- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt
+++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Sc20Monitor.kt
@@ -22,8 +22,6 @@ class Sc20Monitor(
     }
 
     suspend fun onVmStateChanged(server: Server) {
-        println("[${simulationContext.clock.millis()}] ${server.uid} ${server.state}")
-
         if (server.state == ServerState.ERROR) {
             failedInSlice++
         }
@@ -44,6 +42,8 @@ class Sc20Monitor(
             )
         }
 
+        println("[${simulationContext.clock.millis()}] HOST ${server.uid} ${server.state}")
+
         lastServerStates[server] = Pair(server.state, simulationContext.clock.millis())
     }
 
diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt
index d697cde3..ed971df5 100644
--- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt
+++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/TestExperiment.kt
@@ -28,10 +28,13 @@ import com.atlarge.odcsim.Domain
 import com.atlarge.odcsim.SimulationEngineProvider
 import com.atlarge.odcsim.simulationContext
 import com.atlarge.opendc.compute.core.Flavor
+import com.atlarge.opendc.compute.core.Server
 import com.atlarge.opendc.compute.core.ServerEvent
+import com.atlarge.opendc.compute.core.ServerState
 import com.atlarge.opendc.compute.metal.NODE_CLUSTER
 import com.atlarge.opendc.compute.metal.service.ProvisioningService
 import com.atlarge.opendc.compute.virt.HypervisorEvent
+import com.atlarge.opendc.compute.virt.driver.SimpleVirtDriver
 import com.atlarge.opendc.compute.virt.service.SimpleVirtProvisioningService
 import com.atlarge.opendc.compute.virt.service.allocation.AvailableMemoryAllocationPolicy
 import com.atlarge.opendc.core.failure.CorrelatedFaultInjector
@@ -141,12 +144,28 @@ fun main(args: Array<String>) {
             // Wait for the hypervisors to be spawned
             delay(10)
 
+            val hypervisors = scheduler.drivers()
+            var availableHypervisors = hypervisors.size
+
             // Monitor hypervisor events
-            for (hypervisor in scheduler.drivers()) {
+            for (hypervisor in hypervisors) {
+                // TODO Do not expose VirtDriver directly but use Hypervisor class.
+                monitor.serverStateChanged(hypervisor, (hypervisor as SimpleVirtDriver).server)
+                hypervisor.server.events
+                    .onEach { event ->
+                        when (event) {
+                            is ServerEvent.StateChanged -> {
+                                monitor.serverStateChanged(hypervisor, event.server)
+
+                                if (event.server.state == ServerState.ERROR)
+                                    availableHypervisors -= 1
+                            }
+                        }
+                    }
+                    .launchIn(this)
                 hypervisor.events
                     .onEach { event ->
                         when (event) {
-                            is HypervisorEvent.StateChanged -> monitor.serverStateChanged(event.driver, event.server)
                             is HypervisorEvent.SliceFinished -> monitor.onSliceFinish(
                                 simulationContext.clock.millis(),
                                 event.requestedBurst,
@@ -171,6 +190,9 @@ fun main(args: Array<String>) {
                 }
             }
 
+            val running = mutableSetOf<Server>()
+            val finish = Channel<Unit>(Channel.RENDEZVOUS)
+
             val reader = Sc20TraceReader(File(traceDirectory), performanceInterferenceModel, getSelectedVmList())
             while (reader.hasNext()) {
                 val (time, workload) = reader.next()
@@ -181,12 +203,27 @@ fun main(args: Array<String>) {
                         workload.image.name, workload.image,
                         Flavor(workload.image.cores, workload.image.requiredMemory)
                     )
+                    running += server
                     // Monitor server events
-                    server.events.onEach { if (it is ServerEvent.StateChanged) monitor.onVmStateChanged(it.server) }
+                    server.events
+                        .onEach {
+                            if (it is ServerEvent.StateChanged)
+                                monitor.onVmStateChanged(it.server)
+
+                            // Detect whether the VM has finished running
+                            if (it.server.state == ServerState.ERROR || it.server.state == ServerState.SHUTOFF) {
+                                running -= server
+
+                                if (running.isEmpty() && (!reader.hasNext() || availableHypervisors == 0))
+                                    finish.send(Unit)
+                            }
+                        }
                         .collect()
                 }
             }
 
+            finish.receive()
+            scheduler.terminate()
             println(simulationContext.clock.instant())
         }