summaryrefslogtreecommitdiff
path: root/opendc/opendc-workflows/src
diff options
context:
space:
mode:
authorGeorgios Andreadis <g.andreadis@student.tudelft.nl>2020-02-11 14:46:19 +0100
committerGeorgios Andreadis <g.andreadis@student.tudelft.nl>2020-02-11 14:46:19 +0100
commitcd293b79ef2066ffcb605b9c625d6ab0a9af1d16 (patch)
treef5ea605d60538480705a0561e5152f1ed74b2188 /opendc/opendc-workflows/src
parent65a91a92afd8b6e71f08f5cbe345af30606c4861 (diff)
parent8e16b076e9c7c8c086446853e48dfff80cb45ca1 (diff)
Merge branch 'feat/2.x-model' into 'feat/2.x'
Reimplement OpenDC model using 2.x API See merge request opendc/opendc-simulator!21
Diffstat (limited to 'opendc/opendc-workflows/src')
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowScheduler.kt59
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowSchedulerLogic.kt275
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowScheduler.kt48
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerLogic.kt56
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerMode.kt40
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowService.kt184
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/FifoJobSortingPolicy.kt37
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobAdmissionPolicy.kt48
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobSortingPolicy.kt44
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/NullJobAdmissionPolicy.kt40
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/RandomJobSortingPolicy.kt40
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FirstFitResourceSelectionPolicy.kt40
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FunctionalResourceDynamicFilterPolicy.kt43
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceDynamicFilterPolicy.kt49
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceSelectionPolicy.kt48
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FifoTaskSortingPolicy.kt37
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FunctionalTaskEligibilityPolicy.kt38
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/RandomTaskSortingPolicy.kt40
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskEligibilityPolicy.kt48
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskSortingPolicy.kt45
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Job.kt48
-rw-r--r--opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Task.kt48
22 files changed, 1355 insertions, 0 deletions
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowScheduler.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowScheduler.kt
new file mode 100644
index 00000000..d4240421
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowScheduler.kt
@@ -0,0 +1,59 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service
+
+import com.atlarge.odcsim.ProcessContext
+import com.atlarge.opendc.core.services.provisioning.ProvisioningResponse
+import com.atlarge.opendc.workflows.service.stage.job.JobAdmissionPolicy
+import com.atlarge.opendc.workflows.service.stage.job.JobSortingPolicy
+import com.atlarge.opendc.workflows.service.stage.resource.ResourceDynamicFilterPolicy
+import com.atlarge.opendc.workflows.service.stage.resource.ResourceSelectionPolicy
+import com.atlarge.opendc.workflows.service.stage.task.TaskEligibilityPolicy
+import com.atlarge.opendc.workflows.service.stage.task.TaskSortingPolicy
+import kotlinx.coroutines.CoroutineScope
+
+/**
+ * A [WorkflowScheduler] that distributes work through a multi-stage process based on the Reference Architecture for
+ * Datacenter Scheduling.
+ */
+class StageWorkflowScheduler(
+ private val mode: WorkflowSchedulerMode,
+ private val jobAdmissionPolicy: JobAdmissionPolicy,
+ private val jobSortingPolicy: JobSortingPolicy,
+ private val taskEligibilityPolicy: TaskEligibilityPolicy,
+ private val taskSortingPolicy: TaskSortingPolicy,
+ private val resourceDynamicFilterPolicy: ResourceDynamicFilterPolicy,
+ private val resourceSelectionPolicy: ResourceSelectionPolicy
+) : WorkflowScheduler {
+ override fun invoke(
+ ctx: ProcessContext,
+ self: WorkflowServiceRef,
+ coroutineScope: CoroutineScope,
+ lease: ProvisioningResponse.Lease
+ ): WorkflowSchedulerLogic {
+ return StageWorkflowSchedulerLogic(ctx, self, coroutineScope, lease, mode, jobAdmissionPolicy,
+ jobSortingPolicy, taskEligibilityPolicy, taskSortingPolicy, resourceDynamicFilterPolicy, resourceSelectionPolicy)
+ }
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowSchedulerLogic.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowSchedulerLogic.kt
new file mode 100644
index 00000000..c6162f5e
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/StageWorkflowSchedulerLogic.kt
@@ -0,0 +1,275 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service
+
+import com.atlarge.odcsim.ProcessContext
+import com.atlarge.odcsim.SendPort
+import com.atlarge.odcsim.SendRef
+import com.atlarge.odcsim.sendOnce
+import com.atlarge.opendc.core.resources.compute.MachineEvent
+import com.atlarge.opendc.core.resources.compute.MachineMessage
+import com.atlarge.opendc.core.resources.compute.MachineRef
+import com.atlarge.opendc.core.resources.compute.scheduling.ProcessObserver
+import com.atlarge.opendc.core.resources.compute.scheduling.ProcessState
+import com.atlarge.opendc.core.services.provisioning.ProvisioningResponse
+import com.atlarge.opendc.core.services.resources.HostView
+import com.atlarge.opendc.core.workload.application.Application
+import com.atlarge.opendc.core.workload.application.Pid
+import com.atlarge.opendc.workflows.service.stage.job.JobAdmissionPolicy
+import com.atlarge.opendc.workflows.service.stage.job.JobSortingPolicy
+import com.atlarge.opendc.workflows.service.stage.resource.ResourceDynamicFilterPolicy
+import com.atlarge.opendc.workflows.service.stage.resource.ResourceSelectionPolicy
+import com.atlarge.opendc.workflows.service.stage.task.TaskEligibilityPolicy
+import com.atlarge.opendc.workflows.service.stage.task.TaskSortingPolicy
+import com.atlarge.opendc.workflows.workload.Job
+import com.atlarge.opendc.workflows.workload.Task
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+
+/**
+ * Logic of the [StageWorkflowScheduler].
+ */
+class StageWorkflowSchedulerLogic(
+ ctx: ProcessContext,
+ self: WorkflowServiceRef,
+ coroutineScope: CoroutineScope,
+ lease: ProvisioningResponse.Lease,
+ private val mode: WorkflowSchedulerMode,
+ private val jobAdmissionPolicy: JobAdmissionPolicy,
+ private val jobSortingPolicy: JobSortingPolicy,
+ private val taskEligibilityPolicy: TaskEligibilityPolicy,
+ private val taskSortingPolicy: TaskSortingPolicy,
+ private val resourceDynamicFilterPolicy: ResourceDynamicFilterPolicy,
+ private val resourceSelectionPolicy: ResourceSelectionPolicy
+) : WorkflowSchedulerLogic(ctx, self, coroutineScope, lease) {
+
+ /**
+ * The incoming jobs ready to be processed by the scheduler.
+ */
+ internal val incomingJobs: MutableSet<JobView> = mutableSetOf()
+
+ /**
+ * The active jobs in the system.
+ */
+ internal val activeJobs: MutableSet<JobView> = mutableSetOf()
+
+ /**
+ * The running tasks by [Pid].
+ */
+ internal val taskByPid = mutableMapOf<Pid, TaskView>()
+
+ /**
+ * The available processor cores on the leased machines.
+ */
+ internal val machineCores: MutableMap<HostView, Int> = HashMap()
+
+ private val brokers: MutableMap<SendRef<WorkflowEvent>, SendPort<WorkflowEvent>> = HashMap()
+ private val channel = ctx.open<MachineEvent>()
+
+ init {
+ lease.hosts.forEach { machineCores[it] = it.host.cores.count() }
+ coroutineScope.launch {
+ ProcessObserver(ctx, this@StageWorkflowSchedulerLogic, channel.receive)
+ }
+ }
+
+ override suspend fun submit(job: Job, handler: SendRef<WorkflowEvent>) {
+ val broker = brokers.getOrPut(handler) { ctx.connect(handler) }
+
+ // J1 Incoming Jobs
+ val jobInstance = JobView(job, handler)
+ val instances = job.tasks.associateWith {
+ TaskView(jobInstance, it)
+ }
+
+ for ((task, instance) in instances) {
+ instance.dependencies.addAll(task.dependencies.map { instances[it]!! })
+ task.dependencies.forEach {
+ instances[it]!!.dependents.add(instance)
+ }
+
+ // If the task has no dependency, it is a root task and can immediately be evaluated
+ if (instance.isRoot) {
+ instance.state = ProcessState.READY
+ }
+ }
+
+ jobInstance.tasks = instances.values.toMutableSet()
+ incomingJobs += jobInstance
+ broker.send(WorkflowEvent.JobSubmitted(self, job, ctx.clock.millis()))
+ requestCycle()
+ }
+
+ private var next: kotlinx.coroutines.Job? = null
+
+ /**
+ * Indicate to the scheduler that a scheduling cycle is needed.
+ */
+ private fun requestCycle() {
+ when (mode) {
+ is WorkflowSchedulerMode.Interactive -> {
+ coroutineScope.launch {
+ schedule()
+ }
+ }
+ is WorkflowSchedulerMode.Batch -> {
+ if (next == null) {
+ val job = coroutineScope.launch {
+ delay(mode.quantum)
+ schedule()
+ }
+ next = job
+ job.invokeOnCompletion {
+ next = null
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Perform a scheduling cycle immediately.
+ */
+ override suspend fun schedule() {
+ // J2 Create list of eligible jobs
+ jobAdmissionPolicy.startCycle(this)
+ val eligibleJobs = incomingJobs.filter { jobAdmissionPolicy.shouldAdmit(this, it) }
+ for (jobInstance in eligibleJobs) {
+ incomingJobs -= jobInstance
+ activeJobs += jobInstance
+ brokers.getValue(jobInstance.broker).send(WorkflowEvent.JobStarted(self, jobInstance.job, ctx.clock.millis()))
+ }
+
+ // J3 Sort jobs on criterion
+ val sortedJobs = jobSortingPolicy(this, activeJobs)
+
+ // J4 Per job
+ for (jobInstance in sortedJobs) {
+ // T1 Create list of eligible tasks
+ taskEligibilityPolicy.startCycle(this)
+ val eligibleTasks = jobInstance.tasks.filter { taskEligibilityPolicy.isEligible(this, it) }
+
+ // T2 Sort tasks on criterion
+ val sortedTasks = taskSortingPolicy(this, eligibleTasks)
+
+ // T3 Per task
+ for (instance in sortedTasks) {
+ val hosts = resourceDynamicFilterPolicy(this, lease.hosts, instance)
+ val host = resourceSelectionPolicy.select(this, hosts, instance)
+
+ if (host != null) {
+ // T4 Submit task to machine
+ host.ref.sendOnce(MachineMessage.Submit(instance.task.application, instance, channel.send))
+ instance.host = host
+ instance.state = ProcessState.RUNNING // Assume the application starts running
+ machineCores.merge(host, instance.task.application.cores, Int::minus)
+ } else {
+ return
+ }
+ }
+ }
+ }
+
+ override fun onSubmission(instance: MachineRef, application: Application, key: Any, pid: Pid) {
+ val task = key as TaskView
+ task.pid = pid
+ taskByPid[pid] = task
+
+ brokers.getValue(task.job.broker).send(WorkflowEvent.TaskStarted(self, task.job.job, task.task, ctx.clock.millis()))
+ }
+
+ override fun onTermination(instance: MachineRef, pid: Pid, status: Int) {
+ val task = taskByPid.remove(pid) ?: throw IllegalStateException()
+
+ val job = task.job
+ task.state = ProcessState.TERMINATED
+ job.tasks.remove(task)
+ machineCores.merge(task.host!!, task.task.application.cores, Int::plus)
+ brokers.getValue(job.broker).send(WorkflowEvent.TaskFinished(self, job.job, task.task, status, ctx.clock.millis()))
+
+ if (job.isFinished) {
+ activeJobs -= job
+ brokers.getValue(job.broker).send(WorkflowEvent.JobFinished(self, job.job, ctx.clock.millis()))
+ }
+
+ requestCycle()
+ }
+
+ class JobView(val job: Job, val broker: SendRef<WorkflowEvent>) {
+ /**
+ * A flag to indicate whether this job is finished.
+ */
+ val isFinished: Boolean
+ get() = tasks.isEmpty()
+
+ lateinit var tasks: MutableSet<TaskView>
+ }
+
+ class TaskView(val job: JobView, val task: Task) {
+ /**
+ * The dependencies of this task.
+ */
+ val dependencies = HashSet<TaskView>()
+
+ /**
+ * The dependents of this task.
+ */
+ val dependents = HashSet<TaskView>()
+
+ /**
+ * A flag to indicate whether this workflow task instance is a workflow root.
+ */
+ val isRoot: Boolean
+ get() = dependencies.isEmpty()
+
+ var state: ProcessState = ProcessState.CREATED
+ set(value) {
+ field = value
+
+ // Mark the process as terminated in the graph
+ if (value == ProcessState.TERMINATED) {
+ markTerminated()
+ }
+ }
+
+ var pid: Pid? = null
+
+ var host: HostView? = null
+
+ /**
+ * Mark the specified [TaskView] as terminated.
+ */
+ private fun markTerminated() {
+ for (dependent in dependents) {
+ dependent.dependencies.remove(this)
+
+ if (dependent.isRoot) {
+ dependent.state = ProcessState.READY
+ }
+ }
+ }
+ }
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowScheduler.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowScheduler.kt
new file mode 100644
index 00000000..6d6d4179
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowScheduler.kt
@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service
+
+import com.atlarge.odcsim.ProcessContext
+import com.atlarge.opendc.core.services.provisioning.ProvisioningResponse
+import kotlinx.coroutines.CoroutineScope
+
+/**
+ * A factory interface for constructing a [WorkflowSchedulerLogic].
+ */
+interface WorkflowScheduler {
+ /**
+ * Construct a [WorkflowSchedulerLogic] in the given [ActorContext].
+ *
+ * @param ctx The context in which the scheduler runs.
+ * @param timers The timer scheduler to use.
+ * @param lease The resource lease to use.
+ */
+ operator fun invoke(
+ ctx: ProcessContext,
+ self: WorkflowServiceRef,
+ coroutineScope: CoroutineScope,
+ lease: ProvisioningResponse.Lease
+ ): WorkflowSchedulerLogic
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerLogic.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerLogic.kt
new file mode 100644
index 00000000..0b3ba828
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerLogic.kt
@@ -0,0 +1,56 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service
+
+import com.atlarge.odcsim.ProcessContext
+import com.atlarge.odcsim.SendRef
+import com.atlarge.opendc.core.resources.compute.scheduling.ProcessObserver
+import com.atlarge.opendc.core.services.provisioning.ProvisioningResponse
+import com.atlarge.opendc.workflows.workload.Job
+import kotlinx.coroutines.CoroutineScope
+
+/**
+ * A workflow scheduler interface that schedules jobs across machines.
+ *
+ * @property ctx The context in which the scheduler runs.
+ * @property timers The timer scheduler to use.
+ * @property lease The resource lease to use.
+ */
+abstract class WorkflowSchedulerLogic(
+ protected val ctx: ProcessContext,
+ protected val self: WorkflowServiceRef,
+ protected val coroutineScope: CoroutineScope,
+ protected val lease: ProvisioningResponse.Lease
+) : ProcessObserver {
+ /**
+ * Submit the specified workflow for scheduling.
+ */
+ abstract suspend fun submit(job: Job, handler: SendRef<WorkflowEvent>)
+
+ /**
+ * Trigger an immediate scheduling cycle.
+ */
+ abstract suspend fun schedule()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerMode.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerMode.kt
new file mode 100644
index 00000000..f5060c5c
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowSchedulerMode.kt
@@ -0,0 +1,40 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service
+
+/**
+ * The operating mode of a workflow scheduler.
+ */
+sealed class WorkflowSchedulerMode {
+ /**
+ * An interactive scheduler immediately triggers a new scheduling cycle when a workflow is received.
+ */
+ object Interactive : WorkflowSchedulerMode()
+
+ /**
+ * A batch scheduler triggers a scheduling cycle every time quantum if needed.
+ */
+ data class Batch(val quantum: Long) : WorkflowSchedulerMode()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowService.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowService.kt
new file mode 100644
index 00000000..bed6b93b
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/WorkflowService.kt
@@ -0,0 +1,184 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service
+
+import com.atlarge.odcsim.Channel
+import com.atlarge.odcsim.ProcessContext
+import com.atlarge.odcsim.SendRef
+import com.atlarge.odcsim.ask
+import com.atlarge.opendc.core.Zone
+import com.atlarge.opendc.core.ZoneMessage
+import com.atlarge.opendc.core.find
+import com.atlarge.opendc.core.resources.compute.MachineEvent
+import com.atlarge.opendc.core.services.AbstractService
+import com.atlarge.opendc.core.services.Service
+import com.atlarge.opendc.core.services.ServiceProvider
+import com.atlarge.opendc.core.services.provisioning.ProvisioningMessage
+import com.atlarge.opendc.core.services.provisioning.ProvisioningResponse
+import com.atlarge.opendc.core.services.provisioning.ProvisioningService
+import com.atlarge.opendc.workflows.workload.Job
+import com.atlarge.opendc.workflows.workload.Task
+import java.util.UUID
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+
+/**
+ * A service for cloud workflow management.
+ *
+ * The workflow scheduler is modelled after the Reference Architecture for Datacenter Scheduling by Andreadis et al.
+ */
+class WorkflowService(val scheduler: WorkflowScheduler) : ServiceProvider {
+ override val uid: UUID = UUID.randomUUID()
+ override val name: String = "workflows"
+ override val provides: Set<Service<*>> = setOf(WorkflowService)
+ override val dependencies: Set<Service<*>> = setOf(ProvisioningService)
+
+ /**
+ * Build the runtime [Behavior] for the workflow service, responding to messages of shape [WorkflowMessage].
+ */
+ override suspend fun invoke(ctx: ProcessContext, zone: Zone, zoneRef: SendRef<ZoneMessage>, main: Channel<Any>) {
+ coroutineScope {
+ val inlet = ctx.listen(main.receive)
+ val provisioner = zoneRef.find(ProvisioningService)
+ // Wait for 0.1 sec before asking the provisioner to allow it to initialize. Will return empty response if asked
+ // immediately.
+ delay(10)
+ val lease: ProvisioningResponse.Lease = provisioner.ask { ProvisioningMessage.Request(Int.MAX_VALUE, it) }
+ val schedulerLogic = scheduler(ctx, main.send, this, lease)
+
+ while (isActive) {
+ when (val msg = inlet.receive()) {
+ is WorkflowMessage.Submit -> {
+ schedulerLogic.submit(msg.job, msg.broker)
+ }
+ is MachineEvent.Submitted -> {
+ schedulerLogic.onSubmission(msg.instance, msg.application, msg.key, msg.pid)
+ }
+ is MachineEvent.Terminated -> {
+ schedulerLogic.onTermination(msg.instance, msg.pid, msg.status)
+ }
+ }
+ }
+ }
+ }
+
+ companion object : AbstractService<WorkflowMessage>(UUID.randomUUID(), "workflows")
+}
+
+/**
+ * A reference to the workflow service instance.
+ */
+typealias WorkflowServiceRef = SendRef<WorkflowMessage>
+
+/**
+ * A message protocol for communicating to the workflow service.
+ */
+sealed class WorkflowMessage {
+ /**
+ * Submit the specified [Job] to the workflow service for scheduling.
+ *
+ * @property job The workflow to submit for scheduling.
+ * @property broker The broker that has submitted this workflow on behalf of a user and that needs to be kept
+ * up-to-date.
+ */
+ data class Submit(val job: Job, val broker: SendRef<WorkflowEvent>) : WorkflowMessage()
+}
+
+/**
+ * A message protocol used by the workflow service to respond to [WorkflowMessage]s.
+ */
+sealed class WorkflowEvent {
+ /**
+ * Indicate that the specified [Job] was submitted to the workflow service.
+ *
+ * @property service The reference to the service the job was submitted to.
+ * @property job The job that has been submitted.
+ * @property time A timestamp of the moment the job was received.
+ */
+ data class JobSubmitted(
+ val service: WorkflowServiceRef,
+ val job: Job,
+ val time: Long
+ ) : WorkflowEvent()
+
+ /**
+ * Indicate that the specified [Job] has become active.
+ *
+ * @property service The reference to the service the job was submitted to.
+ * @property job The job that has been submitted.
+ * @property time A timestamp of the moment the job started.
+ */
+ data class JobStarted(
+ val service: WorkflowServiceRef,
+ val job: Job,
+ val time: Long
+ ) : WorkflowEvent()
+
+ /**
+ * Indicate that the specified [Task] has started processing.
+ *
+ * @property service The reference to the service the job was submitted to.
+ * @property job The job that contains this task.
+ * @property task The task that has started processing.
+ * @property time A timestamp of the moment the task started.
+ */
+ data class TaskStarted(
+ val service: WorkflowServiceRef,
+ val job: Job,
+ val task: Task,
+ val time: Long
+ ) : WorkflowEvent()
+
+ /**
+ * Indicate that the specified [Task] has started processing.
+ *
+ * @property service The reference to the service the job was submitted to.
+ * @property job The job that contains this task.
+ * @property task The task that has started processing.
+ * @property status The exit code of the task, where zero means successful.
+ * @property time A timestamp of the moment the task finished.
+ */
+ data class TaskFinished(
+ val service: WorkflowServiceRef,
+ val job: Job,
+ val task: Task,
+ val status: Int,
+ val time: Long
+ ) : WorkflowEvent()
+
+ /**
+ * Indicate that the specified [Job] has finished processing.
+ *
+ * @property service The reference to the service the job was submitted to.
+ * @property job The job that has finished processing.
+ * @property time A timestamp of the moment the task finished.
+ */
+ data class JobFinished(
+ val service: WorkflowServiceRef,
+ val job: Job,
+ val time: Long
+ ) : WorkflowEvent()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/FifoJobSortingPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/FifoJobSortingPolicy.kt
new file mode 100644
index 00000000..333ed35a
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/FifoJobSortingPolicy.kt
@@ -0,0 +1,37 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.job
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * The [FifoJobSortingPolicy] sorts tasks based on the order of arrival in the queue.
+ */
+class FifoJobSortingPolicy : JobSortingPolicy {
+ override fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ jobs: Collection<StageWorkflowSchedulerLogic.JobView>
+ ): List<StageWorkflowSchedulerLogic.JobView> = jobs.toList()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobAdmissionPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobAdmissionPolicy.kt
new file mode 100644
index 00000000..d3a5d9a6
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobAdmissionPolicy.kt
@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.job
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A policy interface for admitting [StageWorkflowSchedulerLogic.JobView]s to a scheduling cycle.
+ */
+interface JobAdmissionPolicy {
+ /**
+ * A method that is invoked at the start of each scheduling cycle.
+ *
+ * @param scheduler The scheduler that started the cycle.
+ */
+ fun startCycle(scheduler: StageWorkflowSchedulerLogic) {}
+
+ /**
+ * Determine whether the specified [StageWorkflowSchedulerLogic.JobView] should be admitted to the scheduling cycle.
+ *
+ * @param scheduler The scheduler that should admit or reject the job.
+ * @param job The workflow that has been submitted.
+ * @return `true` if the workflow may be admitted to the scheduling cycle, `false` otherwise.
+ */
+ fun shouldAdmit(scheduler: StageWorkflowSchedulerLogic, job: StageWorkflowSchedulerLogic.JobView): Boolean
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobSortingPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobSortingPolicy.kt
new file mode 100644
index 00000000..ada3e693
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/JobSortingPolicy.kt
@@ -0,0 +1,44 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.job
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A policy interface for ordering admitted workflows in the scheduling queue.
+ */
+interface JobSortingPolicy {
+ /**
+ * Sort the given collection of jobs on a given criterion.
+ *
+ * @param scheduler The scheduler that started the cycle.
+ * @param jobs The collection of tasks that should be sorted.
+ * @return The sorted list of jobs.
+ */
+ operator fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ jobs: Collection<StageWorkflowSchedulerLogic.JobView>
+ ): List<StageWorkflowSchedulerLogic.JobView>
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/NullJobAdmissionPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/NullJobAdmissionPolicy.kt
new file mode 100644
index 00000000..f877403b
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/NullJobAdmissionPolicy.kt
@@ -0,0 +1,40 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.job
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A [JobAdmissionPolicy] that admits all jobs.
+ */
+object NullJobAdmissionPolicy : JobAdmissionPolicy {
+ /**
+ * Admit every submitted job.
+ */
+ override fun shouldAdmit(
+ scheduler: StageWorkflowSchedulerLogic,
+ job: StageWorkflowSchedulerLogic.JobView
+ ): Boolean = true
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/RandomJobSortingPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/RandomJobSortingPolicy.kt
new file mode 100644
index 00000000..30d5c456
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/job/RandomJobSortingPolicy.kt
@@ -0,0 +1,40 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.job
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+import kotlin.random.Random
+
+/**
+ * The [RandomJobSortingPolicy] sorts tasks randomly.
+ *
+ * @property random The [Random] instance to use when sorting the list of tasks.
+ */
+class RandomJobSortingPolicy(private val random: Random = Random.Default) : JobSortingPolicy {
+ override fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ jobs: Collection<StageWorkflowSchedulerLogic.JobView>
+ ): List<StageWorkflowSchedulerLogic.JobView> = jobs.shuffled(random)
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FirstFitResourceSelectionPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FirstFitResourceSelectionPolicy.kt
new file mode 100644
index 00000000..c3307063
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FirstFitResourceSelectionPolicy.kt
@@ -0,0 +1,40 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.resource
+
+import com.atlarge.opendc.core.services.resources.HostView
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A [ResourceSelectionPolicy] that selects the first machine that is available.
+ */
+class FirstFitResourceSelectionPolicy : ResourceSelectionPolicy {
+ override fun select(
+ scheduler: StageWorkflowSchedulerLogic,
+ machines: List<HostView>,
+ task: StageWorkflowSchedulerLogic.TaskView
+ ): HostView? =
+ machines.firstOrNull()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FunctionalResourceDynamicFilterPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FunctionalResourceDynamicFilterPolicy.kt
new file mode 100644
index 00000000..d742f842
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/FunctionalResourceDynamicFilterPolicy.kt
@@ -0,0 +1,43 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.resource
+
+import com.atlarge.opendc.core.services.resources.HostView
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A [ResourceDynamicFilterPolicy] based on the amount of cores available on the machine and the cores required for
+ * the task.
+ */
+class FunctionalResourceDynamicFilterPolicy : ResourceDynamicFilterPolicy {
+ override fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ machines: List<HostView>,
+ task: StageWorkflowSchedulerLogic.TaskView
+ ): List<HostView> {
+ return machines
+ .filter { scheduler.machineCores[it] ?: 0 >= task.task.application.cores }
+ }
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceDynamicFilterPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceDynamicFilterPolicy.kt
new file mode 100644
index 00000000..8a3b5a1e
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceDynamicFilterPolicy.kt
@@ -0,0 +1,49 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.resource
+
+import com.atlarge.opendc.core.services.resources.HostView
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * This interface represents the **R4** stage of the Reference Architecture for Schedulers and acts as a filter yielding
+ * a list of resources with sufficient resource-capacities, based on fixed or dynamic requirements, and on predicted or
+ * monitored information about processing unit availability, memory occupancy, etc.
+ */
+interface ResourceDynamicFilterPolicy {
+ /**
+ * Filter the list of machines based on dynamic information.
+ *
+ * @param scheduler The scheduler to filter the machines.
+ * @param machines The list of machines in the system.
+ * @param task The task that is to be scheduled.
+ * @return The machines on which the task can be scheduled.
+ */
+ operator fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ machines: List<HostView>,
+ task: StageWorkflowSchedulerLogic.TaskView
+ ): List<HostView>
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceSelectionPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceSelectionPolicy.kt
new file mode 100644
index 00000000..90b2873c
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/resource/ResourceSelectionPolicy.kt
@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.resource
+
+import com.atlarge.opendc.core.services.resources.HostView
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * This interface represents the **R5** stage of the Reference Architecture for Schedulers and matches the the selected
+ * task with a (set of) resource(s), using policies such as First-Fit, Worst-Fit, and Best-Fit.
+ */
+interface ResourceSelectionPolicy {
+ /**
+ * Select a machine on which the task should be scheduled.
+ *
+ * @param scheduler The scheduler to select the machine.
+ * @param machines The list of machines in the system.
+ * @param task The task that is to be scheduled.
+ * @return The selected machine or `null` if no machine could be found.
+ */
+ fun select(
+ scheduler: StageWorkflowSchedulerLogic,
+ machines: List<HostView>,
+ task: StageWorkflowSchedulerLogic.TaskView
+ ): HostView?
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FifoTaskSortingPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FifoTaskSortingPolicy.kt
new file mode 100644
index 00000000..48a1a50d
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FifoTaskSortingPolicy.kt
@@ -0,0 +1,37 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.task
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * The [FifoTaskSortingPolicy] sorts tasks based on the order of arrival in the queue.
+ */
+class FifoTaskSortingPolicy : TaskSortingPolicy {
+ override fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ tasks: Collection<StageWorkflowSchedulerLogic.TaskView>
+ ): List<StageWorkflowSchedulerLogic.TaskView> = tasks.toList()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FunctionalTaskEligibilityPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FunctionalTaskEligibilityPolicy.kt
new file mode 100644
index 00000000..1672633e
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/FunctionalTaskEligibilityPolicy.kt
@@ -0,0 +1,38 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.task
+
+import com.atlarge.opendc.core.resources.compute.scheduling.ProcessState
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A [TaskEligibilityPolicy] that marks tasks as eligible if they are tasks roots within the job.
+ */
+class FunctionalTaskEligibilityPolicy : TaskEligibilityPolicy {
+ override fun isEligible(
+ scheduler: StageWorkflowSchedulerLogic,
+ task: StageWorkflowSchedulerLogic.TaskView
+ ): Boolean = task.state == ProcessState.READY
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/RandomTaskSortingPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/RandomTaskSortingPolicy.kt
new file mode 100644
index 00000000..36ef3a50
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/RandomTaskSortingPolicy.kt
@@ -0,0 +1,40 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.task
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+import kotlin.random.Random
+
+/**
+ * The [RandomTaskSortingPolicy] sorts tasks randomly.
+ *
+ * @property random The [Random] instance to use when sorting the list of tasks.
+ */
+class RandomTaskSortingPolicy(private val random: Random = Random.Default) : TaskSortingPolicy {
+ override fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ tasks: Collection<StageWorkflowSchedulerLogic.TaskView>
+ ): List<StageWorkflowSchedulerLogic.TaskView> = tasks.shuffled(random)
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskEligibilityPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskEligibilityPolicy.kt
new file mode 100644
index 00000000..19f0240b
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskEligibilityPolicy.kt
@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.task
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * A policy interface for determining the eligibility of tasks in a scheduling cycle.
+ */
+interface TaskEligibilityPolicy {
+ /**
+ * A method that is invoked at the start of each scheduling cycle.
+ *
+ * @param scheduler The scheduler that started the cycle.
+ */
+ fun startCycle(scheduler: StageWorkflowSchedulerLogic) {}
+
+ /**
+ * Determine whether the specified [StageWorkflowSchedulerLogic.TaskView] is eligible to be scheduled.
+ *
+ * @param scheduler The scheduler that is determining whether the task is eligible.
+ * @param task The task instance to schedule.
+ * @return `true` if the task eligible to be scheduled, `false` otherwise.
+ */
+ fun isEligible(scheduler: StageWorkflowSchedulerLogic, task: StageWorkflowSchedulerLogic.TaskView): Boolean
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskSortingPolicy.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskSortingPolicy.kt
new file mode 100644
index 00000000..6a65ed69
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/service/stage/task/TaskSortingPolicy.kt
@@ -0,0 +1,45 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.service.stage.task
+
+import com.atlarge.opendc.workflows.service.StageWorkflowSchedulerLogic
+
+/**
+ * This interface represents the **T2** stage of the Reference Architecture for Datacenter Schedulers and provides the
+ * scheduler with a sorted list of tasks to schedule.
+ */
+interface TaskSortingPolicy {
+ /**
+ * Sort the given list of tasks on a given criterion.
+ *
+ * @param scheduler The scheduler that is sorting the tasks.
+ * @param tasks The collection of tasks that should be sorted.
+ * @return The sorted list of tasks.
+ */
+ operator fun invoke(
+ scheduler: StageWorkflowSchedulerLogic,
+ tasks: Collection<StageWorkflowSchedulerLogic.TaskView>
+ ): List<StageWorkflowSchedulerLogic.TaskView>
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Job.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Job.kt
new file mode 100644
index 00000000..dece875c
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Job.kt
@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.workload
+
+import com.atlarge.opendc.core.User
+import com.atlarge.opendc.core.workload.Workload
+import java.util.UUID
+
+/**
+ * A workload that represents a directed acyclic graph (DAG) of tasks with control and data dependencies between tasks.
+ *
+ * @property uid A unique identified of this workflow.
+ * @property name The name of this workflow.
+ * @property owner The owner of the workflow.
+ * @property tasks The tasks that are part of this workflow.
+ */
+data class Job(
+ override val uid: UUID,
+ override val name: String,
+ override val owner: User,
+ val tasks: Set<Task>
+) : Workload {
+ override fun equals(other: Any?): Boolean = other is Job && uid == other.uid
+
+ override fun hashCode(): Int = uid.hashCode()
+}
diff --git a/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Task.kt b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Task.kt
new file mode 100644
index 00000000..25fe7348
--- /dev/null
+++ b/opendc/opendc-workflows/src/main/kotlin/com/atlarge/opendc/workflows/workload/Task.kt
@@ -0,0 +1,48 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2019 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.workflows.workload
+
+import com.atlarge.opendc.core.Identity
+import com.atlarge.opendc.core.workload.application.Application
+import java.util.UUID
+
+/**
+ * A stage of a [Job].
+ *
+ * @property uid A unique identified of this task.
+ * @property name The name of this task.
+ * @property application The application to run as part of this workflow task.
+ * @property dependencies The dependencies of this task in order for it to execute.
+ */
+data class Task(
+ override val uid: UUID,
+ override val name: String,
+ val application: Application,
+ val dependencies: Set<Task>
+) : Identity {
+ override fun equals(other: Any?): Boolean = other is Task && uid == other.uid
+
+ override fun hashCode(): Int = uid.hashCode()
+}