Convert to pull-based workload model

This change converts the low-level workload model to be pull-based. This
reduces the overhead that we experienced with our previous co-routine
based approach.

7 files changed, 102 insertions, 99 deletions

diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimBareMetalDriver.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimBareMetalDriver.kt
index 97f550ba..7a978a53 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimBareMetalDriver.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimBareMetalDriver.kt
@@ -41,6 +41,7 @@ import org.opendc.core.services.ServiceRegistry
 import org.opendc.simulator.compute.SimBareMetalMachine
 import org.opendc.simulator.compute.SimExecutionContext
 import org.opendc.simulator.compute.SimMachineModel
+import org.opendc.simulator.compute.workload.SimResourceCommand
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.simulator.failures.FailureDomain
 import org.opendc.utils.flow.EventFlow
@@ -139,15 +140,31 @@ public class SimBareMetalDriver(
             events
         )
 
+        val delegate = (node.image as SimWorkloadImage).workload
         // Wrap the workload to pass in a ComputeSimExecutionContext
         val workload = object : SimWorkload {
-            override suspend fun run(ctx: SimExecutionContext) {
-                val wrappedCtx = object : ComputeSimExecutionContext, SimExecutionContext by ctx {
+            lateinit var wrappedCtx: ComputeSimExecutionContext
+
+            override fun onStart(ctx: SimExecutionContext) {
+                wrappedCtx = object : ComputeSimExecutionContext, SimExecutionContext by ctx {
                     override val server: Server
                         get() = nodeState.value.server!!
+
+                    override fun toString(): String = "WrappedSimExecutionContext"
                 }
-                (node.image as SimWorkloadImage).workload.run(wrappedCtx)
+
+                delegate.onStart(wrappedCtx)
+            }
+
+            override fun onStart(ctx: SimExecutionContext, cpu: Int): SimResourceCommand {
+                return delegate.onStart(wrappedCtx, cpu)
+            }
+
+            override fun onNext(ctx: SimExecutionContext, cpu: Int, remainingWork: Double): SimResourceCommand {
+                return delegate.onNext(wrappedCtx, cpu, remainingWork)
             }
+
+            override fun toString(): String = "SimWorkloadWrapper(delegate=$delegate)"
         }
 
         job = coroutineScope.launch {
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriver.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriver.kt
index 249979a8..f7eb9248 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriver.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriver.kt
@@ -36,25 +36,25 @@ import org.opendc.simulator.compute.*
 import org.opendc.simulator.compute.interference.IMAGE_PERF_INTERFERENCE_MODEL
 import org.opendc.simulator.compute.interference.PerformanceInterferenceModel
 import org.opendc.simulator.compute.model.MemoryUnit
+import org.opendc.simulator.compute.workload.SimResourceCommand
 import org.opendc.simulator.compute.workload.SimWorkload
 import org.opendc.utils.flow.EventFlow
-import java.time.Clock
 import java.util.*
 
 /**
  * A [VirtDriver] that is simulates virtual machines on a physical machine using [SimHypervisor].
  */
-public class SimVirtDriver(
-    private val coroutineScope: CoroutineScope,
-    clock: Clock,
-    private val ctx: SimExecutionContext
-) : VirtDriver {
+public class SimVirtDriver(private val coroutineScope: CoroutineScope) : VirtDriver, SimWorkload {
+    /**
+     * The execution context in which the [VirtDriver] runs.
+     */
+    private lateinit var ctx: ComputeSimExecutionContext
 
     /**
      * The server hosting this hypervisor.
      */
     public val server: Server
-        get() = (ctx as ComputeSimExecutionContext).server
+        get() = ctx.server
 
     /**
      * The [EventFlow] to emit the events.
@@ -66,35 +66,33 @@ public class SimVirtDriver(
     /**
      * Current total memory use of the images on this hypervisor.
      */
-    private var availableMemory: Long = ctx.machine.memory.map { it.size }.sum()
+    private var availableMemory: Long = 0
 
     /**
      * The hypervisor to run multiple workloads.
      */
-    private val hypervisor = SimFairSharedHypervisor(
-        coroutineScope,
-        clock,
+    private val hypervisor = SimFairShareHypervisor(
         object : SimHypervisor.Listener {
             override fun onSliceFinish(
                 hypervisor: SimHypervisor,
-                requestedBurst: Long,
-                grantedBurst: Long,
-                overcommissionedBurst: Long,
-                interferedBurst: Long,
+                requestedWork: Long,
+                grantedWork: Long,
+                overcommittedWork: Long,
+                interferedWork: Long,
                 cpuUsage: Double,
                 cpuDemand: Double
             ) {
                 eventFlow.emit(
                     HypervisorEvent.SliceFinished(
                         this@SimVirtDriver,
-                        requestedBurst,
-                        grantedBurst,
-                        overcommissionedBurst,
-                        interferedBurst,
+                        requestedWork,
+                        grantedWork,
+                        overcommittedWork,
+                        interferedWork,
                         cpuUsage,
                         cpuDemand,
                         vms.size,
-                        (ctx as ComputeSimExecutionContext).server
+                        ctx.server
                     )
                 )
             }
@@ -158,14 +156,31 @@ public class SimVirtDriver(
         val performanceInterferenceModel: PerformanceInterferenceModel? = server.image.tags[IMAGE_PERF_INTERFERENCE_MODEL] as? PerformanceInterferenceModel?
 
         val job = coroutineScope.launch {
+            val delegate = (server.image as SimWorkloadImage).workload
+            // Wrap the workload to pass in a ComputeSimExecutionContext
             val workload = object : SimWorkload {
-                override suspend fun run(ctx: SimExecutionContext) {
-                    val wrappedCtx = object : ComputeSimExecutionContext, SimExecutionContext by ctx {
+                lateinit var wrappedCtx: ComputeSimExecutionContext
+
+                override fun onStart(ctx: SimExecutionContext) {
+                    wrappedCtx = object : ComputeSimExecutionContext, SimExecutionContext by ctx {
                         override val server: Server
-                            get() = this@VirtualMachine.server
+                            get() = server
+
+                        override fun toString(): String = "WrappedSimExecutionContext"
                     }
-                    (server.image as SimWorkloadImage).workload.run(wrappedCtx)
+
+                    delegate.onStart(wrappedCtx)
+                }
+
+                override fun onStart(ctx: SimExecutionContext, cpu: Int): SimResourceCommand {
+                    return delegate.onStart(wrappedCtx, cpu)
                 }
+
+                override fun onNext(ctx: SimExecutionContext, cpu: Int, remainingWork: Double): SimResourceCommand {
+                    return delegate.onNext(wrappedCtx, cpu, remainingWork)
+                }
+
+                override fun toString(): String = "SimWorkloadWrapper(delegate=$delegate)"
             }
 
             delay(1) // TODO Introduce boot time
@@ -206,7 +221,17 @@ public class SimVirtDriver(
         }
     }
 
-    public suspend fun run() {
-        hypervisor.run(ctx)
+    override fun onStart(ctx: SimExecutionContext) {
+        this.ctx = ctx as ComputeSimExecutionContext
+        this.availableMemory = ctx.machine.memory.map { it.size }.sum()
+        this.hypervisor.onStart(ctx)
+    }
+
+    override fun onStart(ctx: SimExecutionContext, cpu: Int): SimResourceCommand {
+        return hypervisor.onStart(ctx, cpu)
+    }
+
+    override fun onNext(ctx: SimExecutionContext, cpu: Int, remainingWork: Double): SimResourceCommand {
+        return hypervisor.onNext(ctx, cpu, remainingWork)
     }
 }
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriverWorkload.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriverWorkload.kt
deleted file mode 100644
index 58b9408a..00000000
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtDriverWorkload.kt
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Copyright (c) 2020 AtLarge Research
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all
- * copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package org.opendc.compute.simulator
-
-import kotlinx.coroutines.coroutineScope
-import org.opendc.simulator.compute.SimExecutionContext
-import org.opendc.simulator.compute.workload.SimWorkload
-
-public class SimVirtDriverWorkload : SimWorkload {
-    public lateinit var driver: SimVirtDriver
-
-    override suspend fun run(ctx: SimExecutionContext) {
-        coroutineScope {
-            driver = SimVirtDriver(this, ctx.clock, ctx)
-            driver.run()
-        }
-    }
-}
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtProvisioningService.kt b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtProvisioningService.kt
index 17de3de7..f3a69f48 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtProvisioningService.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/SimVirtProvisioningService.kt
@@ -107,7 +107,7 @@ public class SimVirtProvisioningService(
         coroutineScope.launch {
             val provisionedNodes = provisioningService.nodes()
             provisionedNodes.forEach { node ->
-                val workload = SimVirtDriverWorkload()
+                val workload = SimVirtDriver(coroutineScope)
                 val hypervisorImage = SimWorkloadImage(UUID.randomUUID(), "vmm", emptyMap(), workload)
                 launch {
                     var init = false
@@ -125,7 +125,7 @@ public class SimVirtProvisioningService(
                     }.launchIn(this)
 
                     delay(1)
-                    onHypervisorAvailable(server, workload.driver)
+                    onHypervisorAvailable(server, workload)
                 }
             }
         }
@@ -208,7 +208,7 @@ public class SimVirtProvisioningService(
                             submittedVms,
                             runningVms,
                             finishedVms,
-                            queuedVms,
+                            --queuedVms,
                             ++unscheduledVms
                         )
                     )
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimBareMetalDriverTest.kt b/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimBareMetalDriverTest.kt
index 0f1bd444..fb8a5f47 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimBareMetalDriverTest.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimBareMetalDriverTest.kt
@@ -64,7 +64,7 @@ internal class SimBareMetalDriverTest {
 
         testScope.launch {
             val driver = SimBareMetalDriver(this, clock, UUID.randomUUID(), "test", emptyMap(), machineModel)
-            val image = SimWorkloadImage(UUID.randomUUID(), "<unnamed>", emptyMap(), SimFlopsWorkload(4_000, 2, utilization = 1.0))
+            val image = SimWorkloadImage(UUID.randomUUID(), "<unnamed>", emptyMap(), SimFlopsWorkload(4_000, utilization = 1.0))
 
             // Batch driver commands
             withContext(coroutineContext) {
@@ -84,6 +84,6 @@ internal class SimBareMetalDriverTest {
 
         testScope.advanceUntilIdle()
         assertEquals(ServerState.SHUTOFF, finalState)
-        assertEquals(1001, finalTime)
+        assertEquals(501, finalTime)
     }
 }
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimProvisioningServiceTest.kt b/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimProvisioningServiceTest.kt
index def78ce7..a33a4e5f 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimProvisioningServiceTest.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimProvisioningServiceTest.kt
@@ -64,7 +64,7 @@ internal class SimProvisioningServiceTest {
         val clock = DelayControllerClockAdapter(testScope)
 
         testScope.launch {
-            val image = SimWorkloadImage(UUID.randomUUID(), "<unnamed>", emptyMap(), SimFlopsWorkload(1000, 2))
+            val image = SimWorkloadImage(UUID.randomUUID(), "<unnamed>", emptyMap(), SimFlopsWorkload(1000))
             val driver = SimBareMetalDriver(this, clock, UUID.randomUUID(), "test", emptyMap(), machineModel)
 
             val provisioner = SimpleProvisioningService()
diff --git a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimVirtDriverTest.kt b/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimVirtDriverTest.kt
index 394e87c6..163b326a 100644
--- a/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimVirtDriverTest.kt
+++ b/simulator/opendc-compute/opendc-compute-simulator/src/test/kotlin/org/opendc/compute/simulator/SimVirtDriverTest.kt
@@ -66,17 +66,17 @@ internal class SimVirtDriverTest {
     }
 
     /**
-     * Test overcommissioning of a hypervisor.
+     * Test overcommitting of resources by the hypervisor.
      */
     @Test
-    fun overcommission() {
-        var requestedBurst = 0L
-        var grantedBurst = 0L
-        var overcommissionedBurst = 0L
+    fun testOvercommitted() {
+        var requestedWork = 0L
+        var grantedWork = 0L
+        var overcommittedWork = 0L
 
         scope.launch {
-            val virtDriverWorkload = SimVirtDriverWorkload()
-            val vmm = SimWorkloadImage(UUID.randomUUID(), "vmm", emptyMap(), virtDriverWorkload)
+            val virtDriver = SimVirtDriver(this)
+            val vmm = SimWorkloadImage(UUID.randomUUID(), "vmm", emptyMap(), virtDriver)
             val duration = 5 * 60L
             val vmImageA = SimWorkloadImage(
                 UUID.randomUUID(),
@@ -84,10 +84,10 @@ internal class SimVirtDriverTest {
                 emptyMap(),
                 SimTraceWorkload(
                     sequenceOf(
-                        SimTraceWorkload.Fragment(0, 28L * duration, duration * 1000, 28.0, 2),
-                        SimTraceWorkload.Fragment(0, 3500L * duration, duration * 1000, 3500.0, 2),
-                        SimTraceWorkload.Fragment(0, 0, duration * 1000, 0.0, 2),
-                        SimTraceWorkload.Fragment(0, 183L * duration, duration * 1000, 183.0, 2)
+                        SimTraceWorkload.Fragment(duration * 1000, 28.0, 2),
+                        SimTraceWorkload.Fragment(duration * 1000, 3500.0, 2),
+                        SimTraceWorkload.Fragment(duration * 1000, 0.0, 2),
+                        SimTraceWorkload.Fragment(duration * 1000, 183.0, 2)
                     ),
                 )
             )
@@ -97,10 +97,10 @@ internal class SimVirtDriverTest {
                 emptyMap(),
                 SimTraceWorkload(
                     sequenceOf(
-                        SimTraceWorkload.Fragment(0, 28L * duration, duration * 1000, 28.0, 2),
-                        SimTraceWorkload.Fragment(0, 3100L * duration, duration * 1000, 3100.0, 2),
-                        SimTraceWorkload.Fragment(0, 0, duration * 1000, 0.0, 2),
-                        SimTraceWorkload.Fragment(0, 73L * duration, duration * 1000, 73.0, 2)
+                        SimTraceWorkload.Fragment(duration * 1000, 28.0, 2),
+                        SimTraceWorkload.Fragment(duration * 1000, 3100.0, 2),
+                        SimTraceWorkload.Fragment(duration * 1000, 0.0, 2),
+                        SimTraceWorkload.Fragment(duration * 1000, 73.0, 2)
                     )
                 ),
             )
@@ -115,31 +115,30 @@ internal class SimVirtDriverTest {
             delay(5)
 
             val flavor = Flavor(2, 0)
-            val vmDriver = virtDriverWorkload.driver
-            vmDriver.events
+            virtDriver.events
                 .onEach { event ->
                     when (event) {
                         is HypervisorEvent.SliceFinished -> {
-                            requestedBurst += event.requestedBurst
-                            grantedBurst += event.grantedBurst
-                            overcommissionedBurst += event.overcommissionedBurst
+                            requestedWork += event.requestedBurst
+                            grantedWork += event.grantedBurst
+                            overcommittedWork += event.overcommissionedBurst
                         }
                     }
                 }
                 .launchIn(this)
 
-            vmDriver.spawn("a", vmImageA, flavor)
-            vmDriver.spawn("b", vmImageB, flavor)
+            virtDriver.spawn("a", vmImageA, flavor)
+            virtDriver.spawn("b", vmImageB, flavor)
         }
 
         scope.advanceUntilIdle()
 
         assertAll(
             { assertEquals(emptyList<Throwable>(), scope.uncaughtExceptions, "No errors") },
-            { assertEquals(2082000, requestedBurst, "Requested Burst does not match") },
-            { assertEquals(2013600, grantedBurst, "Granted Burst does not match") },
-            { assertEquals(60000, overcommissionedBurst, "Overcommissioned Burst does not match") },
-            { assertEquals(1200007, scope.currentTime) }
+            { assertEquals(4197600, requestedWork, "Requested work does not match") },
+            { assertEquals(3057600, grantedWork, "Granted work does not match") },
+            { assertEquals(1140000, overcommittedWork, "Overcommitted work does not match") },
+            { assertEquals(1200006, scope.currentTime) }
         )
     }
 }