diff options
Diffstat (limited to 'opendc-compute/opendc-compute-simulator/src/main')
10 files changed, 269 insertions, 115 deletions
diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java index c7478c84..6fa6af60 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java +++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ComputeService.java @@ -31,6 +31,7 @@ import java.util.Collections; import java.util.Deque; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Objects; @@ -49,6 +50,9 @@ import org.opendc.compute.simulator.host.HostModel; import org.opendc.compute.simulator.host.HostState; import org.opendc.compute.simulator.host.SimHost; import org.opendc.compute.simulator.scheduler.ComputeScheduler; +import org.opendc.compute.simulator.scheduler.SchedulingRequest; +import org.opendc.compute.simulator.scheduler.SchedulingResult; +import org.opendc.compute.simulator.scheduler.SchedulingResultType; import org.opendc.compute.simulator.telemetry.ComputeMetricReader; import org.opendc.compute.simulator.telemetry.SchedulerStats; import org.opendc.simulator.compute.power.SimPowerSource; @@ -204,6 +208,7 @@ public final class ComputeService implements AutoCloseable { } if (task.getState() == TaskState.COMPLETED || task.getState() == TaskState.TERMINATED) { + scheduler.removeTask(task, hv); setTaskToBeRemoved(task); } @@ -430,38 +435,37 @@ public final class ComputeService implements AutoCloseable { private void doSchedule() { // reorder tasks - while (!taskQueue.isEmpty()) { - SchedulingRequest request = taskQueue.peek(); - - if (request.isCancelled) { - taskQueue.poll(); - tasksPending--; - continue; + for (Iterator<SchedulingRequest> iterator = taskQueue.iterator(); + iterator.hasNext(); + iterator = taskQueue.iterator()) { + final SchedulingResult result = scheduler.select(iterator); + if (result.getResultType() == SchedulingResultType.EMPTY) { + break; } - - final ServiceTask task = request.task; + final HostView hv = result.getHost(); + final SchedulingRequest req = result.getReq(); + final ServiceTask task = req.getTask(); + final ServiceFlavor flavor = task.getFlavor(); if (task.getNumFailures() >= maxNumFailures) { LOGGER.warn("task {} has been terminated because it failed {} times", task, task.getNumFailures()); - taskQueue.poll(); + taskQueue.remove(req); tasksPending--; tasksTerminated++; task.setState(TaskState.TERMINATED); + scheduler.removeTask(task, hv); this.setTaskToBeRemoved(task); continue; } - final ServiceFlavor flavor = task.getFlavor(); - final HostView hv = scheduler.select(request.task); - - if (hv == null || !hv.getHost().canFit(task)) { + if (result.getResultType() == SchedulingResultType.FAILURE) { LOGGER.trace("Task {} selected for scheduling but no capacity available for it at the moment", task); if (flavor.getMemorySize() > maxMemory || flavor.getCoreCount() > maxCores) { // Remove the incoming image - taskQueue.poll(); + taskQueue.remove(req); tasksPending--; tasksTerminated++; @@ -472,6 +476,7 @@ public final class ComputeService implements AutoCloseable { this.setTaskToBeRemoved(task); continue; } else { + // VM fits, but we don't have enough capacity break; } } @@ -479,7 +484,7 @@ public final class ComputeService implements AutoCloseable { SimHost host = hv.getHost(); // Remove request from queue - taskQueue.poll(); + taskQueue.remove(req); tasksPending--; LOGGER.info("Assigned task {} to host {}", task, host); @@ -488,7 +493,6 @@ public final class ComputeService implements AutoCloseable { task.host = host; host.spawn(task); - // host.start(task); tasksActive++; attemptsSuccess++; @@ -500,6 +504,7 @@ public final class ComputeService implements AutoCloseable { activeTasks.put(task, host); } catch (Exception cause) { LOGGER.error("Failed to deploy VM", cause); + scheduler.removeTask(task, hv); attemptsFailure++; } } @@ -679,19 +684,4 @@ public final class ComputeService implements AutoCloseable { internalTask.start(); } } - - /** - * A request to schedule a {@link ServiceTask} onto one of the {@link SimHost}s. - */ - static class SchedulingRequest { - final ServiceTask task; - final long submitTime; - - boolean isCancelled; - - SchedulingRequest(ServiceTask task, long submitTime) { - this.task = task; - this.submitTime = submitTime; - } - } } diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/HostView.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/HostView.java index f4aa9c70..7c548add 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/HostView.java +++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/HostView.java @@ -34,6 +34,22 @@ public class HostView { int provisionedCores; /** + * Scheduler bookkeeping + * Use by schedulers which use a priority queue data structure + * to keep track of the order of hosts to scheduler tasks on. + * {@link MemorizingScheduler} for example. + * MemorizingScheduler has an array of lists + * The 0th index of the array has a list of hosts with 0 tasks, + * 1st index of the array has hosts with 1 task, and so on. + * The priorityIndex points to the index of this the list this host + * belongs to in the array. + * The listIndex is the position of this host in the list. + */ + public int priorityIndex; + + public int listIndex; + + /** * Construct a {@link HostView} instance. * * @param host The host to create a view of. diff --git a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceTask.java b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceTask.java index f39142eb..06d6535d 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceTask.java +++ b/opendc-compute/opendc-compute-simulator/src/main/java/org/opendc/compute/simulator/service/ServiceTask.java @@ -34,6 +34,7 @@ import org.jetbrains.annotations.Nullable; import org.opendc.compute.api.TaskState; import org.opendc.compute.simulator.TaskWatcher; import org.opendc.compute.simulator.host.SimHost; +import org.opendc.compute.simulator.scheduler.SchedulingRequest; import org.opendc.simulator.compute.workload.Workload; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -59,7 +60,7 @@ public class ServiceTask { Instant createdAt; Instant finishedAt; SimHost host = null; - private ComputeService.SchedulingRequest request = null; + private SchedulingRequest request = null; private int numFailures = 0; @@ -221,10 +222,10 @@ public class ServiceTask { * Cancel the provisioning request if active. */ private void cancelProvisioningRequest() { - final ComputeService.SchedulingRequest request = this.request; + final SchedulingRequest request = this.request; if (request != null) { this.request = null; - request.isCancelled = true; + request.setCancelled(true); } } } diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeScheduler.kt index f0a2c3b4..f702ace9 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeScheduler.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeScheduler.kt @@ -40,10 +40,43 @@ public interface ComputeScheduler { public fun removeHost(host: HostView) /** - * Select a host for the specified [task]. + * Select a host for the specified [iter]. + * We implicity assume that the task has been scheduled onto the host. * - * @param task The server to select a host for. + * @param iter The server to select a host for. * @return The host to schedule the server on or `null` if no server is available. */ - public fun select(task: ServiceTask): HostView? + public fun select(iter: MutableIterator<SchedulingRequest>): SchedulingResult + + /** + * Inform the scheduler that a [task] has been removed from the [host]. + * Could be due to completion or failure. + */ + public fun removeTask( + task: ServiceTask, + host: HostView?, + ) +} + +/** + * A request to schedule a [ServiceTask] onto one of the [SimHost]s. + */ +public data class SchedulingRequest internal constructor( + public val task: ServiceTask, + public val submitTime: Long, +) { + public var isCancelled: Boolean = false + public var timesSkipped: Int = 0 } + +public enum class SchedulingResultType { + SUCCESS, + FAILURE, + EMPTY, +} + +public data class SchedulingResult( + val resultType: SchedulingResultType, + val host: HostView? = null, + val req: SchedulingRequest? = null, +) diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt index ec3aedcb..7f4f2f07 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ComputeSchedulers.kt @@ -44,15 +44,13 @@ public enum class ComputeSchedulerEnum { ProvisionedCores, ProvisionedCoresInv, Random, - Replay, } public fun createComputeScheduler( name: String, seeder: RandomGenerator, - placements: Map<String, String> = emptyMap(), ): ComputeScheduler { - return createComputeScheduler(ComputeSchedulerEnum.valueOf(name.uppercase()), seeder, placements) + return createComputeScheduler(ComputeSchedulerEnum.valueOf(name.uppercase()), seeder) } /** @@ -61,7 +59,6 @@ public fun createComputeScheduler( public fun createComputeScheduler( name: ComputeSchedulerEnum, seeder: RandomGenerator, - placements: Map<String, String> = emptyMap(), ): ComputeScheduler { val cpuAllocationRatio = 1.0 val ramAllocationRatio = 1.5 @@ -113,6 +110,5 @@ public fun createComputeScheduler( subsetSize = Int.MAX_VALUE, random = SplittableRandom(seeder.nextLong()), ) - ComputeSchedulerEnum.Replay -> ReplayScheduler(placements) } } diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt index 9fd3a862..832482eb 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/FilterScheduler.kt @@ -65,7 +65,20 @@ public class FilterScheduler( hosts.remove(host) } - override fun select(task: ServiceTask): HostView? { + override fun select(iter: MutableIterator<SchedulingRequest>): SchedulingResult { + var req = iter.next() + + while (req.isCancelled) { + iter.remove() + if (iter.hasNext()) { + req = iter.next() + } else { + // No tasks in queue + return SchedulingResult(SchedulingResultType.EMPTY) + } + } + + val task = req.task val hosts = hosts val filteredHosts = hosts.filter { host -> filters.all { filter -> filter.test(host, task) } } @@ -102,10 +115,18 @@ public class FilterScheduler( } // fixme: currently finding no matching hosts can result in an error - return when (val maxSize = min(subsetSize, subset.size)) { - 0 -> null - 1 -> subset[0] - else -> subset[random.nextInt(maxSize)] + val maxSize = min(subsetSize, subset.size) + if (maxSize == 0) { + return SchedulingResult(SchedulingResultType.FAILURE, null, req) + } else { + iter.remove() + return SchedulingResult(SchedulingResultType.SUCCESS, subset[random.nextInt(maxSize)], req) } } + + override fun removeTask( + task: ServiceTask, + host: HostView?, + ) { + } } diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt new file mode 100644 index 00000000..d3b590f7 --- /dev/null +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/MemorizingScheduler.kt @@ -0,0 +1,158 @@ +/* + * Copyright (c) 2024 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.scheduler + +import org.opendc.compute.simulator.scheduler.filters.HostFilter +import org.opendc.compute.simulator.service.HostView +import org.opendc.compute.simulator.service.ServiceTask +import java.util.SplittableRandom +import java.util.random.RandomGenerator + +/* +This scheduler records the number of tasks scheduled on each host. +When scheduling a new task, it assign the next task to the host with the least number of tasks. +We filter hosts to check if the specific task can actually run on the host. + */ +public class MemorizingScheduler( + private val filters: List<HostFilter>, + private val random: RandomGenerator = SplittableRandom(0), + private val maxTimesSkipped: Int = 7, +) : ComputeScheduler { + // We assume that there will be max 200 tasks per host. + // The index of a host list is the number of tasks on that host. + private val hostsQueue = List(200, { mutableListOf<HostView>() }) + private var minAvailableHost = 0 + private var numHosts = 0 + + override fun addHost(host: HostView) { + val zeroQueue = hostsQueue[0] + zeroQueue.add(host) + host.priorityIndex = 0 + host.listIndex = zeroQueue.size - 1 + numHosts++ + minAvailableHost = 0 + } + + override fun removeHost(host: HostView) { + val priorityIdx = host.priorityIndex + val listIdx = host.listIndex + val chosenList = hostsQueue[priorityIdx] + + if (listIdx == chosenList.size - 1) { + chosenList.removeLast() + if (listIdx == minAvailableHost) { + for (i in minAvailableHost + 1..hostsQueue.lastIndex) { + if (hostsQueue[i].size > 0) { + minAvailableHost = i + break + } + } + } + } else { + val lastItem = chosenList.removeLast() + chosenList[listIdx] = lastItem + lastItem.listIndex = listIdx + } + numHosts-- + } + + override fun select(iter: MutableIterator<SchedulingRequest>): SchedulingResult { + if (numHosts == 0) { + return SchedulingResult(SchedulingResultType.FAILURE) + } + + var chosenList: MutableList<HostView>? = null + var chosenHost: HostView? = null + + var result: SchedulingResult? = null + taskloop@ for (req in iter) { + if (req.isCancelled) { + iter.remove() + } + + for (chosenListIndex in minAvailableHost until hostsQueue.size) { + chosenList = hostsQueue[chosenListIndex] + + for (host in chosenList) { + val satisfied = filters.all { filter -> filter.test(host, req.task) } + if (satisfied) { + iter.remove() + chosenHost = host + result = SchedulingResult(SchedulingResultType.SUCCESS, host, req) + break@taskloop + } else if (req.timesSkipped >= maxTimesSkipped) { + return SchedulingResult(SchedulingResultType.FAILURE, null, req) + } + } + } + req.timesSkipped++ + } + + if (result == null) return SchedulingResult(SchedulingResultType.EMPTY) // No tasks to schedule that fit + + // Bookkeeping to maintain the calendar priority queue + val listIdx = chosenHost!!.listIndex + + if (listIdx == chosenList!!.size - 1) { + chosenList.removeLast() + if (chosenList.isEmpty()) minAvailableHost++ + } else { + val lastItem = chosenList.removeLast() + chosenList[listIdx] = lastItem + lastItem.listIndex = listIdx + } + + val nextList = hostsQueue[chosenHost.priorityIndex + 1] + nextList.add(chosenHost) + chosenHost.priorityIndex++ + chosenHost.listIndex = nextList.size - 1 + + return result + } + + override fun removeTask( + task: ServiceTask, + host: HostView?, + ) { + if (host == null) return + + val priorityIdx = host.priorityIndex + val listIdx = host.listIndex + val chosenList = hostsQueue[priorityIdx] + val nextList = hostsQueue[priorityIdx - 1] + + if (listIdx == chosenList.size - 1) { + chosenList.removeLast() + if (priorityIdx == minAvailableHost) { + minAvailableHost-- + } + } else { + val lastItem = chosenList.removeLast() + chosenList[listIdx] = lastItem + lastItem.listIndex = listIdx + } + nextList.add(host) + host.priorityIndex-- + host.listIndex = nextList.size - 1 + } +} diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ReplayScheduler.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ReplayScheduler.kt deleted file mode 100644 index 43e366d9..00000000 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/ReplayScheduler.kt +++ /dev/null @@ -1,65 +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.scheduler - -import mu.KotlinLogging -import org.opendc.compute.simulator.service.HostView -import org.opendc.compute.simulator.service.ServiceTask - -/** - * Policy replaying VM-cluster assignment. - * - * Within each cluster, the active servers on each node determine which node gets - * assigned the VM image. - */ -public class ReplayScheduler(private val vmPlacements: Map<String, String>) : ComputeScheduler { - private val logger = KotlinLogging.logger {} - - /** - * The pool of hosts available to the scheduler. - */ - private val hosts = mutableListOf<HostView>() - - override fun addHost(host: HostView) { - hosts.add(host) - } - - override fun removeHost(host: HostView) { - hosts.remove(host) - } - - override fun select(task: ServiceTask): HostView? { - val clusterName = - vmPlacements[task.name] - ?: throw IllegalStateException("Could not find placement data in VM placement file for VM ${task.name}") - val machinesInCluster = hosts.filter { it.host.getName().contains(clusterName) } - - if (machinesInCluster.isEmpty()) { - logger.info { "Could not find any machines belonging to cluster $clusterName for image ${task.name}, assigning randomly." } - return hosts.maxByOrNull { it.availableMemory } - } - - return machinesInCluster.maxByOrNull { it.availableMemory } - ?: throw IllegalStateException("Cloud not find any machine and could not randomly assign") - } -} diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VCpuCapacityFilter.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VCpuCapacityFilter.kt index 256caa94..4e63baf4 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VCpuCapacityFilter.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/filters/VCpuCapacityFilter.kt @@ -37,6 +37,10 @@ public class VCpuCapacityFilter : HostFilter { val requiredCapacity = task.flavor.meta["cpu-capacity"] as? Double val availableCapacity = host.host.getModel().cpuCapacity - return requiredCapacity == null || availableCapacity >= (requiredCapacity / task.flavor.coreCount) + return ( + requiredCapacity == null || + (availableCapacity / host.host.getModel().coreCount) + >= (requiredCapacity / task.flavor.coreCount) + ) } } diff --git a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/weights/CoreRamWeigher.kt b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/weights/CoreRamWeigher.kt index b6c43c10..aa6fdf3b 100644 --- a/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/weights/CoreRamWeigher.kt +++ b/opendc-compute/opendc-compute-simulator/src/main/kotlin/org/opendc/compute/simulator/scheduler/weights/CoreRamWeigher.kt @@ -37,7 +37,7 @@ public class CoreRamWeigher(override val multiplier: Double = 1.0) : HostWeigher host: HostView, task: ServiceTask, ): Double { - return host.availableMemory.toDouble() + return multiplier * (host.availableMemory.toDouble() / host.host.getModel().coreCount) } override fun toString(): String = "CoreRamWeigher" |