diff options
Diffstat (limited to 'opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler')
3 files changed, 299 insertions, 0 deletions
diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/FifoScheduler.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/FifoScheduler.kt new file mode 100644 index 00000000..1c3dc869 --- /dev/null +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/FifoScheduler.kt @@ -0,0 +1,118 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 nl.atlarge.opendc.platform.scheduler + +import nl.atlarge.opendc.kernel.Context +import nl.atlarge.opendc.topology.Entity +import nl.atlarge.opendc.topology.machine.Machine +import nl.atlarge.opendc.platform.workload.Task +import java.util.* + +/** + * A [Scheduler] that distributes work according to the first-in-first-out principle. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +class FifoScheduler : Scheduler { + /** + * The name of this scheduler. + */ + override val name: String = "FIFO" + + /** + * The set of machines the scheduler knows of. + */ + val machines: MutableSet<Machine> = HashSet() + + /** + * The queue of [Task]s that need to be scheduled. + */ + val queue: Queue<Task> = ArrayDeque() + + /** + * (Re)schedule the tasks submitted to the scheduler over the specified set of machines. + */ + override suspend fun <E : Entity<*>> Context<E>.schedule() { + if (queue.isEmpty()) { + return + } + + // The tasks that need to be rescheduled + val rescheduled = ArrayDeque<Task>() + val iterator = queue.iterator() + + machines + .forEach { machine -> + while (iterator.hasNext()) { + val task = iterator.next() + + // TODO What to do with tasks that are not ready yet to be processed + if (!task.isReady()) { + iterator.remove() + rescheduled.add(task) + continue + } else if (task.finished) { + iterator.remove() + continue + } + + machine.send(task) + break + } + } + + // Reschedule all tasks that are not ready yet + while (!rescheduled.isEmpty()) { + submit(rescheduled.poll()) + } + } + + /** + * Submit a [Task] to this scheduler. + * + * @param task The task to submit to the scheduler. + */ + override fun submit(task: Task) { + queue.add(task) + } + + /** + * Register a [Machine] to this scheduler. + * + * @param machine The machine to register. + */ + override fun register(machine: Machine) { + machines.add(machine) + } + + /** + * Deregister a [Machine] from this scheduler. + * + * @param machine The machine to deregister. + */ + override fun deregister(machine: Machine) { + machines.remove(machine) + } +} diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/Scheduler.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/Scheduler.kt new file mode 100644 index 00000000..bf988802 --- /dev/null +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/Scheduler.kt @@ -0,0 +1,71 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 nl.atlarge.opendc.platform.scheduler + +import nl.atlarge.opendc.kernel.Context +import nl.atlarge.opendc.topology.Entity +import nl.atlarge.opendc.topology.machine.Machine +import nl.atlarge.opendc.platform.workload.Task + +/** + * A task scheduler that is coupled to an [Entity] in the topology of the cloud network. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +interface Scheduler { + /** + * The name of this scheduler. + */ + val name: String + + /** + * (Re)schedule the tasks submitted to the scheduler over the specified set of machines. + * + * This method should be invoked at some interval to allow the scheduler to reschedule existing tasks and schedule + * new tasks. + */ + suspend fun <E: Entity<*>> Context<E>.schedule() + + /** + * Submit a [Task] to this scheduler. + * + * @param task The task to submit to the scheduler. + */ + fun submit(task: Task) + + /** + * Register a [Machine] to this scheduler. + * + * @param machine The machine to register. + */ + fun register(machine: Machine) + + /** + * Deregister a [Machine] from this scheduler. + * + * @param machine The machine to deregister. + */ + fun deregister(machine: Machine) +} diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/SrtfScheduler.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/SrtfScheduler.kt new file mode 100644 index 00000000..b2660964 --- /dev/null +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/scheduler/SrtfScheduler.kt @@ -0,0 +1,110 @@ +/* + * MIT License + * + * Copyright (c) 2017 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 nl.atlarge.opendc.platform.scheduler + +import nl.atlarge.opendc.kernel.Context +import nl.atlarge.opendc.topology.Entity +import nl.atlarge.opendc.topology.machine.Machine +import nl.atlarge.opendc.platform.workload.Task +import java.util.* + +/** + * A [Scheduler] that distributes work according to the shortest job first policy. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +class SrtfScheduler : Scheduler { + /** + * The name of this scheduler. + */ + override val name: String = "SRTF" + + /** + * The set of machines the scheduler knows of. + */ + val machines: MutableSet<Machine> = HashSet() + + /** + * The set of [Task]s that need to be scheduled. + */ + val tasks: MutableSet<Task> = HashSet() + + /** + * (Re)schedule the tasks submitted to the scheduler over the specified set of machines. + */ + override suspend fun <E : Entity<*>> Context<E>.schedule() { + if (tasks.isEmpty()) { + return + } + + val iterator = tasks.sortedBy { it.remaining }.iterator() + + machines + .forEach { machine -> + while (iterator.hasNext()) { + val task = iterator.next() + + // TODO What to do with tasks that are not ready yet to be processed + if (!task.isReady()) { + submit(task) + continue + } else if (task.finished) { + tasks.remove(task) + continue + } + + machine.send(task) + break + } + } + } + + /** + * Submit a [Task] to this scheduler. + * + * @param task The task to submit to the scheduler. + */ + override fun submit(task: Task) { + tasks.add(task) + } + + /** + * Register a [Machine] to this scheduler. + * + * @param machine The machine to register. + */ + override fun register(machine: Machine) { + machines.add(machine) + } + + /** + * Deregister a [Machine] from this scheduler. + * + * @param machine The machine to deregister. + */ + override fun deregister(machine: Machine) { + machines.remove(machine) + } +} |