diff options
Diffstat (limited to 'opendc-stdlib/src')
8 files changed, 100 insertions, 41 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 index 1c3dc869..c45ed5e6 100644 --- 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 @@ -25,6 +25,7 @@ package nl.atlarge.opendc.platform.scheduler import nl.atlarge.opendc.kernel.Context +import nl.atlarge.opendc.platform.workload.Job import nl.atlarge.opendc.topology.Entity import nl.atlarge.opendc.topology.machine.Machine import nl.atlarge.opendc.platform.workload.Task @@ -69,7 +70,7 @@ class FifoScheduler : Scheduler { val task = iterator.next() // TODO What to do with tasks that are not ready yet to be processed - if (!task.isReady()) { + if (!task.ready) { iterator.remove() rescheduled.add(task) continue @@ -85,7 +86,7 @@ class FifoScheduler : Scheduler { // Reschedule all tasks that are not ready yet while (!rescheduled.isEmpty()) { - submit(rescheduled.poll()) + queue.add(rescheduled.poll()) } } 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 index bf988802..578bef9c 100644 --- 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 @@ -25,9 +25,9 @@ package nl.atlarge.opendc.platform.scheduler import nl.atlarge.opendc.kernel.Context +import nl.atlarge.opendc.platform.workload.Task 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. 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 index b2660964..03f37b50 100644 --- 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 @@ -25,9 +25,9 @@ package nl.atlarge.opendc.platform.scheduler import nl.atlarge.opendc.kernel.Context +import nl.atlarge.opendc.platform.workload.Task import nl.atlarge.opendc.topology.Entity import nl.atlarge.opendc.topology.machine.Machine -import nl.atlarge.opendc.platform.workload.Task import java.util.* /** @@ -67,8 +67,8 @@ class SrtfScheduler : Scheduler { val task = iterator.next() // TODO What to do with tasks that are not ready yet to be processed - if (!task.isReady()) { - submit(task) + if (!task.ready) { + tasks.add(task) continue } else if (task.finished) { tasks.remove(task) diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Job.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Job.kt index eb173229..df2f2b6a 100644 --- a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Job.kt +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Job.kt @@ -25,7 +25,7 @@ package nl.atlarge.opendc.platform.workload /** - * A job that is submitted to a cloud network, which consists of one or multiple [Task]s. + * A bag of tasks which are submitted by a [User] to the cloud network. * * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) */ diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Task.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Task.kt index 140a5d5d..e740cdd8 100644 --- a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Task.kt +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/Task.kt @@ -24,12 +24,11 @@ package nl.atlarge.opendc.platform.workload -import nl.atlarge.opendc.kernel.Context -import nl.atlarge.opendc.topology.container.Datacenter +import nl.atlarge.opendc.kernel.time.Instant import nl.atlarge.opendc.topology.machine.Machine /** - * A task represents some computation that is part of a [Job]. + * A task that runs as part of a [Job] on a [Machine]. * * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) */ @@ -55,51 +54,39 @@ interface Task { val parallelizable: Boolean /** - * The remaining amount of flops to compute. + * The remaining flops for this task. */ val remaining: Long /** - * A flag to indicate the task has been accepted by the datacenter. + * The state of the task. */ - val accepted: Boolean + val state: TaskState /** - * A flag to indicate the task has been started. + * A flag to indicate whether the task is ready to be started. */ - val started: Boolean + val ready: Boolean + get() = !dependencies.any { !it.finished } /** - * A flag to indicate whether the task is finished. + * A flag to indicate whether the task has finished. */ val finished: Boolean + get() = state is TaskState.Finished /** - * Determine whether the task is ready to be processed. + * This method is invoked when a task has arrived at a datacenter. * - * @return `true` if the task is ready to be processed, `false` otherwise. + * @param time The moment in time the task has arrived at the datacenter. */ - fun isReady() = dependencies.all { it.finished } - - /** - * Accept the task into the scheduling queue. - */ - fun Context<Datacenter>.accept() - - /** - * Start a task. - */ - fun Context<Machine>.start() + fun arrive(time: Instant) /** * Consume the given amount of flops of this task. * + * @param time The current moment in time of the consumption. * @param flops The total amount of flops to consume. */ - fun Context<Machine>.consume(flops: Long) - - /** - * Finalise the task. - */ - fun Context<Machine>.finalize() + fun consume(time: Instant, flops: Long) } diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/TaskState.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/TaskState.kt new file mode 100644 index 00000000..d1f908af --- /dev/null +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/platform/workload/TaskState.kt @@ -0,0 +1,72 @@ +/* + * 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.workload + +import nl.atlarge.opendc.kernel.time.Instant + + +/** + * This class hierarchy describes the states of a [Task]. + * + * @author Fabian Mastenbroek (f.s.mastenbroek@student.tudelft.nl) + */ +sealed class TaskState { + /** + * A state to indicate the task has not yet arrived at the [Datacenter]. + */ + object Underway : TaskState() + + /** + * A state to indicate the task has arrived at the [Datacenter]. + * + * @property at The moment in time the task has arrived. + */ + data class Queued(val at: Instant) : TaskState() + + /** + * A state to indicate the task has started running on a machine. + * + * @property previous The previous state of the task. + * @property at The moment in time the task started. + */ + data class Running(val previous: Queued, val at: Instant) : TaskState() + + /** + * A state to indicate the task has finished. + * + * @property previous The previous state of the task. + * @property at The moment in time the task finished. + */ + data class Finished(val previous: Running, val at: Instant) : TaskState() + + /** + * A state to indicate the task has failed. + * + * @property previous The previous state of the task. + * @property at The moment in time the task failed. + * @property reason The reason of the failure. + */ + data class Failed(val previous: Running, val at: Instant, val reason: String) : TaskState() +} diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/container/Datacenter.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/container/Datacenter.kt index 333609bb..b3d67568 100644 --- a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/container/Datacenter.kt +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/container/Datacenter.kt @@ -30,9 +30,10 @@ import nl.atlarge.opendc.kernel.Context import nl.atlarge.opendc.kernel.Process import nl.atlarge.opendc.kernel.time.Duration import nl.atlarge.opendc.platform.scheduler.Scheduler +import nl.atlarge.opendc.platform.workload.Job +import nl.atlarge.opendc.platform.workload.Task import nl.atlarge.opendc.topology.Entity import nl.atlarge.opendc.topology.machine.Machine -import nl.atlarge.opendc.platform.workload.Task import java.util.* /** @@ -88,6 +89,7 @@ interface Datacenter : Entity<Unit>, Process<Datacenter> { while (queue.isNotEmpty()) { val msg = queue.poll() if (msg is Task) { + msg.arrive(time) scheduler.submit(msg) } } diff --git a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/machine/Machine.kt b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/machine/Machine.kt index b1d881a1..761f14b1 100644 --- a/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/machine/Machine.kt +++ b/opendc-stdlib/src/main/kotlin/nl/atlarge/opendc/topology/machine/Machine.kt @@ -83,16 +83,14 @@ open class Machine : Entity<Machine.State>, Process<Machine> { var task: Task = receiveTask() update(State(Status.RUNNING, task, load = 1.0, memory = state.memory + 50, temperature = 30.0)) - task.run { start() } while (true) { - if (task.remaining <= 0) { - task.run { finalize() } + if (task.finished) { logger.info { "${entity.id}: Task ${task.id} finished. Machine idle at $time" } update(State(Status.IDLE)) task = receiveTask() } else { - task.run { consume(speed * delta) } + task.consume(time, speed * delta) } // Check if we have received a new order in the meantime. @@ -100,7 +98,6 @@ open class Machine : Entity<Machine.State>, Process<Machine> { if (msg is Task) { task = msg update(State(Status.RUNNING, task, load = 1.0, memory = state.memory + 50, temperature = 30.0)) - task.run { start() } } } } |