feat: Add more VM allocation policies

11 files changed, 166 insertions, 19 deletions

diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/HypervisorView.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/HypervisorView.kt
index 97842f18..e52a1698 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/HypervisorView.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/HypervisorView.kt
@@ -2,11 +2,14 @@ package com.atlarge.opendc.compute.virt.service
 
 import com.atlarge.opendc.compute.core.Server
 import com.atlarge.opendc.compute.virt.driver.VirtDriver
+import java.util.UUID
 
 class HypervisorView(
+    val uid: UUID,
     var server: Server,
     var numberOfActiveServers: Int,
-    var availableMemory: Long
+    var availableMemory: Long,
+    var provisionedCores: Int
 ) {
     lateinit var driver: VirtDriver
 }
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 b8966275..07693200 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
@@ -56,6 +56,16 @@ class SimpleVirtProvisioningService(
 
     override val hypervisorEvents: Flow<HypervisorEvent> = EventFlow()
 
+    /**
+     * The internal event flow to emit to.
+     */
+    private val internalEventFlow = EventFlow<VirtProvisioningServiceEvent>()
+
+    /**
+     * The allocation comparator to use.
+     */
+    private val allocationComparator = allocationPolicy(CoroutineScope(coroutineContext), internalEventFlow)
+
     init {
         launch {
             val provisionedNodes = provisioningService.nodes()
@@ -112,7 +122,7 @@ class SimpleVirtProvisioningService(
         val imagesToBeScheduled = incomingImages.toSet()
 
         for (imageInstance in imagesToBeScheduled) {
-            val selectedHv = availableHypervisors.minWith(allocationPolicy().thenBy { it.server.uid }) ?: break
+            val selectedHv = availableHypervisors.minWith(allocationComparator.thenBy { it.server.uid }) ?: break
             try {
                 println("Spawning ${imageInstance.image}")
                 incomingImages -= imageInstance
@@ -120,6 +130,7 @@ class SimpleVirtProvisioningService(
                 // Speculatively update the hypervisor view information to prevent other images in the queue from
                 // deciding on stale values.
                 selectedHv.numberOfActiveServers++
+                selectedHv.provisionedCores += imageInstance.flavor.cpuCount
                 selectedHv.availableMemory -= (imageInstance.image as VmImage).requiredMemory // XXX Temporary hack
 
                 val server = selectedHv.driver.spawn(
@@ -134,6 +145,7 @@ class SimpleVirtProvisioningService(
                 println("Unable to deploy image due to insufficient memory")
 
                 selectedHv.numberOfActiveServers--
+                selectedHv.provisionedCores -= imageInstance.flavor.cpuCount
                 selectedHv.availableMemory += (imageInstance.image as VmImage).requiredMemory
             }
         }
@@ -142,16 +154,21 @@ class SimpleVirtProvisioningService(
     private fun stateChanged(server: Server) {
         when (server.state) {
             ServerState.ACTIVE -> {
-                val hvView = HypervisorView(
+                val hv = HypervisorView(
+                    server.uid,
                     server,
                     0,
-                    server.flavor.memorySize
+                    server.flavor.memorySize,
+                    0
                 )
-                hypervisors[server] = hvView
+                hypervisors[server] = hv
+                internalEventFlow.emit(VirtProvisioningServiceEvent.HypervisorAdded(hv.uid))
             }
             ServerState.SHUTOFF, ServerState.ERROR -> {
                 val hv = hypervisors[server] ?: return
+                hv.provisionedCores -= server.flavor.cpuCount
                 availableHypervisors -= hv
+                internalEventFlow.emit(VirtProvisioningServiceEvent.HypervisorRemoved(hv.uid))
                 requestCycle()
             }
             else -> throw IllegalStateException()
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt
index fb200f88..550048a4 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningService.kt
@@ -12,6 +12,9 @@ import kotlinx.coroutines.flow.Flow
  * A service for VM provisioning on a cloud.
  */
 interface VirtProvisioningService {
+    /**
+     * The policy used for allocating a VM on the available hypervisors.
+     */
     val allocationPolicy: AllocationPolicy
 
     /**
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningServiceEvent.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningServiceEvent.kt
new file mode 100644
index 00000000..bab43b90
--- /dev/null
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/VirtProvisioningServiceEvent.kt
@@ -0,0 +1,43 @@
+/*
+ * MIT License
+ *
+ * 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 com.atlarge.opendc.compute.virt.service
+
+import java.util.UUID
+
+/**
+ * An internal event emitted by the [VirtProvisioningService] for the allocation policies to keep track of the
+ * available hypervisors.
+ */
+public sealed class VirtProvisioningServiceEvent {
+    /**
+     * This event is emitted when a hypervisor is added to the pool of schedulable hypervisors.
+     */
+    public data class HypervisorAdded(val uid: UUID) : VirtProvisioningServiceEvent()
+
+    /**
+     * This event is emitted when a hypervisor is removed from the pool of schedulable hypervisors.
+     */
+    public data class HypervisorRemoved(val uid: UUID) : VirtProvisioningServiceEvent()
+}
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AllocationPolicy.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AllocationPolicy.kt
index e2871cca..b1ec53b3 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AllocationPolicy.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AllocationPolicy.kt
@@ -2,6 +2,9 @@ package com.atlarge.opendc.compute.virt.service.allocation
 
 import com.atlarge.opendc.compute.metal.Node
 import com.atlarge.opendc.compute.virt.service.HypervisorView
+import com.atlarge.opendc.compute.virt.service.VirtProvisioningServiceEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
 
 /**
  * A policy for selecting the [Node] an image should be deployed to,
@@ -10,5 +13,5 @@ interface AllocationPolicy {
     /**
      * Builds the logic of the policy.
      */
-    operator fun invoke(): Comparator<HypervisorView>
+    operator fun invoke(scope: CoroutineScope, events: Flow<VirtProvisioningServiceEvent>): Comparator<HypervisorView>
 }
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableCoreMemoryAllocationPolicy.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableCoreMemoryAllocationPolicy.kt
new file mode 100644
index 00000000..6232d087
--- /dev/null
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableCoreMemoryAllocationPolicy.kt
@@ -0,0 +1,38 @@
+/*
+ * MIT License
+ *
+ * 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 com.atlarge.opendc.compute.virt.service.allocation
+
+import com.atlarge.opendc.compute.virt.service.HypervisorView
+import com.atlarge.opendc.compute.virt.service.VirtProvisioningServiceEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
+
+/**
+ * An [AllocationPolicy] that selects the machine with the highest amount of memory per core.
+ */
+public class AvailableCoreMemoryAllocationPolicy : AllocationPolicy {
+    override fun invoke(scope: CoroutineScope, events: Flow<VirtProvisioningServiceEvent>): Comparator<HypervisorView> =
+        compareBy { -it.availableMemory / it.server.flavor.cpuCount }
+}
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableMemoryAllocationPolicy.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableMemoryAllocationPolicy.kt
index f095849b..06c82209 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableMemoryAllocationPolicy.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/AvailableMemoryAllocationPolicy.kt
@@ -1,12 +1,14 @@
 package com.atlarge.opendc.compute.virt.service.allocation
 
 import com.atlarge.opendc.compute.virt.service.HypervisorView
+import com.atlarge.opendc.compute.virt.service.VirtProvisioningServiceEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
 
 /**
  * Allocation policy that selects the node with the most available memory.
  */
-class AvailableMemoryAllocationPolicy : AllocationPolicy {
-    override fun invoke(): Comparator<HypervisorView> = Comparator { o1, o2 ->
-        compareValuesBy(o1, o2) { -it.availableMemory }
-    }
+public class AvailableMemoryAllocationPolicy : AllocationPolicy {
+    override fun invoke(scope: CoroutineScope, events: Flow<VirtProvisioningServiceEvent>): Comparator<HypervisorView> =
+        compareBy { -it.availableMemory }
 }
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/NumberOfActiveServersAllocationPolicy.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/NumberOfActiveServersAllocationPolicy.kt
index 59e48465..b95d4260 100644
--- a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/NumberOfActiveServersAllocationPolicy.kt
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/NumberOfActiveServersAllocationPolicy.kt
@@ -1,15 +1,14 @@
 package com.atlarge.opendc.compute.virt.service.allocation
 
 import com.atlarge.opendc.compute.virt.service.HypervisorView
-import kotlinx.coroutines.runBlocking
+import com.atlarge.opendc.compute.virt.service.VirtProvisioningServiceEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
 
 /**
  * Allocation policy that selects the node with the least amount of active servers.
  */
-class NumberOfActiveServersAllocationPolicy : AllocationPolicy {
-    override fun invoke(): Comparator<HypervisorView> = Comparator { o1, o2 ->
-        runBlocking {
-            compareValuesBy(o1, o2) { it.numberOfActiveServers }
-        }
-    }
+public class NumberOfActiveServersAllocationPolicy : AllocationPolicy {
+    override fun invoke(scope: CoroutineScope, events: Flow<VirtProvisioningServiceEvent>): Comparator<HypervisorView> =
+        compareBy { it.numberOfActiveServers }
 }
diff --git a/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/ProvisionedCoresAllocationPolicy.kt b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/ProvisionedCoresAllocationPolicy.kt
new file mode 100644
index 00000000..816d85bf
--- /dev/null
+++ b/opendc/opendc-compute/src/main/kotlin/com/atlarge/opendc/compute/virt/service/allocation/ProvisionedCoresAllocationPolicy.kt
@@ -0,0 +1,39 @@
+/*
+ * MIT License
+ *
+ * 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 com.atlarge.opendc.compute.virt.service.allocation
+
+import com.atlarge.opendc.compute.virt.service.HypervisorView
+import com.atlarge.opendc.compute.virt.service.VirtProvisioningServiceEvent
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
+
+/**
+ * An [AllocationPolicy] that takes into account the number of vCPUs that have been provisioned on this machine
+ * relative to its core count.
+ */
+class ProvisionedCoresAllocationPolicy : AllocationPolicy {
+    override fun invoke(scope: CoroutineScope, events: Flow<VirtProvisioningServiceEvent>): Comparator<HypervisorView> =
+        compareBy { it.provisionedCores / it.server.flavor.cpuCount }
+}
diff --git a/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/CorrelatedFaultInjector.kt b/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/CorrelatedFaultInjector.kt
index da4dee12..c5189764 100644
--- a/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/CorrelatedFaultInjector.kt
+++ b/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/CorrelatedFaultInjector.kt
@@ -44,7 +44,7 @@ public class CorrelatedFaultInjector(
     private val iatShape: Double,
     private val sizeScale: Double,
     private val sizeShape: Double,
-    random: Random = Random
+    random: Random = Random(0)
 ) : FaultInjector {
     /**
      * The active failure domains that have been registered.
diff --git a/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/UncorrelatedFaultInjector.kt b/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/UncorrelatedFaultInjector.kt
index 3883eb11..1b896858 100644
--- a/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/UncorrelatedFaultInjector.kt
+++ b/opendc/opendc-core/src/main/kotlin/com/atlarge/opendc/core/failure/UncorrelatedFaultInjector.kt
@@ -35,7 +35,7 @@ import kotlin.random.Random
  * A [FaultInjector] that injects uncorrelated faults into the system, meaning that failures of the subsystems are
  * independent.
  */
-public class UncorrelatedFaultInjector(private val alpha: Double, private val beta: Double, private val random: Random = Random) : FaultInjector {
+public class UncorrelatedFaultInjector(private val alpha: Double, private val beta: Double, private val random: Random = Random(0)) : FaultInjector {
     /**
      * Enqueue the specified [FailureDomain] to fail some time in the future.
      */