merge: Improve the OpenDC compute model (#37)

This pull request contains various improvements to the OpenDC compute simulation model.

- Support filtering hosts based on CPU capacity
- Do not allocate lambda in fast-path
- Redesign VM interference algorithm
- Report provisioning time of virtual machines
- Prevent allocations during collection cycle 
- Use correct flow input capacity for counters
- Support running workloads without coroutines

**Breaking API Changes**

- `VirtualMachine` now requires `cpuCapacity` parameter.
- `VmInterferenceModel` needs to be constructed using `VmInterferenceModel.Builder` and can't be passed a list of groups anymore.
- Scheduling latency is not collected anymore. Instead, use the boot time and provisioning time to derive the scheduling latency.
- Telemetry data is recorded using `*TableReader` interfaces as opposed to the `*Data` classes. These classes are re-used per row and should not be shared with other threads, since the underlying data may change.
- `SimMachine` does not implement `AutoCloseable` anymore. Machines can be removed from a `SimHypervisor` using the `removeMachine` method.
- `SimMachine.run` is moved to an extension method called `runWorkload`. Users can now also choose to run a workload using the asynchronous `SimMachine.startWorkload`.

3 files changed, 56 insertions, 48 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 b9d02185..908a58e9 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
@@ -47,6 +47,7 @@ import org.opendc.simulator.compute.model.MemoryUnit
 import org.opendc.simulator.compute.power.ConstantPowerModel
 import org.opendc.simulator.compute.power.PowerDriver
 import org.opendc.simulator.compute.power.SimplePowerDriver
+import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.flow.FlowEngine
 import java.util.*
 import kotlin.coroutines.CoroutineContext
@@ -121,7 +122,7 @@ public class SimHost(
             field = value
         }
 
-    override val model: HostModel = HostModel(model.cpus.size, model.memory.sumOf { it.size })
+    override val model: HostModel = HostModel(model.cpus.sumOf { it.frequency }, model.cpus.size, model.memory.sumOf { it.size })
 
     /**
      * The [GuestListener] that listens for guest events.
@@ -136,11 +137,6 @@ public class SimHost(
         }
     }
 
-    /**
-     * The [Job] that represents the machine running the hypervisor.
-     */
-    private var _job: Job? = null
-
     init {
         launch()
 
@@ -188,7 +184,7 @@ public class SimHost(
     }
 
     override fun canFit(server: Server): Boolean {
-        val sufficientMemory = model.memorySize >= server.flavor.memorySize
+        val sufficientMemory = model.memoryCapacity >= server.flavor.memorySize
         val enoughCpus = model.cpuCount >= server.flavor.cpuCount
         val canFit = hypervisor.canFit(server.flavor.toMachineModel())
 
@@ -199,11 +195,12 @@ public class SimHost(
         val guest = guests.computeIfAbsent(server) { key ->
             require(canFit(key)) { "Server does not fit" }
 
-            val machine = hypervisor.createMachine(key.flavor.toMachineModel(), key.name)
+            val machine = hypervisor.newMachine(key.flavor.toMachineModel(), key.name)
             val newGuest = Guest(
                 scope.coroutineContext,
                 clock,
                 this,
+                hypervisor,
                 mapper,
                 guestListener,
                 server,
@@ -249,7 +246,7 @@ public class SimHost(
     override fun close() {
         reset()
         scope.cancel()
-        machine.close()
+        machine.cancel()
     }
 
     override fun toString(): String = "SimHost[uid=$uid,name=$name,model=$model]"
@@ -276,26 +273,39 @@ public class SimHost(
     }
 
     /**
+     * The [Job] that represents the machine running the hypervisor.
+     */
+    private var _ctx: SimMachineContext? = null
+
+    /**
      * Launch the hypervisor.
      */
     private fun launch() {
-        check(_job == null) { "Concurrent hypervisor running" }
+        check(_ctx == null) { "Concurrent hypervisor running" }
 
         // Launch hypervisor onto machine
-        _job = scope.launch {
-            try {
-                _bootTime = clock.millis()
-                _state = HostState.UP
-                machine.run(hypervisor, emptyMap())
-            } catch (_: CancellationException) {
-                // Ignored
-            } catch (cause: Throwable) {
-                logger.error(cause) { "Host failed" }
-                throw cause
-            } finally {
-                _state = HostState.DOWN
+        _ctx = machine.startWorkload(object : SimWorkload {
+            override fun onStart(ctx: SimMachineContext) {
+                try {
+                    _bootTime = clock.millis()
+                    _state = HostState.UP
+                    hypervisor.onStart(ctx)
+                } catch (cause: Throwable) {
+                    _state = HostState.DOWN
+                    _ctx = null
+                    throw cause
+                }
             }
-        }
+
+            override fun onStop(ctx: SimMachineContext) {
+                try {
+                    hypervisor.onStop(ctx)
+                } finally {
+                    _state = HostState.DOWN
+                    _ctx = null
+                }
+            }
+        })
     }
 
     /**
@@ -305,12 +315,7 @@ public class SimHost(
         updateUptime()
 
         // Stop the hypervisor
-        val job = _job
-        if (job != null) {
-            job.cancel()
-            _job = null
-        }
-
+        _ctx?.close()
         _state = HostState.DOWN
     }
 
@@ -319,8 +324,9 @@ public class SimHost(
      */
     private fun Flavor.toMachineModel(): MachineModel {
         val originalCpu = machine.model.cpus[0]
+        val cpuCapacity = (this.meta["cpu-capacity"] as? Double ?: Double.MAX_VALUE).coerceAtMost(originalCpu.frequency)
         val processingNode = originalCpu.node.copy(coreCount = cpuCount)
-        val processingUnits = (0 until cpuCount).map { originalCpu.copy(id = it, node = processingNode) }
+        val processingUnits = (0 until cpuCount).map { originalCpu.copy(id = it, node = processingNode, frequency = cpuCapacity) }
         val memoryUnits = listOf(MemoryUnit("Generic", "Generic", 3200.0, memorySize))
 
         return MachineModel(processingUnits, memoryUnits).optimize()
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 5ea1860d..9f3122db 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
@@ -34,7 +34,9 @@ import org.opendc.compute.api.Server
 import org.opendc.compute.api.ServerState
 import org.opendc.compute.simulator.SimHost
 import org.opendc.compute.simulator.SimWorkloadMapper
+import org.opendc.simulator.compute.kernel.SimHypervisor
 import org.opendc.simulator.compute.kernel.SimVirtualMachine
+import org.opendc.simulator.compute.runWorkload
 import org.opendc.simulator.compute.workload.SimWorkload
 import java.time.Clock
 import kotlin.coroutines.CoroutineContext
@@ -46,6 +48,7 @@ internal class Guest(
     context: CoroutineContext,
     private val clock: Clock,
     val host: SimHost,
+    private val hypervisor: SimHypervisor,
     private val mapper: SimWorkloadMapper,
     private val listener: GuestListener,
     val server: Server,
@@ -114,8 +117,7 @@ internal class Guest(
         stop()
 
         state = ServerState.DELETED
-
-        machine.close()
+        hypervisor.removeMachine(machine)
         scope.cancel()
     }
 
@@ -191,7 +193,7 @@ internal class Guest(
      */
     private suspend fun runMachine(workload: SimWorkload) {
         delay(1) // TODO Introduce model for boot time
-        machine.run(workload, mapOf("driver" to host, "server" to server))
+        machine.runWorkload(workload, mapOf("driver" to host, "server" to server))
     }
 
     /**
@@ -248,7 +250,7 @@ internal class Guest(
      */
     fun collectBootTime(result: ObservableLongMeasurement) {
         if (_bootTime != Long.MIN_VALUE) {
-            result.observe(_bootTime)
+            result.observe(_bootTime, attributes)
         }
     }
 
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 a0ff9228..799a8cf0 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
@@ -46,8 +46,8 @@ import org.opendc.simulator.core.runBlockingSimulation
 import org.opendc.simulator.flow.FlowEngine
 import org.opendc.telemetry.compute.ComputeMetricExporter
 import org.opendc.telemetry.compute.HOST_ID
-import org.opendc.telemetry.compute.table.HostData
-import org.opendc.telemetry.compute.table.ServerData
+import org.opendc.telemetry.compute.table.HostTableReader
+import org.opendc.telemetry.compute.table.ServerTableReader
 import org.opendc.telemetry.sdk.metrics.export.CoroutineMetricReader
 import org.opendc.telemetry.sdk.toOtelClock
 import java.time.Duration
@@ -140,10 +140,10 @@ internal class SimHostTest {
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
             object : ComputeMetricExporter() {
-                override fun record(data: HostData) {
-                    activeTime += data.cpuActiveTime
-                    idleTime += data.cpuIdleTime
-                    stealTime += data.cpuStealTime
+                override fun record(reader: HostTableReader) {
+                    activeTime += reader.cpuActiveTime
+                    idleTime += reader.cpuIdleTime
+                    stealTime += reader.cpuStealTime
                 }
             },
             exportInterval = Duration.ofSeconds(duration)
@@ -236,16 +236,16 @@ internal class SimHostTest {
         val reader = CoroutineMetricReader(
             this, listOf(meterProvider as MetricProducer),
             object : ComputeMetricExporter() {
-                override fun record(data: HostData) {
-                    activeTime += data.cpuActiveTime
-                    idleTime += data.cpuIdleTime
-                    uptime += data.uptime
-                    downtime += data.downtime
+                override fun record(reader: HostTableReader) {
+                    activeTime += reader.cpuActiveTime
+                    idleTime += reader.cpuIdleTime
+                    uptime += reader.uptime
+                    downtime += reader.downtime
                 }
 
-                override fun record(data: ServerData) {
-                    guestUptime += data.uptime
-                    guestDowntime += data.downtime
+                override fun record(reader: ServerTableReader) {
+                    guestUptime += reader.uptime
+                    guestDowntime += reader.downtime
                 }
             },
             exportInterval = Duration.ofSeconds(duration)